diff --git a/docs/roadmap/implementation.md b/docs/roadmap/implementation.md index 0b55b55..7113dfc 100644 --- a/docs/roadmap/implementation.md +++ b/docs/roadmap/implementation.md @@ -20,7 +20,7 @@ implementation sequence. | Stage 2 | Complete | | Stage 3 | Complete | | Stage 4 | Complete | -| Stage 5 | Not started | +| Stage 5 | Complete | | Stage 6 | Not started | | Stage 7 | Not started | | Stage 8 | Not started | diff --git a/internal/app/runner.go b/internal/app/runner.go index d61e995..5f20c25 100644 --- a/internal/app/runner.go +++ b/internal/app/runner.go @@ -8,6 +8,7 @@ import ( "strings" "gitea.maximumdirect.net/eric/narratio/internal/adapters/audita" + "gitea.maximumdirect.net/eric/narratio/internal/adapters/notarius" "gitea.maximumdirect.net/eric/narratio/internal/adapters/notify" "gitea.maximumdirect.net/eric/narratio/internal/adapters/scriptorium" "gitea.maximumdirect.net/eric/narratio/internal/adapters/seriatim" @@ -80,6 +81,9 @@ func executeStages(ctx context.Context, cfg *config.Config, stages []stage.Stage } env.Audita = runner } + if env.Notarius == nil && needsNotariusForRun(env.Config, stages) { + env.Notarius = notarius.NewSubprocessRunner() + } if env.Scriptorium == nil { env.Scriptorium = scriptorium.NewSubprocessRunner() } @@ -672,6 +676,18 @@ func needsObjectStoreForRun(cfg *config.Config, stages []stage.Stage) bool { return true } +func needsNotariusForRun(cfg *config.Config, stages []stage.Stage) bool { + if cfg == nil || cfg.Pipeline == nil || cfg.Pipeline.Notarius == nil || !cfg.Pipeline.Notarius.Enabled { + return false + } + for _, candidate := range stages { + if candidate != nil && candidate.Name() == "extract" { + return true + } + } + return false +} + func needsRemoteLocksForRun(cfg *config.Config, stages []stage.Stage) bool { if cfg == nil || cfg.Pipeline == nil || cfg.Session == nil { return false diff --git a/internal/app/runner_test.go b/internal/app/runner_test.go index 98773c6..5425cc1 100644 --- a/internal/app/runner_test.go +++ b/internal/app/runner_test.go @@ -70,6 +70,17 @@ type captureSelectedArtifactsStage struct { captured *[]string } +type captureNotariusStage struct { + captured *bool +} + +func (s captureNotariusStage) Name() string { return "extract" } +func (s captureNotariusStage) Declares() stage.IODecl { return stage.IODecl{} } +func (s captureNotariusStage) Run(_ context.Context, env *stage.Env, _ *manifest.Manifest) (*stage.StageResult, error) { + *s.captured = env.Notarius != nil + return &stage.StageResult{}, nil +} + func (s captureSelectedArtifactsStage) Name() string { return s.name } func (s captureSelectedArtifactsStage) Declares() stage.IODecl { return stage.IODecl{} } func (s captureSelectedArtifactsStage) Run(_ context.Context, env *stage.Env, _ *manifest.Manifest) (*stage.StageResult, error) { @@ -154,6 +165,28 @@ func TestExecuteStagesPropagatesSelectedArtifactsToEnv(t *testing.T) { } } +func TestExecuteStagesComposesNotariusOnlyForEnabledExtraction(t *testing.T) { + cfg := testConfig(t) + cfg.Pipeline.Notarius = &config.NotariusConfig{Enabled: true} + captured := false + + _, err := executeStages(context.Background(), cfg, []stage.Stage{captureNotariusStage{captured: &captured}}, RunOptions{}) + if err != nil { + t.Fatalf("executeStages() error = %v", err) + } + if !captured { + t.Fatal("extract stage did not receive the default Notarius runner") + } + + if needsNotariusForRun(cfg, []stage.Stage{countingStage{name: "analyze", runs: new(int)}}) { + t.Fatal("Notarius runner requested without extract in the selected plan") + } + cfg.Pipeline.Notarius.Enabled = false + if needsNotariusForRun(cfg, []stage.Stage{captureNotariusStage{captured: new(bool)}}) { + t.Fatal("Notarius runner requested while extraction is disabled") + } +} + func TestExecuteStagesAnalyzeOutputsPersistAsScriptoriumArtifacts(t *testing.T) { cfg := testConfig(t) storeForPaths := artifacts.NewLocalStore(cfg.Pipeline.Workspace.Root) diff --git a/internal/artifacts/paths.go b/internal/artifacts/paths.go index 37f96f9..4dc6399 100644 --- a/internal/artifacts/paths.go +++ b/internal/artifacts/paths.go @@ -86,6 +86,36 @@ func SessionRunStageDirForCampaign(rootDir, campaign, sessionID, runID, stageNam return filepath.Join(SessionRunRootForCampaign(rootDir, campaign, sessionID, runID), stageName) } +// SessionRunExtractDirForCampaign returns the invocation-local extraction directory. +func SessionRunExtractDirForCampaign(rootDir, campaign, sessionID, runID string) string { + return SessionRunStageDirForCampaign(rootDir, campaign, sessionID, runID, "extract") +} + +// SessionRunNotariusReceiptPathForCampaign returns the invocation-local receipt path. +func SessionRunNotariusReceiptPathForCampaign(rootDir, campaign, sessionID, runID string) string { + return filepath.Join(SessionRunExtractDirForCampaign(rootDir, campaign, sessionID, runID), "notarius.receipt.json") +} + +// SessionRunNotariusLogPathForCampaign returns the invocation-local stderr log path. +func SessionRunNotariusLogPathForCampaign(rootDir, campaign, sessionID, runID string) string { + return filepath.Join(SessionRunExtractDirForCampaign(rootDir, campaign, sessionID, runID), "notarius.stderr.log") +} + +// SessionRunNotariusOutputRootForCampaign returns the invocation-local Notarius output root. +func SessionRunNotariusOutputRootForCampaign(rootDir, campaign, sessionID, runID string) string { + return filepath.Join(SessionRunExtractDirForCampaign(rootDir, campaign, sessionID, runID), "notarius-output") +} + +// SessionNotariusBundleDirForCampaign returns one immutable durable bundle destination. +func SessionNotariusBundleDirForCampaign(rootDir, campaign, sessionID, runID string) string { + return filepath.Join( + SessionWorkDirForCampaign(rootDir, campaign, sessionID), + config.PathArtifactsDirSegment, + "notarius", + runID, + ) +} + // SessionSpoolAudioDir returns the campaign/session/run scoped local spool audio path. func SessionSpoolAudioDir(spoolRoot, campaign, sessionID, runID string) string { return filepath.Join(spoolRoot, campaign, sessionID, runID, config.PathAudioDirSegment) diff --git a/internal/artifacts/paths_model_test.go b/internal/artifacts/paths_model_test.go index 54ca020..1a73ed5 100644 --- a/internal/artifacts/paths_model_test.go +++ b/internal/artifacts/paths_model_test.go @@ -49,6 +49,33 @@ func TestSessionRunManifestPathForCampaign(t *testing.T) { } } +func TestSessionNotariusPathsForCampaign(t *testing.T) { + root := "/tmp/workspace" + campaign := "forsaken" + sessionID := "2026-04-19" + runID := "20260515T031522Z-a1b2c3d4" + extractDir := filepath.Join(root, "work", campaign, sessionID, "runs", runID, "extract") + + tests := []struct { + name string + got string + want string + }{ + {name: "extract directory", got: SessionRunExtractDirForCampaign(root, campaign, sessionID, runID), want: extractDir}, + {name: "receipt", got: SessionRunNotariusReceiptPathForCampaign(root, campaign, sessionID, runID), want: filepath.Join(extractDir, "notarius.receipt.json")}, + {name: "stderr", got: SessionRunNotariusLogPathForCampaign(root, campaign, sessionID, runID), want: filepath.Join(extractDir, "notarius.stderr.log")}, + {name: "output root", got: SessionRunNotariusOutputRootForCampaign(root, campaign, sessionID, runID), want: filepath.Join(extractDir, "notarius-output")}, + {name: "durable bundle", got: SessionNotariusBundleDirForCampaign(root, campaign, sessionID, runID), want: filepath.Join(root, "work", campaign, sessionID, "artifacts", "notarius", runID)}, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + if test.got != test.want { + t.Fatalf("path = %q, want %q", test.got, test.want) + } + }) + } +} + func TestSessionPreviousPathsForCampaign(t *testing.T) { root := "/tmp/workspace" previousDir := SessionPreviousDirForCampaign(root, "forsaken", "2026-04-19") diff --git a/internal/stage/extract.go b/internal/stage/extract.go new file mode 100644 index 0000000..ffaf47b --- /dev/null +++ b/internal/stage/extract.go @@ -0,0 +1,462 @@ +package stage + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "fmt" + "os" + "os/exec" + "path/filepath" + "sort" + "strings" + "time" + + "gitea.maximumdirect.net/eric/narratio/internal/adapters/notarius" + "gitea.maximumdirect.net/eric/narratio/internal/artifactmodel" + "gitea.maximumdirect.net/eric/narratio/internal/artifacts" + "gitea.maximumdirect.net/eric/narratio/internal/config" + "gitea.maximumdirect.net/eric/narratio/internal/fileops" + "gitea.maximumdirect.net/eric/narratio/internal/manifest" + "gitea.maximumdirect.net/eric/narratio/internal/pathsafe" +) + +const ( + extractSkipReason = "notarius_disabled" + extractLaneOutputKind = "notarius_lane" + extractIndexOutputKind = "notarius_index" + maxDiagnosticSummaries = 100 +) + +type extractStage struct{} + +func (extractStage) Name() string { return "extract" } + +func (extractStage) Declares() IODecl { + return IODecl{ + Inputs: []artifacts.Ref{{ + Kind: artifactmodel.TranscriptOutputKindFinalTrimmed, + SourceID: artifactmodel.SourceTranscriptFinalTrimmed, + Category: "transcripts", + RelativePath: artifactmodel.TranscriptPathFinalTrimmed, + }}, + Outputs: []artifacts.Ref{ + {Kind: extractLaneOutputKind, Category: "artifacts", RelativePath: "artifacts/notarius//lanes/*.json"}, + {Kind: extractIndexOutputKind, Category: "artifacts", RelativePath: "artifacts/notarius//index.json"}, + }, + } +} + +func (extractStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*StageResult, error) { + if env == nil || env.Config == nil || env.Config.Pipeline == nil || env.Config.Session == nil { + return nil, fmt.Errorf("extract: resolved stage environment config is required") + } + notariusConfig := env.Config.Pipeline.Notarius + if notariusConfig == nil || !notariusConfig.Enabled { + return &StageResult{ + Disposition: StageDispositionSkipped, + SkipReason: extractSkipReason, + Metadata: map[string]any{ + "stage": "extract", + "notarius_enabled": false, + "reason": extractSkipReason, + }, + }, nil + } + if env.ArtifactStore == nil { + return nil, fmt.Errorf("extract: artifact store is required") + } + if env.Notarius == nil { + return nil, fmt.Errorf("extract: notarius adapter is required") + } + if m == nil { + return nil, fmt.Errorf("extract: session manifest is required") + } + + sessionID := strings.TrimSpace(m.SessionID) + if sessionID == "" { + sessionID = strings.TrimSpace(env.Config.Session.SessionID) + } + campaign := strings.TrimSpace(m.Campaign) + if campaign == "" { + campaign = strings.TrimSpace(env.Config.Session.Campaign) + } + runID := strings.TrimSpace(m.RunID) + if sessionID == "" || campaign == "" || runID == "" { + return nil, fmt.Errorf("extract: session id, campaign, and run id are required") + } + if !safePathSegment(runID) { + return nil, fmt.Errorf("extract: run id %q is not a safe path segment", runID) + } + + paths := sessionPathsForEnv(env, sessionID) + input, err := artifacts.ResolveSessionArtifact(paths, m, artifacts.ArtifactTranscriptFinalTrimmed) + if err != nil { + return nil, fmt.Errorf("extract: resolve final-trimmed transcript: %w", err) + } + + timeout, err := time.ParseDuration(strings.TrimSpace(notariusConfig.Timeout)) + if err != nil || timeout <= 0 { + return nil, fmt.Errorf("extract: invalid notarius timeout %q", notariusConfig.Timeout) + } + resolvedBinary, err := resolveExecutable(notariusConfig.Binary) + if err != nil { + return nil, fmt.Errorf("extract: resolve notarius binary: %w", err) + } + configPath, err := absolutePath(notariusConfig.ConfigPath) + if err != nil { + return nil, fmt.Errorf("extract: resolve notarius config path: %w", err) + } + inputPath, err := absolutePath(input.Path) + if err != nil { + return nil, fmt.Errorf("extract: resolve transcript input path: %w", err) + } + workingDirectory, err := absolutePath(notariusConfig.WorkingDirectory) + if err != nil { + return nil, fmt.Errorf("extract: resolve notarius working directory: %w", err) + } + + workspaceRoot := strings.TrimSpace(paths.WorkspaceRoot) + if workspaceRoot == "" { + workspaceRoot = env.Config.Pipeline.Workspace.Root + } + receiptPath, err := absolutePath(artifacts.SessionRunNotariusReceiptPathForCampaign(workspaceRoot, campaign, sessionID, runID)) + if err != nil { + return nil, fmt.Errorf("extract: resolve receipt path: %w", err) + } + logPath, err := absolutePath(artifacts.SessionRunNotariusLogPathForCampaign(workspaceRoot, campaign, sessionID, runID)) + if err != nil { + return nil, fmt.Errorf("extract: resolve stderr path: %w", err) + } + outputRoot, err := absolutePath(artifacts.SessionRunNotariusOutputRootForCampaign(workspaceRoot, campaign, sessionID, runID)) + if err != nil { + return nil, fmt.Errorf("extract: resolve output root: %w", err) + } + durableBundle, err := absolutePath(artifacts.SessionNotariusBundleDirForCampaign(workspaceRoot, campaign, sessionID, runID)) + if err != nil { + return nil, fmt.Errorf("extract: resolve durable bundle path: %w", err) + } + for _, directory := range []string{filepath.Dir(receiptPath), outputRoot, filepath.Dir(durableBundle)} { + if err := os.MkdirAll(directory, 0o755); err != nil { + return nil, fmt.Errorf("extract: create directory %q: %w", directory, err) + } + } + + fingerprint, err := extractionFingerprint(resolvedBinary, configPath, notariusConfig, timeout, workingDirectory) + if err != nil { + return nil, fmt.Errorf("extract: build configuration fingerprint: %w", err) + } + request := notarius.RunRequest{ + Binary: resolvedBinary, ConfigPath: configPath, PipelineID: notariusConfig.PipelineID, + InputPath: inputPath, OutputRoot: outputRoot, WorkingDirectory: workingDirectory, + ReceiptPath: receiptPath, LogPath: logPath, Timeout: timeout, + } + adapterResult, err := env.Notarius.Run(ctx, request) + if err != nil { + return nil, fmt.Errorf("extract: run notarius: %w", err) + } + if strings.TrimSpace(adapterResult.BundleRoot) == "" || strings.TrimSpace(adapterResult.Index.Path) == "" { + return nil, fmt.Errorf("extract: notarius result is missing bundle or index path") + } + if adapterResult.Receipt.RunID == "" || adapterResult.Receipt.PipelineID != notariusConfig.PipelineID { + return nil, fmt.Errorf("extract: notarius receipt identity is missing or incompatible") + } + + selected, err := selectRequiredNotariusLanes(env.ArtifactStore, notariusConfig.Outputs, adapterResult) + if err != nil { + return nil, err + } + indexRelative, err := pathsafe.SlashRelativeFromRoot(adapterResult.BundleRoot, adapterResult.Index.Path) + if err != nil { + return nil, fmt.Errorf("extract: resolve staging index relative path: %w", err) + } + rejectionsRelative, err := pathsafe.SlashRelativeFromRoot(adapterResult.BundleRoot, adapterResult.Index.RejectedPath) + if err != nil { + return nil, fmt.Errorf("extract: resolve staging rejections relative path: %w", err) + } + warningsRelative, err := pathsafe.SlashRelativeFromRoot(adapterResult.BundleRoot, adapterResult.Index.WarningsPath) + if err != nil { + return nil, fmt.Errorf("extract: resolve staging warnings relative path: %w", err) + } + stagingIndexChecksum, err := checksumRegularFile(adapterResult.Index.Path, false) + if err != nil { + return nil, fmt.Errorf("extract: validate staging index: %w", err) + } + + if err := fileops.PromoteDirectory(adapterResult.BundleRoot, durableBundle); err != nil { + return nil, fmt.Errorf("extract: promote notarius bundle: %w", err) + } + promotedIndexPath, err := pathsafe.JoinSlashRelativeUnderRoot(durableBundle, indexRelative) + if err != nil { + return nil, fmt.Errorf("extract: resolve promoted index: %w", err) + } + promotedIndexChecksum, err := checksumRegularFile(promotedIndexPath, false) + if err != nil { + return nil, fmt.Errorf("extract: checksum promoted index: %w", err) + } + if promotedIndexChecksum != stagingIndexChecksum { + return nil, fmt.Errorf("extract: promoted index checksum differs from staging index") + } + promotedRejectionsPath, err := pathsafe.JoinSlashRelativeUnderRoot(durableBundle, rejectionsRelative) + if err != nil { + return nil, fmt.Errorf("extract: resolve promoted rejections: %w", err) + } + promotedWarningsPath, err := pathsafe.JoinSlashRelativeUnderRoot(durableBundle, warningsRelative) + if err != nil { + return nil, fmt.Errorf("extract: resolve promoted warnings: %w", err) + } + + outputs := make([]artifacts.Ref, 0, len(selected)+1) + for _, lane := range selected { + promotedPath, err := pathsafe.JoinSlashRelativeUnderRoot(durableBundle, lane.RelativePath) + if err != nil { + return nil, fmt.Errorf("extract: resolve promoted lane %q: %w", lane.Descriptor.LaneID, err) + } + checksum, err := checksumRegularFile(promotedPath, true) + if err != nil { + return nil, fmt.Errorf("extract: validate promoted lane %q: %w", lane.Descriptor.LaneID, err) + } + if checksum != lane.StagingChecksum { + return nil, fmt.Errorf("extract: promoted lane %q checksum differs from staging payload", lane.Descriptor.LaneID) + } + relativePath, err := pathsafe.SlashRelativeFromRoot(paths.Root, promotedPath) + if err != nil { + return nil, fmt.Errorf("extract: derive lane %q session-relative path: %w", lane.Descriptor.LaneID, err) + } + outputs = append(outputs, artifacts.Ref{ + Kind: extractLaneOutputKind, SourceID: artifacts.ExtractionArtifactSourceID(lane.Key), + Category: "artifacts", SessionID: sessionID, RelativePath: relativePath, + AbsolutePath: promotedPath, Checksum: checksum, + Contract: &artifactmodel.ContractMetadata{ + MediaType: lane.Descriptor.MediaType, SchemaID: lane.Descriptor.SchemaID, + SchemaVersion: lane.Descriptor.SchemaVersion, ModuleKey: lane.Descriptor.ModuleKey, + }, + ExternalProvenance: &artifactmodel.ExternalProvenance{ + System: "notarius", RunID: adapterResult.Receipt.RunID, + PipelineID: adapterResult.Receipt.PipelineID, ArtifactID: lane.Descriptor.LaneID, + }, + }) + } + indexSessionRelative, err := pathsafe.SlashRelativeFromRoot(paths.Root, promotedIndexPath) + if err != nil { + return nil, fmt.Errorf("extract: derive index session-relative path: %w", err) + } + outputs = append(outputs, artifacts.Ref{ + Kind: extractIndexOutputKind, Category: "artifacts", SessionID: sessionID, + RelativePath: indexSessionRelative, AbsolutePath: promotedIndexPath, Checksum: promotedIndexChecksum, + }) + + metadata := map[string]any{ + "stage": "extract", + "notarius_enabled": true, + "bundle_root": durableBundle, + "receipt_path": receiptPath, + "diagnostic_path": logPath, + "rejections_path": promotedRejectionsPath, + "warnings_path": promotedWarningsPath, + "narratio_run_id": runID, + "configuration_fingerprint": fingerprint, + "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, + }, + "rejections": boundedRejectionMetadata(adapterResult.Rejections), + "warnings": boundedWarningMetadata(adapterResult.Warnings), + } + return &StageResult{ + Outputs: outputs, + Logs: []string{receiptPath, logPath}, + Metadata: metadata, + }, nil +} + +type selectedNotariusLane struct { + Key string + Descriptor notarius.LaneDescriptor + RelativePath string + StagingChecksum string +} + +func selectRequiredNotariusLanes( + store artifacts.Store, + required map[string]config.NotariusOutputConfig, + result notarius.RunResult, +) ([]selectedNotariusLane, error) { + keys := make([]string, 0, len(required)) + for key := range required { + keys = append(keys, key) + } + sort.Strings(keys) + selected := make([]selectedNotariusLane, 0, len(keys)) + for _, key := range keys { + expected := required[key] + for _, rejection := range result.Rejections { + if rejection.LaneID == expected.LaneID { + return nil, fmt.Errorf("extract: required lane %q was rejected (reason_code=%q)", expected.LaneID, rejection.ReasonCode) + } + } + matches := make([]notarius.LaneDescriptor, 0, 1) + for _, descriptor := range result.Index.Lanes { + if descriptor.LaneID == expected.LaneID { + matches = append(matches, descriptor) + } + } + if len(matches) != 1 { + return nil, fmt.Errorf("extract: required lane %q has %d descriptors, want exactly one", expected.LaneID, len(matches)) + } + descriptor := matches[0] + if descriptor.MediaType != expected.MediaType || descriptor.SchemaID != expected.SchemaID || + descriptor.SchemaVersion != expected.SchemaVersion || + (expected.ModuleKey != "" && descriptor.ModuleKey != expected.ModuleKey) { + return nil, fmt.Errorf("extract: required lane %q descriptor contract is incompatible", expected.LaneID) + } + checksum, err := checksumRegularFile(descriptor.Path, true) + if err != nil { + return nil, fmt.Errorf("extract: validate required lane %q: %w", expected.LaneID, err) + } + if store != nil { + storeChecksum, err := store.Checksum(descriptor.Path) + if err != nil { + return nil, fmt.Errorf("extract: checksum required lane %q: %w", expected.LaneID, err) + } + if storeChecksum != checksum { + return nil, fmt.Errorf("extract: inconsistent staging checksum for required lane %q", expected.LaneID) + } + } + relative, err := pathsafe.SlashRelativeFromRoot(result.BundleRoot, descriptor.Path) + if err != nil { + return nil, fmt.Errorf("extract: resolve required lane %q relative path: %w", expected.LaneID, err) + } + selected = append(selected, selectedNotariusLane{ + Key: key, Descriptor: descriptor, RelativePath: relative, StagingChecksum: checksum, + }) + } + return selected, nil +} + +func checksumRegularFile(path string, requireJSON bool) (string, error) { + info, err := os.Lstat(path) + if err != nil { + return "", err + } + if info.Mode()&os.ModeSymlink != 0 || !info.Mode().IsRegular() { + return "", fmt.Errorf("path %q must be a regular file without symlinks", path) + } + if info.Size() == 0 { + return "", fmt.Errorf("path %q must be non-empty", path) + } + data, err := os.ReadFile(path) + if err != nil { + return "", err + } + if requireJSON && !json.Valid(data) { + return "", fmt.Errorf("path %q is not valid JSON", path) + } + digest := sha256.Sum256(data) + return hex.EncodeToString(digest[:]), nil +} + +type fingerprintOutput struct { + Key string `json:"key"` + LaneID string `json:"lane_id"` + MediaType string `json:"media_type"` + SchemaID string `json:"schema_id"` + SchemaVersion string `json:"schema_version"` + ModuleKey string `json:"module_key,omitempty"` +} + +type fingerprintDocument struct { + Binary string `json:"binary"` + ConfigPath string `json:"config_path"` + PipelineID string `json:"pipeline_id"` + Timeout string `json:"timeout"` + WorkingDirectory string `json:"working_directory"` + Outputs []fingerprintOutput `json:"outputs"` +} + +func extractionFingerprint( + binary, configPath string, + cfg *config.NotariusConfig, + timeout time.Duration, + workingDirectory string, +) (string, error) { + keys := make([]string, 0, len(cfg.Outputs)) + for key := range cfg.Outputs { + keys = append(keys, key) + } + sort.Strings(keys) + outputs := make([]fingerprintOutput, 0, len(keys)) + for _, key := range keys { + output := cfg.Outputs[key] + outputs = append(outputs, fingerprintOutput{ + Key: key, LaneID: output.LaneID, MediaType: output.MediaType, + SchemaID: output.SchemaID, SchemaVersion: output.SchemaVersion, ModuleKey: output.ModuleKey, + }) + } + payload, err := json.Marshal(fingerprintDocument{ + Binary: binary, ConfigPath: configPath, PipelineID: cfg.PipelineID, + Timeout: timeout.String(), WorkingDirectory: workingDirectory, Outputs: outputs, + }) + if err != nil { + return "", err + } + digest := sha256.Sum256(payload) + return hex.EncodeToString(digest[:]), nil +} + +func resolveExecutable(value string) (string, error) { + resolved, err := exec.LookPath(strings.TrimSpace(value)) + if err != nil { + return "", err + } + return absolutePath(resolved) +} + +func absolutePath(value string) (string, error) { + trimmed := strings.TrimSpace(value) + if trimmed == "" { + return "", fmt.Errorf("path is required") + } + resolved, err := filepath.Abs(trimmed) + if err != nil { + return "", err + } + return filepath.Clean(resolved), nil +} + +func safePathSegment(value string) bool { + return value != "" && value != "." && value != ".." && filepath.Base(value) == value && + !strings.ContainsAny(value, `/\\`) +} + +func boundedRejectionMetadata(values []notarius.RejectionSummary) []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, + "validator_name": value.ValidatorName, "reason_code": value.ReasonCode, + }) + } + return result +} + +func boundedWarningMetadata(values []notarius.WarningSummary) []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{"scope": value.Scope, "reason_code": value.ReasonCode}) + } + return result +} diff --git a/internal/stage/extract_test.go b/internal/stage/extract_test.go new file mode 100644 index 0000000..5e21dbc --- /dev/null +++ b/internal/stage/extract_test.go @@ -0,0 +1,414 @@ +package stage + +import ( + "context" + "errors" + "os" + "path/filepath" + "reflect" + "strings" + "testing" + "time" + + "gitea.maximumdirect.net/eric/narratio/internal/adapters/notarius" + "gitea.maximumdirect.net/eric/narratio/internal/artifactmodel" + "gitea.maximumdirect.net/eric/narratio/internal/artifacts" + "gitea.maximumdirect.net/eric/narratio/internal/config" + "gitea.maximumdirect.net/eric/narratio/internal/manifest" +) + +func TestExtractStageDisabledReturnsExplicitSkip(t *testing.T) { + for _, test := range []struct { + name string + absent bool + }{ + {name: "disabled"}, + {name: "absent", absent: true}, + } { + t.Run(test.name, func(t *testing.T) { + env, m, fake := setupExtractEnv(t) + if test.absent { + env.Config.Pipeline.Notarius = nil + } else { + env.Config.Pipeline.Notarius.Enabled = false + } + + result, err := (extractStage{}).Run(context.Background(), env, m) + if err != nil { + t.Fatalf("Run() error = %v", err) + } + if result.Disposition != StageDispositionSkipped || result.SkipReason != extractSkipReason { + t.Fatalf("result disposition = %q reason = %q", result.Disposition, result.SkipReason) + } + if len(result.Outputs) != 0 || len(fake.Requests) != 0 { + t.Fatalf("outputs = %#v; requests = %#v", result.Outputs, fake.Requests) + } + }) + } +} + +func TestExtractStageIsNotInCanonicalPlansYet(t *testing.T) { + for _, candidate := range All() { + if candidate.Name() == "extract" { + t.Fatal("extract must remain directly executable until lifecycle integration is implemented") + } + } +} + +func TestExtractStageResolvesManifestInputAndBuildsExactRequest(t *testing.T) { + env, m, fake := setupExtractEnv(t) + result, err := (extractStage{}).Run(context.Background(), env, m) + if err != nil { + t.Fatalf("Run() error = %v", err) + } + if len(fake.Requests) != 1 { + t.Fatalf("requests = %d, want 1", len(fake.Requests)) + } + req := fake.Requests[0] + fixture := extractFixtureFromEnv(t, env, m) + if req.InputPath != fixture.inputPath { + t.Fatalf("input path = %q, want manifest path %q", req.InputPath, fixture.inputPath) + } + wantReceipt := artifacts.SessionRunNotariusReceiptPathForCampaign(fixture.workspace, fixture.campaign, fixture.sessionID, fixture.runID) + wantLog := artifacts.SessionRunNotariusLogPathForCampaign(fixture.workspace, fixture.campaign, fixture.sessionID, fixture.runID) + wantOutputRoot := artifacts.SessionRunNotariusOutputRootForCampaign(fixture.workspace, fixture.campaign, fixture.sessionID, fixture.runID) + if req.ConfigPath != env.Config.Pipeline.Notarius.ConfigPath || req.PipelineID != "dnd-session" || + req.OutputRoot != wantOutputRoot || req.ReceiptPath != wantReceipt || req.LogPath != wantLog || + req.WorkingDirectory != env.Config.Pipeline.Notarius.WorkingDirectory || req.Timeout != 45*time.Minute { + t.Fatalf("request = %#v", req) + } + if !filepath.IsAbs(req.Binary) { + t.Fatalf("binary = %q, want absolute resolved path", req.Binary) + } + if len(result.Outputs) != 2 { + t.Fatalf("outputs = %#v, want lane and index", result.Outputs) + } +} + +func TestExtractStageProducesImmutableManifestReadyOutputs(t *testing.T) { + env, m, fake := setupExtractEnv(t) + fixture := extractFixtureFromEnv(t, env, m) + stagingBundle := fake.Result.BundleRoot + + result, err := (extractStage{}).Run(context.Background(), env, m) + if err != nil { + t.Fatalf("Run() error = %v", err) + } + durableBundle := artifacts.SessionNotariusBundleDirForCampaign(fixture.workspace, fixture.campaign, fixture.sessionID, fixture.runID) + if result.Metadata["bundle_root"] != durableBundle || result.Metadata["narratio_run_id"] != fixture.runID { + t.Fatalf("metadata = %#v", result.Metadata) + } + if fingerprint, _ := result.Metadata["configuration_fingerprint"].(string); len(fingerprint) != 64 { + t.Fatalf("configuration fingerprint = %#v", result.Metadata["configuration_fingerprint"]) + } + 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") { + t.Fatalf("diagnostic metadata = %#v", result.Metadata) + } + if len(result.Outputs) != 2 { + t.Fatalf("outputs = %#v", result.Outputs) + } + lane := result.Outputs[0] + if lane.Kind != extractLaneOutputKind || lane.SourceID != "narratio.extraction.npc_registry" { + t.Fatalf("lane identity = %#v", lane) + } + if lane.AbsolutePath != filepath.Join(durableBundle, "lanes", "npc.json") || lane.Checksum == "" { + t.Fatalf("lane path/checksum = %#v", lane) + } + wantContract := &artifactmodel.ContractMetadata{ + MediaType: "application/json", SchemaID: "notarius.dnd.npc_registry", + SchemaVersion: "v1", ModuleKey: "dnd/npc-registry", + } + if !reflect.DeepEqual(lane.Contract, wantContract) { + t.Fatalf("lane contract = %#v, want %#v", lane.Contract, wantContract) + } + wantProvenance := &artifactmodel.ExternalProvenance{ + System: "notarius", RunID: "notarius-run-1", PipelineID: "dnd-session", ArtifactID: "npc-registry", + } + if !reflect.DeepEqual(lane.ExternalProvenance, wantProvenance) { + t.Fatalf("lane provenance = %#v, want %#v", lane.ExternalProvenance, wantProvenance) + } + index := result.Outputs[1] + if index.Kind != extractIndexOutputKind || index.SourceID != "" || index.AbsolutePath != filepath.Join(durableBundle, "index.json") || index.Checksum == "" { + t.Fatalf("index output = %#v", index) + } + if !strings.HasPrefix(lane.AbsolutePath, durableBundle+string(filepath.Separator)) || strings.HasPrefix(lane.AbsolutePath, stagingBundle+string(filepath.Separator)) { + 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", + "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 { + t.Fatalf("promoted file %q missing: %v", relative, err) + } + } + if _, err := os.Stat(filepath.Join(stagingBundle, "unknown", "private-debug.json")); err != nil { + t.Fatalf("source bundle was not preserved: %v", err) + } + for _, output := range result.Outputs { + if strings.Contains(output.AbsolutePath, "unconfigured") || strings.Contains(output.AbsolutePath, "private-debug") { + t.Fatalf("unknown file registered as output: %#v", output) + } + } + + rejections, _ := result.Metadata["rejections"].([]map[string]any) + warnings, _ := result.Metadata["warnings"].([]map[string]any) + if len(rejections) != 1 || rejections[0]["reason_code"] != "optional_rejected" || len(warnings) != 1 { + t.Fatalf("diagnostic summaries = rejections %#v warnings %#v", rejections, warnings) + } +} + +func TestExtractStageRejectsMissingOrInvalidFinalTrimmedInputBeforeInvocation(t *testing.T) { + tests := []struct { + name string + content string + remove bool + }{ + {name: "missing", remove: true}, + {name: "invalid json", content: "not-json"}, + {name: "missing segments", content: `{}`}, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + env, m, fake := setupExtractEnv(t) + fixture := extractFixtureFromEnv(t, env, m) + if test.remove { + if err := os.Remove(fixture.inputPath); err != nil { + t.Fatalf("Remove(input) error = %v", err) + } + } else if err := os.WriteFile(fixture.inputPath, []byte(test.content), 0o644); err != nil { + t.Fatalf("WriteFile(input) error = %v", err) + } + if _, err := (extractStage{}).Run(context.Background(), env, m); err == nil { + t.Fatal("Run() error = nil") + } + if len(fake.Requests) != 0 { + t.Fatalf("adapter requests = %d, want 0", len(fake.Requests)) + } + }) + } +} + +func TestExtractStageEnforcesRequiredLanePolicy(t *testing.T) { + tests := []struct { + name string + mutate func(*notarius.RunResult) + want string + }{ + {name: "missing", mutate: func(result *notarius.RunResult) { result.Index.Lanes = result.Index.Lanes[1:] }, want: "0 descriptors"}, + {name: "rejected", mutate: func(result *notarius.RunResult) { + result.Rejections = append(result.Rejections, notarius.RejectionSummary{LaneID: "npc-registry", ReasonCode: "invalid_npc"}) + }, want: "was rejected"}, + {name: "duplicate", mutate: func(result *notarius.RunResult) { + result.Index.Lanes = append(result.Index.Lanes, result.Index.Lanes[0]) + }, want: "2 descriptors"}, + {name: "media type", mutate: func(result *notarius.RunResult) { result.Index.Lanes[0].MediaType = "text/plain" }, want: "incompatible"}, + {name: "schema id", mutate: func(result *notarius.RunResult) { result.Index.Lanes[0].SchemaID = "other" }, want: "incompatible"}, + {name: "schema version", mutate: func(result *notarius.RunResult) { result.Index.Lanes[0].SchemaVersion = "v2" }, want: "incompatible"}, + {name: "module key", mutate: func(result *notarius.RunResult) { result.Index.Lanes[0].ModuleKey = "other" }, want: "incompatible"}, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + env, m, fake := setupExtractEnv(t) + test.mutate(&fake.Result) + result, err := (extractStage{}).Run(context.Background(), env, m) + if err == nil || !strings.Contains(err.Error(), test.want) { + t.Fatalf("Run() result = %#v error = %v, want %q", result, err, test.want) + } + if result != nil && len(result.Outputs) != 0 { + t.Fatalf("failure returned outputs: %#v", result.Outputs) + } + }) + } +} + +func TestExtractStageValidatesSelectedLaneJSON(t *testing.T) { + for _, test := range []struct { + name string + content string + want string + }{ + {name: "empty", content: "", want: "non-empty"}, + {name: "invalid", content: "not-json", want: "valid JSON"}, + } { + t.Run(test.name, func(t *testing.T) { + env, m, fake := setupExtractEnv(t) + if err := os.WriteFile(fake.Result.Index.Lanes[0].Path, []byte(test.content), 0o644); err != nil { + t.Fatalf("WriteFile(lane) error = %v", err) + } + if _, err := (extractStage{}).Run(context.Background(), env, m); err == nil || !strings.Contains(err.Error(), test.want) { + t.Fatalf("Run() error = %v, want %q", err, test.want) + } + }) + } +} + +func TestExtractStageAdapterAndPromotionFailuresReturnNoOutputs(t *testing.T) { + t.Run("adapter", func(t *testing.T) { + env, m, fake := setupExtractEnv(t) + fake.Err = errors.New("adapter failed") + result, err := (extractStage{}).Run(context.Background(), env, m) + if err == nil || !strings.Contains(err.Error(), "adapter failed") || result != nil { + t.Fatalf("Run() result = %#v error = %v", result, err) + } + }) + + t.Run("promotion", func(t *testing.T) { + env, m, fake := setupExtractEnv(t) + fixture := extractFixtureFromEnv(t, env, m) + destination := artifacts.SessionNotariusBundleDirForCampaign(fixture.workspace, fixture.campaign, fixture.sessionID, fixture.runID) + if err := os.MkdirAll(destination, 0o755); err != nil { + t.Fatalf("MkdirAll(destination) error = %v", err) + } + result, err := (extractStage{}).Run(context.Background(), env, m) + if err == nil || !strings.Contains(err.Error(), "promote") || result != nil { + t.Fatalf("Run() result = %#v error = %v", result, err) + } + if _, err := os.Stat(fake.Result.BundleRoot); err != nil { + t.Fatalf("failed promotion removed staging bundle: %v", err) + } + }) +} + +func TestExtractionFingerprintIsIndependentOfOutputMapOrder(t *testing.T) { + first := &config.NotariusConfig{PipelineID: "pipeline", Outputs: map[string]config.NotariusOutputConfig{ + "zeta": {LaneID: "z", MediaType: "application/json", SchemaID: "z", SchemaVersion: "v1"}, + "alpha": {LaneID: "a", MediaType: "application/json", SchemaID: "a", SchemaVersion: "v1"}, + }} + second := &config.NotariusConfig{PipelineID: "pipeline", Outputs: map[string]config.NotariusOutputConfig{ + "alpha": first.Outputs["alpha"], "zeta": first.Outputs["zeta"], + }} + one, err := extractionFingerprint("/bin/notarius", "/etc/notarius.yml", first, time.Minute, "/work") + if err != nil { + t.Fatalf("extractionFingerprint(first) error = %v", err) + } + two, err := extractionFingerprint("/bin/notarius", "/etc/notarius.yml", second, time.Minute, "/work") + if err != nil { + t.Fatalf("extractionFingerprint(second) error = %v", err) + } + if one != two { + t.Fatalf("fingerprints differ: %q != %q", one, two) + } +} + +type extractFixture struct { + workspace string + campaign string + sessionID string + runID string + inputPath string +} + +func extractFixtureFromEnv(t *testing.T, env *Env, m *manifest.Manifest) extractFixture { + t.Helper() + paths := sessionPathsForEnv(env, m.SessionID) + return extractFixture{ + workspace: paths.WorkspaceRoot, campaign: m.Campaign, sessionID: m.SessionID, + runID: m.RunID, inputPath: filepath.Join(paths.ArtifactsDir, "trimmed.from-manifest.json"), + } +} + +func setupExtractEnv(t *testing.T) (*Env, *manifest.Manifest, *notarius.FakeRunner) { + t.Helper() + workspace := t.TempDir() + campaign := "campaign-a" + sessionID := "2026-08-09" + runID := "20260809T010203Z-abcdef12" + store := artifacts.NewLocalStore(workspace) + paths, err := store.EnsureLayoutFor(campaign, 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":[{"id":1}]}`), 0o644); err != nil { + t.Fatalf("WriteFile(input) error = %v", err) + } + configPath := filepath.Join(workspace, "notarius.yml") + if err := os.WriteFile(configPath, []byte("pipelines: {}\n"), 0o644); err != nil { + t.Fatalf("WriteFile(config) error = %v", err) + } + workingDirectory := filepath.Join(workspace, "notarius-work") + if err := os.Mkdir(workingDirectory, 0o755); err != nil { + t.Fatalf("Mkdir(working directory) error = %v", err) + } + binary := filepath.Join(workspace, "notarius") + if err := os.WriteFile(binary, []byte("#!/bin/sh\nexit 0\n"), 0o755); err != nil { + t.Fatalf("WriteFile(binary) error = %v", err) + } + notariusConfig := &config.NotariusConfig{ + Enabled: true, 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", + }, + }, + } + m := manifest.New(sessionID, time.Now().UTC()) + m.Campaign = campaign + m.RunID = runID + m.MarkStageSucceeded("trim", time.Now().UTC(), []manifest.ArtifactRecord{{ + Kind: artifactmodel.TranscriptOutputKindFinalTrimmed, SourceID: artifactmodel.SourceTranscriptFinalTrimmed, + LocalPath: inputPath, + }}) + + outputRoot := artifacts.SessionRunNotariusOutputRootForCampaign(workspace, campaign, sessionID, runID) + bundle := filepath.Join(outputRoot, "notarius-run-1") + if err := os.MkdirAll(filepath.Join(bundle, "lanes"), 0o755); err != nil { + t.Fatalf("MkdirAll(bundle lanes) error = %v", err) + } + if err := os.MkdirAll(filepath.Join(bundle, "unknown"), 0o755); err != nil { + t.Fatalf("MkdirAll(bundle unknown) error = %v", err) + } + files := map[string]string{ + "index.json": `{ "manifest_file": "manifest.json" }`, + "manifest.json": `{}`, + "rejected.json": `{"rejected":[]}`, + "warnings.json": `{"warnings":[]}`, + "lanes/npc.json": `{"npcs":[]}`, + "lanes/unconfigured.json": `{"spells":[]}`, + "chunk-map.json": `{"chunks":[]}`, + "evidence-context.json": `{"source_units":[]}`, + "unknown/private-debug.json": `{"private":true}`, + } + for relative, content := range files { + if err := os.WriteFile(filepath.Join(bundle, filepath.FromSlash(relative)), []byte(content), 0o644); err != nil { + t.Fatalf("WriteFile(%q) error = %v", relative, err) + } + } + fake := ¬arius.FakeRunner{Result: notarius.RunResult{ + 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", + }, + BundleRoot: bundle, + Index: notarius.Index{ + Path: filepath.Join(bundle, "index.json"), RejectedPath: filepath.Join(bundle, "rejected.json"), + WarningsPath: filepath.Join(bundle, "warnings.json"), + Lanes: []notarius.LaneDescriptor{ + { + LaneID: "npc-registry", File: "lanes/npc.json", Path: filepath.Join(bundle, "lanes", "npc.json"), + MediaType: "application/json", ModuleKey: "dnd/npc-registry", + SchemaID: "notarius.dnd.npc_registry", SchemaVersion: "v1", + }, + {LaneID: "unconfigured", File: "lanes/unconfigured.json", Path: filepath.Join(bundle, "lanes", "unconfigured.json")}, + }, + }, + Rejections: []notarius.RejectionSummary{{LaneID: "optional", ReasonCode: "optional_rejected"}}, + Warnings: []notarius.WarningSummary{{Scope: "lane:npc-registry", ReasonCode: "normalized_name"}}, + }} + env := &Env{ + Config: &config.Config{ + Pipeline: &config.PipelineConfig{Workspace: config.WorkspaceConfig{Root: workspace}, Notarius: notariusConfig}, + Session: &config.SessionConfig{SessionID: sessionID, Campaign: campaign}, + }, + ArtifactStore: store, + Notarius: fake, + } + return env, m, fake +} diff --git a/internal/stage/stage.go b/internal/stage/stage.go index d63c0d0..b8ee604 100644 --- a/internal/stage/stage.go +++ b/internal/stage/stage.go @@ -5,6 +5,7 @@ import ( "log/slog" "gitea.maximumdirect.net/eric/narratio/internal/adapters/audita" + "gitea.maximumdirect.net/eric/narratio/internal/adapters/notarius" "gitea.maximumdirect.net/eric/narratio/internal/adapters/notify" "gitea.maximumdirect.net/eric/narratio/internal/adapters/scriptorium" "gitea.maximumdirect.net/eric/narratio/internal/adapters/seriatim" @@ -26,6 +27,7 @@ type Env struct { WhisperX whisperx.Client Seriatim seriatim.Runner Audita audita.Runner + Notarius notarius.Runner Scriptorium scriptorium.Runner ObjectStore storage.ObjectStore Notifier notify.Sender