narratio/internal/app/extract_lifecycle_test.go

package app

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"os"
	"path/filepath"
	"reflect"
	"sort"
	"strings"
	"testing"
	"time"

	configkit "gitea.maximumdirect.net/eric/ConfigKit"
	"gitea.maximumdirect.net/eric/narratio/internal/adapters/notarius"
	"gitea.maximumdirect.net/eric/narratio/internal/artifactmodel"
	"gitea.maximumdirect.net/eric/narratio/internal/artifactpolicy"
	"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
}

type assertExtractionSourcesStage struct {
	keys []string
	runs *int
}

func (s assertExtractionSourcesStage) Name() string { return "analyze" }

func (s assertExtractionSourcesStage) Run(_ context.Context, env *stage.Env, m *manifest.Manifest) (*stage.StageResult, error) {
	definitions := artifacts.ExtractionDefinitionsFromConfig(env.Config.Pipeline.Notarius)
	catalog, err := artifacts.BootstrapRuntimeCatalog(nil, env.EffectiveArtifacts, definitions)
	if err != nil {
		return nil, err
	}
	paths, err := env.ArtifactStore.EnsureLayoutFor(env.Config.Session.Campaign, env.Config.Session.SessionID)
	if err != nil {
		return nil, err
	}
	catalog.HydrateExtractionArtifacts(paths, m, definitions)
	for _, key := range s.keys {
		sourceID := artifacts.ExtractionArtifactSourceID(key)
		entry, ok := catalog.Lookup(sourceID)
		if !ok || !entry.Available || entry.SourceID != sourceID || entry.Path == "" {
			return nil, fmt.Errorf("extraction source %q unavailable: %#v, present=%v", sourceID, entry, ok)
		}
	}
	if s.runs != nil {
		*s.runs = *s.runs + 1
	}
	return &stage.StageResult{}, nil
}

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"):    `{"schema_version":"notarius.warnings.v2","group_count":0,"occurrence_count":0,"groups":[]}`,
		filepath.Join(bundle, "diagnostics.json"): `{"schema_version":"notarius.diagnostics.v1","group_count":0,"occurrence_count":0,"truncated":false,"unrepresented_occurrence_count":0,"groups":[]}`,
	} {
		if err := os.WriteFile(path, []byte(content), 0o644); err != nil {
			return notarius.RunResult{}, err
		}
	}
	keys := make([]string, 0, len(r.cfg.Outputs))
	for key := range r.cfg.Outputs {
		keys = append(keys, key)
	}
	sort.Strings(keys)
	lanes := make([]notarius.LaneDescriptor, 0, len(keys))
	for _, key := range keys {
		output := r.cfg.Outputs[key]
		filename := key + ".json"
		path := filepath.Join(lanesDir, filename)
		if err := os.WriteFile(path, []byte(`{"records":[]}`), 0o644); err != nil {
			return notarius.RunResult{}, err
		}
		lanes = append(lanes, notarius.LaneDescriptor{
			LaneID: output.LaneID, File: filepath.ToSlash(filepath.Join("lanes", filename)), Path: path,
			MediaType: output.MediaType, SchemaID: output.SchemaID,
			SchemaVersion: output.SchemaVersion, ModuleKey: output.ModuleKey,
		})
	}
	return notarius.RunResult{
		Receipt: notarius.Receipt{
			SchemaVersion: notarius.ReceiptSchemaVersion, RunID: externalRunID,
			PipelineID: req.PipelineID, OutputDirectory: bundle, IndexFile: "index.json",
			NormalizedOutputCount: len(lanes), ValidationStatus: "approved",
		},
		BundleRoot: bundle,
		Index: notarius.Index{
			Path: filepath.Join(bundle, "index.json"), RejectedPath: filepath.Join(bundle, "rejected.json"),
			WarningsPath:    filepath.Join(bundle, "warnings.json"),
			DiagnosticsPath: filepath.Join(bundle, "diagnostics.json"),
			Lanes:           lanes,
		},
	}, 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 TestExtractLifecyclePrepareBindsVerifiedReferenceSnapshots(t *testing.T) {
	cfg, env, runner := extractionLifecycleFixture(t, true)
	originalPaths := configureLifecycleReferences(t, cfg)

	summary, err := executeStages(context.Background(), cfg, prepareExtractLifecyclePlan(t), RunOptions{Env: env})
	if err != nil {
		t.Fatalf("executeStages() error = %v", err)
	}
	if len(summary.Executed) != 3 || len(runner.requests) != 1 {
		t.Fatalf("summary = %#v requests=%d", summary, len(runner.requests))
	}
	paths, err := artifacts.NewLocalStore(cfg.Pipeline.Workspace.Root).EnsureLayoutFor(cfg.Session.Campaign, cfg.Session.SessionID)
	if err != nil {
		t.Fatalf("EnsureLayoutFor() error = %v", err)
	}
	want := []struct {
		selector string
		sourceID string
		filename string
	}{
		{selector: "glossary", sourceID: artifactpolicy.SourceInputGlossary, filename: "glossary.yml"},
		{selector: "party", sourceID: artifactpolicy.SourceInputParty, filename: "party.yml"},
		{selector: "players", sourceID: artifactpolicy.SourceInputPlayers, filename: "players.yml"},
		{selector: "spells", sourceID: artifactpolicy.SourceInputSpellCatalog, filename: "spell_catalog.json"},
	}
	request := runner.requests[0]
	if len(request.References) != len(want) {
		t.Fatalf("references = %#v", request.References)
	}
	loaded := loadLifecycleManifest(t, cfg)
	for index, expected := range want {
		binding := request.References[index]
		canonical := filepath.Join(paths.InputsDir, expected.filename)
		descriptor, ok := artifactpolicy.DescribePreparedInputSource(expected.sourceID)
		if !ok {
			t.Fatalf("DescribePreparedInputSource(%q) ok = false", expected.sourceID)
		}
		snapshot, err := artifacts.SessionNotariusReferenceSnapshotPath(
			paths, lifecycleInputChecksum(t, loaded, descriptor.ManifestKind), expected.filename,
		)
		if err != nil {
			t.Fatalf("SessionNotariusReferenceSnapshotPath(%q) error = %v", expected.sourceID, err)
		}
		if binding.Selector != expected.selector || binding.Path != snapshot || binding.Path == canonical || binding.Path == originalPaths[expected.sourceID] {
			t.Fatalf("reference[%d] = %#v, want selector %q snapshot %q and not prepared/source paths", index, binding, expected.selector, snapshot)
		}
	}

	extract := loaded.Stages["extract"]
	if extract == nil || extract.Status != manifest.StatusSucceeded || extract.Metadata["reference_count"] != float64(len(want)) {
		t.Fatalf("extract record = %#v", extract)
	}
	references, ok := extract.Metadata["references"].([]any)
	if !ok || len(references) != len(want) || len(references) > config.MaxNotariusReferenceBindings {
		t.Fatalf("reference metadata = %#v", extract.Metadata["references"])
	}
	for index, raw := range references {
		entry, ok := raw.(map[string]any)
		if !ok || len(entry) != 5 || entry["selector"] != want[index].selector || entry["source_id"] != want[index].sourceID {
			t.Fatalf("reference metadata[%d] = %#v", index, raw)
		}
	}
}

