diff --git a/docs/cli.md b/docs/cli.md index f104db7..75447c7 100644 --- a/docs/cli.md +++ b/docs/cli.md @@ -75,7 +75,8 @@ narratio run [--force] [--artifacts ] [...common co Behavior: - evaluates full stage order; -- skips already-succeeded stages unless `--force` is set; +- skips already-succeeded stages unless `--force` is set or a stage-specific + resume check finds its durable result obsolete; - continues interrupted or partially completed sessions by running non-succeeded stages; - writes session and run manifests. @@ -93,6 +94,7 @@ Valid stage names: - `polish` - `normalize` - `trim` +- `extract` - `render` - `analyze` - `publish` diff --git a/docs/internal/overview.md b/docs/internal/overview.md index c83c7b1..62b82ba 100644 --- a/docs/internal/overview.md +++ b/docs/internal/overview.md @@ -53,10 +53,11 @@ The implemented canonical order is: 4. [`polish`](stage-polish.md) 5. [`normalize`](stage-normalize.md) 6. [`trim`](stage-trim.md) -7. [`render`](stage-render.md) -8. [`analyze`](stage-analyze.md) -9. [`publish`](stage-publish.md) -10. `notify` (placeholder) +7. `extract` +8. [`render`](stage-render.md) +9. [`analyze`](stage-analyze.md) +10. [`publish`](stage-publish.md) +11. `notify` (placeholder) `notify` currently has optional notifier call behavior and no persisted pipeline outputs; its default collaborator is a no-op sender. The focused stage diff --git a/docs/roadmap/implementation.md b/docs/roadmap/implementation.md index 7113dfc..315dbe0 100644 --- a/docs/roadmap/implementation.md +++ b/docs/roadmap/implementation.md @@ -21,7 +21,7 @@ implementation sequence. | Stage 3 | Complete | | Stage 4 | Complete | | Stage 5 | Complete | -| Stage 6 | Not started | +| Stage 6 | Complete | | Stage 7 | Not started | | Stage 8 | Not started | | Stage 9 | Not started | diff --git a/internal/app/analyze_artifacts_commands_test.go b/internal/app/analyze_artifacts_commands_test.go index 27c2393..d881dd0 100644 --- a/internal/app/analyze_artifacts_commands_test.go +++ b/internal/app/analyze_artifacts_commands_test.go @@ -21,7 +21,7 @@ func TestExecuteRunStageArtifactsUnsupportedStageFails(t *testing.T) { var stdout bytes.Buffer var stderr bytes.Buffer code := Execute( - []string{"run-stage", "polish", "2026-05-03", "--config", pipelinePath, "--campaign-file", campaignPath, "--session", sessionPath, "--artifacts", "session_recap"}, + []string{"run-stage", "extract", "2026-05-03", "--config", pipelinePath, "--campaign-file", campaignPath, "--session", sessionPath, "--artifacts", "session_recap"}, &stdout, &stderr, ) @@ -143,7 +143,7 @@ func TestRunArtifactsWithSucceededAnalyzeSkipsUnlessForced(t *testing.T) { if err != nil { t.Fatalf("Run() error = %v", err) } - if !strings.Contains(out.String(), "executed=0 skipped=10") { + if !strings.Contains(out.String(), "executed=1 skipped=11") { t.Fatalf("output = %q, want all stages skipped", out.String()) } } diff --git a/internal/app/commands_test.go b/internal/app/commands_test.go index 98288d7..ec4398a 100644 --- a/internal/app/commands_test.go +++ b/internal/app/commands_test.go @@ -31,8 +31,8 @@ func TestExecuteValidCommands(t *testing.T) { args []string wantOut string }{ - {name: "run", args: []string{"run", "2026-05-03", "--config", pipelinePath, "--campaign-file", campaignPath, "--session", sessionPath}, wantOut: "narratio run: session 2026-05-03; executed=10 skipped=0; manifest="}, - {name: "session plan", args: []string{"session", "plan", "2026-05-03", "--config", pipelinePath, "--campaign-file", campaignPath, "--session", sessionPath}, wantOut: "prepare: skip\ntranscribe: skip\nmerge: skip\npolish: skip\nnormalize: skip\ntrim: skip\nrender: skip\nanalyze: skip\npublish: skip\nnotify: skip"}, + {name: "run", args: []string{"run", "2026-05-03", "--config", pipelinePath, "--campaign-file", campaignPath, "--session", sessionPath}, wantOut: "narratio run: session 2026-05-03; executed=11 skipped=1; manifest="}, + {name: "session plan", args: []string{"session", "plan", "2026-05-03", "--config", pipelinePath, "--campaign-file", campaignPath, "--session", sessionPath}, wantOut: "prepare: skip\ntranscribe: skip\nmerge: skip\npolish: skip\nnormalize: skip\ntrim: skip\nextract: run\nrender: skip\nanalyze: skip\npublish: skip\nnotify: skip"}, {name: "session status", args: []string{"session", "status", "2026-05-03", "--config", pipelinePath, "--campaign-file", campaignPath, "--session", sessionPath}, wantOut: "Session: 2026-05-03"}, {name: "run-stage", args: []string{"run-stage", "polish", "2026-05-03", "--config", pipelinePath, "--campaign-file", campaignPath, "--session", sessionPath}, wantOut: "narratio run-stage: stage=polish executed=0 skipped=1 force=false; manifest="}, } @@ -335,7 +335,7 @@ func TestExecuteUsesDefaultPipelineConfigPathWhenConfigFlagOmitted(t *testing.T) if code != 0 { t.Fatalf("exit code = %d, want 0; stderr=%q", code, stderr.String()) } - if !strings.Contains(stdout.String(), "narratio run: session 2026-05-03; executed=10 skipped=0; manifest=") { + if !strings.Contains(stdout.String(), "narratio run: session 2026-05-03; executed=11 skipped=1; manifest=") { t.Fatalf("stdout = %q, want successful run output", stdout.String()) } } diff --git a/internal/app/extract_lifecycle_test.go b/internal/app/extract_lifecycle_test.go new file mode 100644 index 0000000..ff08755 --- /dev/null +++ b/internal/app/extract_lifecycle_test.go @@ -0,0 +1,239 @@ +package app + +import ( + "context" + "encoding/json" + "fmt" + "os" + "path/filepath" + "strings" + "testing" + "time" + + "gitea.maximumdirect.net/eric/narratio/internal/adapters/notarius" + "gitea.maximumdirect.net/eric/narratio/internal/artifactmodel" + "gitea.maximumdirect.net/eric/narratio/internal/artifacts" + "gitea.maximumdirect.net/eric/narratio/internal/config" + "gitea.maximumdirect.net/eric/narratio/internal/manifest" + "gitea.maximumdirect.net/eric/narratio/internal/stage" +) + +type materializingNotariusRunner struct { + cfg *config.NotariusConfig + requests []notarius.RunRequest +} + +func (r *materializingNotariusRunner) Run(_ context.Context, req notarius.RunRequest) (notarius.RunResult, error) { + r.requests = append(r.requests, req) + externalRunID := fmt.Sprintf("notarius-run-%d", len(r.requests)) + bundle := filepath.Join(req.OutputRoot, externalRunID) + lanesDir := filepath.Join(bundle, "lanes") + if err := os.MkdirAll(lanesDir, 0o755); err != nil { + return notarius.RunResult{}, err + } + for path, content := range map[string]string{ + filepath.Join(bundle, "index.json"): `{"manifest_file":"manifest.json"}`, + filepath.Join(bundle, "manifest.json"): `{}`, + filepath.Join(bundle, "rejected.json"): `{"rejected":[]}`, + filepath.Join(bundle, "warnings.json"): `{"warnings":[]}`, + filepath.Join(lanesDir, "npcs.json"): `{"npcs":[]}`, + } { + if err := os.WriteFile(path, []byte(content), 0o644); err != nil { + return notarius.RunResult{}, err + } + } + output := r.cfg.Outputs["npc_registry"] + return notarius.RunResult{ + Receipt: notarius.Receipt{ + SchemaVersion: notarius.ReceiptSchemaVersion, RunID: externalRunID, + PipelineID: req.PipelineID, OutputDirectory: bundle, IndexFile: "index.json", + NormalizedOutputCount: 1, ValidationStatus: "valid", + }, + BundleRoot: bundle, + Index: notarius.Index{ + Path: filepath.Join(bundle, "index.json"), RejectedPath: filepath.Join(bundle, "rejected.json"), + WarningsPath: filepath.Join(bundle, "warnings.json"), + Lanes: []notarius.LaneDescriptor{{ + LaneID: output.LaneID, File: "lanes/npcs.json", Path: filepath.Join(lanesDir, "npcs.json"), + MediaType: output.MediaType, SchemaID: output.SchemaID, + SchemaVersion: output.SchemaVersion, ModuleKey: output.ModuleKey, + }}, + }, + }, nil +} + +func TestExtractLifecycleDisabledThenEnabled(t *testing.T) { + cfg, env, runner := extractionLifecycleFixture(t, false) + plan, err := BuildSingleStagePlan("extract") + if err != nil { + t.Fatalf("BuildSingleStagePlan(extract) error = %v", err) + } + + first, err := executeStages(context.Background(), cfg, plan, RunOptions{Env: env}) + if err != nil { + t.Fatalf("disabled executeStages() error = %v", err) + } + if len(first.Executed) != 1 || len(first.Skipped) != 1 || first.Skipped[0] != "extract" || len(runner.requests) != 0 { + t.Fatalf("disabled summary = %#v requests=%d", first, len(runner.requests)) + } + + cfg.Pipeline.Notarius.Enabled = true + second, err := executeStages(context.Background(), cfg, plan, RunOptions{Env: env}) + if err != nil { + t.Fatalf("enabled executeStages() error = %v", err) + } + if len(second.Executed) != 1 || len(second.Skipped) != 0 || len(runner.requests) != 1 { + t.Fatalf("enabled summary = %#v requests=%d", second, len(runner.requests)) + } + loaded, err := (&manifest.LocalStore{}).Load(context.Background(), second.ManifestPath) + if err != nil { + t.Fatalf("Load() error = %v", err) + } + if loaded.Stages["extract"] == nil || loaded.Stages["extract"].Status != manifest.StatusSucceeded || len(loaded.Stages["extract"].Outputs) != 2 { + t.Fatalf("extract record = %#v, want succeeded manifest-ready outputs", loaded.Stages["extract"]) + } +} + +func TestExtractLifecycleSkipsCurrentResumableResult(t *testing.T) { + cfg, env, runner := extractionLifecycleFixture(t, true) + plan, _ := BuildSingleStagePlan("extract") + if _, err := executeStages(context.Background(), cfg, plan, RunOptions{Env: env}); err != nil { + t.Fatalf("first executeStages() error = %v", err) + } + resumed, err := executeStages(context.Background(), cfg, plan, RunOptions{Env: env}) + if err != nil { + t.Fatalf("resume executeStages() error = %v", err) + } + if len(resumed.Executed) != 0 || len(resumed.Skipped) != 1 || resumed.Skipped[0] != "extract" || len(runner.requests) != 1 { + t.Fatalf("resume summary = %#v requests=%d", resumed, len(runner.requests)) + } +} + +func TestExtractLifecycleResumesAndRerunsObsoleteResults(t *testing.T) { + for _, test := range []struct { + name string + mutate func(*testing.T, *config.Config, *manifest.Manifest) + }{ + {name: "configuration changed", mutate: func(_ *testing.T, cfg *config.Config, _ *manifest.Manifest) { + output := cfg.Pipeline.Notarius.Outputs["npc_registry"] + output.SchemaVersion = "v2" + cfg.Pipeline.Notarius.Outputs["npc_registry"] = output + }}, + {name: "payload missing", mutate: func(t *testing.T, _ *config.Config, m *manifest.Manifest) { + if err := os.Remove(m.Stages["extract"].Outputs[0].LocalPath); err != nil { + t.Fatalf("Remove() error = %v", err) + } + }}, + {name: "payload tampered", mutate: func(t *testing.T, _ *config.Config, m *manifest.Manifest) { + if err := os.WriteFile(m.Stages["extract"].Outputs[0].LocalPath, []byte(`{"npcs":["tampered"]}`), 0o644); err != nil { + t.Fatalf("WriteFile() error = %v", err) + } + }}, + {name: "record incompatible", mutate: func(_ *testing.T, _ *config.Config, m *manifest.Manifest) { + m.Stages["extract"].Outputs[0].Contract.SchemaID = "incompatible" + }}, + } { + t.Run(test.name, func(t *testing.T) { + cfg, env, runner := extractionLifecycleFixture(t, true) + plan, _ := BuildSingleStagePlan("extract") + first, err := executeStages(context.Background(), cfg, plan, RunOptions{Env: env}) + if err != nil { + t.Fatalf("first executeStages() error = %v", err) + } + persisted, err := (&manifest.LocalStore{}).Load(context.Background(), first.ManifestPath) + if err != nil { + t.Fatalf("Load() error = %v", err) + } + + test.mutate(t, cfg, persisted) + if err := (&manifest.LocalStore{}).Save(context.Background(), first.ManifestPath, persisted); err != nil { + t.Fatalf("Save(mutated) error = %v", err) + } + rerun, err := executeStages(context.Background(), cfg, plan, RunOptions{Env: env}) + if err != nil { + t.Fatalf("rerun executeStages() error = %v", err) + } + if len(rerun.Executed) != 1 || len(rerun.Skipped) != 0 || len(runner.requests) != 2 { + t.Fatalf("rerun summary = %#v requests=%d", rerun, len(runner.requests)) + } + }) + } +} + +func TestExtractLifecycleUnsafeResumeErrorPreservesSuccess(t *testing.T) { + cfg, env, runner := extractionLifecycleFixture(t, true) + plan, _ := BuildSingleStagePlan("extract") + first, err := executeStages(context.Background(), cfg, plan, RunOptions{Env: env}) + if err != nil { + t.Fatalf("first executeStages() error = %v", err) + } + store := &manifest.LocalStore{} + persisted, err := store.Load(context.Background(), first.ManifestPath) + if err != nil { + t.Fatalf("Load() error = %v", err) + } + persisted.Stages["extract"].Outputs[0].LocalPath = filepath.Join(cfg.Pipeline.Workspace.Root, "outside.json") + if err := store.Save(context.Background(), first.ManifestPath, persisted); err != nil { + t.Fatalf("Save() error = %v", err) + } + before, _ := json.Marshal(persisted.Stages["extract"]) + + if _, err := executeStages(context.Background(), cfg, plan, RunOptions{Env: env}); err == nil || !strings.Contains(err.Error(), "unsafe") { + t.Fatalf("executeStages() error = %v, want unsafe resume failure", err) + } + afterManifest, err := store.Load(context.Background(), first.ManifestPath) + if err != nil { + t.Fatalf("Load(after) error = %v", err) + } + after, _ := json.Marshal(afterManifest.Stages["extract"]) + if string(before) != string(after) || len(runner.requests) != 1 { + t.Fatalf("successful extract record changed: before=%s after=%s requests=%d", before, after, len(runner.requests)) + } +} + +func extractionLifecycleFixture(t *testing.T, enabled bool) (*config.Config, *stage.Env, *materializingNotariusRunner) { + t.Helper() + cfg := testConfig(t) + root := cfg.Pipeline.Workspace.Root + binary := filepath.Join(root, "notarius") + configPath := filepath.Join(root, "notarius.yml") + workingDirectory := filepath.Join(root, "notarius-work") + if err := os.WriteFile(binary, []byte("#!/bin/sh\nexit 0\n"), 0o755); err != nil { + t.Fatalf("WriteFile(binary) error = %v", err) + } + if err := os.WriteFile(configPath, []byte("pipelines: {}\n"), 0o644); err != nil { + t.Fatalf("WriteFile(config) error = %v", err) + } + if err := os.Mkdir(workingDirectory, 0o755); err != nil { + t.Fatalf("Mkdir(working directory) error = %v", err) + } + cfg.Pipeline.Notarius = &config.NotariusConfig{ + Enabled: enabled, Binary: binary, ConfigPath: configPath, PipelineID: "dnd-session", + Timeout: "45m", WorkingDirectory: workingDirectory, + Outputs: map[string]config.NotariusOutputConfig{ + "npc_registry": { + LaneID: "npc-registry", MediaType: "application/json", SchemaID: "notarius.dnd.npc_registry", + SchemaVersion: "v1", ModuleKey: "dnd/npc-registry", + }, + }, + } + paths, err := artifacts.NewLocalStore(root).EnsureLayoutFor(cfg.Session.Campaign, cfg.Session.SessionID) + if err != nil { + t.Fatalf("EnsureLayoutFor() error = %v", err) + } + inputPath := filepath.Join(paths.ArtifactsDir, "trimmed.from-manifest.json") + if err := os.WriteFile(inputPath, []byte(`{"segments":[]}`), 0o644); err != nil { + t.Fatalf("WriteFile(input) error = %v", err) + } + m := manifest.New(cfg.Session.SessionID, time.Now().UTC()) + m.Campaign = cfg.Session.Campaign + m.MarkStageSucceeded("trim", time.Now().UTC(), []manifest.ArtifactRecord{{ + Kind: artifactmodel.TranscriptOutputKindFinalTrimmed, SourceID: artifactmodel.SourceTranscriptFinalTrimmed, + LocalPath: inputPath, + }}) + if err := (&manifest.LocalStore{}).Save(context.Background(), paths.ManifestPath, m); err != nil { + t.Fatalf("Save(seed) error = %v", err) + } + runner := &materializingNotariusRunner{cfg: cfg.Pipeline.Notarius} + return cfg, &stage.Env{Notarius: runner}, runner +} diff --git a/internal/app/operator_helpers_test.go b/internal/app/operator_helpers_test.go index fad625e..a78a2d3 100644 --- a/internal/app/operator_helpers_test.go +++ b/internal/app/operator_helpers_test.go @@ -1006,7 +1006,7 @@ func TestExecutePublishLoadsRemoteLocks(t *testing.T) { restoreAppConfigTestGlobals(t, fake, &storeInitCalls, []string{sessionPath}) workRoot := filepath.Join(workspaceRoot, "work", "sample-campaign", "2026-05-03") - for _, stageName := range []string{"prepare", "transcribe", "merge", "polish", "normalize", "trim", "render", "analyze"} { + for _, stageName := range []string{"prepare", "transcribe", "merge", "polish", "normalize", "trim", "extract", "render", "analyze"} { // The publish stage only checks the manifest statuses and source files. _ = stageName } @@ -1076,7 +1076,7 @@ func writeValidPublishRunConfigFiles(t *testing.T, workspaceRoot string) (string m := manifest.New("2026-05-03", nowUTC()) m.Campaign = "sample-campaign" m.RunID = "20260521T160000Z-test" - for _, name := range []string{"prepare", "transcribe", "merge", "polish", "normalize", "trim", "render", "analyze"} { + for _, name := range []string{"prepare", "transcribe", "merge", "polish", "normalize", "trim", "extract", "render", "analyze"} { m.MarkStageSucceeded(name, nowUTC(), nil) } path := artifacts.SessionManifestPathForCampaign(cfg.Pipeline.Workspace.Root, cfg.Session.Campaign, cfg.Session.SessionID) diff --git a/internal/app/plan.go b/internal/app/plan.go index 6d1c4ba..8b734a1 100644 --- a/internal/app/plan.go +++ b/internal/app/plan.go @@ -22,7 +22,7 @@ func Plan(ctx context.Context, args []string, out io.Writer) error { var flags commonConfigFlags var force bool addCommonConfigFlags(fs, &flags) - fs.BoolVar(&force, "force", false, "force stage execution (reserved for future behavior)") + fs.BoolVar(&force, "force", false, "show all stages as scheduled to rerun") if err := parseSessionAwareFlags("plan", fs, args, &flags.sessionID); err != nil { return err diff --git a/internal/app/plan_test.go b/internal/app/plan_test.go index 60e5db9..6a694db 100644 --- a/internal/app/plan_test.go +++ b/internal/app/plan_test.go @@ -27,12 +27,12 @@ func TestPlanCreatesAndReusesWorkdir(t *testing.T) { if !strings.Contains(got, "narratio session plan: workdir prepared at") { t.Fatalf("first output = %q, want workdir prepared", got) } - for _, name := range []string{"prepare", "transcribe", "merge", "polish", "normalize", "trim", "render", "analyze", "publish", "notify"} { + for _, name := range []string{"prepare", "transcribe", "merge", "polish", "normalize", "trim", "extract", "render", "analyze", "publish", "notify"} { if !strings.Contains(got, name+": run") { t.Fatalf("first output = %q, missing stage %q", got, name) } } - if !strings.Contains(got, "totals: run=10 skip=0") { + if !strings.Contains(got, "totals: run=11 skip=0") { t.Fatalf("first output = %q, want totals", got) } @@ -84,8 +84,8 @@ func TestPlanShowsRunAndSkipFromManifest(t *testing.T) { if !strings.Contains(got, "trim: run") { t.Fatalf("output = %q, want trim run", got) } - if !strings.Contains(got, "totals: run=8 skip=2") { - t.Fatalf("output = %q, want totals run=8 skip=2", got) + if !strings.Contains(got, "totals: run=9 skip=2") { + t.Fatalf("output = %q, want totals run=9 skip=2", got) } } diff --git a/internal/app/planner_test.go b/internal/app/planner_test.go index a68b4da..a76f6f4 100644 --- a/internal/app/planner_test.go +++ b/internal/app/planner_test.go @@ -4,7 +4,7 @@ import "testing" func TestBuildFullPlanOrder(t *testing.T) { got := BuildFullPlan() - want := []string{"prepare", "transcribe", "merge", "polish", "normalize", "trim", "render", "analyze", "publish", "notify"} + want := []string{"prepare", "transcribe", "merge", "polish", "normalize", "trim", "extract", "render", "analyze", "publish", "notify"} if len(got) != len(want) { t.Fatalf("len(plan) = %d, want %d", len(got), len(want)) } diff --git a/internal/app/post_publish_cleanup_test.go b/internal/app/post_publish_cleanup_test.go index 9f54e7a..b64e83a 100644 --- a/internal/app/post_publish_cleanup_test.go +++ b/internal/app/post_publish_cleanup_test.go @@ -355,7 +355,7 @@ func publishStageCleanupFixture(t *testing.T) (*config.Config, cleanupSeed, stri if err != nil { t.Fatalf("Load() error = %v", err) } - for _, name := range []string{"prepare", "transcribe", "merge", "polish", "normalize", "trim", "render", "analyze"} { + for _, name := range []string{"prepare", "transcribe", "merge", "polish", "normalize", "trim", "extract", "render", "analyze"} { seedManifest.MarkStageSucceeded(name, time.Now().UTC(), nil) } seedManifest.S3SessionPrefix = artifacts.S3SessionPrefix("dnd", cfg.Session.Campaign, cfg.Session.SessionID) diff --git a/internal/app/run.go b/internal/app/run.go index 53fa3b6..f4f0274 100644 --- a/internal/app/run.go +++ b/internal/app/run.go @@ -18,7 +18,7 @@ func Run(ctx context.Context, args []string, out io.Writer) error { var force bool var selectedArtifacts artifactSelectionFlag addCommonConfigFlags(fs, &flags) - fs.BoolVar(&force, "force", false, "force stage execution (reserved for future behavior)") + fs.BoolVar(&force, "force", false, "rerun stages even when already succeeded") fs.Var(&selectedArtifacts, "artifacts", "configured artifact names to execute and publish (comma-separated or repeatable)") if err := parseSessionAwareFlags("run", fs, args, &flags.sessionID); err != nil { diff --git a/internal/app/run_control.go b/internal/app/run_control.go index abeae11..8dbfcff 100644 --- a/internal/app/run_control.go +++ b/internal/app/run_control.go @@ -92,6 +92,10 @@ func downstreamStageNames(stageName string) []string { } func invalidateDownstreamSucceededStages(m *manifest.Manifest, upstreamStage string, at time.Time) []string { + return invalidateDownstreamSucceededStagesWithReason(m, upstreamStage, at, "upstream stage rerun with force") +} + +func invalidateDownstreamSucceededStagesWithReason(m *manifest.Manifest, upstreamStage string, at time.Time, reason string) []string { if m == nil || m.Stages == nil { return nil } @@ -102,7 +106,7 @@ func invalidateDownstreamSucceededStages(m *manifest.Manifest, upstreamStage str if sr == nil || sr.Status != manifest.StatusSucceeded { continue } - m.MarkStageStale(downstream, at, "upstream stage rerun with force") + m.MarkStageStale(downstream, at, reason) invalidated = append(invalidated, downstream) } return invalidated diff --git a/internal/app/run_control_test.go b/internal/app/run_control_test.go index 2927e55..31d536e 100644 --- a/internal/app/run_control_test.go +++ b/internal/app/run_control_test.go @@ -32,7 +32,7 @@ func TestDecideStageActions(t *testing.T) { func TestDownstreamStageNames(t *testing.T) { got := downstreamStageNames("polish") - want := []string{"normalize", "trim", "render", "analyze", "publish", "notify"} + want := []string{"normalize", "trim", "extract", "render", "analyze", "publish", "notify"} if !reflect.DeepEqual(got, want) { t.Fatalf("downstreamStageNames(polish) = %#v, want %#v", got, want) } @@ -52,13 +52,14 @@ func TestInvalidateDownstreamSucceededStages(t *testing.T) { m.MarkStageSucceeded("polish", now, nil) m.MarkStageSucceeded("normalize", now, nil) m.MarkStageSucceeded("trim", now, nil) + m.MarkStageSucceeded("extract", now, nil) m.MarkStageSucceeded("render", now, nil) m.MarkStageFailed("analyze", now, "analysis failed") m.MarkStageSucceeded("publish", now, nil) m.MarkStageSucceeded("notify", now, nil) got := invalidateDownstreamSucceededStages(m, "polish", now.Add(1*time.Second)) - want := []string{"normalize", "trim", "render", "publish", "notify"} + want := []string{"normalize", "trim", "extract", "render", "publish", "notify"} if !reflect.DeepEqual(got, want) { t.Fatalf("invalidateDownstreamSucceededStages() = %#v, want %#v", got, want) } @@ -75,3 +76,30 @@ func TestInvalidateDownstreamSucceededStages(t *testing.T) { t.Fatalf("prepare status = %q, want succeeded", m.Stages["prepare"].Status) } } + +func TestExtractionPositionControlsForceInvalidation(t *testing.T) { + now := time.Now().UTC() + tests := []struct { + upstream string + want []string + }{ + {upstream: "trim", want: []string{"extract", "render", "analyze", "publish", "notify"}}, + {upstream: "extract", want: []string{"render", "analyze", "publish", "notify"}}, + {upstream: "render", want: []string{"analyze", "publish", "notify"}}, + } + for _, test := range tests { + t.Run(test.upstream, func(t *testing.T) { + m := manifest.New("2026-05-03", now) + for _, name := range canonicalStageNames() { + m.MarkStageSucceeded(name, now, nil) + } + got := invalidateDownstreamSucceededStages(m, test.upstream, now.Add(time.Second)) + if !reflect.DeepEqual(got, test.want) { + t.Fatalf("invalidated = %#v, want %#v", got, test.want) + } + if test.upstream == "render" && m.Stages["extract"].Status != manifest.StatusSucceeded { + t.Fatalf("forcing render changed extract: %#v", m.Stages["extract"]) + } + }) + } +} diff --git a/internal/app/run_stage.go b/internal/app/run_stage.go index 05d5219..a92b55f 100644 --- a/internal/app/run_stage.go +++ b/internal/app/run_stage.go @@ -27,7 +27,7 @@ func RunStage(ctx context.Context, args []string, out io.Writer) error { var force bool var selectedArtifacts artifactSelectionFlag addCommonConfigFlags(fs, &flags) - fs.BoolVar(&force, "force", false, "force stage execution (reserved for future behavior)") + fs.BoolVar(&force, "force", false, "rerun the stage even when already succeeded") fs.Var(&selectedArtifacts, "artifacts", "configured artifact names to execute or publish (comma-separated or repeatable)") if err := fs.Parse(args); err != nil { diff --git a/internal/app/run_stage_test.go b/internal/app/run_stage_test.go index e2c36b3..7bef0e6 100644 --- a/internal/app/run_stage_test.go +++ b/internal/app/run_stage_test.go @@ -36,8 +36,8 @@ func TestRunContinuesAfterCompletedStages(t *testing.T) { if err != nil { t.Fatalf("Run() error = %v", err) } - if !strings.Contains(out.String(), "executed=8 skipped=2") { - t.Fatalf("output = %q, want executed=8 skipped=2", out.String()) + if !strings.Contains(out.String(), "executed=9 skipped=3") { + t.Fatalf("output = %q, want executed=9 skipped=3", out.String()) } loaded, err := store.Load(context.Background(), manifestPath) @@ -68,8 +68,15 @@ func TestRunNoRemainingStagesRecordsSkippedStages(t *testing.T) { if err != nil { t.Fatalf("Run() error = %v", err) } - if !strings.Contains(out.String(), "executed=0 skipped=10") { - t.Fatalf("output = %q, want executed=0 skipped=10", out.String()) + if !strings.Contains(out.String(), "executed=1 skipped=11") { + t.Fatalf("output = %q, want disabled extraction to self-skip", out.String()) + } + loaded, err := store.Load(context.Background(), manifestPath) + if err != nil { + t.Fatalf("load migrated manifest: %v", err) + } + if loaded.Stages["extract"] == nil || loaded.Stages["extract"].Status != manifest.StatusSkipped { + t.Fatalf("legacy manifest extract record = %#v, want newly executed disabled skip", loaded.Stages["extract"]) } } @@ -85,7 +92,7 @@ func TestRunForceRerunsSucceeded(t *testing.T) { store := &manifest.LocalStore{} m := manifest.New("2026-05-03", time.Date(2026, 5, 3, 10, 0, 0, 0, time.UTC)) - for _, name := range []string{"prepare", "transcribe", "merge", "polish", "normalize", "trim", "render", "analyze", "publish", "notify"} { + for _, name := range []string{"prepare", "transcribe", "merge", "polish", "normalize", "trim", "extract", "render", "analyze", "publish", "notify"} { m.MarkStageSucceeded(name, time.Date(2026, 5, 3, 10, 1, 0, 0, time.UTC), nil) } if err := store.Save(context.Background(), manifestPath, m); err != nil { @@ -97,7 +104,7 @@ func TestRunForceRerunsSucceeded(t *testing.T) { if err != nil { t.Fatalf("Run() error = %v", err) } - if !strings.Contains(out.String(), "executed=10 skipped=0") { + if !strings.Contains(out.String(), "executed=11 skipped=1") { t.Fatalf("output = %q, want forced full rerun", out.String()) } } @@ -132,6 +139,27 @@ func TestRunStageExecutesOnlySelectedStage(t *testing.T) { } } +func TestRunStageExtractIsAcceptedAndSelfSkipsWhenDisabled(t *testing.T) { + workspaceRoot := t.TempDir() + pipelinePath, campaignPath, sessionPath := writeValidConfigFiles(t, workspaceRoot) + + var out bytes.Buffer + err := RunStage(context.Background(), []string{"extract", "2026-05-03", "--config", pipelinePath, "--campaign-file", campaignPath, "--session", sessionPath}, &out) + if err != nil { + t.Fatalf("RunStage(extract) error = %v", err) + } + if !strings.Contains(out.String(), "stage=extract executed=1 skipped=1 force=false") { + t.Fatalf("output = %q, want disabled extraction self-skip", out.String()) + } + loaded, err := (&manifest.LocalStore{}).Load(context.Background(), filepath.Join(workspaceRoot, "work", "sample-campaign", "2026-05-03", "manifest.json")) + if err != nil { + t.Fatalf("Load() error = %v", err) + } + if loaded.Stages["extract"] == nil || loaded.Stages["extract"].Status != manifest.StatusSkipped { + t.Fatalf("extract record = %#v, want skipped", loaded.Stages["extract"]) + } +} + func TestRunStageSkipAndForce(t *testing.T) { workspaceRoot := t.TempDir() pipelinePath, campaignPath, sessionPath := writeValidConfigFiles(t, workspaceRoot) @@ -176,7 +204,7 @@ func TestRunStageForceMarksDownstreamStaleAndRunContinuesFromStale(t *testing.T) store := &manifest.LocalStore{} seed := manifest.New("2026-05-03", time.Date(2026, 5, 3, 10, 0, 0, 0, time.UTC)) - for _, name := range []string{"prepare", "transcribe", "merge", "polish", "normalize", "trim", "render", "analyze", "publish", "notify"} { + 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) } if err := store.Save(context.Background(), manifestPath, seed); err != nil { @@ -196,7 +224,7 @@ func TestRunStageForceMarksDownstreamStaleAndRunContinuesFromStale(t *testing.T) if err != nil { t.Fatalf("load manifest after force: %v", err) } - for _, name := range []string{"normalize", "trim", "render", "analyze", "publish", "notify"} { + for _, name := range []string{"normalize", "trim", "extract", "render", "analyze", "publish", "notify"} { if afterForce.Stages[name] == nil || afterForce.Stages[name].Status != manifest.StatusStale { t.Fatalf("stage %q = %#v, want stale", name, afterForce.Stages[name]) } @@ -207,7 +235,7 @@ func TestRunStageForceMarksDownstreamStaleAndRunContinuesFromStale(t *testing.T) if err != nil { t.Fatalf("Run() error = %v", err) } - if !strings.Contains(out.String(), "executed=6 skipped=4") { + if !strings.Contains(out.String(), "executed=7 skipped=5") { t.Fatalf("output = %q, want run to execute stale downstream stages", out.String()) } } diff --git a/internal/app/runner.go b/internal/app/runner.go index 5f20c25..e701a2b 100644 --- a/internal/app/runner.go +++ b/internal/app/runner.go @@ -171,6 +171,31 @@ func executeStages(ctx context.Context, cfg *config.Config, stages []stage.Stage for _, d := range decisions { s := d.Stage runNames = append(runNames, s.Name()) + if d.Action == stageActionSkip && !stageSucceeded(m, s.Name()) { + d.Action = stageActionRun + } + + if d.Action == stageActionSkip { + if validator, ok := s.(stage.ResumeValidator); ok { + validation, err := validator.ValidateResume(ctx, stageEnv, m) + if err != nil { + return nil, fmt.Errorf("validate resume for stage %q: %w", s.Name(), err) + } + validation = validation.Normalized() + if !validation.Resumable { + staleAt := nowUTC() + m.MarkStageStale(s.Name(), staleAt, validation.Reason) + invalidateDownstreamSucceededStagesWithReason( + m, s.Name(), staleAt, "upstream stage result was not resumable", + ) + if err := env.ManifestStore.Save(ctx, manifestPath, m); err != nil { + return nil, fmt.Errorf("save manifest after resume validation for stage %q: %w", s.Name(), err) + } + env.Logger.Info("stage result is not resumable", "stage", s.Name(), "reason", validation.Reason) + d.Action = stageActionRun + } + } + } if d.Action == stageActionSkip { skipped = append(skipped, s.Name()) diff --git a/internal/app/runner_test.go b/internal/app/runner_test.go index 5425cc1..9c12cfe 100644 --- a/internal/app/runner_test.go +++ b/internal/app/runner_test.go @@ -2,6 +2,7 @@ package app import ( "context" + "encoding/json" "errors" "fmt" "os" @@ -74,6 +75,23 @@ type captureNotariusStage struct { captured *bool } +type resumeCheckingStage struct { + name string + validation stage.ResumeValidation + validateErr error + runs *int +} + +func (s resumeCheckingStage) Name() string { return s.name } +func (s resumeCheckingStage) Declares() stage.IODecl { return stage.IODecl{} } +func (s resumeCheckingStage) Run(_ context.Context, _ *stage.Env, _ *manifest.Manifest) (*stage.StageResult, error) { + *s.runs++ + return &stage.StageResult{}, nil +} +func (s resumeCheckingStage) ValidateResume(_ context.Context, _ *stage.Env, _ *manifest.Manifest) (stage.ResumeValidation, error) { + return s.validation, s.validateErr +} + func (s captureNotariusStage) Name() string { return "extract" } func (s captureNotariusStage) Declares() stage.IODecl { return stage.IODecl{} } func (s captureNotariusStage) Run(_ context.Context, env *stage.Env, _ *manifest.Manifest) (*stage.StageResult, error) { @@ -381,7 +399,7 @@ func TestExecuteStagesPublishSkipsRequiredUnselectedConfiguredOutput(t *testing. manifestPath := manifestPathFor(cfg) seed := manifest.New(cfg.Session.SessionID, time.Now().UTC()) seed.Campaign = cfg.Session.Campaign - for _, stageName := range []string{"prepare", "transcribe", "merge", "polish", "normalize", "trim", "render"} { + for _, stageName := range []string{"prepare", "transcribe", "merge", "polish", "normalize", "trim", "extract", "render"} { seed.MarkStageSucceeded(stageName, time.Now().UTC(), nil) } if err := os.MkdirAll(filepath.Dir(manifestPath), 0o755); err != nil { @@ -440,8 +458,8 @@ func TestExecuteStagesPlaceholderSuccessUpdatesManifest(t *testing.T) { if err != nil { t.Fatalf("executeStages() error = %v", err) } - if len(summary.StageNames) != 10 || len(summary.Executed) != 10 || len(summary.Skipped) != 0 { - t.Fatalf("summary = %#v, want all 10 executed", summary) + if len(summary.StageNames) != 11 || len(summary.Executed) != 11 || len(summary.Skipped) != 1 || summary.Skipped[0] != "extract" { + t.Fatalf("summary = %#v, want full plan with disabled extraction self-skip", summary) } store := &manifest.LocalStore{} @@ -450,11 +468,17 @@ func TestExecuteStagesPlaceholderSuccessUpdatesManifest(t *testing.T) { t.Fatalf("Load manifest error = %v", err) } - for _, name := range []string{"prepare", "transcribe", "merge", "polish", "normalize", "trim", "render", "analyze", "publish", "notify"} { + for _, name := range []string{"prepare", "transcribe", "merge", "polish", "normalize", "trim", "extract", "render", "analyze", "publish", "notify"} { sr := m.Stages[name] if sr == nil { t.Fatalf("missing stage record %q", name) } + if name == "extract" { + if sr.Status != manifest.StatusSkipped || sr.Error == nil || sr.Error.Message != "notarius_disabled" { + t.Fatalf("extract stage = %#v, want disabled skip", sr) + } + continue + } if sr.Status != manifest.StatusSucceeded { t.Fatalf("stage %q status = %q, want %q", name, sr.Status, manifest.StatusSucceeded) } @@ -605,6 +629,91 @@ func TestExecuteStagesSkipSucceededWhenNotForced(t *testing.T) { } } +func TestExecuteStagesUsesOptionalResumeValidation(t *testing.T) { + for _, test := range []struct { + name string + validation stage.ResumeValidation + wantRuns int + wantSkipped int + }{ + {name: "resumable", validation: stage.Resumable(), wantSkipped: 1}, + {name: "rerun", validation: stage.NonResumable("durable output changed"), wantRuns: 1}, + } { + t.Run(test.name, func(t *testing.T) { + cfg := testConfig(t) + store := &manifest.LocalStore{} + seed := manifest.New(cfg.Session.SessionID, time.Now().UTC()) + seed.MarkStageSucceeded("checked", time.Now().UTC(), nil) + if err := store.Save(context.Background(), manifestPathFor(cfg), seed); err != nil { + t.Fatalf("Save() error = %v", err) + } + runs := 0 + candidate := resumeCheckingStage{name: "checked", validation: test.validation, runs: &runs} + summary, err := executeStages(context.Background(), cfg, []stage.Stage{candidate}, RunOptions{}) + if err != nil { + t.Fatalf("executeStages() error = %v", err) + } + if runs != test.wantRuns || len(summary.Skipped) != test.wantSkipped { + t.Fatalf("runs = %d summary = %#v", runs, summary) + } + }) + } +} + +func TestExecuteStagesNonResumableResultRerunsSucceededDownstream(t *testing.T) { + cfg := testConfig(t) + store := &manifest.LocalStore{} + seed := manifest.New(cfg.Session.SessionID, time.Now().UTC()) + seed.MarkStageSucceeded("extract", time.Now().UTC(), nil) + seed.MarkStageSucceeded("render", time.Now().UTC(), nil) + if err := store.Save(context.Background(), manifestPathFor(cfg), seed); err != nil { + t.Fatalf("Save() error = %v", err) + } + extractRuns, renderRuns := 0, 0 + stages := []stage.Stage{ + resumeCheckingStage{name: "extract", validation: stage.NonResumable("checksum changed"), runs: &extractRuns}, + countingStage{name: "render", runs: &renderRuns}, + } + summary, err := executeStages(context.Background(), cfg, stages, RunOptions{}) + if err != nil { + t.Fatalf("executeStages() error = %v", err) + } + if extractRuns != 1 || renderRuns != 1 || len(summary.Executed) != 2 || len(summary.Skipped) != 0 { + t.Fatalf("extract runs=%d render runs=%d summary=%#v", extractRuns, renderRuns, summary) + } +} + +func TestExecuteStagesResumeValidationErrorPreservesSucceededRecord(t *testing.T) { + cfg := testConfig(t) + store := &manifest.LocalStore{} + seed := manifest.New(cfg.Session.SessionID, time.Now().UTC()) + seed.MarkStageSucceeded("checked", time.Now().UTC(), []manifest.ArtifactRecord{{Kind: "kept", LocalPath: "kept.json"}}) + seed.Stages["checked"].Metadata = map[string]any{"kept": true} + if err := store.Save(context.Background(), manifestPathFor(cfg), seed); err != nil { + t.Fatalf("Save() error = %v", err) + } + before, err := json.Marshal(seed.Stages["checked"]) + if err != nil { + t.Fatalf("Marshal(before) error = %v", err) + } + runs := 0 + candidate := resumeCheckingStage{name: "checked", validateErr: errors.New("inspection unavailable"), runs: &runs} + if _, err := executeStages(context.Background(), cfg, []stage.Stage{candidate}, RunOptions{}); err == nil || !strings.Contains(err.Error(), "inspection unavailable") { + t.Fatalf("executeStages() error = %v", err) + } + loaded, err := store.Load(context.Background(), manifestPathFor(cfg)) + if err != nil { + t.Fatalf("Load() error = %v", err) + } + after, err := json.Marshal(loaded.Stages["checked"]) + if err != nil { + t.Fatalf("Marshal(after) error = %v", err) + } + if string(after) != string(before) || runs != 0 { + t.Fatalf("succeeded record changed: before=%s after=%s runs=%d", before, after, runs) + } +} + func TestExecuteStagesForceRerunsSucceeded(t *testing.T) { cfg := testConfig(t) manifestPath := manifestPathFor(cfg) @@ -639,7 +748,7 @@ func TestExecuteStagesForceSuccessInvalidatesDownstreamSucceededStages(t *testin store := &manifest.LocalStore{} existing := manifest.New(cfg.Session.SessionID, time.Date(2026, 5, 3, 1, 0, 0, 0, time.UTC)) - for _, stageName := range []string{"prepare", "transcribe", "merge", "polish", "normalize", "trim", "render", "publish", "notify"} { + for _, stageName := range []string{"prepare", "transcribe", "merge", "polish", "normalize", "trim", "extract", "render", "publish", "notify"} { existing.MarkStageSucceeded(stageName, time.Date(2026, 5, 3, 1, 1, 0, 0, time.UTC), nil) } existing.MarkStageFailed("analyze", time.Date(2026, 5, 3, 1, 1, 0, 0, time.UTC), "previous analyze failure") @@ -670,7 +779,7 @@ func TestExecuteStagesForceSuccessInvalidatesDownstreamSucceededStages(t *testin if loaded.Stages["polish"] == nil || loaded.Stages["polish"].Status != manifest.StatusSucceeded { t.Fatalf("polish status = %#v, want succeeded", loaded.Stages["polish"]) } - for _, stageName := range []string{"normalize", "trim", "render", "publish", "notify"} { + for _, stageName := range []string{"normalize", "trim", "extract", "render", "publish", "notify"} { if loaded.Stages[stageName] == nil || loaded.Stages[stageName].Status != manifest.StatusStale { t.Fatalf("%s status = %#v, want stale", stageName, loaded.Stages[stageName]) } @@ -1214,7 +1323,7 @@ func TestAdapterBackedStageFailureMarksManifestFailed(t *testing.T) { seed.S3Bucket = "my-dnd-archive" seed.S3SessionPrefix = "dnd/campaigns/" + cfg.Session.Campaign + "/sessions/" + cfg.Session.SessionID + "/" seed.S3RunPrefix = seed.S3SessionPrefix + "runs/" + runID + "/" - for _, name := range []string{"prepare", "transcribe", "merge", "polish", "normalize", "trim", "render", "analyze"} { + for _, name := range []string{"prepare", "transcribe", "merge", "polish", "normalize", "trim", "extract", "render", "analyze"} { seed.MarkStageSucceeded(name, time.Now().UTC(), nil) } if err := tc.env.ManifestStore.Save(context.Background(), manifestPathFor(cfg), seed); err != nil { diff --git a/internal/stage/extract_resume.go b/internal/stage/extract_resume.go new file mode 100644 index 0000000..e1e0500 --- /dev/null +++ b/internal/stage/extract_resume.go @@ -0,0 +1,250 @@ +package stage + +import ( + "context" + "fmt" + "os" + "path/filepath" + "strings" + "time" + + "gitea.maximumdirect.net/eric/narratio/internal/artifactmodel" + "gitea.maximumdirect.net/eric/narratio/internal/artifacts" + "gitea.maximumdirect.net/eric/narratio/internal/config" + "gitea.maximumdirect.net/eric/narratio/internal/manifest" + "gitea.maximumdirect.net/eric/narratio/internal/pathsafe" +) + +func (extractStage) ValidateResume(_ context.Context, env *Env, m *manifest.Manifest) (ResumeValidation, error) { + if env == nil || env.Config == nil || env.Config.Pipeline == nil || env.Config.Session == nil { + return ResumeValidation{}, fmt.Errorf("extract resume: resolved stage environment config is required") + } + cfg := env.Config.Pipeline.Notarius + if cfg == nil || !cfg.Enabled { + return NonResumable("Notarius extraction is disabled"), nil + } + if m == nil { + return ResumeValidation{}, fmt.Errorf("extract resume: session manifest is required") + } + record := m.Stages[(extractStage{}).Name()] + if record == nil || record.Status != manifest.StatusSucceeded { + return NonResumable("extract stage has no succeeded result"), nil + } + if record.Name != (extractStage{}).Name() { + return NonResumable("extract stage record identity is inconsistent"), nil + } + + producerRunID := metadataString(record.Metadata, "narratio_run_id") + if producerRunID == "" { + return NonResumable("extract result is missing its producing run ID"), nil + } + if !safePathSegment(producerRunID) { + return NonResumable("extract result has an invalid producing run ID"), nil + } + + timeout, err := time.ParseDuration(strings.TrimSpace(cfg.Timeout)) + if err != nil || timeout <= 0 { + return ResumeValidation{}, fmt.Errorf("extract resume: invalid Notarius timeout %q", cfg.Timeout) + } + resolvedBinary, err := resolveExecutable(cfg.Binary) + if err != nil { + return ResumeValidation{}, fmt.Errorf("extract resume: resolve Notarius binary: %w", err) + } + configPath, err := absolutePath(cfg.ConfigPath) + if err != nil { + return ResumeValidation{}, fmt.Errorf("extract resume: resolve Notarius config path: %w", err) + } + workingDirectory, err := absolutePath(cfg.WorkingDirectory) + if err != nil { + return ResumeValidation{}, fmt.Errorf("extract resume: resolve Notarius working directory: %w", err) + } + fingerprint, err := extractionFingerprint(resolvedBinary, configPath, cfg, timeout, workingDirectory) + if err != nil { + return ResumeValidation{}, fmt.Errorf("extract resume: build configuration fingerprint: %w", err) + } + if metadataString(record.Metadata, "configuration_fingerprint") != fingerprint { + return NonResumable("Notarius invocation contract changed"), nil + } + + sessionID := strings.TrimSpace(m.SessionID) + if sessionID == "" { + sessionID = strings.TrimSpace(env.Config.Session.SessionID) + } + campaign := strings.TrimSpace(m.Campaign) + if campaign == "" { + campaign = strings.TrimSpace(env.Config.Session.Campaign) + } + if sessionID == "" || campaign == "" { + return ResumeValidation{}, fmt.Errorf("extract resume: session ID and campaign are required") + } + paths := sessionPathsForEnv(env, sessionID) + workspaceRoot := strings.TrimSpace(paths.WorkspaceRoot) + if workspaceRoot == "" { + workspaceRoot = env.Config.Pipeline.Workspace.Root + } + bundleRoot, err := absolutePath(artifacts.SessionNotariusBundleDirForCampaign(workspaceRoot, campaign, sessionID, producerRunID)) + if err != nil { + return ResumeValidation{}, fmt.Errorf("extract resume: resolve durable bundle path: %w", err) + } + storedBundleRoot := metadataString(record.Metadata, "bundle_root") + if storedBundleRoot == "" || !filepath.IsAbs(storedBundleRoot) || filepath.Clean(storedBundleRoot) != bundleRoot { + return NonResumable("extract result does not identify its canonical immutable bundle"), nil + } + bundleRelative, err := pathsafe.SlashRelativeFromRoot(paths.Root, bundleRoot) + if err != nil { + return ResumeValidation{}, fmt.Errorf("extract resume: immutable bundle path is unsafe: %w", err) + } + if err := rejectSymlinkComponents(paths.Root, bundleRelative); err != nil { + return ResumeValidation{}, err + } + bundleInfo, err := os.Lstat(bundleRoot) + if os.IsNotExist(err) { + return NonResumable("immutable Notarius bundle is missing"), nil + } + if err != nil { + return ResumeValidation{}, fmt.Errorf("extract resume: inspect immutable bundle: %w", err) + } + if bundleInfo.Mode()&os.ModeSymlink != 0 { + return ResumeValidation{}, fmt.Errorf("extract resume: immutable bundle must not be a symlink") + } + if !bundleInfo.IsDir() { + return NonResumable("immutable Notarius bundle is not a directory"), nil + } + + receiptRunID, receiptPipelineID := receiptIdentity(record.Metadata) + if receiptRunID == "" || receiptPipelineID != cfg.PipelineID { + return NonResumable("extract result has incompatible Notarius receipt identity"), nil + } + + expectedSources := make(map[string]config.NotariusOutputConfig, len(cfg.Outputs)) + for key, output := range cfg.Outputs { + expectedSources[artifacts.ExtractionArtifactSourceID(key)] = output + } + seenSources := make(map[string]struct{}, len(expectedSources)) + indexSeen := false + for _, output := range record.Outputs { + if output.ProducerRunID != producerRunID { + return NonResumable("extract output producer identity is inconsistent"), nil + } + if output.SourceID == "" { + if indexSeen || output.Kind != extractIndexOutputKind { + return NonResumable("extract result has an unexpected non-selectable output"), nil + } + indexSeen = true + expectedIndex := filepath.Join(bundleRoot, "index.json") + if filepath.Clean(output.LocalPath) != expectedIndex { + return NonResumable("extract index path is not canonical"), nil + } + validation, err := validateResumePayload(bundleRoot, output.LocalPath, output.Checksum) + if err != nil || !validation.Resumable { + return validation, err + } + continue + } + + expected, ok := expectedSources[output.SourceID] + if !ok || output.Kind != extractLaneOutputKind { + return NonResumable("extract result source set differs from current configuration"), nil + } + if _, duplicate := seenSources[output.SourceID]; duplicate { + return NonResumable("extract result contains a duplicate configured source"), nil + } + seenSources[output.SourceID] = struct{}{} + if !compatibleExtractionContract(output.Contract, expected) { + return NonResumable("extract output contract is incompatible with current configuration"), nil + } + if !compatibleExtractionProvenance(output.ExternalProvenance, receiptRunID, receiptPipelineID, expected.LaneID) { + return NonResumable("extract output has incompatible Notarius provenance"), nil + } + validation, err := validateResumePayload(bundleRoot, output.LocalPath, output.Checksum) + if err != nil || !validation.Resumable { + return validation, err + } + } + if !indexSeen { + return NonResumable("extract result is missing its canonical index"), nil + } + if len(seenSources) != len(expectedSources) || len(record.Outputs) != len(expectedSources)+1 { + return NonResumable("extract result is missing configured sources"), nil + } + return Resumable(), nil +} + +func validateResumePayload(bundleRoot, path, checksum string) (ResumeValidation, error) { + if !filepath.IsAbs(path) { + return ResumeValidation{}, fmt.Errorf("extract resume: persisted output path must be absolute") + } + relative, err := pathsafe.SlashRelativeFromRoot(bundleRoot, path) + if err != nil { + return ResumeValidation{}, fmt.Errorf("extract resume: persisted output path is unsafe: %w", err) + } + if err := rejectSymlinkComponents(bundleRoot, relative); err != nil { + return ResumeValidation{}, err + } + info, err := os.Lstat(path) + if os.IsNotExist(err) { + return NonResumable("extract output is missing"), nil + } + if err != nil { + return ResumeValidation{}, fmt.Errorf("extract resume: inspect output %q: %w", path, err) + } + if info.Mode()&os.ModeSymlink != 0 || !info.Mode().IsRegular() { + return ResumeValidation{}, fmt.Errorf("extract resume: output %q must be a regular file without symlinks", path) + } + if strings.TrimSpace(checksum) == "" { + return NonResumable("extract output is missing its checksum"), nil + } + actual, err := artifacts.SHA256File(path) + if err != nil { + return ResumeValidation{}, fmt.Errorf("extract resume: checksum output %q: %w", path, err) + } + if actual != checksum { + return NonResumable("extract output checksum does not match durable bytes"), nil + } + return Resumable(), nil +} + +func rejectSymlinkComponents(root, slashRelative string) error { + current := filepath.Clean(root) + parts := strings.Split(filepath.FromSlash(slashRelative), string(filepath.Separator)) + for _, part := range parts[:len(parts)-1] { + current = filepath.Join(current, part) + info, err := os.Lstat(current) + if os.IsNotExist(err) { + return nil + } + if err != nil { + return fmt.Errorf("extract resume: inspect output directory %q: %w", current, err) + } + if info.Mode()&os.ModeSymlink != 0 { + return fmt.Errorf("extract resume: output directory %q must not be a symlink", current) + } + } + return nil +} + +func compatibleExtractionContract(got *artifactmodel.ContractMetadata, want config.NotariusOutputConfig) bool { + return got != nil && got.MediaType == want.MediaType && got.SchemaID == want.SchemaID && + got.SchemaVersion == want.SchemaVersion && (want.ModuleKey == "" || got.ModuleKey == want.ModuleKey) +} + +func compatibleExtractionProvenance(got *artifactmodel.ExternalProvenance, runID, pipelineID, laneID string) bool { + return got != nil && got.System == "notarius" && got.RunID == runID && + got.PipelineID == pipelineID && got.ArtifactID == laneID +} + +func metadataString(metadata map[string]any, key string) string { + if metadata == nil { + return "" + } + value, _ := metadata[key].(string) + return strings.TrimSpace(value) +} + +func receiptIdentity(metadata map[string]any) (string, string) { + if metadata == nil { + return "", "" + } + receipt, _ := metadata["receipt"].(map[string]any) + return metadataString(receipt, "run_id"), metadataString(receipt, "pipeline_id") +} diff --git a/internal/stage/extract_test.go b/internal/stage/extract_test.go index 5e21dbc..7876679 100644 --- a/internal/stage/extract_test.go +++ b/internal/stage/extract_test.go @@ -47,14 +47,6 @@ func TestExtractStageDisabledReturnsExplicitSkip(t *testing.T) { } } -func TestExtractStageIsNotInCanonicalPlansYet(t *testing.T) { - for _, candidate := range All() { - if candidate.Name() == "extract" { - t.Fatal("extract must remain directly executable until lifecycle integration is implemented") - } - } -} - func TestExtractStageResolvesManifestInputAndBuildsExactRequest(t *testing.T) { env, m, fake := setupExtractEnv(t) result, err := (extractStage{}).Run(context.Background(), env, m) @@ -294,6 +286,110 @@ func TestExtractionFingerprintIsIndependentOfOutputMapOrder(t *testing.T) { } } +func TestExtractStageResumeValidationAcceptsCurrentImmutableResult(t *testing.T) { + env, m, _ := setupExtractEnv(t) + seedSucceededExtractResult(t, env, m) + m.RunID = "20260810T020304Z-fedcba98" + + validation, err := (extractStage{}).ValidateResume(context.Background(), env, m) + if err != nil { + t.Fatalf("ValidateResume() error = %v", err) + } + if !validation.Resumable || validation.Reason != "" { + t.Fatalf("validation = %#v, want resumable", validation) + } +} + +func TestExtractStageResumeValidationRejectsObsoleteResults(t *testing.T) { + tests := []struct { + name string + mutate func(*testing.T, *Env, *manifest.Manifest) + }{ + {name: "disabled", mutate: func(_ *testing.T, env *Env, _ *manifest.Manifest) { + env.Config.Pipeline.Notarius.Enabled = false + }}, + {name: "config changed", mutate: func(_ *testing.T, env *Env, _ *manifest.Manifest) { + env.Config.Pipeline.Notarius.PipelineID = "changed" + }}, + {name: "missing lane", mutate: func(t *testing.T, _ *Env, m *manifest.Manifest) { + if err := os.Remove(m.Stages["extract"].Outputs[0].LocalPath); err != nil { + t.Fatalf("Remove(lane) error = %v", err) + } + }}, + {name: "tampered lane", mutate: func(t *testing.T, _ *Env, m *manifest.Manifest) { + if err := os.WriteFile(m.Stages["extract"].Outputs[0].LocalPath, []byte(`{"tampered":true}`), 0o644); err != nil { + t.Fatalf("WriteFile(lane) error = %v", err) + } + }}, + {name: "incompatible contract", mutate: func(_ *testing.T, _ *Env, m *manifest.Manifest) { + m.Stages["extract"].Outputs[0].Contract.SchemaVersion = "v2" + }}, + {name: "missing source", mutate: func(_ *testing.T, _ *Env, m *manifest.Manifest) { + m.Stages["extract"].Outputs = m.Stages["extract"].Outputs[1:] + }}, + {name: "producer mismatch", mutate: func(_ *testing.T, _ *Env, m *manifest.Manifest) { + m.Stages["extract"].Outputs[0].ProducerRunID = "different-run" + }}, + {name: "provenance mismatch", mutate: func(_ *testing.T, _ *Env, m *manifest.Manifest) { + m.Stages["extract"].Outputs[0].ExternalProvenance.RunID = "different-run" + }}, + {name: "missing index", mutate: func(t *testing.T, _ *Env, m *manifest.Manifest) { + index := m.Stages["extract"].Outputs[1] + if err := os.Remove(index.LocalPath); err != nil { + t.Fatalf("Remove(index) error = %v", err) + } + }}, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + env, m, _ := setupExtractEnv(t) + seedSucceededExtractResult(t, env, m) + test.mutate(t, env, m) + + validation, err := (extractStage{}).ValidateResume(context.Background(), env, m) + if err != nil { + t.Fatalf("ValidateResume() error = %v", err) + } + if validation.Resumable || validation.Reason == "" || len(validation.Reason) > maxResumeReasonLength { + t.Fatalf("validation = %#v, want bounded non-resumable result", validation) + } + }) + } +} + +func TestExtractStageResumeValidationRejectsUnsafePathWithError(t *testing.T) { + env, m, _ := setupExtractEnv(t) + seedSucceededExtractResult(t, env, m) + prior := *m.Stages["extract"] + m.Stages["extract"].Outputs[0].LocalPath = filepath.Join(env.Config.Pipeline.Workspace.Root, "outside.json") + + if _, err := (extractStage{}).ValidateResume(context.Background(), env, m); err == nil || !strings.Contains(err.Error(), "unsafe") { + t.Fatalf("ValidateResume() error = %v, want unsafe path error", err) + } + if m.Stages["extract"].Status != prior.Status || m.Stages["extract"].Error != prior.Error { + t.Fatalf("validation mutated stage record: %#v", m.Stages["extract"]) + } +} + +func seedSucceededExtractResult(t *testing.T, env *Env, m *manifest.Manifest) { + t.Helper() + producerRunID := m.RunID + result, err := (extractStage{}).Run(context.Background(), env, m) + if err != nil { + t.Fatalf("Run() error = %v", err) + } + records := make([]manifest.ArtifactRecord, 0, len(result.Outputs)) + for _, output := range result.Outputs { + records = append(records, manifest.ArtifactRecord{ + Kind: output.Kind, SourceID: output.SourceID, LocalPath: output.AbsolutePath, + Contract: output.Contract, ExternalProvenance: output.ExternalProvenance, + ProducerRunID: producerRunID, Checksum: output.Checksum, + }) + } + m.MarkStageSucceeded("extract", time.Now().UTC(), records) + m.Stages["extract"].Metadata = result.Metadata +} + type extractFixture struct { workspace string campaign string diff --git a/internal/stage/placeholders.go b/internal/stage/placeholders.go index 9d39d5a..e191f73 100644 --- a/internal/stage/placeholders.go +++ b/internal/stage/placeholders.go @@ -74,6 +74,7 @@ func All() []Stage { polishStage{}, normalizeStage{}, trimStage{}, + extractStage{}, renderStage{}, analyzeStage{}, publishStage{}, diff --git a/internal/stage/placeholders_test.go b/internal/stage/placeholders_test.go index ab9de29..9dcce14 100644 --- a/internal/stage/placeholders_test.go +++ b/internal/stage/placeholders_test.go @@ -113,7 +113,7 @@ func TestStagesReturnExpectedMetadata(t *testing.T) { } m := manifest.New("2026-05-03", time.Now().UTC()) - for _, name := range []string{"prepare", "transcribe", "merge", "polish", "normalize", "trim", "render", "analyze"} { + for _, name := range []string{"prepare", "transcribe", "merge", "polish", "normalize", "trim", "extract", "render", "analyze"} { m.MarkStageSucceeded(name, time.Now().UTC(), nil) } m.RunID = "20260516T000000Z-abcdef12" @@ -206,6 +206,12 @@ func TestStagesReturnExpectedMetadata(t *testing.T) { } continue } + if s.Name() == "extract" { + if result.Disposition != StageDispositionSkipped || result.SkipReason != extractSkipReason { + t.Fatalf("extract result = %#v, want disabled self-skip", result) + } + continue + } if s.Name() == "render" { if result.Metadata["stage"] != "render" { t.Fatalf("render metadata = %#v, want stage=render", result.Metadata) diff --git a/internal/stage/stage.go b/internal/stage/stage.go index b8ee604..bff6a4f 100644 --- a/internal/stage/stage.go +++ b/internal/stage/stage.go @@ -3,6 +3,8 @@ package stage import ( "context" "log/slog" + "strings" + "unicode/utf8" "gitea.maximumdirect.net/eric/narratio/internal/adapters/audita" "gitea.maximumdirect.net/eric/narratio/internal/adapters/notarius" @@ -46,6 +48,48 @@ type Stage interface { Run(ctx context.Context, env *Env, m *manifest.Manifest) (*StageResult, error) } +const maxResumeReasonLength = 512 + +// ResumeValidation reports whether a previously succeeded stage can be reused. +type ResumeValidation struct { + Resumable bool + Reason string +} + +// Normalized returns a result with a bounded reason and no reason on success. +func (r ResumeValidation) Normalized() ResumeValidation { + if r.Resumable { + return Resumable() + } + return NonResumable(r.Reason) +} + +// ResumeValidator is implemented by stages that validate persisted success before reuse. +type ResumeValidator interface { + ValidateResume(ctx context.Context, env *Env, m *manifest.Manifest) (ResumeValidation, error) +} + +// Resumable reports a successful resume validation. +func Resumable() ResumeValidation { + return ResumeValidation{Resumable: true} +} + +// NonResumable reports a bounded reason that persisted success must be rerun. +func NonResumable(reason string) ResumeValidation { + reason = strings.TrimSpace(reason) + if reason == "" { + reason = "persisted stage result is not reusable" + } + if len(reason) > maxResumeReasonLength { + cutoff := maxResumeReasonLength + for cutoff > 0 && !utf8.ValidString(reason[:cutoff]) { + cutoff-- + } + reason = reason[:cutoff] + } + return ResumeValidation{Reason: reason} +} + // StageDisposition describes the outcome of a stage that returned without an error. type StageDisposition string