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

Protect initial pipeline stage resume semantics

Browse Source
This commit is contained in:
Eric Rakestraw
2026-08-30 12:54:38 +00:00
parent 82cb53e107
commit 4d6086fefb
13 changed files with 549 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
internal/app/analyze_artifacts_commands_test.go
View File

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

1
internal/app/plan_test.go
View File

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

3
internal/app/run_stage_test.go
View File

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

3
internal/app/runner_test.go
View File

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

36
internal/app/runner_test_helpers_test.go
View File

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

88
internal/app/semantic_resume_test.go
View File

@@ -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[:])}