func TestExtractLifecyclePreparedReferenceChangeRerunsExtractionAndInvalidatesDownstream(t *testing.T) {
	cfg, env, runner := extractionLifecycleFixture(t, true)
	configureLifecycleReferences(t, cfg)
	if _, err := executeStages(context.Background(), cfg, prepareExtractLifecyclePlan(t), RunOptions{Env: env}); err != nil {
		t.Fatalf("initial executeStages() error = %v", err)
	}
	before := loadLifecycleManifest(t, cfg)
	beforeChecksum := lifecycleInputChecksum(t, before, "party")
	for _, name := range []string{"render", "analyze", "publish"} {
		before.MarkStageSucceeded(name, time.Now().UTC(), nil)
	}
	if err := (&manifest.LocalStore{}).Save(context.Background(), manifestPathFor(cfg), before); err != nil {
		t.Fatalf("Save(downstream success) error = %v", err)
	}
	partyDocument, err := config.ParseParty([]byte(`schema_version: narratio.party.v1
characters:
  changed:
    player: {name: Eric}
    character:
      name: Changed
      classes: [{name: wizard}]
`))
	if err != nil {
		t.Fatalf("ParseParty(changed party) error = %v", err)
	}
	cfg.Party.Canonical = partyDocument

	prepared := loadLifecycleManifest(t, cfg)
	prepare, err := stage.Select("prepare")
	if err != nil {
		t.Fatalf("stage.Select(prepare) error = %v", err)
	}
	if _, err := prepare.Run(context.Background(), env, prepared); err != nil {
		t.Fatalf("prepare.Run() error = %v", err)
	}
	if lifecycleInputChecksum(t, prepared, "party") == beforeChecksum {
		t.Fatal("prepared party checksum did not change")
	}
	if err := (&manifest.LocalStore{}).Save(context.Background(), manifestPathFor(cfg), prepared); err != nil {
		t.Fatalf("Save(reprepared manifest) error = %v", err)
	}

	plan, err := BuildSingleStagePlan("extract")
	if err != nil {
		t.Fatalf("BuildSingleStagePlan(extract) error = %v", err)
	}
	run, err := executeStages(context.Background(), cfg, plan, RunOptions{Env: env})
	if err != nil {
		t.Fatalf("rerun executeStages() error = %v", err)
	}
	if len(run.Executed) != 1 || len(run.Skipped) != 0 || len(runner.requests) != 2 {
		t.Fatalf("rerun summary = %#v requests=%d", run, len(runner.requests))
	}
	after := loadLifecycleManifest(t, cfg)
	if after.Stages["extract"].Status != manifest.StatusSucceeded {
		t.Fatalf("extract status = %#v", after.Stages["extract"])
	}
	if after.Stages["render"] == nil || after.Stages["render"].Status != manifest.StatusSucceeded {
		t.Fatalf("render status = %#v, want succeeded sibling", after.Stages["render"])
	}
	for _, name := range []string{"analyze", "publish"} {
		if after.Stages[name] == nil || after.Stages[name].Status != manifest.StatusStale {
			t.Fatalf("%s status = %#v, want stale", name, after.Stages[name])
		}
	}
}

