From 4d6086fefb651704730f63a8d96f638990ac9bbb Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Sun, 30 Aug 2026 12:54:38 +0000 Subject: [PATCH] Protect initial pipeline stage resume semantics --- docs/internal/stage-merge.md | 16 +- docs/internal/stage-prepare.md | 20 +- docs/internal/stage-transcribe.md | 17 +- docs/operations.md | 13 +- docs/roadmap/implementation.md | 2 +- .../app/analyze_artifacts_commands_test.go | 1 + internal/app/plan_test.go | 1 + internal/app/run_stage_test.go | 3 + internal/app/runner_test.go | 3 + internal/app/runner_test_helpers_test.go | 36 ++++ internal/app/semantic_resume_test.go | 88 +++++++++ internal/stage/semantic_contracts_initial.go | 180 ++++++++++++++++++ .../stage/semantic_contracts_initial_test.go | 174 +++++++++++++++++ 13 files changed, 549 insertions(+), 5 deletions(-) create mode 100644 internal/stage/semantic_contracts_initial.go create mode 100644 internal/stage/semantic_contracts_initial_test.go diff --git a/docs/internal/stage-merge.md b/docs/internal/stage-merge.md index 22b86d5..395dafa 100644 --- a/docs/internal/stage-merge.md +++ b/docs/internal/stage-merge.md @@ -29,9 +29,23 @@ Normalize raw transcript inputs and merge into base transcript via Seriatim. - base transcript must validate before stage success. - report output is config-gated. +## Resume Evidence + +Merge records a versioned semantic-configuration fingerprint for the Seriatim +merge operation, output schema, coalesce gap, and every configured advanced +merge transformation. A change reruns merge and stales only its fixed +descendants; prepare and transcribe remain reusable. Binary path, timeout, +report emission, logs, and diagnostic retention are operational exclusions. + +Configuration or resources loaded privately inside Seriatim are outside +Narratio's observable contract and require `--force` when changed. An existing +successful merge record without evidence reruns once when selected. + ## Related Contracts And Tests - [Seriatim](../integrations/seriatim.md) owns subprocess and output semantics. - [Configuration](../config.md#pipeline) owns operator-selected Seriatim values. - Implementation and tests: `internal/stage/merge.go`, - `internal/stage/merge_test.go` + `internal/stage/merge_test.go`, + `internal/stage/semantic_contracts_initial.go`, and + `internal/stage/semantic_contracts_initial_test.go` diff --git a/docs/internal/stage-prepare.md b/docs/internal/stage-prepare.md index fb37408..f39e045 100644 --- a/docs/internal/stage-prepare.md +++ b/docs/internal/stage-prepare.md @@ -57,6 +57,21 @@ mapping, while the isolated legacy reader rejects ambiguous fallback matches. - managed `previous/` state represents only the current requirement set. - `manifest.inputs` ordering is deterministic (`kind`, `path`). +## Resume Evidence + +Prepare records a versioned semantic-configuration fingerprint for the +resolved campaign/session selection, local-versus-S3 audio mode and canonical +audio names, stable-input ownership/presence, previous-session identity, and +the effective previous-artifact requirement set. A change reruns prepare and +stales its fixed descendants. Existing successful records without this +evidence rerun once when selected. + +Workspace, spool, and cache placement and absolute source relocation are not +semantic when logical selection, canonical names, and bytes are equivalent. +The fingerprint deliberately does not read or rehash large audio. Prepared +input checksums remain the content provenance; force prepare after changing +source bytes that are not otherwise reflected by the semantic selection. + ## Related Contracts And Tests - [Configuration](../config.md) owns audio selection, stable input fields, and @@ -66,5 +81,8 @@ mapping, while the isolated legacy reader rejects ambiguous fallback matches. - [Storage Internals](storage.md) and [Artifact Internals](artifacts.md) explain the internal collaborators. - Implementation and tests: `internal/stage/prepare.go`, - `internal/stage/prepare_test.go`, `internal/audio/s3_audio_test.go`, + `internal/stage/prepare_test.go`, + `internal/stage/semantic_contracts_initial.go`, + `internal/stage/semantic_contracts_initial_test.go`, + `internal/audio/s3_audio_test.go`, `internal/previouscache/*_test.go` diff --git a/docs/internal/stage-transcribe.md b/docs/internal/stage-transcribe.md index 15f98e1..e8a8acb 100644 --- a/docs/internal/stage-transcribe.md +++ b/docs/internal/stage-transcribe.md @@ -31,6 +31,19 @@ Generate raw per-speaker transcripts from prepared audio using WhisperX. - each successful output is validated before stage success, and cancellation or incomplete dispatch cannot be reported as a successful result. +## Resume Evidence + +Transcribe records a versioned semantic-configuration fingerprint containing +the Narratio-visible WhisperX service URL and recognition language. Changes to +either rerun transcription and stale its fixed descendants while leaving +prepare reusable. Retry count/delay, concurrency, timeout, credentials, and +diagnostic locations are operational and do not change this evidence. + +WhisperX models or private service configuration not exposed by Narratio's +adapter contract cannot be fingerprinted; use `--force` after changing them. +An existing successful transcribe record without evidence reruns once when +selected. + ## Related Contracts And Tests - [WhisperX](../integrations/whisperx.md) owns HTTP, retry, timeout, and @@ -38,4 +51,6 @@ Generate raw per-speaker transcripts from prepared audio using WhisperX. - [Configuration](../config.md#pipeline) owns concurrency and other operator-selected values. - Implementation and tests: `internal/stage/transcribe.go`, - `internal/stage/transcribe_test.go` + `internal/stage/transcribe_test.go`, + `internal/stage/semantic_contracts_initial.go`, and + `internal/stage/semantic_contracts_initial_test.go` diff --git a/docs/operations.md b/docs/operations.md index 124049e..eac291c 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -88,7 +88,9 @@ Canonical stage order: Execution rules: -- succeeded stages are skipped unless `--force` is set; +- succeeded stages are skipped unless `--force` is set; stages with semantic + configuration contracts additionally require matching versioned evidence, + and missing legacy evidence causes a safe one-time rerun; - `run` continues interrupted or partially completed sessions by running non-succeeded stages; - forcing a stage marks succeeded transitive dependents as `stale` before the replacement runs; render and extract are independent siblings; and @@ -97,6 +99,15 @@ Execution rules: repeated self-skip with the same reason and no outputs is stable and does not perpetually rerun dependent work. +Prepare, transcribe, and merge currently provide semantic-configuration +evidence. Changing prepare selection semantics reruns all fixed descendants; +changing WhisperX language/service identity reuses prepare; and changing a +Seriatim merge transformation reuses prepare and transcribe. Operational +timeouts, retry/concurrency tuning, executable paths, workspace/cache/spool +placement, reports, diagnostics, and secret values are excluded. Configuration, +models, prompts, modules, or resources loaded privately inside external tools +remain unobservable to Narratio and require an explicit `--force` after change. + An explicit self-skip is a durable `skipped` stage outcome that later runs reconsider. It differs from successful no-output execution: disabled `render` and `publish`, and absent or no-executable `analyze`, record `succeeded` with diff --git a/docs/roadmap/implementation.md b/docs/roadmap/implementation.md index 61e5d62..15606ca 100644 --- a/docs/roadmap/implementation.md +++ b/docs/roadmap/implementation.md @@ -303,7 +303,7 @@ configuration before selectable profiles can change those values. ## Stage 4 — Prepare, Transcribe, And Merge Semantic Contracts -**Status: Pending** +**Status: Completed** ### Goal diff --git a/internal/app/analyze_artifacts_commands_test.go b/internal/app/analyze_artifacts_commands_test.go index f8f419a..9933f73 100644 --- a/internal/app/analyze_artifacts_commands_test.go +++ b/internal/app/analyze_artifacts_commands_test.go @@ -130,6 +130,7 @@ func TestRunArtifactsWithSucceededAnalyzeSkipsUnlessForced(t *testing.T) { seed.MarkStageSucceeded(stageName, time.Date(2026, 5, 3, 10, 1, 0, 0, time.UTC), nil) } seed.MarkStageSkipped("extract", time.Date(2026, 5, 3, 10, 1, 0, 0, time.UTC), "notarius_disabled") + seedCurrentSemanticEvidence(t, loadConfigForSemanticEvidence(t, pipelinePath, campaignPath, sessionPath), seed, "prepare", "transcribe", "merge") if err := store.Save(context.Background(), manifestPath, seed); err != nil { t.Fatalf("save manifest: %v", err) } diff --git a/internal/app/plan_test.go b/internal/app/plan_test.go index c1ce5ce..7bf150d 100644 --- a/internal/app/plan_test.go +++ b/internal/app/plan_test.go @@ -64,6 +64,7 @@ func TestPlanShowsRunAndSkipFromManifest(t *testing.T) { m := manifest.New("2026-05-03", time.Date(2026, 5, 3, 10, 0, 0, 0, time.UTC)) m.MarkStageSucceeded("prepare", time.Date(2026, 5, 3, 10, 1, 0, 0, time.UTC), nil) m.MarkStageSucceeded("transcribe", time.Date(2026, 5, 3, 10, 2, 0, 0, time.UTC), nil) + seedCurrentSemanticEvidence(t, loadConfigForSemanticEvidence(t, pipelinePath, campaignPath, sessionPath), m, "prepare", "transcribe") if err := store.Save(context.Background(), manifestPath, m); err != nil { t.Fatalf("save manifest: %v", err) } diff --git a/internal/app/run_stage_test.go b/internal/app/run_stage_test.go index da28a16..e94632f 100644 --- a/internal/app/run_stage_test.go +++ b/internal/app/run_stage_test.go @@ -22,6 +22,7 @@ func TestRunContinuesAfterCompletedStages(t *testing.T) { m := manifest.New("2026-05-03", time.Date(2026, 5, 3, 10, 0, 0, 0, time.UTC)) m.MarkStageSucceeded("prepare", time.Date(2026, 5, 3, 10, 1, 0, 0, time.UTC), nil) m.MarkStageSucceeded("transcribe", time.Date(2026, 5, 3, 10, 2, 0, 0, time.UTC), nil) + seedCurrentSemanticEvidence(t, loadConfigForSemanticEvidence(t, pipelinePath, campaignPath, sessionPath), m, "prepare", "transcribe") if err := store.Save(context.Background(), manifestPath, m); err != nil { t.Fatalf("save manifest: %v", err) } @@ -60,6 +61,7 @@ func TestRunNoRemainingStagesRecordsSkippedStages(t *testing.T) { m.MarkStageSucceeded(name, time.Date(2026, 5, 3, 10, 1, 0, 0, time.UTC), nil) } m.MarkStageSkipped("extract", time.Date(2026, 5, 3, 10, 1, 0, 0, time.UTC), "notarius_disabled") + seedCurrentSemanticEvidence(t, loadConfigForSemanticEvidence(t, pipelinePath, campaignPath, sessionPath), m, "prepare", "transcribe", "merge") if err := store.Save(context.Background(), manifestPath, m); err != nil { t.Fatalf("save manifest: %v", err) } @@ -208,6 +210,7 @@ func TestRunStageForceMarksDownstreamStaleAndRunContinuesFromStale(t *testing.T) for _, name := range []string{"prepare", "transcribe", "merge", "polish", "normalize", "trim", "extract", "render", "analyze", "publish", "notify"} { seed.MarkStageSucceeded(name, time.Date(2026, 5, 3, 10, 1, 0, 0, time.UTC), nil) } + seedCurrentSemanticEvidence(t, loadConfigForSemanticEvidence(t, pipelinePath, campaignPath, sessionPath), seed, "prepare", "transcribe", "merge") if err := store.Save(context.Background(), manifestPath, seed); err != nil { t.Fatalf("save manifest: %v", err) } diff --git a/internal/app/runner_test.go b/internal/app/runner_test.go index 95a4c0c..c6965a3 100644 --- a/internal/app/runner_test.go +++ b/internal/app/runner_test.go @@ -642,6 +642,7 @@ func TestExecuteStagesForceRerunsSucceeded(t *testing.T) { existing := manifest.New(cfg.Session.SessionID, time.Date(2026, 5, 3, 1, 0, 0, 0, time.UTC)) existing.MarkStageSucceeded("transcribe", time.Date(2026, 5, 3, 1, 1, 0, 0, time.UTC), nil) + seedCurrentSemanticEvidence(t, cfg, existing, "transcribe") if err := os.MkdirAll(filepath.Dir(manifestPath), 0o755); err != nil { t.Fatalf("MkdirAll() error = %v", err) } @@ -983,6 +984,7 @@ func TestExecuteStagesRunManifestRecordsSkippedStage(t *testing.T) { manifestPath := manifestPathFor(cfg) existing := manifest.New(cfg.Session.SessionID, time.Date(2026, 5, 3, 1, 0, 0, 0, time.UTC)) existing.MarkStageSucceeded("transcribe", time.Date(2026, 5, 3, 1, 1, 0, 0, time.UTC), nil) + seedCurrentSemanticEvidence(t, cfg, existing, "transcribe") if err := os.MkdirAll(filepath.Dir(manifestPath), 0o755); err != nil { t.Fatalf("MkdirAll() error = %v", err) } @@ -1232,6 +1234,7 @@ func TestExecuteStagesSkippedStagePreservesExistingOutputsProvenance(t *testing. ProducerRunID: "20260501T000000Z-deadbeef", }, }) + seedCurrentSemanticEvidence(t, cfg, existing, "transcribe") if err := os.MkdirAll(filepath.Dir(manifestPath), 0o755); err != nil { t.Fatalf("MkdirAll() error = %v", err) } diff --git a/internal/app/runner_test_helpers_test.go b/internal/app/runner_test_helpers_test.go index 85cf1da..3dde60e 100644 --- a/internal/app/runner_test_helpers_test.go +++ b/internal/app/runner_test_helpers_test.go @@ -2,8 +2,10 @@ package app import ( "context" + "testing" "gitea.maximumdirect.net/eric/narratio/internal/config" + "gitea.maximumdirect.net/eric/narratio/internal/manifest" "gitea.maximumdirect.net/eric/narratio/internal/stage" ) @@ -13,3 +15,37 @@ func executeStages(ctx context.Context, cfg *config.Config, stages []stage.Stage plan := BoundedPlan{stages: append([]stage.Stage(nil), stages...)} return executePlan(ctx, cfg, plan, opts) } + +func seedCurrentSemanticEvidence(t *testing.T, cfg *config.Config, m *manifest.Manifest, names ...string) { + t.Helper() + providers := make(map[string]stage.SemanticConfigFingerprinter) + for _, candidate := range stage.All() { + if provider, ok := candidate.(stage.SemanticConfigFingerprinter); ok { + providers[candidate.Name()] = provider + } + } + for _, name := range names { + record := m.Stages[name] + if record == nil { + t.Fatalf("stage %q must exist before semantic evidence is seeded", name) + } + provider, ok := providers[name] + if !ok { + t.Fatalf("stage %q has no semantic fingerprint provider", name) + } + fingerprint, err := provider.SemanticConfigFingerprint(&stage.Env{Config: cfg}) + if err != nil { + t.Fatalf("fingerprint stage %q: %v", name, err) + } + record.SemanticConfig = &fingerprint + } +} + +func loadConfigForSemanticEvidence(t *testing.T, pipelinePath, campaignPath, sessionPath string) *config.Config { + t.Helper() + cfg, err := config.Load(pipelinePath, campaignPath, sessionPath) + if err != nil { + t.Fatalf("load config for semantic evidence: %v", err) + } + return cfg +} diff --git a/internal/app/semantic_resume_test.go b/internal/app/semantic_resume_test.go index 41bfffb..5e4f903 100644 --- a/internal/app/semantic_resume_test.go +++ b/internal/app/semantic_resume_test.go @@ -9,6 +9,7 @@ import ( "testing" "time" + "gitea.maximumdirect.net/eric/narratio/internal/config" "gitea.maximumdirect.net/eric/narratio/internal/manifest" "gitea.maximumdirect.net/eric/narratio/internal/stage" ) @@ -39,6 +40,21 @@ type semanticResumeCheckingStage struct { validationCalls *int } +type semanticContractRunStub struct { + name string + provider stage.SemanticConfigFingerprinter + runs *int +} + +func (s semanticContractRunStub) Name() string { return s.name } +func (s semanticContractRunStub) Run(_ context.Context, _ *stage.Env, _ *manifest.Manifest) (*stage.StageResult, error) { + *s.runs++ + return &stage.StageResult{}, nil +} +func (s semanticContractRunStub) SemanticConfigFingerprint(env *stage.Env) (manifest.SemanticConfigFingerprint, error) { + return s.provider.SemanticConfigFingerprint(env) +} + func (s semanticResumeCheckingStage) ValidateResume(_ context.Context, _ *stage.Env, _ *manifest.Manifest) (stage.ResumeValidation, error) { if s.validationCalls != nil { *s.validationCalls++ @@ -295,6 +311,78 @@ func TestSemanticMismatchInvalidatesOnlyFixedDependents(t *testing.T) { } } +func TestInitialPipelineSemanticChangesRerunOnlyAffectedLineage(t *testing.T) { + for _, test := range []struct { + name string + mutate func(*config.Config) + wantRuns [3]int + }{ + {name: "prepare selection", mutate: func(cfg *config.Config) { + cfg.Session.PreviousSessionID = "2026-04-26" + }, wantRuns: [3]int{1, 1, 1}}, + {name: "transcribe language", mutate: func(cfg *config.Config) { + cfg.Pipeline.WhisperX.Language = "fr" + }, wantRuns: [3]int{0, 1, 1}}, + {name: "merge transformation", mutate: func(cfg *config.Config) { + value := 1.75 + cfg.Pipeline.Seriatim.CoalesceGap = &value + }, wantRuns: [3]int{0, 0, 1}}, + } { + t.Run(test.name, func(t *testing.T) { + cfg := testConfig(t) + names := []string{"prepare", "transcribe", "merge"} + providers := make([]stage.SemanticConfigFingerprinter, len(names)) + for index, name := range names { + providers[index] = canonicalSemanticProvider(t, name) + } + seed := manifest.New(cfg.Session.SessionID, time.Now().UTC()) + for _, candidate := range stage.All() { + seed.MarkStageSucceeded(candidate.Name(), time.Now().UTC(), nil) + } + for index, name := range names { + fingerprint, err := providers[index].SemanticConfigFingerprint(&stage.Env{Config: cfg}) + if err != nil { + t.Fatal(err) + } + seed.Stages[name].SemanticConfig = &fingerprint + } + store := &manifest.LocalStore{} + if err := store.Save(context.Background(), manifestPathFor(cfg), seed); err != nil { + t.Fatal(err) + } + test.mutate(cfg) + + runs := [3]int{} + selected := make([]stage.Stage, 0, len(names)) + for index, name := range names { + selected = append(selected, semanticContractRunStub{name: name, provider: providers[index], runs: &runs[index]}) + } + if _, err := executeStages(context.Background(), cfg, selected, RunOptions{}); err != nil { + t.Fatal(err) + } + if runs != test.wantRuns { + t.Fatalf("runs = %v, want %v", runs, test.wantRuns) + } + }) + } +} + +func canonicalSemanticProvider(t *testing.T, name string) stage.SemanticConfigFingerprinter { + t.Helper() + for _, candidate := range stage.All() { + if candidate.Name() != name { + continue + } + provider, ok := candidate.(stage.SemanticConfigFingerprinter) + if !ok { + t.Fatalf("canonical stage %q does not implement semantic fingerprinting", name) + } + return provider + } + t.Fatalf("canonical stage %q not found", name) + return nil +} + func semanticFingerprint(version int, seed string) manifest.SemanticConfigFingerprint { digest := sha256.Sum256([]byte(seed)) return manifest.SemanticConfigFingerprint{Version: version, Digest: hex.EncodeToString(digest[:])} diff --git a/internal/stage/semantic_contracts_initial.go b/internal/stage/semantic_contracts_initial.go new file mode 100644 index 0000000..8bccaa2 --- /dev/null +++ b/internal/stage/semantic_contracts_initial.go @@ -0,0 +1,180 @@ +package stage + +import ( + "fmt" + "path/filepath" + "sort" + "strings" + + "gitea.maximumdirect.net/eric/narratio/internal/config" + "gitea.maximumdirect.net/eric/narratio/internal/manifest" +) + +const ( + prepareSemanticConfigVersion = 1 + transcribeSemanticConfigVersion = 1 + mergeSemanticConfigVersion = 1 +) + +type prepareSemanticConfig struct { + CampaignID string `json:"campaign_id"` + SessionID string `json:"session_id"` + PreviousSession string `json:"previous_session_id"` + Audio prepareAudioSelection `json:"audio"` + StableInputs []prepareStableInputSelection `json:"stable_inputs"` + PreviousArtifacts []preparePreviousRequirement `json:"previous_artifacts"` +} + +type prepareAudioSelection struct { + Mode string `json:"mode"` + LocalNames []string `json:"local_names,omitempty"` + S3Bucket string `json:"s3_bucket,omitempty"` + S3RootPrefix string `json:"s3_root_prefix,omitempty"` + S3AudioPrefix string `json:"s3_audio_prefix,omitempty"` +} + +type prepareStableInputSelection struct { + Kind string `json:"kind"` + Source string `json:"source"` + Present bool `json:"present"` +} + +type preparePreviousRequirement struct { + Name string `json:"name"` + Required bool `json:"required"` + OutputPath string `json:"output_path,omitempty"` +} + +type transcribeSemanticConfig struct { + ServiceURL string `json:"service_url"` + Language string `json:"language"` +} + +type mergeSemanticConfig struct { + Operation string `json:"operation"` + OutputSchema string `json:"output_schema"` + CoalesceGap *float64 `json:"coalesce_gap,omitempty"` + OverlapWordRunGap *float64 `json:"overlap_word_run_gap,omitempty"` + OverlapWordRunReorderWindow *float64 `json:"overlap_word_run_reorder_window,omitempty"` + BackchannelMaxDuration *float64 `json:"backchannel_max_duration,omitempty"` + FillerMaxDuration *float64 `json:"filler_max_duration,omitempty"` +} + +func (prepareStage) SemanticConfigFingerprint(env *Env) (manifest.SemanticConfigFingerprint, error) { + payload, err := buildPrepareSemanticConfig(env) + if err != nil { + return manifest.SemanticConfigFingerprint{}, err + } + return FingerprintSemanticConfig(prepareSemanticConfigVersion, payload) +} + +func (transcribeStage) SemanticConfigFingerprint(env *Env) (manifest.SemanticConfigFingerprint, error) { + if env == nil || env.Config == nil || env.Config.Pipeline == nil { + return manifest.SemanticConfigFingerprint{}, fmt.Errorf("transcribe semantic configuration requires resolved pipeline config") + } + whisper := env.Config.Pipeline.WhisperX + return FingerprintSemanticConfig(transcribeSemanticConfigVersion, transcribeSemanticConfig{ + ServiceURL: strings.TrimSpace(whisper.TranscribeURL), + Language: strings.TrimSpace(whisper.Language), + }) +} + +func (mergeStage) SemanticConfigFingerprint(env *Env) (manifest.SemanticConfigFingerprint, error) { + if env == nil || env.Config == nil || env.Config.Pipeline == nil { + return manifest.SemanticConfigFingerprint{}, fmt.Errorf("merge semantic configuration requires resolved pipeline config") + } + seriatim := env.Config.Pipeline.Seriatim + return FingerprintSemanticConfig(mergeSemanticConfigVersion, mergeSemanticConfig{ + Operation: "merge", + OutputSchema: strings.TrimSpace(seriatim.OutputSchema), + CoalesceGap: cloneFloat64(seriatim.CoalesceGap), + OverlapWordRunGap: cloneFloat64(seriatim.Env.OverlapWordRunGap), + OverlapWordRunReorderWindow: cloneFloat64(seriatim.Env.OverlapWordRunReorderWindow), + BackchannelMaxDuration: cloneFloat64(seriatim.Env.BackchannelMaxDuration), + FillerMaxDuration: cloneFloat64(seriatim.Env.FillerMaxDuration), + }) +} + +func buildPrepareSemanticConfig(env *Env) (prepareSemanticConfig, error) { + if env == nil || env.Config == nil || env.Config.Pipeline == nil || env.Config.Session == nil { + return prepareSemanticConfig{}, fmt.Errorf("prepare semantic configuration requires resolved pipeline and session config") + } + cfg := env.Config + requirements, err := collectPreparePreviousRequirements(cfg, env.EffectiveArtifacts) + if err != nil { + return prepareSemanticConfig{}, fmt.Errorf("resolve previous artifact requirements: %w", err) + } + payload := prepareSemanticConfig{ + CampaignID: strings.TrimSpace(cfg.Session.Campaign), + SessionID: strings.TrimSpace(cfg.Session.SessionID), + PreviousSession: strings.TrimSpace(cfg.Session.PreviousSessionID), + Audio: prepareAudioSemantics(cfg), + StableInputs: prepareStableInputSemantics(cfg), + } + for _, requirement := range requirements { + entry := preparePreviousRequirement{Name: strings.TrimSpace(requirement.Name), Required: requirement.Required} + if cfg.Pipeline.Scriptorium != nil { + entry.OutputPath = strings.TrimSpace(cfg.Pipeline.Scriptorium.Artifacts[requirement.Name].OutputPath) + } + payload.PreviousArtifacts = append(payload.PreviousArtifacts, entry) + } + return payload, nil +} + +func prepareAudioSemantics(cfg *config.Config) prepareAudioSelection { + inputs := cfg.Session.Inputs + if inputs.AudioS3 != nil { + selection := prepareAudioSelection{Mode: "s3", S3AudioPrefix: strings.TrimSpace(inputs.AudioS3.Prefix)} + if cfg.Pipeline.Storage.S3 != nil { + selection.S3Bucket = strings.TrimSpace(cfg.Pipeline.Storage.S3.Bucket) + selection.S3RootPrefix = strings.TrimSpace(cfg.Pipeline.Storage.S3.RootPrefix) + } + return selection + } + selection := prepareAudioSelection{Mode: "local_directory"} + if len(inputs.AudioFiles) > 0 { + selection.Mode = "local_files" + destinations := localAudioDestinations(inputs.AudioFiles) + for _, name := range destinations { + selection.LocalNames = append(selection.LocalNames, filepath.ToSlash(name)) + } + sort.Strings(selection.LocalNames) + } + return selection +} + +func prepareStableInputSemantics(cfg *config.Config) []prepareStableInputSelection { + stable := cfg.StableInputs + return []prepareStableInputSelection{ + prepareStableInput("speakers", stable.SpeakersFile, cfg.Session.Inputs.SpeakersFile), + prepareStableInput("autocorrect", stable.AutocorrectFile, cfg.Session.Inputs.AutocorrectFile), + prepareStableInput("glossary", stable.GlossaryFile, cfg.Session.Inputs.GlossaryFile), + prepareStableInput("players", stable.PlayersFile, cfg.Session.Inputs.PlayersFile), + prepareStableInput("party", stable.PartyFile, cfg.Session.Inputs.PartyFile), + prepareStableInput("spell_catalog", stable.SpellCatalogFile, cfg.Session.Inputs.SpellCatalogFile), + } +} + +func prepareStableInput(kind string, resolved config.ResolvedInputFile, fallback string) prepareStableInputSelection { + source := strings.TrimSpace(resolved.Source) + if source == "" { + source = "session_config" + } + path := resolved.Path + if strings.TrimSpace(path) == "" { + path = fallback + } + return prepareStableInputSelection{Kind: kind, Source: source, Present: strings.TrimSpace(path) != ""} +} + +func cloneFloat64(value *float64) *float64 { + if value == nil { + return nil + } + copy := *value + return © +} + +var _ SemanticConfigFingerprinter = prepareStage{} +var _ SemanticConfigFingerprinter = transcribeStage{} +var _ SemanticConfigFingerprinter = mergeStage{} diff --git a/internal/stage/semantic_contracts_initial_test.go b/internal/stage/semantic_contracts_initial_test.go new file mode 100644 index 0000000..8d545d1 --- /dev/null +++ b/internal/stage/semantic_contracts_initial_test.go @@ -0,0 +1,174 @@ +package stage + +import ( + "testing" + + "gitea.maximumdirect.net/eric/narratio/internal/config" + "gitea.maximumdirect.net/eric/narratio/internal/manifest" +) + +func TestPrepareSemanticConfigSensitivity(t *testing.T) { + assertSemanticFingerprintChanges(t, prepareStage{}, initialSemanticEnv(), []struct { + name string + mutate func(*Env) + }{ + {name: "audio mode", mutate: func(env *Env) { + env.Config.Session.Inputs.AudioFiles = nil + env.Config.Session.Inputs.AudioS3 = &config.SessionAudioS3Input{Prefix: "sessions/audio"} + env.Config.Pipeline.Storage.S3 = &config.StorageS3Config{Bucket: "archive", RootPrefix: "dnd"} + }}, + {name: "audio canonical name", mutate: func(env *Env) { + env.Config.Session.Inputs.AudioFiles = []string{"/audio/other.flac"} + }}, + {name: "previous session", mutate: func(env *Env) { + env.Config.Session.PreviousSessionID = "2026-04-26" + }}, + {name: "stable input owner", mutate: func(env *Env) { + env.Config.StableInputs.PartyFile.Source = "campaign_config" + }}, + }) + + assertSemanticFingerprintUnchanged(t, prepareStage{}, initialSemanticEnv(), []struct { + name string + mutate func(*Env) + }{ + {name: "workspace and spool", mutate: func(env *Env) { + env.Config.Pipeline.Workspace.Root = "/different/work" + env.Config.Pipeline.Spool.Root = "/different/spool" + env.Config.Pipeline.Cache.Root = "/different/cache" + }}, + {name: "absolute stable source path", mutate: func(env *Env) { + env.Config.StableInputs.PartyFile.Path = "/relocated/party.yml" + env.Config.StableInputs.PartyFile.ConfigPath = "/relocated/campaign.yml" + }}, + {name: "absolute unique audio source", mutate: func(env *Env) { + env.Config.Session.Inputs.AudioFiles = []string{"/relocated/speaker.flac"} + }}, + }) +} + +func TestTranscribeSemanticConfigSensitivity(t *testing.T) { + assertSemanticFingerprintChanges(t, transcribeStage{}, initialSemanticEnv(), []struct { + name string + mutate func(*Env) + }{ + {name: "language", mutate: func(env *Env) { env.Config.Pipeline.WhisperX.Language = "fr" }}, + {name: "service", mutate: func(env *Env) { env.Config.Pipeline.WhisperX.TranscribeURL = "https://other.example/transcribe" }}, + }) + assertSemanticFingerprintUnchanged(t, transcribeStage{}, initialSemanticEnv(), []struct { + name string + mutate func(*Env) + }{ + {name: "transport tuning", mutate: func(env *Env) { + env.Config.Pipeline.WhisperX.Timeout = "30m" + env.Config.Pipeline.WhisperX.Retries = intPointer(9) + env.Config.Pipeline.WhisperX.Concurrency = intPointer(12) + env.Config.Pipeline.WhisperX.RetryDelay = "10s" + }}, + }) +} + +func TestMergeSemanticConfigSensitivity(t *testing.T) { + assertSemanticFingerprintChanges(t, mergeStage{}, initialSemanticEnv(), []struct { + name string + mutate func(*Env) + }{ + {name: "output schema", mutate: func(env *Env) { env.Config.Pipeline.Seriatim.OutputSchema = "seriatim.transcript.v2" }}, + {name: "coalesce gap", mutate: func(env *Env) { env.Config.Pipeline.Seriatim.CoalesceGap = floatPointer(1.25) }}, + {name: "overlap transformation", mutate: func(env *Env) { env.Config.Pipeline.Seriatim.Env.OverlapWordRunGap = floatPointer(0.75) }}, + }) + assertSemanticFingerprintUnchanged(t, mergeStage{}, initialSemanticEnv(), []struct { + name string + mutate func(*Env) + }{ + {name: "process and diagnostics", mutate: func(env *Env) { + env.Config.Pipeline.Seriatim.Binary = "/opt/seriatim" + env.Config.Pipeline.Seriatim.Timeout = "45m" + env.Config.Pipeline.Seriatim.Report = boolPointer(false) + }}, + }) +} + +func initialSemanticEnv() *Env { + return &Env{Config: &config.Config{ + Pipeline: &config.PipelineConfig{ + Workspace: config.WorkspaceConfig{Root: "/work"}, + Spool: config.SpoolConfig{Root: "/spool"}, + Cache: config.CacheConfig{Root: "/cache"}, + WhisperX: config.WhisperXConfig{ + TranscribeURL: "https://whisper.example/transcribe", Language: "en", + Timeout: "10m", Retries: intPointer(3), RetryDelay: "1s", Concurrency: intPointer(2), + }, + Seriatim: config.SeriatimConfig{ + Binary: "seriatim", Timeout: "10m", OutputSchema: "seriatim.transcript.v1", + CoalesceGap: floatPointer(0.5), Report: boolPointer(true), + Env: config.SeriatimEnvConfig{OverlapWordRunGap: floatPointer(0.25)}, + }, + }, + Session: &config.SessionConfig{ + SessionID: "2026-05-03", Campaign: "campaign", + Inputs: config.SessionInputsConfig{ + AudioFiles: []string{"/audio/speaker.flac"}, PartyFile: "party.yml", + }, + }, + StableInputs: config.ResolvedStableInputs{ + PartyFile: config.ResolvedInputFile{Path: "/campaign/party.yml", ConfigPath: "/campaign/campaign.yml", Source: "session_config"}, + }, + }} +} + +type semanticFingerprintProvider interface { + SemanticConfigFingerprint(*Env) (manifest.SemanticConfigFingerprint, error) +} + +func assertSemanticFingerprintChanges(t *testing.T, provider semanticFingerprintProvider, base *Env, tests []struct { + name string + mutate func(*Env) +}) { + t.Helper() + want, err := provider.SemanticConfigFingerprint(base) + if err != nil { + t.Fatal(err) + } + for _, test := range tests { + t.Run("semantic "+test.name, func(t *testing.T) { + candidate := initialSemanticEnv() + test.mutate(candidate) + got, err := provider.SemanticConfigFingerprint(candidate) + if err != nil { + t.Fatal(err) + } + if want.Equal(got) { + t.Fatalf("semantic change %q retained fingerprint %q", test.name, got.Digest) + } + }) + } +} + +func assertSemanticFingerprintUnchanged(t *testing.T, provider semanticFingerprintProvider, base *Env, tests []struct { + name string + mutate func(*Env) +}) { + t.Helper() + want, err := provider.SemanticConfigFingerprint(base) + if err != nil { + t.Fatal(err) + } + for _, test := range tests { + t.Run("operational "+test.name, func(t *testing.T) { + candidate := initialSemanticEnv() + test.mutate(candidate) + got, err := provider.SemanticConfigFingerprint(candidate) + if err != nil { + t.Fatal(err) + } + if !want.Equal(got) { + t.Fatalf("operational change %q changed fingerprint: want %#v got %#v", test.name, want, got) + } + }) + } +} + +func intPointer(value int) *int { return &value } +func floatPointer(value float64) *float64 { return &value } +func boolPointer(value bool) *bool { return &value }