Prove prepared reference extraction lifecycle

docs/roadmap/implementation.md

@@ -21,7 +21,7 @@ different contract.
 | 3 | Centralize manifest-authoritative prepared-input resolution and migrate analyze to it. | Completed |
 | 4 | Add deterministic Notarius v0.6 reference arguments at the subprocess adapter boundary. | Completed |
 | 5 | Resolve references in extract and bind fingerprints, resume, and metadata to their identities. | Completed |
-| 6 | Prove assembled extraction lifecycle and downstream invalidation behavior. | Pending |
+| 6 | Prove assembled extraction lifecycle and downstream invalidation behavior. | Completed |
 | 7 | Update canonical documentation and maintained examples for the completed feature. | Pending |
 | 8 | Perform compatibility, quality, and repository-wide closure validation. | Pending |
 

internal/app/extract_lifecycle_test.go

@@ -7,12 +7,14 @@ import (
 	"fmt"
 	"os"
 	"path/filepath"
+	"sort"
 	"strings"
 	"testing"
 	"time"
 
 	"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"
@@ -25,6 +27,37 @@ type materializingNotariusRunner struct {
 	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 {
@@ -43,29 +76,42 @@ func (r *materializingNotariusRunner) Run(_ context.Context, req notarius.RunReq
 		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":[]}`,
-		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"]
+	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: 1, ValidationStatus: "approved",
+			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: []notarius.LaneDescriptor{{
+			Lanes:           lanes,
-				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
 }
@@ -102,6 +148,176 @@ func TestExtractLifecycleDisabledThenEnabled(t *testing.T) {
 	}
 }
 
+func TestExtractLifecyclePrepareBindsCanonicalPreparedReferences(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) != 2 || 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)
+	}
+	for index, expected := range want {
+		binding := request.References[index]
+		canonical := filepath.Join(paths.InputsDir, expected.filename)
+		if binding.Selector != expected.selector || binding.Path != canonical || binding.Path == originalPaths[expected.sourceID] {
+			t.Fatalf("reference[%d] = %#v, want selector %q canonical %q and not source %q", index, binding, expected.selector, canonical, originalPaths[expected.sourceID])
+		}
+	}
+
+	loaded := loadLifecycleManifest(t, cfg)
+	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)
+	originalPaths := 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)
+	}
+	if err := os.WriteFile(originalPaths[artifactpolicy.SourceInputParty], []byte("changed party bytes\n"), 0o644); err != nil {
+		t.Fatalf("WriteFile(party source) error = %v", err)
+	}
+
+	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"])
+	}
+	for _, name := range []string{"render", "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 TestExtractLifecycleSessionOverrideBytesReachCanonicalReference(t *testing.T) {
+	cfg, env, runner := extractionLifecycleFixture(t, true)
+	configureLifecycleReferences(t, cfg)
+	overridePath := filepath.Join(filepath.Dir(cfg.SessionPath), "session-party.yml")
+	if err := os.WriteFile(overridePath, []byte("session override party\n"), 0o644); err != nil {
+		t.Fatalf("WriteFile(session override) error = %v", err)
+	}
+	cfg.StableInputs.PartyFile = config.ResolvedInputFile{
+		Path: "./session-party.yml", ConfigPath: cfg.SessionPath, Source: "session_config",
+	}
+	cfg.Session.Inputs.PartyFile = "./session-party.yml"
+
+	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) != "session override party\n" || partyPath == overridePath {
+		t.Fatalf("prepared party path=%q contents=%q override=%q", partyPath, contents, overridePath)
+	}
+}
+
+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
@@ -396,6 +612,73 @@ func markLifecycleStageSucceeded(t *testing.T, cfg *config.Config, name string)
 	}
 }
 
+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(prepare, extract...)
+}
+
+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 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)