func TestExtractLifecycleCanonicalPartyBytesReachCanonicalReference(t *testing.T) {
	cfg, env, runner := extractionLifecycleFixture(t, true)
	configureLifecycleReferences(t, cfg)
	partyBytes := []byte(`schema_version: narratio.party.v1
characters:
  session:
    player: {name: Eric}
    character:
      name: Session
      classes: [{name: wizard}]
`)
	partyDocument, err := config.ParseParty(partyBytes)
	if err != nil {
		t.Fatalf("ParseParty(canonical party) error = %v", err)
	}
	cfg.Party.Canonical = partyDocument

	if _, err := executeStages(context.Background(), cfg, prepareExtractLifecyclePlan(t), RunOptions{Env: env}); err != nil {
		t.Fatalf("executeStages() error = %v", err)
	}
	if len(runner.requests) != 1 {
		t.Fatalf("requests = %d", len(runner.requests))
	}
	var partyPath string
	for _, binding := range runner.requests[0].References {
		if binding.Selector == "party" {
			partyPath = binding.Path
		}
	}
	contents, err := os.ReadFile(partyPath)
	if err != nil {
		t.Fatalf("ReadFile(prepared party) error = %v", err)
	}
	if string(contents) != string(partyBytes) || partyPath == cfg.Party.Source.Path {
		t.Fatalf("prepared party path=%q contents=%q source=%q", partyPath, contents, cfg.Party.Source.Path)
	}
}

func TestExtractLifecycleEmptyReferencesPreserveAllDndExtractionSources(t *testing.T) {
	cfg, env, runner := extractionLifecycleFixture(t, true)
	cfg.Pipeline.Notarius.Outputs = lifecycleDndOutputs()
	keys := make([]string, 0, len(cfg.Pipeline.Notarius.Outputs))
	for key := range cfg.Pipeline.Notarius.Outputs {
		keys = append(keys, key)
	}
	sort.Strings(keys)
	analyzeRuns := 0
	extractPlan, err := BuildSingleStagePlan("extract")
	if err != nil {
		t.Fatalf("BuildSingleStagePlan(extract) error = %v", err)
	}
	plan := append(extractPlan, assertExtractionSourcesStage{keys: keys, runs: &analyzeRuns})

	summary, err := executeStages(context.Background(), cfg, plan, RunOptions{Env: env})
	if err != nil {
		t.Fatalf("executeStages() error = %v", err)
	}
	if len(summary.Executed) != 2 || len(runner.requests) != 1 || len(runner.requests[0].References) != 0 || analyzeRuns != 1 {
		t.Fatalf("summary=%#v requests=%#v analyze=%d", summary, runner.requests, analyzeRuns)
	}
	loaded := loadLifecycleManifest(t, cfg)
	if got := len(loaded.Stages["extract"].Outputs); got != len(keys)+1 {
		t.Fatalf("extract outputs = %d, want %d lanes plus index", got, len(keys))
	}
}

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 TestExtractLifecycleRetryReusesStableSnapshotsAndChangedReferenceSelectsNewPath(t *testing.T) {
	cfg, env, runner := extractionLifecycleFixture(t, true)
	configureLifecycleReferences(t, cfg)
	runner.failuresRemaining = 1
	plan := prepareExtractLifecyclePlan(t)

	if _, err := executeStages(context.Background(), cfg, plan, RunOptions{Env: env}); err == nil {
		t.Fatal("first executeStages() error = nil, want extraction failure")
	}
	if len(runner.requests) != 1 {
		t.Fatalf("failed requests = %d, want 1", len(runner.requests))
	}
	firstPaths := lifecycleReferencePaths(runner.requests[0])
	if _, err := executeStages(context.Background(), cfg, plan, RunOptions{Env: env}); err != nil {
		t.Fatalf("retry executeStages() error = %v", err)
	}
	if len(runner.requests) != 2 {
		t.Fatalf("retry requests = %d, want 2", len(runner.requests))
	}
	if secondPaths := lifecycleReferencePaths(runner.requests[1]); !reflect.DeepEqual(secondPaths, firstPaths) {
		t.Fatalf("snapshot paths changed on retry:\nfirst %#v\nsecond %#v", firstPaths, secondPaths)
	}

	partyDocument, err := config.ParseParty([]byte(`schema_version: narratio.party.v1
characters:
  changed:
    player: {name: Eric}
    character:
      name: Changed
      classes: [{name: wizard}]
`))
	if err != nil {
		t.Fatalf("ParseParty(changed party) error = %v", err)
	}
	cfg.Party.Canonical = partyDocument
	persisted := loadLifecycleManifest(t, cfg)
	prepare, err := stage.Select("prepare")
	if err != nil {
		t.Fatalf("stage.Select(prepare) error = %v", err)
	}
	if _, err := prepare.Run(context.Background(), env, persisted); err != nil {
		t.Fatalf("prepare.Run() error = %v", err)
	}
	if err := (&manifest.LocalStore{}).Save(context.Background(), manifestPathFor(cfg), persisted); err != nil {
		t.Fatalf("Save(reprepared manifest) error = %v", err)
	}
	extractPlan, _ := BuildSingleStagePlan("extract")
	if _, err := executeStages(context.Background(), cfg, extractPlan, RunOptions{Env: env}); err != nil {
		t.Fatalf("changed-reference executeStages() error = %v", err)
	}
	if len(runner.requests) != 3 {
		t.Fatalf("changed-reference requests = %d, want 3", len(runner.requests))
	}
	thirdPaths := lifecycleReferencePaths(runner.requests[2])
	if thirdPaths["party"] == firstPaths["party"] {
		t.Fatalf("changed party retained snapshot path %q", thirdPaths["party"])
	}
	if thirdPaths["players"] == firstPaths["players"] {
		t.Fatalf("changed party retained derived players snapshot path %q", thirdPaths["players"])
	}
	for selector, firstPath := range firstPaths {
		if selector != "party" && selector != "players" && thirdPaths[selector] != firstPath {
			t.Fatalf("unchanged %s snapshot path = %q, want %q", selector, thirdPaths[selector], firstPath)
		}
		if _, err := os.Stat(firstPath); err != nil {
			t.Fatalf("old %s snapshot %q was not retained: %v", selector, firstPath, err)
		}
	}
}

