From 7b0bf99ed13221b874bf4099f6c74e84d7b29c51 Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Mon, 21 Sep 2026 16:01:19 +0000 Subject: [PATCH] Add durable speaker transcription state --- docs/roadmap/implementation.md | 9 +- internal/manifest/manifest.go | 30 +-- internal/manifest/store.go | 6 + internal/manifest/transcribe_state.go | 128 +++++++++++ internal/manifest/transcribe_state_test.go | 248 +++++++++++++++++++++ 5 files changed, 405 insertions(+), 16 deletions(-) create mode 100644 internal/manifest/transcribe_state.go create mode 100644 internal/manifest/transcribe_state_test.go diff --git a/docs/roadmap/implementation.md b/docs/roadmap/implementation.md index 2d8f7ed..a6a103f 100644 --- a/docs/roadmap/implementation.md +++ b/docs/roadmap/implementation.md @@ -1,6 +1,6 @@ # Implementation Plan: Speaker Checkpoints and Replayable Uploads -Status: Ready; all stages pending. +Status: Implementation in progress; Stage 1 complete. ## Scope and Instructions @@ -118,7 +118,12 @@ body-preserving redirects. Do not introduce a new redirect restriction. ## Stage 1: Add Durable Speaker Completion State -Status: Pending. +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`. Preparation: `internal/manifest/manifest.go`, `store.go`, `analyze_state.go`, their tests, and `docs/internal/manifest.md`. diff --git a/internal/manifest/manifest.go b/internal/manifest/manifest.go index e6dbebc..d207b70 100644 --- a/internal/manifest/manifest.go +++ b/internal/manifest/manifest.go @@ -50,20 +50,22 @@ type ArtifactRecord struct { // StageRecord tracks lifecycle and provenance for one pipeline stage. type StageRecord struct { - Name string `json:"name"` - Status StageStatus `json:"status"` - CreatedAt time.Time `json:"created_at"` - UpdatedAt time.Time `json:"updated_at"` - StartedAt *time.Time `json:"started_at,omitempty"` - CompletedAt *time.Time `json:"completed_at,omitempty"` - Outputs []ArtifactRecord `json:"outputs,omitempty"` - Logs []string `json:"logs,omitempty"` - GeneratedConfigs []string `json:"generated_configs,omitempty"` - Error *ErrorRecord `json:"error,omitempty"` - Metadata map[string]any `json:"metadata,omitempty"` - SemanticConfig *SemanticConfigFingerprint `json:"semantic_config,omitempty"` - AnalyzeStateVersion int `json:"analyze_state_version,omitempty"` - AnalyzeArtifacts map[string]AnalyzeArtifactRecord `json:"analyze_artifacts,omitempty"` + Name string `json:"name"` + Status StageStatus `json:"status"` + CreatedAt time.Time `json:"created_at"` + UpdatedAt time.Time `json:"updated_at"` + StartedAt *time.Time `json:"started_at,omitempty"` + CompletedAt *time.Time `json:"completed_at,omitempty"` + Outputs []ArtifactRecord `json:"outputs,omitempty"` + Logs []string `json:"logs,omitempty"` + GeneratedConfigs []string `json:"generated_configs,omitempty"` + Error *ErrorRecord `json:"error,omitempty"` + Metadata map[string]any `json:"metadata,omitempty"` + SemanticConfig *SemanticConfigFingerprint `json:"semantic_config,omitempty"` + TranscribeStateVersion int `json:"transcribe_state_version,omitempty"` + TranscribeSpeakers map[string]TranscribeSpeakerRecord `json:"transcribe_speakers,omitempty"` + AnalyzeStateVersion int `json:"analyze_state_version,omitempty"` + AnalyzeArtifacts map[string]AnalyzeArtifactRecord `json:"analyze_artifacts,omitempty"` } // EffectiveConfigProvenance records bounded non-secret configuration identity diff --git a/internal/manifest/store.go b/internal/manifest/store.go index 5cc56e5..8a4fa7e 100644 --- a/internal/manifest/store.go +++ b/internal/manifest/store.go @@ -119,6 +119,9 @@ func (s *LocalStore) Save(ctx context.Context, path string, m *Manifest) error { } for name, stage := range m.Stages { normalizeAnalyzeArtifactCollection(stage.AnalyzeArtifacts) + if err := validateTranscribeStageState(name, stage.TranscribeStateVersion, stage.TranscribeSpeakers); err != nil { + return fmt.Errorf("save manifest: stages.%s: %w", name, err) + } if err := validateAnalyzeStageState(name, stage.AnalyzeStateVersion, stage.AnalyzeArtifacts); err != nil { return fmt.Errorf("save manifest: stages.%s: %w", name, err) } @@ -160,6 +163,9 @@ func validateLoadedManifest(m *Manifest) error { return err } for name, stage := range m.Stages { + if err := validateTranscribeStageState(name, stage.TranscribeStateVersion, stage.TranscribeSpeakers); err != nil { + return fmt.Errorf("stages.%s: %w", name, err) + } if err := validateAnalyzeStageState(name, stage.AnalyzeStateVersion, stage.AnalyzeArtifacts); err != nil { return fmt.Errorf("stages.%s: %w", name, err) } diff --git a/internal/manifest/transcribe_state.go b/internal/manifest/transcribe_state.go new file mode 100644 index 0000000..816e124 --- /dev/null +++ b/internal/manifest/transcribe_state.go @@ -0,0 +1,128 @@ +package manifest + +import ( + "fmt" + + "gitea.maximumdirect.net/eric/narratio/internal/pathsafe" +) + +const ( + // TranscribeStateContractVersion identifies the supported per-speaker + // completion representation owned by the transcribe stage. + TranscribeStateContractVersion = 1 + // TranscribeFingerprintContractVersion identifies the fingerprint + // representation stored by the supported transcribe state contract. + TranscribeFingerprintContractVersion = 1 +) + +// TranscribeSpeakerRecord binds one completed raw transcript to the audio and +// observable recognition settings that produced it. +type TranscribeSpeakerRecord struct { + FingerprintVersion int `json:"fingerprint_version"` + Fingerprint string `json:"fingerprint"` + AudioPath string `json:"audio_path"` + AudioSHA256 string `json:"audio_sha256"` + AudioSize int64 `json:"audio_size"` + Output ArtifactRecord `json:"output"` + OutputSize int64 `json:"output_size"` + ProducerRunID string `json:"producer_run_id"` +} + +// HasVersionedTranscribeState reports whether a transcribe stage record carries +// the supported per-speaker authority. An aggregate-only record returns false. +func (s *StageRecord) HasVersionedTranscribeState() bool { + return s != nil && s.Name == "transcribe" && s.TranscribeStateVersion == TranscribeStateContractVersion +} + +// ValidateTranscribeSpeakerCollection validates one complete session +// collection independently of its containing manifest. +func ValidateTranscribeSpeakerCollection(version int, records map[string]TranscribeSpeakerRecord) error { + if version != TranscribeStateContractVersion { + return fmt.Errorf("unsupported transcribe state version %d", version) + } + for speakerID, record := range records { + if err := validateTranscribeSpeakerRecord(speakerID, record); err != nil { + return fmt.Errorf("transcribe speaker %q: %w", speakerID, err) + } + } + return nil +} + +// CloneTranscribeSpeakerCollection returns a deep copy so checkpoint candidates +// and installed authority cannot share mutable artifact metadata. +func CloneTranscribeSpeakerCollection(records map[string]TranscribeSpeakerRecord) map[string]TranscribeSpeakerRecord { + if records == nil { + return nil + } + cloned := make(map[string]TranscribeSpeakerRecord, len(records)) + for speakerID, record := range records { + output := cloneArtifactRecord(&record.Output) + record.Output = *output + cloned[speakerID] = record + } + return cloned +} + +func validateTranscribeStageState(stageName string, version int, records map[string]TranscribeSpeakerRecord) error { + if stageName != "transcribe" { + if version != 0 || records != nil { + return fmt.Errorf("stage %q cannot contain transcribe-owned state", stageName) + } + return nil + } + if version == 0 && records == nil { + return nil + } + return ValidateTranscribeSpeakerCollection(version, records) +} + +func validateTranscribeSpeakerRecord(speakerID string, record TranscribeSpeakerRecord) error { + if speakerID == "" { + return fmt.Errorf("speaker ID is required") + } + if record.FingerprintVersion != TranscribeFingerprintContractVersion { + return fmt.Errorf("unsupported fingerprint version %d", record.FingerprintVersion) + } + if err := validateSHA256("fingerprint", record.Fingerprint); err != nil { + return err + } + if err := validateCanonicalRelativePath("audio_path", record.AudioPath); err != nil { + return err + } + if err := validateSHA256("audio_sha256", record.AudioSHA256); err != nil { + return err + } + if record.AudioSize <= 0 { + return fmt.Errorf("audio_size must be positive") + } + if record.Output.Kind != "transcript_raw" { + return fmt.Errorf("output kind %q must be transcript_raw", record.Output.Kind) + } + if err := validateCanonicalRelativePath("output local_path", record.Output.LocalPath); err != nil { + return err + } + if err := validateSHA256("output checksum", record.Output.Checksum); err != nil { + return err + } + if record.OutputSize <= 0 { + return fmt.Errorf("output_size must be positive") + } + if err := pathsafe.ValidateOpaqueSegment(record.ProducerRunID); err != nil { + return fmt.Errorf("producer_run_id is invalid: %w", err) + } + if record.Output.ProducerRunID != record.ProducerRunID { + return fmt.Errorf("output producer_run_id %q does not match record", record.Output.ProducerRunID) + } + return nil +} + +func validateCanonicalRelativePath(field, value string) error { + normalized, err := pathsafe.NormalizeRelativeDestination(value) + if err != nil { + return fmt.Errorf("%s is unsafe: %w", field, err) + } + if normalized != value { + return fmt.Errorf("%s %q is not canonical %q", field, value, normalized) + } + return nil +} diff --git a/internal/manifest/transcribe_state_test.go b/internal/manifest/transcribe_state_test.go new file mode 100644 index 0000000..4f73af7 --- /dev/null +++ b/internal/manifest/transcribe_state_test.go @@ -0,0 +1,248 @@ +package manifest + +import ( + "context" + "encoding/json" + "path/filepath" + "strings" + "testing" + "time" +) + +func TestTranscribeSpeakerStateRoundTripsPopulatedAndVersionedEmptyCollections(t *testing.T) { + now := time.Date(2026, 9, 21, 12, 0, 0, 0, time.UTC) + store := &LocalStore{} + + for _, test := range []struct { + name string + records map[string]TranscribeSpeakerRecord + }{ + { + name: "populated", + records: map[string]TranscribeSpeakerRecord{ + "Alice Table 1": validTranscribeSpeakerRecord("Alice Table 1"), + }, + }, + {name: "versioned empty", records: map[string]TranscribeSpeakerRecord{}}, + } { + t.Run(test.name, func(t *testing.T) { + m := New("session", now) + m.Stages["transcribe"] = &StageRecord{ + Name: "transcribe", Status: StatusSucceeded, CreatedAt: now, UpdatedAt: now, + TranscribeStateVersion: TranscribeStateContractVersion, + TranscribeSpeakers: test.records, + } + path := filepath.Join(t.TempDir(), "manifest.json") + if err := store.Save(context.Background(), path, m); err != nil { + t.Fatalf("Save() error = %v", err) + } + loaded, err := store.Load(context.Background(), path) + if err != nil { + t.Fatalf("Load() error = %v", err) + } + stage := loaded.Stages["transcribe"] + if !stage.HasVersionedTranscribeState() { + t.Fatalf("transcribe state = %#v, want supported version", stage) + } + if len(stage.TranscribeSpeakers) != len(test.records) { + t.Fatalf("speaker count = %d, want %d", len(stage.TranscribeSpeakers), len(test.records)) + } + if len(test.records) != 0 { + got := stage.TranscribeSpeakers["Alice Table 1"] + if got.AudioPath != "inputs/audio/Alice Table 1.flac" || got.Output.LocalPath != "transcripts/raw/Alice Table 1.json" { + t.Fatalf("round-tripped record = %#v", got) + } + } + }) + } +} + +func TestTranscribeSpeakerStateAllowsLegacyAggregateOnlyManifest(t *testing.T) { + legacy := `{ + "schema_version": "narratio.session-manifest.v1", + "session_id": "session", + "created_at": "2026-09-21T12:00:00Z", + "updated_at": "2026-09-21T12:01:00Z", + "stages": { + "transcribe": { + "name": "transcribe", + "status": "succeeded", + "created_at": "2026-09-21T12:00:00Z", + "updated_at": "2026-09-21T12:01:00Z", + "outputs": [{"kind":"transcript_raw","local_path":"transcripts/raw/alice.json"}] + } + } +}` + loaded, err := (&LocalStore{}).LoadReader(context.Background(), strings.NewReader(legacy)) + if err != nil { + t.Fatalf("LoadReader() error = %v", err) + } + if loaded.Stages["transcribe"].HasVersionedTranscribeState() { + t.Fatal("legacy aggregate-only manifest unexpectedly gained per-speaker authority") + } + if err := (&LocalStore{}).Save(context.Background(), filepath.Join(t.TempDir(), "manifest.json"), loaded); err != nil { + t.Fatalf("Save(legacy aggregate-only manifest) error = %v", err) + } +} + +func TestValidateTranscribeSpeakerCollectionRejectsMalformedEvidence(t *testing.T) { + valid := validTranscribeSpeakerRecord("alice") + tests := []struct { + name string + speakerID string + mutate func(*TranscribeSpeakerRecord) + want string + }{ + {name: "empty speaker ID", speakerID: "", want: "speaker ID is required"}, + {name: "unsupported fingerprint version", speakerID: "alice", mutate: func(r *TranscribeSpeakerRecord) { r.FingerprintVersion = 2 }, want: "unsupported fingerprint version"}, + {name: "bad fingerprint", speakerID: "alice", mutate: func(r *TranscribeSpeakerRecord) { r.Fingerprint = "bad" }, want: "fingerprint must be"}, + {name: "unsafe audio path", speakerID: "alice", mutate: func(r *TranscribeSpeakerRecord) { r.AudioPath = "../alice.flac" }, want: "audio_path is unsafe"}, + {name: "noncanonical audio path", speakerID: "alice", mutate: func(r *TranscribeSpeakerRecord) { r.AudioPath = "inputs/audio/../alice.flac" }, want: "not canonical"}, + {name: "bad audio checksum", speakerID: "alice", mutate: func(r *TranscribeSpeakerRecord) { r.AudioSHA256 = strings.Repeat("G", 64) }, want: "audio_sha256 must be"}, + {name: "bad audio size", speakerID: "alice", mutate: func(r *TranscribeSpeakerRecord) { r.AudioSize = 0 }, want: "audio_size must be positive"}, + {name: "bad output kind", speakerID: "alice", mutate: func(r *TranscribeSpeakerRecord) { r.Output.Kind = "transcript" }, want: "must be transcript_raw"}, + {name: "unsafe output path", speakerID: "alice", mutate: func(r *TranscribeSpeakerRecord) { r.Output.LocalPath = "/tmp/alice.json" }, want: "output local_path is unsafe"}, + {name: "bad output checksum", speakerID: "alice", mutate: func(r *TranscribeSpeakerRecord) { r.Output.Checksum = "bad" }, want: "output checksum must be"}, + {name: "bad output size", speakerID: "alice", mutate: func(r *TranscribeSpeakerRecord) { r.OutputSize = 0 }, want: "output_size must be positive"}, + {name: "bad producer run", speakerID: "alice", mutate: func(r *TranscribeSpeakerRecord) { r.ProducerRunID = "bad/run" }, want: "producer_run_id is invalid"}, + {name: "producer mismatch", speakerID: "alice", mutate: func(r *TranscribeSpeakerRecord) { r.Output.ProducerRunID = "other-run" }, want: "does not match record"}, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + record := valid + if test.mutate != nil { + test.mutate(&record) + } + err := ValidateTranscribeSpeakerCollection(TranscribeStateContractVersion, map[string]TranscribeSpeakerRecord{test.speakerID: record}) + if err == nil || !strings.Contains(err.Error(), test.want) { + t.Fatalf("error = %v, want %q", err, test.want) + } + }) + } + if err := ValidateTranscribeSpeakerCollection(99, nil); err == nil || !strings.Contains(err.Error(), "unsupported transcribe state version") { + t.Fatalf("unsupported version error = %v", err) + } +} + +func TestTranscribeSpeakerStateValidationAppliesOnSaveAndLoad(t *testing.T) { + now := time.Date(2026, 9, 21, 12, 0, 0, 0, time.UTC) + invalid := validTranscribeSpeakerRecord("alice") + invalid.AudioSHA256 = "bad" + m := New("session", now) + m.Stages["transcribe"] = &StageRecord{ + Name: "transcribe", Status: StatusSucceeded, CreatedAt: now, UpdatedAt: now, + TranscribeStateVersion: TranscribeStateContractVersion, + TranscribeSpeakers: map[string]TranscribeSpeakerRecord{"alice": invalid}, + } + store := &LocalStore{} + if err := store.Save(context.Background(), filepath.Join(t.TempDir(), "manifest.json"), m); err == nil || !strings.Contains(err.Error(), "audio_sha256") { + t.Fatalf("Save() error = %v, want audio evidence validation", err) + } + + data, err := json.Marshal(m) + if err != nil { + t.Fatalf("Marshal() error = %v", err) + } + if _, err := store.LoadReader(context.Background(), strings.NewReader(string(data))); err == nil || !strings.Contains(err.Error(), "audio_sha256") { + t.Fatalf("LoadReader() error = %v, want audio evidence validation", err) + } +} + +func TestTranscribeSpeakerStateOwnershipAndVersionAreValidated(t *testing.T) { + now := time.Date(2026, 9, 21, 12, 0, 0, 0, time.UTC) + store := &LocalStore{} + + for _, test := range []struct { + name string + stage *StageRecord + want string + }{ + { + name: "wrong owner", + stage: &StageRecord{Name: "merge", Status: StatusSucceeded, CreatedAt: now, UpdatedAt: now, + TranscribeStateVersion: TranscribeStateContractVersion, + TranscribeSpeakers: map[string]TranscribeSpeakerRecord{}}, + want: "cannot contain transcribe-owned state", + }, + { + name: "missing version", + stage: &StageRecord{Name: "transcribe", Status: StatusSucceeded, CreatedAt: now, UpdatedAt: now, + TranscribeSpeakers: map[string]TranscribeSpeakerRecord{"alice": validTranscribeSpeakerRecord("alice")}}, + want: "unsupported transcribe state version 0", + }, + { + name: "unsupported version", + stage: &StageRecord{Name: "transcribe", Status: StatusSucceeded, CreatedAt: now, UpdatedAt: now, + TranscribeStateVersion: 2}, + want: "unsupported transcribe state version 2", + }, + } { + t.Run(test.name, func(t *testing.T) { + m := New("session", now) + m.Stages[test.stage.Name] = test.stage + if err := store.Save(context.Background(), filepath.Join(t.TempDir(), "manifest.json"), m); err == nil || !strings.Contains(err.Error(), test.want) { + t.Fatalf("Save() error = %v, want %q", err, test.want) + } + }) + } +} + +func TestCloneTranscribeSpeakerCollectionIsIsolated(t *testing.T) { + original := map[string]TranscribeSpeakerRecord{"alice": validTranscribeSpeakerRecord("alice")} + cloned := CloneTranscribeSpeakerCollection(original) + record := cloned["alice"] + record.AudioPath = "inputs/audio/replacement.flac" + record.Output.Checksum = strings.Repeat("c", 64) + cloned["alice"] = record + + if got := original["alice"]; got.AudioPath != "inputs/audio/alice.flac" || got.Output.Checksum != strings.Repeat("b", 64) { + t.Fatalf("original changed through clone: %#v", got) + } +} + +func TestTranscribeSpeakerStateSurvivesAggregateLifecycleTransitions(t *testing.T) { + now := time.Date(2026, 9, 21, 12, 0, 0, 0, time.UTC) + for _, transition := range []struct { + name string + apply func(*Manifest) + }{ + {name: "running", apply: func(m *Manifest) { m.MarkStageRunning("transcribe", now.Add(time.Minute)) }}, + {name: "failed", apply: func(m *Manifest) { m.MarkStageFailed("transcribe", now.Add(time.Minute), "request failed") }}, + {name: "stale", apply: func(m *Manifest) { m.MarkStageStale("transcribe", now.Add(time.Minute), "audio changed") }}, + } { + t.Run(transition.name, func(t *testing.T) { + m := New("session", now) + m.Stages["transcribe"] = &StageRecord{ + Name: "transcribe", Status: StatusSucceeded, CreatedAt: now, UpdatedAt: now, + Outputs: []ArtifactRecord{{Kind: "transcript_raw", LocalPath: "transcripts/raw/old.json"}}, + Logs: []string{"runs/old/transcribe.log"}, Metadata: map[string]any{"attempts": 1}, + TranscribeStateVersion: TranscribeStateContractVersion, + TranscribeSpeakers: map[string]TranscribeSpeakerRecord{"alice": validTranscribeSpeakerRecord("alice")}, + } + transition.apply(m) + stage := m.Stages["transcribe"] + if !stage.HasVersionedTranscribeState() || stage.TranscribeSpeakers["alice"].AudioPath != "inputs/audio/alice.flac" { + t.Fatalf("speaker completion was lost during %s: %#v", transition.name, stage) + } + if transition.name != "stale" && (len(stage.Outputs) != 0 || len(stage.Logs) != 0 || len(stage.Metadata) != 0) { + t.Fatalf("aggregate details survived %s: %#v", transition.name, stage) + } + }) + } +} + +func validTranscribeSpeakerRecord(speakerID string) TranscribeSpeakerRecord { + runID := "run-1" + return TranscribeSpeakerRecord{ + FingerprintVersion: TranscribeFingerprintContractVersion, + Fingerprint: strings.Repeat("a", 64), + AudioPath: "inputs/audio/" + speakerID + ".flac", + AudioSHA256: strings.Repeat("1", 64), + AudioSize: 1024, + Output: ArtifactRecord{ + Kind: "transcript_raw", LocalPath: "transcripts/raw/" + speakerID + ".json", + Checksum: strings.Repeat("b", 64), ProducerRunID: runID, + }, + OutputSize: 256, ProducerRunID: runID, + } +}