package app import ( "context" "encoding/json" "errors" "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 failuresRemaining int } func (r *materializingNotariusRunner) Run(_ context.Context, req notarius.RunRequest) (notarius.RunResult, error) { r.requests = append(r.requests, req) if r.failuresRemaining > 0 { r.failuresRemaining-- return notarius.RunResult{}, errors.New("notarius execution failed") } 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 TestExtractLifecycleChangedOutcomeRerunsSucceededDownstream(t *testing.T) { cfg, env, runner := extractionLifecycleFixture(t, false) analyzeRuns := 0 plan := extractionLifecyclePlan(t, &analyzeRuns) if _, err := executeStages(context.Background(), cfg, plan, RunOptions{Env: env}); err != nil { t.Fatalf("disabled executeStages() error = %v", err) } if analyzeRuns != 1 || len(runner.requests) != 0 { t.Fatalf("disabled run analyze=%d Notarius=%d, want 1 and 0", analyzeRuns, len(runner.requests)) } cfg.Pipeline.Notarius.Enabled = true summary, err := executeStages(context.Background(), cfg, plan, RunOptions{Env: env}) if err != nil { t.Fatalf("enabled executeStages() error = %v", err) } if analyzeRuns != 2 || len(runner.requests) != 1 { t.Fatalf("enabled run analyze=%d Notarius=%d, want 2 and 1", analyzeRuns, len(runner.requests)) } if len(summary.Executed) != 2 || len(summary.Skipped) != 0 { t.Fatalf("enabled summary = %#v, want extract and analyze executed", summary) } } func TestExtractLifecycleFailureInvalidatesAndOrdinaryRetryRerunsDownstream(t *testing.T) { cfg, env, runner := extractionLifecycleFixture(t, true) runner.failuresRemaining = 1 markLifecycleStageSucceeded(t, cfg, "analyze") analyzeRuns := 0 plan := extractionLifecyclePlan(t, &analyzeRuns) if _, err := executeStages(context.Background(), cfg, plan, RunOptions{Env: env}); err == nil || !strings.Contains(err.Error(), "notarius execution failed") { t.Fatalf("failed executeStages() error = %v", err) } failed := loadLifecycleManifest(t, cfg) if failed.Stages["extract"].Status != manifest.StatusFailed || failed.Stages["analyze"].Status != manifest.StatusStale { t.Fatalf("failed lifecycle extract=%#v analyze=%#v", failed.Stages["extract"], failed.Stages["analyze"]) } if failed.Stages["analyze"].Error == nil || failed.Stages["analyze"].Error.Message != staleReasonFailure { t.Fatalf("analyze stale reason = %#v, want %q", failed.Stages["analyze"].Error, staleReasonFailure) } summary, err := executeStages(context.Background(), cfg, plan, RunOptions{Env: env}) if err != nil { t.Fatalf("retry executeStages() error = %v", err) } if analyzeRuns != 1 || len(runner.requests) != 2 || len(summary.Executed) != 2 { t.Fatalf("retry analyze=%d Notarius=%d summary=%#v", analyzeRuns, len(runner.requests), summary) } } func TestExtractLifecycleForcedSelfSkipInvalidatesDownstream(t *testing.T) { cfg, env, _ := extractionLifecycleFixture(t, false) markLifecycleStageSucceeded(t, cfg, "analyze") plan, _ := BuildSingleStagePlan("extract") if _, err := executeStages(context.Background(), cfg, plan, RunOptions{Env: env, Force: true}); err != nil { t.Fatalf("executeStages() error = %v", err) } loaded := loadLifecycleManifest(t, cfg) if loaded.Stages["extract"].Status != manifest.StatusSkipped || loaded.Stages["analyze"].Status != manifest.StatusStale { t.Fatalf("forced self-skip extract=%#v analyze=%#v", loaded.Stages["extract"], loaded.Stages["analyze"]) } if loaded.Stages["analyze"].Error == nil || loaded.Stages["analyze"].Error.Message != staleReasonForcedReplacement { t.Fatalf("analyze stale reason = %#v, want %q", loaded.Stages["analyze"].Error, staleReasonForcedReplacement) } } func TestExtractLifecycleForcedFailureInvalidatesDownstream(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("initial executeStages() error = %v", err) } succeeded := loadLifecycleManifest(t, cfg).Stages["extract"] if succeeded == nil || succeeded.Status != manifest.StatusSucceeded || len(succeeded.Outputs) == 0 || len(succeeded.Logs) == 0 || len(succeeded.Metadata) == 0 { t.Fatalf("initial extraction result = %#v, want succeeded result details", succeeded) } markLifecycleStageSucceeded(t, cfg, "analyze") runner.failuresRemaining = 1 if _, err := executeStages(context.Background(), cfg, plan, RunOptions{Env: env, Force: true}); err == nil { t.Fatal("executeStages() error = nil, want forced extraction failure") } loaded := loadLifecycleManifest(t, cfg) if loaded.Stages["extract"].Status != manifest.StatusFailed || loaded.Stages["analyze"].Status != manifest.StatusStale { t.Fatalf("forced failure extract=%#v analyze=%#v", loaded.Stages["extract"], loaded.Stages["analyze"]) } if loaded.Stages["analyze"].Error == nil || loaded.Stages["analyze"].Error.Message != staleReasonForcedReplacement { t.Fatalf("analyze stale reason = %#v, want %q", loaded.Stages["analyze"].Error, staleReasonForcedReplacement) } failed := loaded.Stages["extract"] if len(failed.Outputs) != 0 || len(failed.Logs) != 0 || len(failed.GeneratedConfigs) != 0 || len(failed.Metadata) != 0 { t.Fatalf("failed replacement inherited extraction result details: %#v", failed) } historical, err := (&manifest.LocalStore{}).LoadRun(context.Background(), first.RunManifestPath) if err != nil { t.Fatalf("LoadRun(initial) error = %v", err) } historicalExtract := historical.Stages["extract"] if historicalExtract == nil || historicalExtract.Status != manifest.StatusSucceeded || len(historicalExtract.Outputs) == 0 || len(historicalExtract.Logs) == 0 || len(historicalExtract.Metadata) == 0 { t.Fatalf("historical extraction result = %#v, want preserved succeeded details", historicalExtract) } if _, err := os.Stat(succeeded.Outputs[0].LocalPath); err != nil { t.Fatalf("durable extraction output was not preserved: %v", err) } } func TestExtractLifecycleRepeatedSelfSkipPreservesSucceededDownstream(t *testing.T) { cfg, env, runner := extractionLifecycleFixture(t, false) analyzeRuns := 0 plan := extractionLifecyclePlan(t, &analyzeRuns) if _, err := executeStages(context.Background(), cfg, plan, RunOptions{Env: env}); err != nil { t.Fatalf("first executeStages() error = %v", err) } second, err := executeStages(context.Background(), cfg, plan, RunOptions{Env: env}) if err != nil { t.Fatalf("second executeStages() error = %v", err) } if analyzeRuns != 1 || len(runner.requests) != 0 { t.Fatalf("repeated disabled run analyze=%d Notarius=%d, want 1 and 0", analyzeRuns, len(runner.requests)) } if len(second.Executed) != 1 || len(second.Skipped) != 2 { t.Fatalf("second summary = %#v, want executed self-skip and skipped analyze", second) } loaded := loadLifecycleManifest(t, cfg) if loaded.Stages["analyze"].Status != manifest.StatusSucceeded { t.Fatalf("analyze = %#v, want succeeded", loaded.Stages["analyze"]) } } func TestExtractLifecycleSkipsCurrentResumableResult(t *testing.T) { cfg, env, runner := extractionLifecycleFixture(t, true) analyzeRuns := 0 plan := extractionLifecyclePlan(t, &analyzeRuns) 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) != 2 || len(runner.requests) != 1 || analyzeRuns != 1 { t.Fatalf("resume summary = %#v requests=%d analyze=%d", resumed, len(runner.requests), analyzeRuns) } } 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) analyzeRuns := 0 plan := extractionLifecyclePlan(t, &analyzeRuns) 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(map[string]*manifest.StageRecord{ "extract": persisted.Stages["extract"], "analyze": persisted.Stages["analyze"], }) 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(map[string]*manifest.StageRecord{ "extract": afterManifest.Stages["extract"], "analyze": afterManifest.Stages["analyze"], }) if string(before) != string(after) || len(runner.requests) != 1 || analyzeRuns != 1 { t.Fatalf("successful records changed: before=%s after=%s requests=%d analyze=%d", before, after, len(runner.requests), analyzeRuns) } } func extractionLifecyclePlan(t *testing.T, analyzeRuns *int) []stage.Stage { t.Helper() plan, err := BuildSingleStagePlan("extract") if err != nil { t.Fatalf("BuildSingleStagePlan(extract) error = %v", err) } return append(plan, countingStage{name: "analyze", runs: analyzeRuns}) } func loadLifecycleManifest(t *testing.T, cfg *config.Config) *manifest.Manifest { t.Helper() loaded, err := (&manifest.LocalStore{}).Load(context.Background(), manifestPathFor(cfg)) if err != nil { t.Fatalf("Load() error = %v", err) } return loaded } func markLifecycleStageSucceeded(t *testing.T, cfg *config.Config, name string) { t.Helper() loaded := loadLifecycleManifest(t, cfg) loaded.MarkStageSucceeded(name, time.Now().UTC(), nil) if err := (&manifest.LocalStore{}).Save(context.Background(), manifestPathFor(cfg), loaded); err != nil { t.Fatalf("Save() error = %v", err) } } 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 }