From 5887839aa1cae54720750bd81809cdd2d7821e28 Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Sat, 29 Aug 2026 20:08:48 +0000 Subject: [PATCH] Add assembled workflow compatibility coverage --- docs/roadmap/implementation.md | 2 + internal/app/assembled_workflow_test.go | 297 ++++++++++++++++++++++++ internal/app/run_control.go | 16 -- internal/app/run_control_test.go | 19 +- internal/app/runner.go | 19 +- 5 files changed, 314 insertions(+), 39 deletions(-) create mode 100644 internal/app/assembled_workflow_test.go diff --git a/docs/roadmap/implementation.md b/docs/roadmap/implementation.md index 46751a1..d7a70ff 100644 --- a/docs/roadmap/implementation.md +++ b/docs/roadmap/implementation.md @@ -713,6 +713,8 @@ freshness using the same read-only decision engine as execution. ## Stage 15 — Assembled Workflow And Compatibility Coverage +**Status: Completed** + ### Goal Verify the complete feature through production composition and representative diff --git a/internal/app/assembled_workflow_test.go b/internal/app/assembled_workflow_test.go new file mode 100644 index 0000000..4a0b501 --- /dev/null +++ b/internal/app/assembled_workflow_test.go @@ -0,0 +1,297 @@ +package app + +import ( + "bytes" + "context" + "path/filepath" + "reflect" + "testing" + "time" + + "gitea.maximumdirect.net/eric/narratio/internal/adapters/scriptorium" + "gitea.maximumdirect.net/eric/narratio/internal/adapters/storage" + "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" +) + +func TestAssembledFullRunUsesCanonicalOrderAndBoundedRunManifests(t *testing.T) { + cfg := testConfig(t) + canonical := []string{ + "prepare", "transcribe", "merge", "polish", "normalize", "trim", + "render", "extract", "analyze", "publish", "notify", + } + var order []string + stages := make([]stage.Stage, 0, len(canonical)) + for _, name := range canonical { + stages = append(stages, resultStage{name: name, result: &stage.StageResult{}, order: &order}) + } + summary, err := executeStages(context.Background(), cfg, stages, RunOptions{}) + if err != nil { + t.Fatalf("executeStages() error = %v", err) + } + if !reflect.DeepEqual(order, canonical) || !reflect.DeepEqual(summary.Executed, canonical) { + t.Fatalf("execution order=%#v summary=%#v, want %#v", order, summary.Executed, canonical) + } + runManifest, err := (&manifest.LocalStore{}).LoadRun(context.Background(), summary.RunManifestPath) + if err != nil { + t.Fatal(err) + } + if !reflect.DeepEqual(runManifest.RequestedStages, canonical) { + t.Fatalf("requested stages = %#v, want canonical order", runManifest.RequestedStages) + } +} + +func TestAssembledCanonicalAndAliasArtifactRegenerationRequestsMatch(t *testing.T) { + workspaceRoot := t.TempDir() + pipelinePath, campaignPath, sessionPath := writeValidConfigFilesWithScriptoriumArtifacts(t, workspaceRoot) + type capturedRequest struct { + stages []string + artifacts []string + force bool + } + var captured []capturedRequest + original := executeStagesFn + t.Cleanup(func() { executeStagesFn = original }) + executeStagesFn = func(_ context.Context, _ *config.Config, stages []stage.Stage, options RunOptions) (*RunSummary, error) { + captured = append(captured, capturedRequest{ + stages: stageNames(stages), artifacts: append([]string(nil), options.SelectedArtifacts...), force: options.Force, + }) + return &RunSummary{SessionID: "2026-05-03", ManifestPath: manifestPathForConfig(workspaceRoot)}, nil + } + base := []string{ + "2026-05-03", "--force", "--from", "extract", "--through", "analyze", + "--config", pipelinePath, "--campaign-file", campaignPath, "--session", sessionPath, + } + if err := Run(context.Background(), base, &bytes.Buffer{}); err != nil { + t.Fatalf("canonical unselected Run() error = %v", err) + } + selected := append(append([]string(nil), base...), "--artifacts", "session_recap") + if err := Run(context.Background(), selected, &bytes.Buffer{}); err != nil { + t.Fatalf("canonical selected Run() error = %v", err) + } + var stdout, stderr bytes.Buffer + alias := []string{ + "regenerate-artifacts", "2026-05-03", "--artifacts", "session_recap", + "--config", pipelinePath, "--campaign-file", campaignPath, "--session", sessionPath, + } + if code := Execute(alias, &stdout, &stderr); code != 0 { + t.Fatalf("alias exit=%d stderr=%q", code, stderr.String()) + } + if len(captured) != 3 { + t.Fatalf("captured requests = %#v", captured) + } + wantStages := []string{"extract", "analyze"} + if !reflect.DeepEqual(captured[0].stages, wantStages) || len(captured[0].artifacts) != 0 || !captured[0].force { + t.Fatalf("unselected request = %#v", captured[0]) + } + if !reflect.DeepEqual(captured[1], captured[2]) || !reflect.DeepEqual(captured[1].stages, wantStages) || + !reflect.DeepEqual(captured[1].artifacts, []string{"session_recap"}) || !captured[1].force { + t.Fatalf("canonical=%#v alias=%#v, want identical bounded request", captured[1], captured[2]) + } +} + +func TestAssembledForcedSiblingIndependenceAndFailureBoundary(t *testing.T) { + for _, selected := range []string{"render", "extract"} { + t.Run(selected, func(t *testing.T) { + cfg := testConfig(t) + seedAllStagesSucceeded(t, cfg) + plan := mustBoundedPlan(t, selected, selected) + runs := 0 + summary, err := executeStages(context.Background(), cfg, []stage.Stage{ + countingStage{name: selected, runs: &runs}, + }, RunOptions{Plan: plan, Force: true}) + if err != nil { + t.Fatal(err) + } + loaded := loadAssembledManifest(t, cfg) + sibling := "render" + if selected == "render" { + sibling = "extract" + } + if loaded.Stages[sibling].Status != manifest.StatusSucceeded { + t.Fatalf("%s sibling = %#v, want succeeded", sibling, loaded.Stages[sibling]) + } + for _, dependent := range []string{"analyze", "publish", "notify"} { + if loaded.Stages[dependent].Status != manifest.StatusStale { + t.Fatalf("%s status = %q, want stale", dependent, loaded.Stages[dependent].Status) + } + } + runManifest, err := (&manifest.LocalStore{}).LoadRun(context.Background(), summary.RunManifestPath) + if err != nil { + t.Fatal(err) + } + if !reflect.DeepEqual(runManifest.RequestedStages, []string{selected}) || len(runManifest.Stages) != 1 { + t.Fatalf("bounded run manifest = %#v", runManifest) + } + }) + } + + t.Run("stop on failure", func(t *testing.T) { + cfg := testConfig(t) + seedAllStagesSucceeded(t, cfg) + plan := mustBoundedPlan(t, "render", "render") + _, err := executeStages(context.Background(), cfg, []stage.Stage{ + failingStage{name: "render", err: context.Canceled}, + }, RunOptions{Plan: plan, Force: true}) + if err == nil { + t.Fatal("forced render failure returned nil") + } + loaded := loadAssembledManifest(t, cfg) + if loaded.Stages["extract"].Status != manifest.StatusSucceeded { + t.Fatalf("extract sibling = %#v", loaded.Stages["extract"]) + } + for _, outside := range []string{"analyze", "publish", "notify"} { + if loaded.Stages[outside].Status != manifest.StatusStale { + t.Fatalf("outside stage %s = %#v, want stale and unexecuted", outside, loaded.Stages[outside]) + } + } + }) +} + +func TestAssembledLegacyAnalyzeTransitionPublishesOnlyCurrentRecords(t *testing.T) { + cfg := testConfig(t) + cfg.Pipeline.Scriptorium = &config.ScriptoriumConfig{Artifacts: map[string]config.ScriptoriumArtifactConfig{ + "player_handout": {Enabled: true, PromptID: "dnd.player_handout", OutputPath: "artifacts/player_handout.md"}, + "session_recap": {Enabled: true, PromptID: "dnd.session_recap", OutputPath: "artifacts/session_recap.md"}, + }} + cfg.Pipeline.Storage.Backend = config.StorageBackendS3 + cfg.Pipeline.Storage.S3 = &config.StorageS3Config{Bucket: "archive", RootPrefix: "dnd"} + cfg.Pipeline.Publish = &config.PublishConfig{ + Enabled: boolPtr(true), UploadRun: boolPtr(true), + Outputs: []config.PublishOutputRule{ + {Source: artifacts.ConfiguredArtifactSourceID("player_handout"), Dest: "artifacts/player_handout.md", Required: boolPtr(true)}, + {Source: artifacts.ConfiguredArtifactSourceID("session_recap"), Dest: "artifacts/session_recap.md", Required: boolPtr(true)}, + }, + } + paths, err := artifacts.NewLocalStore(cfg.Pipeline.Workspace.Root).EnsureLayoutFor(cfg.Session.Campaign, cfg.Session.SessionID) + if err != nil { + t.Fatal(err) + } + legacyHandout := []byte("legacy handout\n") + legacyRecap := []byte("legacy recap\n") + mustWriteTestFile(t, filepath.Join(paths.ArtifactsDir, "player_handout.md"), string(legacyHandout)) + mustWriteTestFile(t, filepath.Join(paths.ArtifactsDir, "session_recap.md"), string(legacyRecap)) + + now := time.Date(2026, 5, 3, 10, 0, 0, 0, time.UTC) + m := manifest.New(cfg.Session.SessionID, now) + m.Campaign = cfg.Session.Campaign + for index, name := range []string{"prepare", "transcribe", "merge", "polish", "normalize", "trim"} { + m.MarkStageSucceeded(name, now.Add(time.Duration(index)*time.Minute), nil) + } + // Historical manifests can show extract completing before render. Status, + // not the old relative timestamps, is the compatibility authority. + m.MarkStageSucceeded("extract", now.Add(10*time.Minute), nil) + m.MarkStageSucceeded("render", now.Add(11*time.Minute), nil) + m.MarkStageSucceeded("analyze", now.Add(12*time.Minute), []manifest.ArtifactRecord{ + {Kind: "player_handout", LocalPath: "artifacts/player_handout.md"}, + {Kind: "session_recap", LocalPath: "artifacts/session_recap.md"}, + }) + if err := (&manifest.LocalStore{}).Save(context.Background(), paths.ManifestPath, m); err != nil { + t.Fatal(err) + } + + analyze, err := stage.Select("analyze") + if err != nil { + t.Fatal(err) + } + fake := &scriptorium.FakeRunner{} + _, err = executeStages(context.Background(), cfg, []stage.Stage{analyze}, RunOptions{ + SelectedArtifacts: []string{"session_recap"}, Env: &Env{Scriptorium: fake}, + }) + if err != nil { + t.Fatalf("partial legacy regeneration: %v", err) + } + if len(fake.RunRequests) != 1 || fake.RunRequests[0].PromptID != "dnd.session_recap" { + t.Fatalf("partial requests = %#v", fake.RunRequests) + } + afterPartial := loadAssembledManifest(t, cfg) + if len(afterPartial.Stages["analyze"].AnalyzeArtifacts) != 1 || + afterPartial.Stages["analyze"].AnalyzeArtifacts["session_recap"].Status != manifest.AnalyzeArtifactCurrent { + t.Fatalf("partial state = %#v", afterPartial.Stages["analyze"].AnalyzeArtifacts) + } + for _, transcriptStage := range []string{"prepare", "transcribe", "merge", "polish", "normalize", "trim", "render"} { + if afterPartial.Stages[transcriptStage].Status != manifest.StatusSucceeded { + t.Fatalf("legacy transition invalidated %s: %#v", transcriptStage, afterPartial.Stages[transcriptStage]) + } + } + configured := artifacts.ConfiguredArtifactDefinitions(cfg.Pipeline.Scriptorium.Artifacts) + effective, err := artifacts.ResolveEffectiveArtifactSet(configured, nil) + if err != nil { + t.Fatal(err) + } + catalog, err := artifacts.BootstrapRuntimeCatalog(configured, effective, nil) + if err != nil { + t.Fatal(err) + } + catalog.HydrateAnalyzeArtifacts(paths, afterPartial, configured) + if entry, ok := catalog.Lookup(artifacts.ConfiguredArtifactSourceID("player_handout")); !ok || entry.Available { + t.Fatalf("legacy unselected handout catalog entry = %#v, present=%v", entry, ok) + } + + full, err := executeStages(context.Background(), cfg, []stage.Stage{analyze}, RunOptions{Env: &Env{Scriptorium: fake}}) + if err != nil { + t.Fatalf("full regeneration: %v", err) + } + if len(fake.RunRequests) != 2 || fake.RunRequests[1].PromptID != "dnd.player_handout" { + t.Fatalf("full requests = %#v, want only missing handout added", fake.RunRequests) + } + fullRun, err := (&manifest.LocalStore{}).LoadRun(context.Background(), full.RunManifestPath) + if err != nil { + t.Fatal(err) + } + if got := analyzeArtifactOutputKeys(fullRun.Stages["analyze"].Outputs); !reflect.DeepEqual(got, []string{"player_handout"}) { + t.Fatalf("full invocation outputs = %#v, want newly generated handout only", got) + } + afterFull := loadAssembledManifest(t, cfg) + for _, key := range []string{"player_handout", "session_recap"} { + if afterFull.Stages["analyze"].AnalyzeArtifacts[key].Status != manifest.AnalyzeArtifactCurrent { + t.Fatalf("%s state = %#v", key, afterFull.Stages["analyze"].AnalyzeArtifacts[key]) + } + } + + publish, err := stage.Select("publish") + if err != nil { + t.Fatal(err) + } + remote := &storage.FakeBackend{} + published, err := executeStages(context.Background(), cfg, []stage.Stage{publish}, RunOptions{Env: &Env{ObjectStore: remote}}) + if err != nil { + t.Fatalf("publish current records: %v", err) + } + afterPublish := loadAssembledManifest(t, cfg) + if got := afterPublish.Stages["publish"].Metadata["published_files_uploaded"]; got != float64(2) { + t.Fatalf("published files = %#v, want 2", got) + } + publishRun, err := (&manifest.LocalStore{}).LoadRun(context.Background(), published.RunManifestPath) + if err != nil { + t.Fatal(err) + } + if !reflect.DeepEqual(publishRun.RequestedStages, []string{"publish"}) || publishRun.Stages["analyze"] != nil { + t.Fatalf("publish run manifest = %#v", publishRun) + } +} + +func seedAllStagesSucceeded(t *testing.T, cfg *config.Config) { + t.Helper() + m := manifest.New(cfg.Session.SessionID, time.Now().UTC()) + m.Campaign = cfg.Session.Campaign + for _, name := range canonicalStageNames() { + m.MarkStageSucceeded(name, time.Now().UTC(), nil) + } + saveBoundedManifest(t, cfg, m) +} + +func loadAssembledManifest(t *testing.T, cfg *config.Config) *manifest.Manifest { + t.Helper() + m, err := (&manifest.LocalStore{}).Load(context.Background(), manifestPathFor(cfg)) + if err != nil { + t.Fatal(err) + } + return m +} + +func manifestPathForConfig(workspaceRoot string) string { + return artifacts.SessionManifestPathForCampaign(workspaceRoot, "sample-campaign", "2026-05-03") +} diff --git a/internal/app/run_control.go b/internal/app/run_control.go index 7d6c2c4..f9ad5d8 100644 --- a/internal/app/run_control.go +++ b/internal/app/run_control.go @@ -19,11 +19,6 @@ const ( stageActionSkip stageAction = "skip" ) -type stageDecision struct { - Stage stage.Stage - Action stageAction -} - const ( staleReasonForcedReplacement = "upstream stage was force-run" staleReasonChangedResult = "upstream stage result changed" @@ -39,17 +34,6 @@ type priorStageOutcome struct { outputs int } -func decideStageActions(stages []stage.Stage, m *manifest.Manifest, force bool) []stageDecision { - out := make([]stageDecision, 0, len(stages)) - for _, s := range stages { - out = append(out, stageDecision{ - Stage: s, - Action: decideStageAction(s, m, force), - }) - } - return out -} - func decideStageAction(s stage.Stage, m *manifest.Manifest, force bool) stageAction { // TODO: incorporate stale detection once checksum/input change tracking is implemented. if !force && stageSucceeded(m, s.Name()) { diff --git a/internal/app/run_control_test.go b/internal/app/run_control_test.go index 4d2978d..0e0e4dc 100644 --- a/internal/app/run_control_test.go +++ b/internal/app/run_control_test.go @@ -11,25 +11,20 @@ import ( "gitea.maximumdirect.net/eric/narratio/internal/stage" ) -func TestDecideStageActions(t *testing.T) { +func TestDecideStageAction(t *testing.T) { stages := BuildFullPlan()[:2] m := manifest.New("2026-05-03", time.Now().UTC()) m.MarkStageSucceeded("prepare", time.Now().UTC(), nil) - got := decideStageActions(stages, m, false) - if len(got) != 2 { - t.Fatalf("len(decisions) = %d, want 2", len(got)) + if got := decideStageAction(stages[0], m, false); got != stageActionSkip { + t.Fatalf("prepare action = %q, want %q", got, stageActionSkip) } - if got[0].Action != stageActionSkip { - t.Fatalf("prepare action = %q, want %q", got[0].Action, stageActionSkip) - } - if got[1].Action != stageActionRun { - t.Fatalf("transcribe action = %q, want %q", got[1].Action, stageActionRun) + if got := decideStageAction(stages[1], m, false); got != stageActionRun { + t.Fatalf("transcribe action = %q, want %q", got, stageActionRun) } - forced := decideStageActions(stages, m, true) - if forced[0].Action != stageActionRun { - t.Fatalf("forced prepare action = %q, want %q", forced[0].Action, stageActionRun) + if got := decideStageAction(stages[0], m, true); got != stageActionRun { + t.Fatalf("forced prepare action = %q, want %q", got, stageActionRun) } } diff --git a/internal/app/runner.go b/internal/app/runner.go index 540f6c3..1cbf923 100644 --- a/internal/app/runner.go +++ b/internal/app/runner.go @@ -246,18 +246,15 @@ func executeStages(ctx context.Context, cfg *config.Config, stages []stage.Stage stageEnv := env - decisions := decideStageActions(stages, m, opts.Force) - - runNames := make([]string, 0, len(decisions)) - executed := make([]string, 0, len(decisions)) - skipped := make([]string, 0, len(decisions)) - for _, d := range decisions { - s := d.Stage + runNames := make([]string, 0, len(stages)) + executed := make([]string, 0, len(stages)) + skipped := make([]string, 0, len(stages)) + for _, s := range stages { stageEnv.Force = opts.Force runNames = append(runNames, s.Name()) - d.Action = decideStageAction(s, m, opts.Force) + action := decideStageAction(s, m, opts.Force) - if d.Action == stageActionSkip { + if action == stageActionSkip { if validator, ok := s.(stage.ResumeValidator); ok { validation, err := validator.ValidateResume(ctx, stageEnv, m) if err != nil { @@ -285,12 +282,12 @@ func executeStages(ctx context.Context, cfg *config.Config, stages []stage.Stage ) } env.Logger.Info("stage result is not resumable", "stage", s.Name(), "reason", validation.Reason) - d.Action = stageActionRun + action = stageActionRun } } } - if d.Action == stageActionSkip { + if action == stageActionSkip { skipped = append(skipped, s.Name()) skipAt := nowUTC() runManifest.SetStageAction(s.Name(), manifest.RunStageActionSkip, skipAt)