From ef8dae776ee532a32e2ed8c0efcbb152eef34b26 Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Mon, 10 Aug 2026 01:50:31 +0000 Subject: [PATCH] Enforce canonical Notarius bundle paths --- docs/roadmap/implementation.md | 2 +- internal/adapters/notarius/subprocess.go | 35 +++++++--- internal/adapters/notarius/subprocess_test.go | 48 +++++++++++-- internal/stage/extract_test.go | 67 ++++++++++++++++++- 4 files changed, 138 insertions(+), 14 deletions(-) diff --git a/docs/roadmap/implementation.md b/docs/roadmap/implementation.md index 563d0db..56ae592 100644 --- a/docs/roadmap/implementation.md +++ b/docs/roadmap/implementation.md @@ -29,7 +29,7 @@ implementation sequence. | Stage 9 | Complete | | Stage 10 | Complete | | Stage 11 | Complete | -| Stage 12 | Pending | +| Stage 12 | Complete | | Stage 13 | Pending | | Stage 14 | Pending | diff --git a/internal/adapters/notarius/subprocess.go b/internal/adapters/notarius/subprocess.go index 028641f..0a8603c 100644 --- a/internal/adapters/notarius/subprocess.go +++ b/internal/adapters/notarius/subprocess.go @@ -15,9 +15,13 @@ import ( ) const ( - maxReceiptBytes = 1 << 20 - maxIndexBytes = 4 << 20 - maxSummaryBytes = 4 << 20 + maxReceiptBytes = 1 << 20 + maxIndexBytes = 4 << 20 + maxSummaryBytes = 4 << 20 + canonicalIndexFile = "index.json" + canonicalManifestFile = "manifest.json" + canonicalRejectedFile = "rejected.json" + canonicalWarningsFile = "warnings.json" ) type subprocessRun func(context.Context, subprocess.RunRequest) (subprocess.RunResult, error) @@ -170,11 +174,14 @@ func loadReceipt(path, pipelineID string) (Receipt, error) { return Receipt{}, fmt.Errorf("unsupported notarius receipt schema version %q", document.SchemaVersion) } if strings.TrimSpace(document.RunID) == "" || strings.TrimSpace(document.PipelineID) == "" || - strings.TrimSpace(document.OutputDirectory) == "" || strings.TrimSpace(document.IndexFile) == "" || - strings.TrimSpace(document.ValidationStatus) == "" || document.NormalizedOutputCount == nil || + strings.TrimSpace(document.OutputDirectory) == "" || strings.TrimSpace(document.ValidationStatus) == "" || + document.NormalizedOutputCount == nil || document.RejectedOutputCount == nil || document.WarningCount == nil { return Receipt{}, fmt.Errorf("notarius receipt is missing required fields") } + if document.IndexFile != canonicalIndexFile { + return Receipt{}, fmt.Errorf("notarius receipt index_file %q is incompatible; want %q", document.IndexFile, canonicalIndexFile) + } if document.PipelineID != pipelineID { return Receipt{}, fmt.Errorf("notarius receipt pipeline id %q does not match requested pipeline %q", document.PipelineID, pipelineID) } @@ -234,9 +241,21 @@ func loadIndex(bundleRoot, indexPath string) (Index, error) { if err := decodeBoundedJSON(indexPath, maxIndexBytes, &document); err != nil { return Index{}, fmt.Errorf("decode notarius index: %w", err) } - if strings.TrimSpace(document.ManifestFile) == "" || document.OutputFiles == nil || - strings.TrimSpace(document.RejectedFile) == "" || strings.TrimSpace(document.WarningsFile) == "" { - return Index{}, fmt.Errorf("notarius index is missing required management paths or output_files") + for _, field := range []struct { + name string + got string + want string + }{ + {name: "manifest_file", got: document.ManifestFile, want: canonicalManifestFile}, + {name: "rejected_file", got: document.RejectedFile, want: canonicalRejectedFile}, + {name: "warnings_file", got: document.WarningsFile, want: canonicalWarningsFile}, + } { + if field.got != field.want { + return Index{}, fmt.Errorf("notarius index %s %q is incompatible; want %q", field.name, field.got, field.want) + } + } + if document.OutputFiles == nil { + return Index{}, fmt.Errorf("notarius index is missing required output_files") } index := Index{ diff --git a/internal/adapters/notarius/subprocess_test.go b/internal/adapters/notarius/subprocess_test.go index be96b23..395bbcc 100644 --- a/internal/adapters/notarius/subprocess_test.go +++ b/internal/adapters/notarius/subprocess_test.go @@ -175,10 +175,11 @@ func TestLoadReceiptValidation(t *testing.T) { "validation_status": "approved", "future_field": true, } tests := []struct { - name string - mutate func(map[string]any) - raw []byte - wantOK bool + name string + mutate func(map[string]any) + raw []byte + wantOK bool + wantError string }{ {name: "unknown fields tolerated", wantOK: true}, {name: "malformed", raw: []byte("{")}, @@ -187,6 +188,14 @@ func TestLoadReceiptValidation(t *testing.T) { {name: "pipeline mismatch", mutate: func(v map[string]any) { v["pipeline_id"] = "other" }}, {name: "relative output", mutate: func(v map[string]any) { v["output_directory"] = "run-1" }}, {name: "negative count", mutate: func(v map[string]any) { v["warning_count"] = -1 }}, + { + name: "nested index", mutate: func(v map[string]any) { v["index_file"] = "nested/index.json" }, + wantError: `index_file "nested/index.json"`, + }, + { + name: "cleanable index", mutate: func(v map[string]any) { v["index_file"] = "./index.json" }, + wantError: `index_file "./index.json"`, + }, } for _, test := range tests { t.Run(test.name, func(t *testing.T) { @@ -209,6 +218,9 @@ func TestLoadReceiptValidation(t *testing.T) { if !test.wantOK && err == nil { t.Fatal("loadReceipt() error = nil, want validation failure") } + if test.wantError != "" && !strings.Contains(err.Error(), test.wantError) { + t.Fatalf("loadReceipt() error = %v, want fragment %q", err, test.wantError) + } }) } @@ -260,10 +272,31 @@ func TestLoadIndexRejectsMalformedUnsafeAndUnsupportedDocuments(t *testing.T) { name string indexValue any prepare func(*testing.T, string) + wantError string }{ {name: "malformed", indexValue: json.RawMessage(`{"manifest_file":`)}, {name: "unsupported output shape", indexValue: map[string]any{"manifest_file": "manifest.json", "output_files": map[string]any{}, "rejected_file": "rejected.json", "warnings_file": "warnings.json"}}, {name: "missing management path", indexValue: map[string]any{"output_files": []any{}, "rejected_file": "rejected.json", "warnings_file": "warnings.json"}}, + {name: "renamed manifest", indexValue: func() any { + value := validIndexValue([]any{}) + value["manifest_file"] = "metadata.json" + return value + }(), wantError: `manifest_file "metadata.json"`}, + {name: "cleanable manifest", indexValue: func() any { + value := validIndexValue([]any{}) + value["manifest_file"] = "./manifest.json" + return value + }(), wantError: `manifest_file "./manifest.json"`}, + {name: "renamed rejections", indexValue: func() any { + value := validIndexValue([]any{}) + value["rejected_file"] = "rejections.json" + return value + }(), wantError: `rejected_file "rejections.json"`}, + {name: "renamed warnings", indexValue: func() any { + value := validIndexValue([]any{}) + value["warnings_file"] = "diagnostics/warnings.json" + return value + }(), wantError: `warnings_file "diagnostics/warnings.json"`}, {name: "duplicate lane", indexValue: validIndexValue([]any{ map[string]any{"lane_id": "npc", "file": "lanes/npc.json"}, map[string]any{"lane_id": "npc", "file": "lanes/npc.json"}, @@ -281,6 +314,11 @@ func TestLoadIndexRejectsMalformedUnsafeAndUnsupportedDocuments(t *testing.T) { t.Skipf("Symlink() unavailable: %v", err) } }}, + {name: "missing management file", indexValue: validIndexValue([]any{}), prepare: func(t *testing.T, bundle string) { + if err := os.Remove(filepath.Join(bundle, "manifest.json")); err != nil { + t.Fatalf("Remove(manifest) error = %v", err) + } + }}, {name: "incomplete pipeline descriptor", indexValue: func() any { value := validIndexValue([]any{}) value["chunk_map"] = map[string]any{"artifact_kind": "chunk_map", "file": "chunk-map.json"} @@ -311,6 +349,8 @@ func TestLoadIndexRejectsMalformedUnsafeAndUnsupportedDocuments(t *testing.T) { } if _, err := loadIndex(bundle, indexPath); err == nil { t.Fatal("loadIndex() error = nil, want failure") + } else if test.wantError != "" && !strings.Contains(err.Error(), test.wantError) { + t.Fatalf("loadIndex() error = %v, want fragment %q", err, test.wantError) } }) } diff --git a/internal/stage/extract_test.go b/internal/stage/extract_test.go index 7876679..718e8a7 100644 --- a/internal/stage/extract_test.go +++ b/internal/stage/extract_test.go @@ -2,6 +2,7 @@ package stage import ( "context" + "encoding/json" "errors" "os" "path/filepath" @@ -300,6 +301,66 @@ func TestExtractStageResumeValidationAcceptsCurrentImmutableResult(t *testing.T) } } +func TestExtractStageAdapterResultIsImmediatelyReusableAndCatalogVisible(t *testing.T) { + env, m, fake := setupExtractEnv(t) + producerRunID := m.RunID + receiptFixture := filepath.Join(t.TempDir(), "receipt.json") + receipt, err := json.Marshal(map[string]any{ + "schema_version": notarius.ReceiptSchemaVersion, + "run_id": "notarius-run-1", "pipeline_id": "dnd-session", + "output_directory": fake.Result.BundleRoot, "index_file": "index.json", + "normalized_output_count": 2, "rejected_output_count": 0, "warning_count": 0, + "validation_status": "approved", "future_field": true, + }) + if err != nil { + t.Fatalf("json.Marshal(receipt) error = %v", err) + } + if err := os.WriteFile(receiptFixture, receipt, 0o644); err != nil { + t.Fatalf("WriteFile(receipt fixture) error = %v", err) + } + if err := os.WriteFile(env.Config.Pipeline.Notarius.Binary, []byte("#!/bin/sh\ncat \"$NARRATIO_NOTARIUS_RECEIPT_FIXTURE\"\n"), 0o755); err != nil { + t.Fatalf("WriteFile(notarius helper) error = %v", err) + } + t.Setenv("NARRATIO_NOTARIUS_RECEIPT_FIXTURE", receiptFixture) + env.Notarius = notarius.NewSubprocessRunner() + + result, err := (extractStage{}).Run(context.Background(), env, m) + if err != nil { + t.Fatalf("Run() error = %v", err) + } + durableIndex := filepath.Join(result.Metadata["bundle_root"].(string), "index.json") + if len(result.Outputs) != 2 || result.Outputs[1].AbsolutePath != durableIndex { + t.Fatalf("outputs = %#v, want canonical durable index %q", result.Outputs, durableIndex) + } + recordSucceededExtractResult(m, producerRunID, result) + m.RunID = "20260810T020304Z-fedcba98" + + validation, err := (extractStage{}).ValidateResume(context.Background(), env, m) + if err != nil { + t.Fatalf("ValidateResume() error = %v", err) + } + if !validation.Resumable { + t.Fatalf("validation = %#v, want resumable", validation) + } + definition := env.Config.Pipeline.Notarius.Outputs["npc_registry"] + definitions := map[string]artifacts.ExtractionArtifactDefinition{ + "npc_registry": { + LaneID: definition.LaneID, PipelineID: env.Config.Pipeline.Notarius.PipelineID, + MediaType: definition.MediaType, SchemaID: definition.SchemaID, + SchemaVersion: definition.SchemaVersion, ModuleKey: definition.ModuleKey, + }, + } + catalog := artifacts.NewArtifactCatalog() + if err := catalog.RegisterExtractionArtifacts(definitions); err != nil { + t.Fatalf("RegisterExtractionArtifacts() error = %v", err) + } + catalog.HydrateExtractionArtifacts(sessionPathsForEnv(env, m.SessionID), m, definitions) + entry, ok := catalog.Lookup(artifacts.ExtractionArtifactSourceID("npc_registry")) + if !ok || !entry.Available || entry.Path != result.Outputs[0].AbsolutePath || entry.ProducerRunID != producerRunID { + t.Fatalf("hydrated extraction entry = %#v, %v", entry, ok) + } +} + func TestExtractStageResumeValidationRejectsObsoleteResults(t *testing.T) { tests := []struct { name string @@ -378,6 +439,10 @@ func seedSucceededExtractResult(t *testing.T, env *Env, m *manifest.Manifest) { if err != nil { t.Fatalf("Run() error = %v", err) } + recordSucceededExtractResult(m, producerRunID, result) +} + +func recordSucceededExtractResult(m *manifest.Manifest, producerRunID string, result *StageResult) { records := make([]manifest.ArtifactRecord, 0, len(result.Outputs)) for _, output := range result.Outputs { records = append(records, manifest.ArtifactRecord{ @@ -461,7 +526,7 @@ func setupExtractEnv(t *testing.T) (*Env, *manifest.Manifest, *notarius.FakeRunn t.Fatalf("MkdirAll(bundle unknown) error = %v", err) } files := map[string]string{ - "index.json": `{ "manifest_file": "manifest.json" }`, + "index.json": `{"manifest_file":"manifest.json","output_files":[{"lane_id":"npc-registry","file":"lanes/npc.json","media_type":"application/json","module_key":"dnd/npc-registry","schema_id":"notarius.dnd.npc_registry","schema_version":"v1","future_field":true},{"lane_id":"unconfigured","file":"lanes/unconfigured.json"}],"rejected_file":"rejected.json","warnings_file":"warnings.json","future_field":true}`, "manifest.json": `{}`, "rejected.json": `{"rejected":[]}`, "warnings.json": `{"warnings":[]}`,