Files
narratio/internal/app/assembled_workflow_test.go

481 lines
20 KiB
Go

package app
import (
"bytes"
"context"
"io"
"os"
"path/filepath"
"reflect"
"strings"
"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, plan BoundedPlan, options RunOptions) (*RunSummary, error) {
captured = append(captured, capturedRequest{
stages: plan.Names(), 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
plan.stages = []stage.Stage{countingStage{name: selected, runs: &runs}}
summary, err := executePlan(context.Background(), cfg, plan, RunOptions{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")
plan.stages = []stage.Stage{failingStage{name: "render", err: context.Canceled}}
_, err := executePlan(context.Background(), cfg, plan, RunOptions{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 TestAssembledSplitBundleCommandsUseOneEffectiveFamilyConfiguration(t *testing.T) {
pipelinePath, campaignPath, sessionPath := assembledSplitBundlePaths()
workspacePath := filepath.Join(filepath.Dir(pipelinePath), "workspace")
if _, err := os.Stat(workspacePath); !os.IsNotExist(err) {
t.Fatalf("example workspace stat = %v, want absent", err)
}
var validated bytes.Buffer
if err := ConfigValidate(context.Background(), []string{"--config", pipelinePath, "--campaign-file", campaignPath}, &validated); err != nil {
t.Fatalf("validate production default: %v", err)
}
if !strings.Contains(validated.String(), "profile=production") {
t.Fatalf("validate output = %q, want production profile", validated.String())
}
for _, command := range []func(context.Context, []string, io.Writer) error{ConfigShow, ConfigSources} {
if err := command(context.Background(), []string{
"--config", pipelinePath, "--campaign-file", campaignPath, "--profile", "testing",
}, io.Discard); err != nil {
t.Fatalf("testing inspection command %T: %v", command, err)
}
}
var diff bytes.Buffer
if err := ConfigDiff(context.Background(), []string{
"production", "testing", "--config", pipelinePath, "--campaign-file", campaignPath,
}, &diff); err != nil {
t.Fatalf("compare profiles: %v", err)
}
if !strings.Contains(diff.String(), "audita.model") || !strings.Contains(diff.String(), "character_items_arannis.profile_id") {
t.Fatalf("profile diff = %q, want model and expanded family changes", diff.String())
}
common := []string{
"2026-05-03", "--config", pipelinePath, "--campaign-file", campaignPath,
"--session", sessionPath, "--profile", "testing",
}
var planned bytes.Buffer
if err := Plan(context.Background(), common, &planned); err != nil {
t.Fatalf("plan testing bundle: %v", err)
}
if !strings.Contains(planned.String(), "profile=testing") {
t.Fatalf("plan output = %q, want testing provenance", planned.String())
}
original := executeStagesFn
t.Cleanup(func() { executeStagesFn = original })
var capturedCfg *config.Config
var capturedPlan BoundedPlan
var capturedOptions RunOptions
executeStagesFn = func(_ context.Context, cfg *config.Config, plan BoundedPlan, options RunOptions) (*RunSummary, error) {
capturedCfg = cfg
capturedPlan = plan
capturedOptions = options
return &RunSummary{SessionID: cfg.Session.SessionID, ManifestPath: manifestPathForConfig(cfg.Pipeline.Workspace.Root)}, nil
}
runArgs := append(append([]string(nil), common...), "--from", "analyze", "--through", "analyze", "--artifacts", "character_items_arannis")
var runOut bytes.Buffer
if err := Run(context.Background(), runArgs, &runOut); err != nil {
t.Fatalf("bounded family run: %v", err)
}
if !reflect.DeepEqual(capturedPlan.Names(), []string{"analyze"}) {
t.Fatalf("bounded plan = %#v, want analyze only", capturedPlan.Names())
}
if profile, ok := config.SelectedPipelineProfile(capturedCfg.Pipeline); !ok || profile.Name != "testing" {
t.Fatalf("captured profile = %#v, selected=%t", profile, ok)
}
if digest := config.EffectivePipelineDigest(capturedCfg.Pipeline); digest == "" || !strings.Contains(runOut.String(), "digest="+digest) {
t.Fatalf("run output = %q, want effective digest %q", runOut.String(), digest)
}
if got := capturedOptions.EffectiveArtifacts.Keys(); !reflect.DeepEqual(got, []string{"character_items_arannis"}) {
t.Fatalf("effective artifact keys = %#v", got)
}
if origin, ok := capturedOptions.EffectiveArtifacts.Origin("character_items_arannis"); !ok || origin.Family != "character_items" || origin.CharacterID != "arannis" {
t.Fatalf("family origin = %#v, present=%t", origin, ok)
}
fullFamily, err := resolveEffectiveArtifacts(capturedCfg, []string{"character_items"})
if err != nil {
t.Fatalf("resolve complete family: %v", err)
}
if got := fullFamily.Keys(); !reflect.DeepEqual(got, []string{"character_items_arannis", "character_items_brenna"}) {
t.Fatalf("full family keys = %#v", got)
}
if _, err := os.Stat(workspacePath); !os.IsNotExist(err) {
t.Fatalf("inspection or bounded run created example workspace: %v", err)
}
}
func TestAssembledSplitBundleProfileSwitchRecordsProvenanceAndLimitsReuse(t *testing.T) {
production := loadAssembledSplitBundleConfig(t, "")
testingCfg := loadAssembledSplitBundleConfig(t, "testing")
workspace := t.TempDir()
production.Pipeline.Workspace.Root = workspace
testingCfg.Pipeline.Workspace.Root = workspace
names := []string{"prepare", "transcribe", "merge", "polish", "normalize", "trim", "render"}
providers := make([]stage.SemanticConfigFingerprinter, len(names))
seed := manifest.New(production.Session.SessionID, time.Now().UTC())
seed.Campaign = production.Session.Campaign
for _, candidate := range stage.All() {
seed.MarkStageSucceeded(candidate.Name(), time.Now().UTC(), nil)
}
for index, name := range names {
providers[index] = canonicalSemanticProvider(t, name)
fingerprint, err := providers[index].SemanticConfigFingerprint(&stage.Env{Config: production})
if err != nil {
t.Fatalf("production %s fingerprint: %v", name, err)
}
seed.Stages[name].SemanticConfig = &fingerprint
}
store := &manifest.LocalStore{}
if err := store.Save(context.Background(), manifestPathForConfig(workspace), seed); err != nil {
t.Fatal(err)
}
productionRuns := make([]int, len(names))
productionStages := assembledSemanticStages(names, providers, productionRuns)
productionSummary, err := executeStages(context.Background(), production, productionStages, RunOptions{})
if err != nil {
t.Fatalf("record production provenance: %v", err)
}
if !reflect.DeepEqual(productionRuns, make([]int, len(names))) {
t.Fatalf("production runs = %v, want complete reuse", productionRuns)
}
assertAssembledProvenance(t, store, productionSummary, "production", config.EffectivePipelineDigest(production.Pipeline))
testingRuns := make([]int, len(names))
testingStages := assembledSemanticStages(names, providers, testingRuns)
testingSummary, err := executeStages(context.Background(), testingCfg, testingStages, RunOptions{})
if err != nil {
t.Fatalf("switch to testing profile: %v", err)
}
if got, want := testingRuns, []int{0, 0, 0, 1, 1, 1, 1}; !reflect.DeepEqual(got, want) {
t.Fatalf("testing profile runs = %v, want %v", got, want)
}
assertAssembledProvenance(t, store, testingSummary, "testing", config.EffectivePipelineDigest(testingCfg.Pipeline))
}
func assembledSplitBundlePaths() (pipelinePath, campaignPath, sessionPath string) {
examplesDir := filepath.Join("..", "..", "examples")
return filepath.Join(examplesDir, "production-testing", "pipeline.yml"),
filepath.Join(examplesDir, "campaigns", "sample-campaign", "campaign.yml"),
filepath.Join(examplesDir, "session.local-audio.yml")
}
func loadAssembledSplitBundleConfig(t *testing.T, profile string) *config.Config {
t.Helper()
pipelinePath, campaignPath, sessionPath := assembledSplitBundlePaths()
options := config.SessionLoadOptions{}
if profile != "" {
options.Profile = &profile
}
cfg, err := config.LoadWithSessionOptions(pipelinePath, campaignPath, sessionPath, options)
if err != nil {
t.Fatalf("load split bundle profile %q: %v", profile, err)
}
return cfg
}
func assembledSemanticStages(names []string, providers []stage.SemanticConfigFingerprinter, runs []int) []stage.Stage {
stages := make([]stage.Stage, 0, len(names))
for index, name := range names {
stages = append(stages, semanticContractRunStub{name: name, provider: providers[index], runs: &runs[index]})
}
return stages
}
func assertAssembledProvenance(t *testing.T, store *manifest.LocalStore, summary *RunSummary, profile, digest string) {
t.Helper()
session, err := store.Load(context.Background(), summary.ManifestPath)
if err != nil {
t.Fatal(err)
}
run, err := store.LoadRun(context.Background(), summary.RunManifestPath)
if err != nil {
t.Fatal(err)
}
for _, value := range []*manifest.EffectiveConfigProvenance{session.EffectiveConfig, run.EffectiveConfig} {
if value == nil || value.SelectedProfile == nil || value.SelectedProfile.Name != profile || value.EffectiveConfigDigest != digest {
t.Fatalf("effective configuration provenance = %#v, want profile=%q digest=%q", value, profile, digest)
}
}
}
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")
}