eric/narratio
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 27 Wiki Activity

Add durable speaker transcription state

Browse Source
This commit is contained in:
Eric Rakestraw
2026-09-21 16:01:19 +00:00
parent 89ada1b269
commit 7b0bf99ed1
5 changed files with 405 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
docs/roadmap/implementation.md
View File

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

30
internal/manifest/manifest.go
View File

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

6
internal/manifest/store.go
View File

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

128
internal/manifest/transcribe_state.go Normal file
View File

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

248
internal/manifest/transcribe_state_test.go Normal file
View File

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