func TestExtractLifecycleCacheControlsAndSnapshotLossDoNotInvalidateCompleteResult(t *testing.T) {
	cfg, env, runner := extractionLifecycleFixture(t, true)
	configureLifecycleReferences(t, cfg)
	if _, err := executeStages(context.Background(), cfg, prepareExtractLifecyclePlan(t), RunOptions{Env: env}); err != nil {
		t.Fatalf("initial executeStages() error = %v", err)
	}
	paths := artifacts.NewLocalStore(cfg.Pipeline.Workspace.Root).SessionPathsFor(cfg.Session.Campaign, cfg.Session.SessionID)
	cacheRoot := artifacts.SessionNotariusReferenceCacheDir(paths)
	if err := os.RemoveAll(cacheRoot); err != nil {
		t.Fatalf("RemoveAll(snapshot cache) error = %v", err)
	}
	cfg.Pipeline.Notarius.Resume = true
	cfg.Pipeline.Notarius.ChunkCache = "bypass"
	extractPlan, _ := BuildSingleStagePlan("extract")
	summary, err := executeStages(context.Background(), cfg, extractPlan, RunOptions{Env: env})
	if err != nil {
		t.Fatalf("resume executeStages() error = %v", err)
	}
	if len(summary.Executed) != 0 || !reflect.DeepEqual(summary.Skipped, []string{"extract"}) || len(runner.requests) != 1 {
		t.Fatalf("resume summary = %#v requests=%d", summary, len(runner.requests))
	}
	if _, err := os.Stat(cacheRoot); !errors.Is(err, os.ErrNotExist) {
		t.Fatalf("skipped extraction recreated snapshot cache: %v", err)
	}
}

func TestAnalyzeOnlyLifecyclePreservesExtractionWithoutInvokingNotarius(t *testing.T) {
	cfg, env, runner := extractionLifecycleFixture(t, true)
	extractPlan, _ := BuildSingleStagePlan("extract")
	if _, err := executeStages(context.Background(), cfg, extractPlan, RunOptions{Env: env}); err != nil {
		t.Fatalf("extract executeStages() error = %v", err)
	}
	before := loadLifecycleManifest(t, cfg).Stages["extract"]
	beforeJSON, _ := json.Marshal(before)
	analyzeRuns := 0
	if _, err := executeStages(context.Background(), cfg, []stage.Stage{countingStage{name: "analyze", runs: &analyzeRuns}}, RunOptions{Env: env, Force: true}); err != nil {
		t.Fatalf("analyze-only executeStages() error = %v", err)
	}
	after := loadLifecycleManifest(t, cfg).Stages["extract"]
	afterJSON, _ := json.Marshal(after)
	if analyzeRuns != 1 || len(runner.requests) != 1 || string(afterJSON) != string(beforeJSON) {
		t.Fatalf("analyze-only lifecycle changed extraction: analyze=%d requests=%d\nbefore=%s\nafter=%s", analyzeRuns, len(runner.requests), beforeJSON, afterJSON)
	}
	for _, output := range after.Outputs {
		if _, err := os.Stat(output.LocalPath); err != nil {
			t.Fatalf("analyze-only lifecycle removed extract output %q: %v", output.LocalPath, err)
		}
	}
}

