eric/narratio
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 16 Wiki Activity

Update Notarius integration and plan references

Browse Source
This commit is contained in:
Eric Rakestraw
2026-08-29 14:41:22 +00:00
parent e433c86203
commit 42ed81cbe1
11 changed files with 1450 additions and 114 deletions
Download Patch File Download Diff File Expand all files Collapse all files

79
internal/stage/extract.go
View File

@@ -161,6 +161,10 @@ func (extractStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*S
if err != nil {
return nil, fmt.Errorf("extract: resolve staging warnings relative path: %w", err)
}
diagnosticsRelative, err := pathsafe.SlashRelativeFromRoot(adapterResult.BundleRoot, adapterResult.Index.DiagnosticsPath)
if err != nil {
return nil, fmt.Errorf("extract: resolve staging diagnostics relative path: %w", err)
}
stagingIndexChecksum, err := checksumRegularFile(adapterResult.Index.Path, false)
if err != nil {
return nil, fmt.Errorf("extract: validate staging index: %w", err)
@@ -188,6 +192,10 @@ func (extractStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*S
if err != nil {
return nil, fmt.Errorf("extract: resolve promoted warnings: %w", err)
}
promotedDiagnosticsPath, err := pathsafe.JoinSlashRelativeUnderRoot(durableBundle, diagnosticsRelative)
if err != nil {
return nil, fmt.Errorf("extract: resolve promoted diagnostics: %w", err)
}
outputs := make([]artifacts.Ref, 0, len(selected)+1)
for _, lane := range selected {
@@ -237,18 +245,25 @@ func (extractStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*S
"diagnostic_path": logPath,
"rejections_path": promotedRejectionsPath,
"warnings_path": promotedWarningsPath,
"diagnostics_path": promotedDiagnosticsPath,
"narratio_run_id": runID,
"configuration_fingerprint": fingerprint,
"direct_input": input.Metadata(),
"receipt": map[string]any{
"run_id": adapterResult.Receipt.RunID, "pipeline_id": adapterResult.Receipt.PipelineID,
"normalized_output_count": adapterResult.Receipt.NormalizedOutputCount,
"rejected_output_count": adapterResult.Receipt.RejectedOutputCount,
"warning_count": adapterResult.Receipt.WarningCount,
"validation_status": adapterResult.Receipt.ValidationStatus,
"normalized_output_count": adapterResult.Receipt.NormalizedOutputCount,
"rejected_output_count": adapterResult.Receipt.RejectedOutputCount,
"warning_group_count": adapterResult.Receipt.WarningGroupCount,
"warning_occurrence_count": adapterResult.Receipt.WarningOccurrenceCount,
"diagnostic_group_count": adapterResult.Receipt.DiagnosticGroupCount,
"diagnostic_occurrence_count": adapterResult.Receipt.DiagnosticOccurrenceCount,
"diagnostics_truncated": adapterResult.Receipt.DiagnosticsTruncated,
"validation_status": adapterResult.Receipt.ValidationStatus,
},
"rejections": boundedRejectionMetadata(adapterResult.Rejections),
"warnings": boundedWarningMetadata(adapterResult.Warnings),
"rejections": boundedRejectionMetadata(adapterResult.Rejections),
"warnings": boundedWarningMetadata(adapterResult.Warnings),
"diagnostics": boundedDiagnosticMetadata(adapterResult.Diagnostics),
"validation_summaries": boundedValidationMetadata(adapterResult.Receipt.ValidationSummaries),
}
return &StageResult{
Outputs: outputs,
@@ -282,6 +297,11 @@ func selectRequiredNotariusLanes(
return nil, fmt.Errorf("extract: required lane %q was rejected (reason_code=%q)", expected.LaneID, rejection.ReasonCode)
}
}
for _, validation := range result.Receipt.ValidationSummaries {
if validation.LaneID == expected.LaneID && validation.Status != "complete" {
return nil, fmt.Errorf("extract: required lane %q validation is %q", expected.LaneID, validation.Status)
}
}
matches := make([]notarius.LaneDescriptor, 0, 1)
for _, descriptor := range result.Index.Lanes {
if descriptor.LaneID == expected.LaneID {
@@ -441,7 +461,52 @@ func boundedWarningMetadata(values []notarius.WarningSummary) []map[string]any {
}
result := make([]map[string]any, 0, limit)
for _, value := range values[:limit] {
result = append(result, map[string]any{"scope": value.Scope, "reason_code": value.ReasonCode})
result = append(result, map[string]any{
"disposition": value.Disposition, "category": value.Category,
"reason_code": value.ReasonCode, "origin": diagnosticOriginMetadata(value.Origin),
"occurrence_count": value.OccurrenceCount,
})
}
return result
}
func boundedDiagnosticMetadata(values []notarius.DiagnosticSummary) []map[string]any {
limit := len(values)
if limit > maxDiagnosticSummaries {
limit = maxDiagnosticSummaries
}
result := make([]map[string]any, 0, limit)
for _, value := range values[:limit] {
result = append(result, map[string]any{
"disposition": value.Disposition, "category": value.Category,
"reason_code": value.ReasonCode, "origin": diagnosticOriginMetadata(value.Origin),
"occurrence_count": value.OccurrenceCount,
})
}
return result
}
func boundedValidationMetadata(values []notarius.ValidationSummary) []map[string]any {
limit := len(values)
if limit > maxDiagnosticSummaries {
limit = maxDiagnosticSummaries
}
result := make([]map[string]any, 0, limit)
for _, value := range values[:limit] {
result = append(result, map[string]any{
"stage": value.Stage, "step_id": value.StepID, "lane_id": value.LaneID,
"module_key": value.ModuleKey, "chunk_id": value.ChunkID, "status": value.Status,
"rejecting_validators": value.RejectingValidators, "reason_codes": value.ReasonCodes,
"incomplete_validators": value.IncompleteValidators,
"producer_attempt_count": value.ProducerAttemptCount, "terminal_action": value.TerminalAction,
})
}
return result
}
func diagnosticOriginMetadata(value notarius.DiagnosticOrigin) map[string]any {
return map[string]any{
"stage": value.Stage, "step_id": value.StepID, "lane_id": value.LaneID,
"module_key": value.ModuleKey, "validator_key": value.ValidatorKey,
}
}

110
internal/stage/extract_test.go
View File

@@ -99,7 +99,8 @@ func TestExtractStageProducesImmutableManifestReadyOutputs(t *testing.T) {
if result.Metadata["receipt_path"] != artifacts.SessionRunNotariusReceiptPathForCampaign(fixture.workspace, fixture.campaign, fixture.sessionID, fixture.runID) ||
result.Metadata["diagnostic_path"] != artifacts.SessionRunNotariusLogPathForCampaign(fixture.workspace, fixture.campaign, fixture.sessionID, fixture.runID) ||
result.Metadata["rejections_path"] != filepath.Join(durableBundle, "rejected.json") ||
result.Metadata["warnings_path"] != filepath.Join(durableBundle, "warnings.json") {
result.Metadata["warnings_path"] != filepath.Join(durableBundle, "warnings.json") ||
result.Metadata["diagnostics_path"] != filepath.Join(durableBundle, "diagnostics.json") {
t.Fatalf("diagnostic metadata = %#v", result.Metadata)
}
if len(result.Outputs) != 2 {
@@ -133,7 +134,7 @@ func TestExtractStageProducesImmutableManifestReadyOutputs(t *testing.T) {
t.Fatalf("lane path was not re-resolved after promotion: %q", lane.AbsolutePath)
}
for _, relative := range []string{
"index.json", "manifest.json", "rejected.json", "warnings.json", "lanes/npc.json",
"index.json", "manifest.json", "rejected.json", "warnings.json", "diagnostics.json", "lanes/npc.json",
"lanes/unconfigured.json", "chunk-map.json", "evidence-context.json", "unknown/private-debug.json",
} {
if _, err := os.Stat(filepath.Join(durableBundle, filepath.FromSlash(relative))); err != nil {
@@ -156,6 +157,81 @@ func TestExtractStageProducesImmutableManifestReadyOutputs(t *testing.T) {
}
}
func TestExtractStageConsumesAllDefaultDndPipelineArtifacts(t *testing.T) {
env, m, fake := setupExtractEnv(t)
contracts := 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"},
}
env.Config.Pipeline.Notarius.Outputs = contracts
fake.Result.Index.Lanes = nil
fake.Result.Rejections = nil
fake.Result.Warnings = nil
fake.Result.Diagnostics = nil
fake.Result.Receipt.NormalizedOutputCount = len(contracts)
fake.Result.Receipt.RejectedOutputCount = 0
fake.Result.Receipt.WarningGroupCount = 0
fake.Result.Receipt.WarningOccurrenceCount = 0
fake.Result.Receipt.DiagnosticGroupCount = 0
fake.Result.Receipt.DiagnosticOccurrenceCount = 0
fake.Result.Receipt.ValidationStatus = "approved"
indexDescriptors := make([]map[string]any, 0, len(contracts))
for _, contract := range contracts {
relative := "lanes/" + contract.LaneID + ".json"
path := filepath.Join(fake.Result.BundleRoot, filepath.FromSlash(relative))
if err := os.WriteFile(path, []byte(`{"records":[]}`), 0o644); err != nil {
t.Fatalf("WriteFile(%q) error = %v", path, err)
}
fake.Result.Index.Lanes = append(fake.Result.Index.Lanes, notarius.LaneDescriptor{
LaneID: contract.LaneID, File: relative, Path: path, MediaType: contract.MediaType,
ModuleKey: contract.ModuleKey, SchemaID: contract.SchemaID, SchemaVersion: contract.SchemaVersion,
})
indexDescriptors = append(indexDescriptors, map[string]any{
"lane_id": contract.LaneID, "file": relative, "media_type": contract.MediaType,
"module_key": contract.ModuleKey, "schema_id": contract.SchemaID,
"schema_version": contract.SchemaVersion,
})
}
if err := writeJSONFile(fake.Result.Index.Path, map[string]any{
"manifest_file": "manifest.json", "output_files": indexDescriptors,
"rejected_file": "rejected.json", "warnings_file": "warnings.json",
"diagnostics_file": "diagnostics.json",
}); err != nil {
t.Fatalf("write complete D&D index: %v", err)
}
result, err := (extractStage{}).Run(context.Background(), env, m)
if err != nil {
t.Fatalf("Run() error = %v", err)
}
if len(result.Outputs) != len(contracts)+1 {
t.Fatalf("output count = %d, want %d", len(result.Outputs), len(contracts)+1)
}
got := make(map[string]*artifactmodel.ContractMetadata, len(contracts))
for _, output := range result.Outputs {
if output.Kind == extractLaneOutputKind {
got[output.SourceID] = output.Contract
}
}
for key, contract := range contracts {
sourceID := artifacts.ExtractionArtifactSourceID(key)
actual := got[sourceID]
if actual == nil || actual.MediaType != contract.MediaType || actual.SchemaID != contract.SchemaID ||
actual.SchemaVersion != contract.SchemaVersion || actual.ModuleKey != contract.ModuleKey {
t.Fatalf("source %q contract = %#v, want %#v", sourceID, actual, contract)
}
}
}
func TestExtractStageRejectsMissingOrInvalidFinalTrimmedInputBeforeInvocation(t *testing.T) {
tests := []struct {
name string
@@ -197,6 +273,9 @@ func TestExtractStageEnforcesRequiredLanePolicy(t *testing.T) {
{name: "rejected", mutate: func(result *notarius.RunResult) {
result.Rejections = append(result.Rejections, notarius.RejectionSummary{LaneID: "npc-registry", ReasonCode: "invalid_npc"})
}, want: "was rejected"},
{name: "incomplete validation", mutate: func(result *notarius.RunResult) {
result.Receipt.ValidationSummaries = []notarius.ValidationSummary{{LaneID: "npc-registry", Status: "incomplete"}}
}, want: `validation is "incomplete"`},
{name: "duplicate", mutate: func(result *notarius.RunResult) {
result.Index.Lanes = append(result.Index.Lanes, result.Index.Lanes[0])
}, want: "2 descriptors"},
@@ -338,8 +417,11 @@ func TestExtractStageAdapterResultIsImmediatelyReusableAndCatalogVisible(t *test
"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,
"normalized_output_count": 2, "rejected_output_count": 0,
"warning_group_count": 0, "warning_occurrence_count": 0,
"diagnostic_group_count": 0, "diagnostic_occurrence_count": 0,
"diagnostics_truncated": false,
"validation_status": "approved", "future_field": true,
})
if err != nil {
t.Fatalf("json.Marshal(receipt) error = %v", err)
@@ -594,10 +676,11 @@ 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","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}`,
"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","diagnostics_file":"diagnostics.json","future_field":true}`,
"manifest.json": `{}`,
"rejected.json": `{"rejected":[]}`,
"warnings.json": `{"warnings":[]}`,
"warnings.json": `{"schema_version":"notarius.warnings.v2","group_count":0,"occurrence_count":0,"groups":[]}`,
"diagnostics.json": `{"schema_version":"notarius.diagnostics.v1","group_count":0,"occurrence_count":0,"truncated":false,"unrepresented_occurrence_count":0,"groups":[]}`,
"lanes/npc.json": `{"npcs":[]}`,
"lanes/unconfigured.json": `{"spells":[]}`,
"chunk-map.json": `{"chunks":[]}`,
@@ -613,12 +696,14 @@ func setupExtractEnv(t *testing.T) (*Env, *manifest.Manifest, *notarius.FakeRunn
Receipt: notarius.Receipt{
SchemaVersion: notarius.ReceiptSchemaVersion, RunID: "notarius-run-1", PipelineID: "dnd-session",
OutputDirectory: bundle, IndexFile: "index.json", NormalizedOutputCount: 2,
RejectedOutputCount: 1, WarningCount: 1, ValidationStatus: "rejected",
RejectedOutputCount: 1, WarningGroupCount: 1, WarningOccurrenceCount: 1,
DiagnosticGroupCount: 1, DiagnosticOccurrenceCount: 1, ValidationStatus: "rejected",
},
BundleRoot: bundle,
Index: notarius.Index{
Path: filepath.Join(bundle, "index.json"), RejectedPath: filepath.Join(bundle, "rejected.json"),
WarningsPath: filepath.Join(bundle, "warnings.json"),
WarningsPath: filepath.Join(bundle, "warnings.json"),
DiagnosticsPath: filepath.Join(bundle, "diagnostics.json"),
Lanes: []notarius.LaneDescriptor{
{
LaneID: "npc-registry", File: "lanes/npc.json", Path: filepath.Join(bundle, "lanes", "npc.json"),
@@ -629,7 +714,14 @@ func setupExtractEnv(t *testing.T) (*Env, *manifest.Manifest, *notarius.FakeRunn
},
},
Rejections: []notarius.RejectionSummary{{LaneID: "optional", ReasonCode: "optional_rejected"}},
Warnings: []notarius.WarningSummary{{Scope: "lane:npc-registry", ReasonCode: "normalized_name"}},
Warnings: []notarius.WarningSummary{{
Disposition: "warning", Category: "degradation", ReasonCode: "normalized_name",
Origin: notarius.DiagnosticOrigin{Stage: "normalize", LaneID: "npc-registry"}, OccurrenceCount: 1,
}},
Diagnostics: []notarius.DiagnosticSummary{{
Disposition: "advisory", Category: "data_quality", ReasonCode: "low_confidence",
Origin: notarius.DiagnosticOrigin{Stage: "normalize", ValidatorKey: "validator"}, OccurrenceCount: 1,
}},
}}
env := &Env{
Config: &config.Config{