func TestForcedArtifactRegenerationUsesOrdinaryExtractAndAnalyzePath(t *testing.T) {
	cfg, env, runner := extractionLifecycleFixture(t, true)
	cfg.Pipeline.Notarius.Resume = true
	cfg.Pipeline.Notarius.ChunkCache = "bypass"
	analyzeRuns := 0
	plan := extractionLifecyclePlan(t, &analyzeRuns)
	if _, err := executeStages(context.Background(), cfg, plan, RunOptions{Env: env, Force: true, SelectedArtifacts: []string{"session_recap"}}); err != nil {
		t.Fatalf("forced artifact regeneration error = %v", err)
	}
	if len(runner.requests) != 1 || runner.requests[0].Resume || runner.requests[0].ChunkCache != "refresh" || analyzeRuns != 1 {
		t.Fatalf("forced path request=%#v analyze=%d", runner.requests, analyzeRuns)
	}
}

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")

	_, 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)
	}
	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 TestExtractLifecycleCredentialChangesRemainResumable(t *testing.T) {
	for _, test := range []struct {
		name   string
		before func(*testing.T) *configkit.Secret
		after  func(*testing.T) *configkit.Secret
	}{
		{
			name: "added",
			after: func(t *testing.T) *configkit.Secret {
				return loadLifecycleNotariusCredential(t, "NARRATIO_LIFECYCLE_ADDED_KEY", "added-value")
			},
		},
		{
			name: "removed",
			before: func(t *testing.T) *configkit.Secret {
				return loadLifecycleNotariusCredential(t, "NARRATIO_LIFECYCLE_REMOVED_KEY", "removed-value")
			},
		},
		{
			name: "renamed",
			before: func(t *testing.T) *configkit.Secret {
				return loadLifecycleNotariusCredential(t, "NARRATIO_LIFECYCLE_FIRST_KEY", "first-value")
			},
			after: func(t *testing.T) *configkit.Secret {
				return loadLifecycleNotariusCredential(t, "NARRATIO_LIFECYCLE_SECOND_KEY", "second-value")
			},
		},
		{
			name: "value rotated",
			before: func(t *testing.T) *configkit.Secret {
				return loadLifecycleNotariusCredential(t, "NARRATIO_LIFECYCLE_ROTATED_KEY", "first-value")
			},
			after: func(t *testing.T) *configkit.Secret {
				return loadLifecycleNotariusCredential(t, "NARRATIO_LIFECYCLE_ROTATED_KEY", "second-value")
			},
		},
	} {
		t.Run(test.name, func(t *testing.T) {
			cfg, env, runner := extractionLifecycleFixture(t, true)
			if test.before != nil {
				cfg.Pipeline.Notarius.LLMAPIKey = test.before(t)
			}
			analyzeRuns := 0
			plan := extractionLifecyclePlan(t, &analyzeRuns)
			if _, err := executeStages(context.Background(), cfg, plan, RunOptions{Env: env}); err != nil {
				t.Fatalf("initial executeStages() error = %v", err)
			}
			cfg.Pipeline.Notarius.LLMAPIKey = nil
			if test.after != nil {
				cfg.Pipeline.Notarius.LLMAPIKey = test.after(t)
			}
			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("credential-only change reexecuted work: summary=%#v requests=%d analyze=%d", resumed, len(runner.requests), analyzeRuns)
			}
			loaded := loadLifecycleManifest(t, cfg)
			if loaded.Stages["extract"].Status != manifest.StatusSucceeded || loaded.Stages["analyze"].Status != manifest.StatusSucceeded {
				t.Fatalf("credential-only change invalidated stages: extract=%q analyze=%q", loaded.Stages["extract"].Status, loaded.Stages["analyze"].Status)
			}
		})
	}
}

func TestExtractLifecycleConfigProfileChangesRerunExtractionAndDependents(t *testing.T) {
	cfg, env, runner := extractionLifecycleFixture(t, true)
	selected := "inexpensive"
	cfg.Pipeline.Notarius.ConfigProfile = &selected
	analyzeRuns := 0
	plan := extractionLifecyclePlan(t, &analyzeRuns)

	if _, err := executeStages(context.Background(), cfg, plan, RunOptions{Env: env}); err != nil {
		t.Fatalf("initial executeStages() error = %v", err)
	}
	if len(runner.requests) != 1 || runner.requests[0].ConfigProfile == nil || *runner.requests[0].ConfigProfile != selected || analyzeRuns != 1 {
		t.Fatalf("initial requests=%#v analyze=%d, want selected extraction and analysis", runner.requests, analyzeRuns)
	}
	reused, err := executeStages(context.Background(), cfg, plan, RunOptions{Env: env})
	if err != nil {
		t.Fatalf("reuse executeStages() error = %v", err)
	}
	if len(reused.Executed) != 0 || len(runner.requests) != 1 || analyzeRuns != 1 {
		t.Fatalf("same-profile reuse summary=%#v requests=%d analyze=%d", reused, len(runner.requests), analyzeRuns)
	}

	markLifecycleStageSucceeded(t, cfg, "render")
	markLifecycleStageSucceeded(t, cfg, "publish")
	before := loadLifecycleManifest(t, cfg)
	trimBefore, err := json.Marshal(before.Stages["trim"])
	if err != nil {
		t.Fatal(err)
	}

	changed := "experimental"
	cfg.Pipeline.Notarius.ConfigProfile = &changed
	rerun, err := executeStages(context.Background(), cfg, plan, RunOptions{Env: env})
	if err != nil {
		t.Fatalf("changed-profile executeStages() error = %v", err)
	}
	if len(rerun.Executed) != 2 || len(runner.requests) != 2 || runner.requests[1].ConfigProfile == nil || *runner.requests[1].ConfigProfile != changed || analyzeRuns != 2 {
		t.Fatalf("changed-profile summary=%#v requests=%#v analyze=%d", rerun, runner.requests, analyzeRuns)
	}
	after := loadLifecycleManifest(t, cfg)
	trimAfter, err := json.Marshal(after.Stages["trim"])
	if err != nil {
		t.Fatal(err)
	}
	if string(trimAfter) != string(trimBefore) {
		t.Fatalf("profile change altered upstream trim record:\nbefore=%s\nafter=%s", trimBefore, trimAfter)
	}
	if after.Stages["render"].Status != manifest.StatusSucceeded || after.Stages["analyze"].Status != manifest.StatusSucceeded || after.Stages["publish"].Status != manifest.StatusStale {
		t.Fatalf("changed-profile lifecycle render=%#v analyze=%#v publish=%#v", after.Stages["render"], after.Stages["analyze"], after.Stages["publish"])
	}

	cfg.Pipeline.Notarius.ConfigProfile = nil
	returned, err := executeStages(context.Background(), cfg, plan, RunOptions{Env: env})
	if err != nil {
		t.Fatalf("return-to-base executeStages() error = %v", err)
	}
	if len(returned.Executed) != 2 || len(runner.requests) != 3 || runner.requests[2].ConfigProfile != nil || analyzeRuns != 3 {
		t.Fatalf("return-to-base summary=%#v requests=%#v analyze=%d", returned, runner.requests, analyzeRuns)
	}
	final := loadLifecycleManifest(t, cfg)
	if final.Stages["trim"].Status != manifest.StatusSucceeded || final.Stages["render"].Status != manifest.StatusSucceeded {
		t.Fatalf("return-to-base invalidated upstream siblings: trim=%#v render=%#v", final.Stages["trim"], final.Stages["render"])
	}
}

func TestExtractLifecycleConfigProfileFailureClearsCurrentResultAndDependents(t *testing.T) {
	cfg, env, runner := extractionLifecycleFixture(t, true)
	selected := "inexpensive"
	cfg.Pipeline.Notarius.ConfigProfile = &selected
	analyzeRuns := 0
	plan := extractionLifecyclePlan(t, &analyzeRuns)
	if _, err := executeStages(context.Background(), cfg, plan, RunOptions{Env: env}); err != nil {
		t.Fatalf("initial executeStages() error = %v", err)
	}
	markLifecycleStageSucceeded(t, cfg, "render")
	markLifecycleStageSucceeded(t, cfg, "publish")

	changed := "missing-profile"
	cfg.Pipeline.Notarius.ConfigProfile = &changed
	runner.failuresRemaining = 1
	if _, err := executeStages(context.Background(), cfg, plan, RunOptions{Env: env}); err == nil {
		t.Fatal("changed-profile executeStages() error = nil, want downstream failure")
	}
	failed := loadLifecycleManifest(t, cfg)
	extract := failed.Stages["extract"]
	if extract.Status != manifest.StatusFailed || len(extract.Outputs) != 0 || len(extract.Logs) != 0 || len(extract.GeneratedConfigs) != 0 || len(extract.Metadata) != 0 {
		t.Fatalf("failed changed-profile extraction advertised prior success: %#v", extract)
	}
	if failed.Stages["trim"].Status != manifest.StatusSucceeded || failed.Stages["render"].Status != manifest.StatusSucceeded ||
		failed.Stages["analyze"].Status != manifest.StatusStale || failed.Stages["publish"].Status != manifest.StatusStale || analyzeRuns != 1 {
		t.Fatalf("failed changed-profile lifecycle trim=%#v render=%#v analyze=%#v publish=%#v analyzeRuns=%d",
			failed.Stages["trim"], failed.Stages["render"], failed.Stages["analyze"], failed.Stages["publish"], analyzeRuns)
	}
	if len(runner.requests) != 2 || runner.requests[1].ConfigProfile == nil || *runner.requests[1].ConfigProfile != changed {
		t.Fatalf("failed request = %#v, want changed selector", runner.requests)
	}
}

func TestExtractLifecycleRealConfigProfilesSelectNotariusExplicitly(t *testing.T) {
	seed, env, runner := extractionLifecycleFixture(t, true)
	workspace := seed.Pipeline.Workspace.Root
	pipelinePath, campaignPath, sessionPath := writeValidConfigFiles(t, workspace)
	pipelineFile, err := os.OpenFile(pipelinePath, os.O_APPEND|os.O_WRONLY, 0)
	if err != nil {
		t.Fatal(err)
	}
	_, writeErr := pipelineFile.WriteString(`notarius:
  enabled: true
  binary: ` + filepath.Join(workspace, "notarius") + `
  config_path: notarius.yml
  pipeline_id: dnd-session
  timeout: 45m
  outputs:
    npc_registry:
      lane_id: npc-registry
      media_type: application/json
      schema_id: notarius.dnd.npc_registry
      schema_version: v1
      module_key: dnd/npc-registry
`)
	closeErr := pipelineFile.Close()
	if writeErr != nil || closeErr != nil {
		t.Fatalf("write Notarius configuration: %v / %v", writeErr, closeErr)
	}
	if err := os.WriteFile(filepath.Join(filepath.Dir(pipelinePath), "notarius.yml"), []byte("pipelines: {}\n"), 0o644); err != nil {
		t.Fatal(err)
	}
	profilesDir := filepath.Join(filepath.Dir(pipelinePath), "profiles")
	if err := os.MkdirAll(profilesDir, 0o755); err != nil {
		t.Fatal(err)
	}
	for name, contents := range map[string]string{
		"testing":           "whisperx:\n  language: fr\nnotarius:\n  config_profile: downstream-economy\n",
		"narratio-only":     "whisperx:\n  language: de\n",
		"narratio-only-two": "whisperx:\n  language: es\n",
	} {
		if err := os.WriteFile(filepath.Join(profilesDir, name+".yml"), []byte(contents), 0o644); err != nil {
			t.Fatal(err)
		}
	}
	load := func(profile string) *config.Config {
		t.Helper()
		loaded, err := config.LoadWithSessionOptions(pipelinePath, campaignPath, sessionPath, config.SessionLoadOptions{Profile: &profile})
		if err != nil {
			t.Fatalf("load profile %q: %v", profile, err)
		}
		return loaded
	}
	plan, err := BuildSingleStagePlan("extract")
	if err != nil {
		t.Fatal(err)
	}

	testingConfig := load("testing")
	runner.cfg = testingConfig.Pipeline.Notarius
	if _, err := executeStages(context.Background(), testingConfig, plan, RunOptions{Env: env}); err != nil {
		t.Fatalf("execute testing profile: %v", err)
	}
	if len(runner.requests) != 1 || runner.requests[0].ConfigProfile == nil || *runner.requests[0].ConfigProfile != "downstream-economy" {
		t.Fatalf("testing profile request = %#v, want explicit differently named downstream selector", runner.requests)
	}

	narratioOnly := load("narratio-only")
	runner.cfg = narratioOnly.Pipeline.Notarius
	if _, err := executeStages(context.Background(), narratioOnly, plan, RunOptions{Env: env}); err != nil {
		t.Fatalf("execute Narratio-only profile: %v", err)
	}
	if len(runner.requests) != 2 || runner.requests[1].ConfigProfile != nil {
		t.Fatalf("Narratio-only profile request = %#v, want no automatic downstream selector", runner.requests)
	}

	narratioOnlyTwo := load("narratio-only-two")
	runner.cfg = narratioOnlyTwo.Pipeline.Notarius
	reused, err := executeStages(context.Background(), narratioOnlyTwo, plan, RunOptions{Env: env})
	if err != nil {
		t.Fatalf("execute second Narratio-only profile: %v", err)
	}
	if len(reused.Executed) != 0 || len(runner.requests) != 2 {
		t.Fatalf("Narratio profile label/effective digest broadened invalidation: summary=%#v requests=%#v", reused, runner.requests)
	}
}

func TestPreparedPipelineAndManifestExcludeNotariusCredentialValue(t *testing.T) {
	const name = "NARRATIO_LIFECYCLE_PERSISTED_KEY"
	const canary = "notarius-persisted-credential-canary"
	cfg, env, _ := extractionLifecycleFixture(t, true)
	cfg.Pipeline.Notarius.LLMAPIKey = loadLifecycleNotariusCredential(t, name, canary)
	if _, err := executeStages(context.Background(), cfg, prepareExtractLifecyclePlan(t), RunOptions{Env: env}); err != nil {
		t.Fatalf("executeStages() error = %v", err)
	}
	paths := artifacts.NewLocalStore(cfg.Pipeline.Workspace.Root).SessionPathsFor(cfg.Session.Campaign, cfg.Session.SessionID)
	for _, path := range []string{paths.ManifestPath, filepath.Join(paths.InputsDir, "pipeline.resolved.yml")} {
		data, err := os.ReadFile(path)
		if err != nil {
			t.Fatalf("ReadFile(%q) error = %v", path, err)
		}
		if strings.Contains(string(data), canary) {
			t.Fatalf("persisted Narratio file %q contains the credential value", path)
		}
	}
}

func TestExtractLifecycleRerunsAfterDirectTranscriptChange(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)
	}

	persisted := loadLifecycleManifest(t, cfg)
	trimmed := persisted.Stages["trim"].Outputs[0].LocalPath
	if err := os.WriteFile(trimmed, []byte(`{"segments":[{"id":"changed"}]}`), 0o644); err != nil {
		t.Fatalf("WriteFile(trimmed transcript) error = %v", err)
	}
	if err := (&manifest.LocalStore{}).Save(context.Background(), manifestPathFor(cfg), persisted); err != nil {
		t.Fatalf("Save(mutated) error = %v", err)
	}

	_, err := executeStages(context.Background(), cfg, plan, RunOptions{Env: env})
	if err == nil || !strings.Contains(err.Error(), "checksum mismatch") {
		t.Fatalf("rerun executeStages() error = %v, want stale transcript evidence rejection", err)
	}
	if len(runner.requests) != 1 || analyzeRuns != 1 {
		t.Fatalf("stale transcript should prevent extraction, requests=%d analyze=%d", 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 prepareExtractLifecyclePlan(t *testing.T) []stage.Stage {
	t.Helper()
	prepare, err := BuildSingleStagePlan("prepare")
	if err != nil {
		t.Fatalf("BuildSingleStagePlan(prepare) error = %v", err)
	}
	extract, err := BuildSingleStagePlan("extract")
	if err != nil {
		t.Fatalf("BuildSingleStagePlan(extract) error = %v", err)
	}
	return append(append(prepare, lifecycleTrimEvidenceStage{}), extract...)
}

type lifecycleTrimEvidenceStage struct{}

func (lifecycleTrimEvidenceStage) Name() string { return "trim" }

func (lifecycleTrimEvidenceStage) Run(_ context.Context, env *stage.Env, m *manifest.Manifest) (*stage.StageResult, error) {
	paths := artifacts.NewLocalStore(env.Config.Pipeline.Workspace.Root).SessionPathsFor(m.Campaign, m.SessionID)
	path := filepath.Join(paths.ArtifactsDir, "trimmed.from-manifest.json")
	checksum, err := artifacts.SHA256File(path)
	if err != nil {
		return nil, err
	}
	return &stage.StageResult{Outputs: []artifacts.Ref{{
		Kind: artifactmodel.TranscriptOutputKindFinalTrimmed, SourceID: artifactmodel.SourceTranscriptFinalTrimmed,
		AbsolutePath: path, Checksum: checksum,
	}}}, nil
}

func configureLifecycleReferences(t *testing.T, cfg *config.Config) map[string]string {
	t.Helper()
	cfg.Pipeline.Notarius.References = map[string]string{
		"party":    artifactpolicy.SourceInputParty,
		"players":  artifactpolicy.SourceInputPlayers,
		"glossary": artifactpolicy.SourceInputGlossary,
		"spells":   artifactpolicy.SourceInputSpellCatalog,
	}
	spellPath := filepath.Join(filepath.Dir(cfg.CampaignPath), "spells.json")
	if err := os.WriteFile(spellPath, []byte(`{"spells":[]}`+"\n"), 0o644); err != nil {
		t.Fatalf("WriteFile(spell catalog) error = %v", err)
	}
	cfg.StableInputs.SpellCatalogFile = config.ResolvedInputFile{
		Path: "./spells.json", ConfigPath: cfg.CampaignPath, Source: "campaign_config",
	}
	cfg.Session.Inputs.SpellCatalogFile = "./spells.json"

	return map[string]string{
		artifactpolicy.SourceInputParty:        filepath.Join(filepath.Dir(cfg.CampaignPath), "party.yml"),
		artifactpolicy.SourceInputPlayers:      filepath.Join(filepath.Dir(cfg.CampaignPath), "players.yml"),
		artifactpolicy.SourceInputGlossary:     filepath.Join(filepath.Dir(cfg.CampaignPath), "glossary.yml"),
		artifactpolicy.SourceInputSpellCatalog: spellPath,
	}
}

func lifecycleInputChecksum(t *testing.T, m *manifest.Manifest, kind string) string {
	t.Helper()
	for _, input := range m.Inputs {
		if input.Kind == kind {
			if strings.TrimSpace(input.Checksum) == "" {
				t.Fatalf("input %q has no checksum: %#v", kind, input)
			}
			return input.Checksum
		}
	}
	t.Fatalf("manifest input %q not found: %#v", kind, m.Inputs)
	return ""
}

func lifecycleReferencePaths(request notarius.RunRequest) map[string]string {
	paths := make(map[string]string, len(request.References))
	for _, reference := range request.References {
		paths[reference.Selector] = reference.Path
	}
	return paths
}

func lifecycleDndOutputs() map[string]config.NotariusOutputConfig {
	return map[string]config.NotariusOutputConfig{
		"item_registry":        {LaneID: "item-registry", MediaType: "application/json", SchemaID: "notarius.dnd.item_registry", SchemaVersion: "v1", ModuleKey: "dnd/item-registry"},
		"npc_registry":         {LaneID: "npc-registry", MediaType: "application/json", SchemaID: "notarius.dnd.npc_registry", SchemaVersion: "v1", ModuleKey: "dnd/npc-registry"},
		"location_registry":    {LaneID: "location-registry", MediaType: "application/json", SchemaID: "notarius.dnd.location_registry", SchemaVersion: "v1", ModuleKey: "dnd/location-registry"},
		"scene_descriptions":   {LaneID: "scene-descriptions", MediaType: "application/json", SchemaID: "notarius.dnd.scene_descriptions", SchemaVersion: "v1", ModuleKey: "dnd/scene-descriptions"},
		"item_occurrences":     {LaneID: "item-occurrences", MediaType: "application/json", SchemaID: "notarius.dnd.item_occurrences", SchemaVersion: "v1", ModuleKey: "dnd/item-occurrences"},
		"spells":               {LaneID: "spells", MediaType: "application/json", SchemaID: "notarius.dnd.spells", SchemaVersion: "v1", ModuleKey: "dnd/spells"},
		"combat_turns":         {LaneID: "combat-turns", MediaType: "application/json", SchemaID: "notarius.dnd.combat_turns", SchemaVersion: "v1", ModuleKey: "dnd/combat-turns"},
		"npc_occurrences":      {LaneID: "npc-occurrences", MediaType: "application/json", SchemaID: "notarius.dnd.npc_occurrences", SchemaVersion: "v1", ModuleKey: "dnd/npc-occurrences"},
		"location_occurrences": {LaneID: "location-occurrences", MediaType: "application/json", SchemaID: "notarius.dnd.location_occurrences", SchemaVersion: "v1", ModuleKey: "dnd/location-occurrences"},
		"enemy_events":         {LaneID: "enemy-events", MediaType: "application/json", SchemaID: "notarius.dnd.enemy_events", SchemaVersion: "v1", ModuleKey: "dnd/enemy-events"},
	}
}

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
	inputChecksum, err := artifacts.SHA256File(inputPath)
	if err != nil {
		t.Fatalf("SHA256File(input) error = %v", err)
	}
	m.MarkStageSucceeded("trim", time.Now().UTC(), []manifest.ArtifactRecord{{
		Kind: artifactmodel.TranscriptOutputKindFinalTrimmed, SourceID: artifactmodel.SourceTranscriptFinalTrimmed,
		LocalPath: inputPath, ProducerRunID: "trim-run", Checksum: inputChecksum,
	}})
	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
}

func loadLifecycleNotariusCredential(t *testing.T, name, value string) *configkit.Secret {
	t.Helper()
	t.Setenv(name, value)
	path := filepath.Join(t.TempDir(), "pipeline.yml")
	if err := os.WriteFile(path, []byte("whisperx:\n  transcribe_url: https://transcription.example.test/transcribe\nnotarius:\n  enabled: false\n  llm_api_key: "+name+"\n"), 0o644); err != nil {
		t.Fatal(err)
	}
	pipeline, err := config.LoadPipeline(path)
	if err != nil {
		t.Fatal(err)
	}
	return pipeline.Notarius.LLMAPIKey
}