Add the Notarius subprocess adapter

2026-08-09 23:47:38 +00:00
parent dce721cdbd
commit f9482639d4
8 changed files with 1246 additions and 3 deletions
--- a/docs/integrations/README.md
+++ b/docs/integrations/README.md
@@ -19,6 +19,8 @@ focused stage documents.
 ## Integration Contracts

 - [Audita](./audita.md): transcript polishing (`audita process`).
+- [Notarius](./notarius.md): complete pipeline execution and safe JSON bundle
+  discovery (`notarius run`).
 - [Seriatim](./seriatim.md): merge, normalize, trim, and render operations.
 - [Scriptorium](./scriptorium.md): artifact generation and debug rendering
  (`scriptorium run|render`).
--- a/docs/integrations/notarius.md
+++ b/docs/integrations/notarius.md
@@ -0,0 +1,68 @@
+# Integration: Notarius
+
+## Purpose
+
+Define the implemented Notarius subprocess and published-bundle discovery
+boundary. Extract-stage orchestration and required-lane policy are not yet
+implemented.
+
+## Invocation Contract
+
+The adapter invokes one complete pipeline with separate argument elements:
+
+```text
+notarius run <pipeline-id> --config <absolute-path> --input <absolute-path> --output-dir <absolute-path> --json
+```
+
+It does not pass `--session-id` or automatically execute
+`notarius config validate`. The configured working directory and inherited
+parent environment apply to the process. Narratio's shared subprocess timeout
+and cancellation handling bounds execution.
+
+Standard output is written directly to the configured receipt path and
+standard error to a separate log path. A nonzero exit, cancellation, or timeout
+returns the subprocess failure without interpreting receipt bytes.
+
+## Receipt Contract
+
+After exit status zero, the adapter accepts the tolerant
+`notarius.run-result.v1` receipt. It requires run and pipeline identity, an
+absolute run-specific output directory, the relative production index path,
+non-negative output counts, and validation status. The receipt pipeline must
+match the requested pipeline.
+
+Receipt, index, rejection, and warning files have explicit read limits.
+Unknown fields are tolerated within supported document shapes.
+
+## Bundle Discovery
+
+The receipt's output directory must be strictly beneath the absolute output
+root supplied by Narratio. The adapter resolves the receipt index and every
+index path beneath that bundle root.
+
+The production index must provide manifest, rejection, warning, and lane-list
+fields. Lane descriptors require unique non-empty lane IDs and relative file
+paths. Optional chunk-map and evidence-context descriptors require their full
+generic contract metadata when present.
+
+Every consumed path must remain lexically confined and identify a regular file
+through directory components that are not symlinks. Absolute logical paths,
+traversal, prefix-confusion escapes, symlinks, directories, and special files
+are rejected.
+
+The adapter returns lane and pipeline-wide descriptors without deciding which
+lanes a caller requires or comparing their contracts to campaign policy. It
+does not read lane payload bodies or interpret D&D data.
+
+## Diagnostic Summaries
+
+`rejected.json` and `warnings.json` are decoded tolerantly after bounded reads.
+The result retains structured fields such as stage, scope, lane ID, and reason
+code. Free-form external messages are validated as required by the upstream
+shape but are not returned in the summaries.
+
+## Internal Implementation
+
+The transport-neutral interface, production runner, and configurable fake live
+in `internal/adapters/notarius`. Composition and adapter ownership are
+documented in [Adapter Internals](../internal/adapters.md).
--- a/docs/internal/adapters.md
+++ b/docs/internal/adapters.md
@@ -16,6 +16,7 @@ Primary adapters:
 - `seriatim.Runner`
 - `audita.Runner`
 - `scriptorium.Runner`
+- `notarius.Runner`
 - `storage.ObjectStore`
 - `notify.Sender`

@@ -43,6 +44,9 @@ Adapters do not own:
 - Noop notifier (`notify.NoopSender`).
 - Object store only when required by selected stages/config.

+The Notarius subprocess adapter is implemented independently of stage policy;
+application composition and extract-stage wiring are not yet implemented.
+
 Object-store construction goes through `newCommandObjectStore`, which loads
 configured filesystem secrets before adapter initialization.

@@ -56,16 +60,18 @@ configured filesystem secrets before adapter initialization.

 - Composition: `internal/app/runner.go`, `internal/app/object_store.go`
 - Shared subprocess mechanics: `internal/adapters/subprocess`
- Focused adapters: `internal/adapters/{whisperx,seriatim,audita,scriptorium,storage,notify}`
+- Focused adapters: `internal/adapters/{whisperx,seriatim,audita,scriptorium,notarius,storage,notify}`
 - `internal/adapters/whisperx/http_test.go`
 - `internal/adapters/seriatim/subprocess_test.go`
 - `internal/adapters/audita/subprocess_test.go`
 - `internal/adapters/scriptorium/subprocess_test.go`
+- `internal/adapters/notarius/subprocess_test.go`
 - `internal/adapters/storage/*_test.go`
 - `internal/app/runner_test.go`

 See the [WhisperX](../integrations/whisperx.md),
 [Seriatim](../integrations/seriatim.md), [Audita](../integrations/audita.md),
-and [Scriptorium](../integrations/scriptorium.md) contracts before changing an
+[Scriptorium](../integrations/scriptorium.md), and
+[Notarius](../integrations/notarius.md) contracts before changing an
 externally visible boundary. Operator-selected values belong in
 [Configuration](../config.md).
--- a/docs/roadmap/implementation.md
+++ b/docs/roadmap/implementation.md
@@ -19,7 +19,7 @@ implementation sequence.
 | Stage 1 | Complete |
 | Stage 2 | Complete |
 | Stage 3 | Complete |
-| Stage 4 | Not started |
+| Stage 4 | Complete |
 | Stage 5 | Not started |
 | Stage 6 | Not started |
 | Stage 7 | Not started |
--- a/internal/adapters/notarius/fake.go
+++ b/internal/adapters/notarius/fake.go
@@ -0,0 +1,22 @@
+package notarius
+
+import "context"
+
+// FakeRunner is a configurable in-memory runner for stage tests.
+type FakeRunner struct {
+	Requests []RunRequest
+	Result   RunResult
+	Err      error
+}
+
+// Run records the request and returns the configured result or error.
+func (f *FakeRunner) Run(ctx context.Context, req RunRequest) (RunResult, error) {
+	if err := ctx.Err(); err != nil {
+		return RunResult{}, err
+	}
+	f.Requests = append(f.Requests, req)
+	if f.Err != nil {
+		return RunResult{}, f.Err
+	}
+	return f.Result, nil
+}
--- a/internal/adapters/notarius/runner.go
+++ b/internal/adapters/notarius/runner.go
@@ -0,0 +1,108 @@
+// Package notarius declares the adapter contract for Notarius CLI invocations.
+package notarius
+
+import (
+	"context"
+	"time"
+)
+
+const ReceiptSchemaVersion = "notarius.run-result.v1"
+
+// Runner is the adapter boundary for a complete Notarius pipeline invocation.
+type Runner interface {
+	Run(ctx context.Context, req RunRequest) (RunResult, error)
+}
+
+// RunRequest contains the resolved inputs and diagnostic destinations for one invocation.
+type RunRequest struct {
+	Binary           string
+	ConfigPath       string
+	PipelineID       string
+	InputPath        string
+	OutputRoot       string
+	WorkingDirectory string
+	ReceiptPath      string
+	LogPath          string
+	Timeout          time.Duration
+}
+
+// Receipt is the transport-neutral successful run receipt.
+type Receipt struct {
+	SchemaVersion         string
+	RunID                 string
+	PipelineID            string
+	OutputDirectory       string
+	IndexFile             string
+	NormalizedOutputCount int
+	RejectedOutputCount   int
+	WarningCount          int
+	ValidationStatus      string
+	DebugDirectory        string
+}
+
+// LaneDescriptor identifies one normalized lane payload discovered through the index.
+type LaneDescriptor struct {
+	LaneID        string
+	File          string
+	Path          string
+	MediaType     string
+	ModuleKey     string
+	SchemaID      string
+	SchemaName    string
+	SchemaVersion string
+}
+
+// PipelineDescriptor identifies a pipeline-wide artifact discovered through the index.
+type PipelineDescriptor struct {
+	ArtifactKind  string
+	File          string
+	Path          string
+	MediaType     string
+	SchemaID      string
+	SchemaName    string
+	SchemaVersion string
+}
+
+// Index describes the validated bundle-management and artifact paths.
+type Index struct {
+	Path            string
+	ManifestFile    string
+	ManifestPath    string
+	RejectedFile    string
+	RejectedPath    string
+	WarningsFile    string
+	WarningsPath    string
+	Lanes           []LaneDescriptor
+	ChunkMap        *PipelineDescriptor
+	EvidenceContext *PipelineDescriptor
+}
+
+// RejectionSummary retains structured rejection identity without free-form messages.
+type RejectionSummary struct {
+	Stage         string
+	StepID        string
+	LaneID        string
+	ModuleKey     string
+	ChunkID       string
+	ValidatorName string
+	ReasonCode    string
+}
+
+// WarningSummary retains structured warning identity without free-form messages.
+type WarningSummary struct {
+	Scope      string
+	ReasonCode string
+}
+
+// RunResult describes a successfully decoded and validated Notarius bundle.
+type RunResult struct {
+	Receipt     Receipt
+	Index       Index
+	BundleRoot  string
+	ReceiptPath string
+	LogPath     string
+	ExitCode    int
+	Duration    time.Duration
+	Rejections  []RejectionSummary
+	Warnings    []WarningSummary
+}
--- a/internal/adapters/notarius/subprocess.go
+++ b/internal/adapters/notarius/subprocess.go
@@ -0,0 +1,505 @@
+package notarius
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"os"
+	"path/filepath"
+	"strings"
+
+	"gitea.maximumdirect.net/eric/narratio/internal/adapters/subprocess"
+	"gitea.maximumdirect.net/eric/narratio/internal/pathsafe"
+)
+
+const (
+	maxReceiptBytes = 1 << 20
+	maxIndexBytes   = 4 << 20
+	maxSummaryBytes = 4 << 20
+)
+
+type subprocessRun func(context.Context, subprocess.RunRequest) (subprocess.RunResult, error)
+
+// SubprocessRunner invokes Notarius through its public CLI.
+type SubprocessRunner struct {
+	run subprocessRun
+}
+
+// NewSubprocessRunner constructs a production Notarius subprocess runner.
+func NewSubprocessRunner() *SubprocessRunner {
+	return &SubprocessRunner{run: subprocess.Run}
+}
+
+// Run executes a complete Notarius pipeline and discovers its published bundle.
+func (r *SubprocessRunner) Run(ctx context.Context, req RunRequest) (RunResult, error) {
+	if r == nil || r.run == nil {
+		return RunResult{}, fmt.Errorf("notarius subprocess runner is nil")
+	}
+	if err := validateRunRequest(req); err != nil {
+		return RunResult{}, err
+	}
+
+	args := []string{
+		"run", req.PipelineID,
+		"--config", req.ConfigPath,
+		"--input", req.InputPath,
+		"--output-dir", req.OutputRoot,
+		"--json",
+	}
+	processResult, err := r.run(ctx, subprocess.RunRequest{
+		Executable:    req.Binary,
+		Args:          args,
+		WorkingDir:    req.WorkingDirectory,
+		Timeout:       req.Timeout,
+		StdoutLogPath: req.ReceiptPath,
+		StderrLogPath: req.LogPath,
+	})
+	baseResult := RunResult{
+		ReceiptPath: req.ReceiptPath,
+		LogPath:     req.LogPath,
+		ExitCode:    processResult.ExitCode,
+		Duration:    processResult.Duration,
+	}
+	if err != nil {
+		return baseResult, fmt.Errorf("run notarius pipeline %q: %w", req.PipelineID, err)
+	}
+
+	receipt, err := loadReceipt(req.ReceiptPath, req.PipelineID)
+	if err != nil {
+		return baseResult, err
+	}
+	bundleRoot, err := validateBundleRoot(req.OutputRoot, receipt.OutputDirectory)
+	if err != nil {
+		return baseResult, err
+	}
+	indexPath, err := resolveRegularFile(bundleRoot, receipt.IndexFile)
+	if err != nil {
+		return baseResult, fmt.Errorf("resolve receipt index file: %w", err)
+	}
+	index, err := loadIndex(bundleRoot, indexPath)
+	if err != nil {
+		return baseResult, err
+	}
+	rejections, err := loadRejections(index.RejectedPath)
+	if err != nil {
+		return baseResult, err
+	}
+	warnings, err := loadWarnings(index.WarningsPath)
+	if err != nil {
+		return baseResult, err
+	}
+
+	baseResult.Receipt = receipt
+	baseResult.Index = index
+	baseResult.BundleRoot = bundleRoot
+	baseResult.Rejections = rejections
+	baseResult.Warnings = warnings
+	return baseResult, nil
+}
+
+func validateRunRequest(req RunRequest) error {
+	if strings.TrimSpace(req.Binary) == "" {
+		return fmt.Errorf("notarius binary is required")
+	}
+	if strings.TrimSpace(req.PipelineID) == "" {
+		return fmt.Errorf("notarius pipeline id is required")
+	}
+	if req.Timeout <= 0 {
+		return fmt.Errorf("notarius timeout must be positive")
+	}
+	for label, path := range map[string]string{
+		"config":            req.ConfigPath,
+		"input":             req.InputPath,
+		"output root":       req.OutputRoot,
+		"working directory": req.WorkingDirectory,
+		"receipt":           req.ReceiptPath,
+		"log":               req.LogPath,
+	} {
+		if strings.TrimSpace(path) == "" {
+			return fmt.Errorf("notarius %s path is required", label)
+		}
+		if !filepath.IsAbs(path) {
+			return fmt.Errorf("notarius %s path must be absolute", label)
+		}
+	}
+	if filepath.Clean(req.ReceiptPath) == filepath.Clean(req.LogPath) {
+		return fmt.Errorf("notarius receipt and log paths must be different")
+	}
+	if err := requireRegularFile(req.ConfigPath); err != nil {
+		return fmt.Errorf("validate notarius config path: %w", err)
+	}
+	if err := requireRegularFile(req.InputPath); err != nil {
+		return fmt.Errorf("validate notarius input path: %w", err)
+	}
+	if err := requireDirectory(req.OutputRoot); err != nil {
+		return fmt.Errorf("validate notarius output root: %w", err)
+	}
+	if err := requireDirectory(req.WorkingDirectory); err != nil {
+		return fmt.Errorf("validate notarius working directory: %w", err)
+	}
+	if err := validateLogDestination(req.ReceiptPath); err != nil {
+		return fmt.Errorf("validate notarius receipt path: %w", err)
+	}
+	if err := validateLogDestination(req.LogPath); err != nil {
+		return fmt.Errorf("validate notarius log path: %w", err)
+	}
+	return nil
+}
+
+type receiptDocument struct {
+	SchemaVersion         string `json:"schema_version"`
+	RunID                 string `json:"run_id"`
+	PipelineID            string `json:"pipeline_id"`
+	OutputDirectory       string `json:"output_directory"`
+	IndexFile             string `json:"index_file"`
+	NormalizedOutputCount *int   `json:"normalized_output_count"`
+	RejectedOutputCount   *int   `json:"rejected_output_count"`
+	WarningCount          *int   `json:"warning_count"`
+	ValidationStatus      string `json:"validation_status"`
+	DebugDirectory        string `json:"debug_directory"`
+}
+
+func loadReceipt(path, pipelineID string) (Receipt, error) {
+	var document receiptDocument
+	if err := decodeBoundedJSON(path, maxReceiptBytes, &document); err != nil {
+		return Receipt{}, fmt.Errorf("decode notarius receipt: %w", err)
+	}
+	if document.SchemaVersion != ReceiptSchemaVersion {
+		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 ||
+		document.RejectedOutputCount == nil || document.WarningCount == nil {
+		return Receipt{}, fmt.Errorf("notarius receipt is missing required fields")
+	}
+	if document.PipelineID != pipelineID {
+		return Receipt{}, fmt.Errorf("notarius receipt pipeline id %q does not match requested pipeline %q", document.PipelineID, pipelineID)
+	}
+	if *document.NormalizedOutputCount < 0 || *document.RejectedOutputCount < 0 || *document.WarningCount < 0 {
+		return Receipt{}, fmt.Errorf("notarius receipt counts must be non-negative")
+	}
+	if !filepath.IsAbs(document.OutputDirectory) {
+		return Receipt{}, fmt.Errorf("notarius receipt output directory must be absolute")
+	}
+	if document.DebugDirectory != "" && !filepath.IsAbs(document.DebugDirectory) {
+		return Receipt{}, fmt.Errorf("notarius receipt debug directory must be absolute when present")
+	}
+	return Receipt{
+		SchemaVersion:         document.SchemaVersion,
+		RunID:                 document.RunID,
+		PipelineID:            document.PipelineID,
+		OutputDirectory:       filepath.Clean(document.OutputDirectory),
+		IndexFile:             document.IndexFile,
+		NormalizedOutputCount: *document.NormalizedOutputCount,
+		RejectedOutputCount:   *document.RejectedOutputCount,
+		WarningCount:          *document.WarningCount,
+		ValidationStatus:      document.ValidationStatus,
+		DebugDirectory:        document.DebugDirectory,
+	}, nil
+}
+
+type indexDocument struct {
+	ManifestFile    string            `json:"manifest_file"`
+	OutputFiles     *[]laneDocument   `json:"output_files"`
+	RejectedFile    string            `json:"rejected_file"`
+	WarningsFile    string            `json:"warnings_file"`
+	ChunkMap        *pipelineDocument `json:"chunk_map"`
+	EvidenceContext *pipelineDocument `json:"evidence_context"`
+}
+
+type laneDocument struct {
+	LaneID        string `json:"lane_id"`
+	File          string `json:"file"`
+	MediaType     string `json:"media_type"`
+	ModuleKey     string `json:"module_key"`
+	SchemaID      string `json:"schema_id"`
+	SchemaName    string `json:"schema_name"`
+	SchemaVersion string `json:"schema_version"`
+}
+
+type pipelineDocument struct {
+	ArtifactKind  string `json:"artifact_kind"`
+	File          string `json:"file"`
+	MediaType     string `json:"media_type"`
+	SchemaID      string `json:"schema_id"`
+	SchemaName    string `json:"schema_name"`
+	SchemaVersion string `json:"schema_version"`
+}
+
+func loadIndex(bundleRoot, indexPath string) (Index, error) {
+	var document indexDocument
+	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")
+	}
+
+	index := Index{
+		Path:         indexPath,
+		ManifestFile: document.ManifestFile,
+		RejectedFile: document.RejectedFile,
+		WarningsFile: document.WarningsFile,
+	}
+	var err error
+	if index.ManifestPath, err = resolveRegularFile(bundleRoot, index.ManifestFile); err != nil {
+		return Index{}, fmt.Errorf("resolve notarius manifest file: %w", err)
+	}
+	if index.RejectedPath, err = resolveRegularFile(bundleRoot, index.RejectedFile); err != nil {
+		return Index{}, fmt.Errorf("resolve notarius rejection file: %w", err)
+	}
+	if index.WarningsPath, err = resolveRegularFile(bundleRoot, index.WarningsFile); err != nil {
+		return Index{}, fmt.Errorf("resolve notarius warning file: %w", err)
+	}
+
+	seenLanes := make(map[string]struct{}, len(*document.OutputFiles))
+	for _, lane := range *document.OutputFiles {
+		if strings.TrimSpace(lane.LaneID) == "" || strings.TrimSpace(lane.File) == "" {
+			return Index{}, fmt.Errorf("notarius lane descriptors require lane_id and file")
+		}
+		if _, exists := seenLanes[lane.LaneID]; exists {
+			return Index{}, fmt.Errorf("notarius index contains duplicate lane id %q", lane.LaneID)
+		}
+		seenLanes[lane.LaneID] = struct{}{}
+		path, err := resolveRegularFile(bundleRoot, lane.File)
+		if err != nil {
+			return Index{}, fmt.Errorf("resolve notarius lane %q file: %w", lane.LaneID, err)
+		}
+		index.Lanes = append(index.Lanes, LaneDescriptor{
+			LaneID: lane.LaneID, File: lane.File, Path: path, MediaType: lane.MediaType,
+			ModuleKey: lane.ModuleKey, SchemaID: lane.SchemaID, SchemaName: lane.SchemaName,
+			SchemaVersion: lane.SchemaVersion,
+		})
+	}
+	if document.ChunkMap != nil {
+		index.ChunkMap, err = resolvePipelineDescriptor(bundleRoot, "chunk_map", *document.ChunkMap)
+		if err != nil {
+			return Index{}, err
+		}
+	}
+	if document.EvidenceContext != nil {
+		index.EvidenceContext, err = resolvePipelineDescriptor(bundleRoot, "evidence_context", *document.EvidenceContext)
+		if err != nil {
+			return Index{}, err
+		}
+	}
+	return index, nil
+}
+
+func resolvePipelineDescriptor(bundleRoot, label string, document pipelineDocument) (*PipelineDescriptor, error) {
+	if strings.TrimSpace(document.ArtifactKind) == "" || strings.TrimSpace(document.File) == "" ||
+		strings.TrimSpace(document.MediaType) == "" || strings.TrimSpace(document.SchemaID) == "" ||
+		strings.TrimSpace(document.SchemaName) == "" || strings.TrimSpace(document.SchemaVersion) == "" {
+		return nil, fmt.Errorf("notarius %s descriptor is missing required fields", label)
+	}
+	path, err := resolveRegularFile(bundleRoot, document.File)
+	if err != nil {
+		return nil, fmt.Errorf("resolve notarius %s file: %w", label, err)
+	}
+	return &PipelineDescriptor{
+		ArtifactKind: document.ArtifactKind, File: document.File, Path: path,
+		MediaType: document.MediaType, SchemaID: document.SchemaID,
+		SchemaName: document.SchemaName, SchemaVersion: document.SchemaVersion,
+	}, nil
+}
+
+type rejectionDocument struct {
+	Rejected *[]struct {
+		Stage         string `json:"stage"`
+		StepID        string `json:"step_id"`
+		LaneID        string `json:"lane_id"`
+		ModuleKey     string `json:"module_key"`
+		ChunkID       string `json:"chunk_id"`
+		ValidatorName string `json:"validator_name"`
+		ReasonCode    string `json:"reason_code"`
+		Message       string `json:"message"`
+	} `json:"rejected"`
+}
+
+func loadRejections(path string) ([]RejectionSummary, error) {
+	var document rejectionDocument
+	if err := decodeBoundedJSON(path, maxSummaryBytes, &document); err != nil {
+		return nil, fmt.Errorf("decode notarius rejections: %w", err)
+	}
+	if document.Rejected == nil {
+		return nil, fmt.Errorf("notarius rejection document is missing rejected array")
+	}
+	summaries := make([]RejectionSummary, 0, len(*document.Rejected))
+	for _, item := range *document.Rejected {
+		if strings.TrimSpace(item.Stage) == "" || strings.TrimSpace(item.Message) == "" {
+			return nil, fmt.Errorf("notarius rejection entries require stage and message")
+		}
+		summaries = append(summaries, RejectionSummary{
+			Stage: item.Stage, StepID: item.StepID, LaneID: item.LaneID,
+			ModuleKey: item.ModuleKey, ChunkID: item.ChunkID,
+			ValidatorName: item.ValidatorName, ReasonCode: item.ReasonCode,
+		})
+	}
+	return summaries, nil
+}
+
+type warningDocument struct {
+	Warnings *[]struct {
+		Scope      string `json:"scope"`
+		ReasonCode string `json:"reason_code"`
+		Message    string `json:"message"`
+	} `json:"warnings"`
+}
+
+func loadWarnings(path string) ([]WarningSummary, error) {
+	var document warningDocument
+	if err := decodeBoundedJSON(path, maxSummaryBytes, &document); err != nil {
+		return nil, fmt.Errorf("decode notarius warnings: %w", err)
+	}
+	if document.Warnings == nil {
+		return nil, fmt.Errorf("notarius warning document is missing warnings array")
+	}
+	summaries := make([]WarningSummary, 0, len(*document.Warnings))
+	for _, item := range *document.Warnings {
+		if strings.TrimSpace(item.ReasonCode) == "" || strings.TrimSpace(item.Message) == "" {
+			return nil, fmt.Errorf("notarius warning entries require reason_code and message")
+		}
+		summaries = append(summaries, WarningSummary{Scope: item.Scope, ReasonCode: item.ReasonCode})
+	}
+	return summaries, nil
+}
+
+func decodeBoundedJSON(path string, limit int64, destination any) error {
+	inspected, err := os.Lstat(path)
+	if err != nil {
+		return err
+	}
+	if inspected.Mode()&os.ModeSymlink != 0 || !inspected.Mode().IsRegular() {
+		return fmt.Errorf("path %q must be a regular file without symlinks", path)
+	}
+	file, err := os.Open(path)
+	if err != nil {
+		return err
+	}
+	defer func() { _ = file.Close() }()
+	opened, err := file.Stat()
+	if err != nil {
+		return err
+	}
+	if !opened.Mode().IsRegular() || !os.SameFile(inspected, opened) {
+		return fmt.Errorf("file %q changed before it could be read", path)
+	}
+	reader := io.LimitReader(file, limit+1)
+	data, err := io.ReadAll(reader)
+	if err != nil {
+		return err
+	}
+	if int64(len(data)) > limit {
+		return fmt.Errorf("file %q exceeds %d-byte limit", path, limit)
+	}
+	if err := json.Unmarshal(data, destination); err != nil {
+		return err
+	}
+	return nil
+}
+
+func validateBundleRoot(outputRoot, bundleRoot string) (string, error) {
+	root := filepath.Clean(outputRoot)
+	bundle := filepath.Clean(bundleRoot)
+	relative, err := filepath.Rel(root, bundle)
+	if err != nil {
+		return "", fmt.Errorf("compare notarius output paths: %w", err)
+	}
+	if relative == "." || relative == ".." || strings.HasPrefix(relative, ".."+string(filepath.Separator)) {
+		return "", fmt.Errorf("notarius output directory %q is not beneath output root %q", bundleRoot, outputRoot)
+	}
+	if err := requireDirectoryTree(root, relative); err != nil {
+		return "", fmt.Errorf("validate notarius output directory: %w", err)
+	}
+	return bundle, nil
+}
+
+func resolveRegularFile(root, logicalPath string) (string, error) {
+	resolved, err := pathsafe.JoinSlashRelativeUnderRoot(root, logicalPath)
+	if err != nil {
+		return "", err
+	}
+	relative, err := filepath.Rel(root, resolved)
+	if err != nil {
+		return "", err
+	}
+	if err := requireRegularFileTree(root, relative); err != nil {
+		return "", err
+	}
+	return resolved, nil
+}
+
+func requireDirectoryTree(root, relative string) error {
+	if err := requireDirectory(root); err != nil {
+		return err
+	}
+	current := root
+	for _, component := range strings.Split(relative, string(filepath.Separator)) {
+		current = filepath.Join(current, component)
+		if err := requireDirectory(current); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func requireRegularFileTree(root, relative string) error {
+	components := strings.Split(relative, string(filepath.Separator))
+	if len(components) == 0 {
+		return fmt.Errorf("regular file path is required")
+	}
+	if err := requireDirectory(root); err != nil {
+		return err
+	}
+	current := root
+	for _, component := range components[:len(components)-1] {
+		current = filepath.Join(current, component)
+		if err := requireDirectory(current); err != nil {
+			return err
+		}
+	}
+	return requireRegularFile(filepath.Join(current, components[len(components)-1]))
+}
+
+func requireDirectory(path string) error {
+	info, err := os.Lstat(path)
+	if err != nil {
+		return err
+	}
+	if info.Mode()&os.ModeSymlink != 0 || !info.IsDir() {
+		return fmt.Errorf("path %q must be a directory without symlinks", path)
+	}
+	return nil
+}
+
+func requireRegularFile(path 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)
+	}
+	return nil
+}
+
+func validateLogDestination(path string) error {
+	if err := requireDirectory(filepath.Dir(path)); err != nil {
+		return err
+	}
+	info, err := os.Lstat(path)
+	if errors.Is(err, os.ErrNotExist) {
+		return nil
+	}
+	if err != nil {
+		return err
+	}
+	if info.Mode()&os.ModeSymlink != 0 || !info.Mode().IsRegular() {
+		return fmt.Errorf("path %q must be absent or a regular file without symlinks", path)
+	}
+	return nil
+}
--- a/internal/adapters/notarius/subprocess_test.go
+++ b/internal/adapters/notarius/subprocess_test.go
@@ -0,0 +1,532 @@
+package notarius
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"os"
+	"path/filepath"
+	"reflect"
+	"strings"
+	"testing"
+	"time"
+
+	sharedsubprocess "gitea.maximumdirect.net/eric/narratio/internal/adapters/subprocess"
+)
+
+func TestSubprocessRunnerBuildsExactInvocationAndDiscoversBundle(t *testing.T) {
+	req := validRunRequest(t)
+	var captured sharedsubprocess.RunRequest
+	runner := &SubprocessRunner{run: func(_ context.Context, processReq sharedsubprocess.RunRequest) (sharedsubprocess.RunResult, error) {
+		captured = processReq
+		writeValidBundleAndReceipt(t, req, true)
+		return sharedsubprocess.RunResult{ExitCode: 0, Duration: 2 * time.Second}, nil
+	}}
+
+	result, err := runner.Run(context.Background(), req)
+	if err != nil {
+		t.Fatalf("Run() error = %v", err)
+	}
+	wantArgs := []string{
+		"run", "dnd-session", "--config", req.ConfigPath, "--input", req.InputPath,
+		"--output-dir", req.OutputRoot, "--json",
+	}
+	if !reflect.DeepEqual(captured.Args, wantArgs) {
+		t.Fatalf("subprocess args = %#v, want %#v", captured.Args, wantArgs)
+	}
+	if captured.Executable != req.Binary || captured.WorkingDir != req.WorkingDirectory || captured.Timeout != req.Timeout {
+		t.Fatalf("subprocess request = %#v", captured)
+	}
+	if captured.StdoutLogPath != req.ReceiptPath || captured.StderrLogPath != req.LogPath {
+		t.Fatalf("stream paths = stdout %q stderr %q", captured.StdoutLogPath, captured.StderrLogPath)
+	}
+	if captured.EnvOverrides != nil {
+		t.Fatalf("environment overrides = %#v, want inherited environment only", captured.EnvOverrides)
+	}
+	for _, arg := range captured.Args {
+		if arg == "--session-id" {
+			t.Fatal("subprocess args unexpectedly contain --session-id")
+		}
+	}
+
+	if result.Receipt.SchemaVersion != ReceiptSchemaVersion || result.Receipt.RunID != "notarius-run-1" {
+		t.Fatalf("receipt = %#v", result.Receipt)
+	}
+	if len(result.Index.Lanes) != 1 || result.Index.Lanes[0].LaneID != "npc-registry" {
+		t.Fatalf("lanes = %#v", result.Index.Lanes)
+	}
+	if result.Index.ChunkMap == nil || result.Index.ChunkMap.ArtifactKind != "chunk_map" {
+		t.Fatalf("chunk map = %#v", result.Index.ChunkMap)
+	}
+	if result.Index.EvidenceContext == nil || result.Index.EvidenceContext.ArtifactKind != "evidence_context" {
+		t.Fatalf("evidence context = %#v", result.Index.EvidenceContext)
+	}
+	if len(result.Rejections) != 1 || result.Rejections[0].LaneID != "spells" || result.Rejections[0].ReasonCode != "invalid_spell" {
+		t.Fatalf("rejections = %#v", result.Rejections)
+	}
+	if len(result.Warnings) != 1 || result.Warnings[0].Scope != "lane:npc-registry" || result.Warnings[0].ReasonCode != "normalized_name" {
+		t.Fatalf("warnings = %#v", result.Warnings)
+	}
+}
+
+func TestSubprocessRunnerInheritsEnvironmentAndSeparatesStreams(t *testing.T) {
+	req := validRunRequest(t)
+	writeValidBundleAndReceipt(t, req, false)
+	receiptFixture := req.ReceiptPath + ".fixture"
+	data, err := os.ReadFile(req.ReceiptPath)
+	if err != nil {
+		t.Fatalf("ReadFile(receipt) error = %v", err)
+	}
+	if err := os.WriteFile(receiptFixture, data, 0o644); err != nil {
+		t.Fatalf("WriteFile(receipt fixture) error = %v", err)
+	}
+	if err := os.Remove(req.ReceiptPath); err != nil {
+		t.Fatalf("Remove(receipt) error = %v", err)
+	}
+
+	captureDir := filepath.Join(filepath.Dir(req.ReceiptPath), "capture")
+	if err := os.Mkdir(captureDir, 0o755); err != nil {
+		t.Fatalf("Mkdir(capture) error = %v", err)
+	}
+	script := writeShellScript(t, `#!/bin/sh
+pwd > "$NOTARIUS_CAPTURE_DIR/working-directory"
+printf '%s' "$NOTARIUS_INHERITED_VALUE" > "$NOTARIUS_CAPTURE_DIR/environment"
+printf 'diagnostic stream\n' >&2
+cat "$NOTARIUS_RECEIPT_FIXTURE"
+`)
+	req.Binary = script
+	t.Setenv("NOTARIUS_CAPTURE_DIR", captureDir)
+	t.Setenv("NOTARIUS_INHERITED_VALUE", "inherited-value")
+	t.Setenv("NOTARIUS_RECEIPT_FIXTURE", receiptFixture)
+
+	if _, err := NewSubprocessRunner().Run(context.Background(), req); err != nil {
+		t.Fatalf("Run() error = %v", err)
+	}
+	assertTextFile(t, filepath.Join(captureDir, "working-directory"), req.WorkingDirectory+"\n")
+	assertTextFile(t, filepath.Join(captureDir, "environment"), "inherited-value")
+	assertTextFile(t, req.LogPath, "diagnostic stream\n")
+	receiptBytes, err := os.ReadFile(req.ReceiptPath)
+	if err != nil {
+		t.Fatalf("ReadFile(receipt) error = %v", err)
+	}
+	if strings.Contains(string(receiptBytes), "diagnostic stream") {
+		t.Fatal("receipt contains stderr output")
+	}
+}
+
+func TestSubprocessRunnerReturnsProcessFailuresWithoutParsingStdout(t *testing.T) {
+	tests := []struct {
+		name       string
+		scriptBody string
+		timeout    time.Duration
+		cancel     bool
+		want       string
+	}{
+		{name: "nonzero", scriptBody: "printf '{malformed receipt'; printf 'failed\\n' >&2; exit 7\n", timeout: time.Second, want: "exit code 7"},
+		{name: "timeout", scriptBody: "sleep 5\n", timeout: 20 * time.Millisecond, want: "timed out"},
+		{name: "cancellation", scriptBody: "sleep 5\n", timeout: time.Second, cancel: true, want: "canceled"},
+	}
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			req := validRunRequest(t)
+			req.Binary = writeShellScript(t, "#!/bin/sh\n"+test.scriptBody)
+			req.Timeout = test.timeout
+			ctx := context.Background()
+			if test.cancel {
+				cancelCtx, cancel := context.WithCancel(ctx)
+				ctx = cancelCtx
+				time.AfterFunc(20*time.Millisecond, cancel)
+			}
+			_, err := NewSubprocessRunner().Run(ctx, req)
+			if err == nil || !strings.Contains(err.Error(), test.want) {
+				t.Fatalf("Run() error = %v, want fragment %q", err, test.want)
+			}
+			if strings.Contains(err.Error(), "decode notarius receipt") {
+				t.Fatalf("Run() parsed stdout after process failure: %v", err)
+			}
+		})
+	}
+}
+
+func TestSubprocessRunnerReturnsSharedSubprocessErrorWithoutReadingReceipt(t *testing.T) {
+	req := validRunRequest(t)
+	if err := os.WriteFile(req.ReceiptPath, []byte("not json"), 0o644); err != nil {
+		t.Fatalf("WriteFile(receipt) error = %v", err)
+	}
+	wantErr := errors.New("process failed")
+	runner := &SubprocessRunner{run: func(context.Context, sharedsubprocess.RunRequest) (sharedsubprocess.RunResult, error) {
+		return sharedsubprocess.RunResult{ExitCode: 9}, wantErr
+	}}
+	_, err := runner.Run(context.Background(), req)
+	if !errors.Is(err, wantErr) {
+		t.Fatalf("Run() error = %v, want wrapped process error", err)
+	}
+	if strings.Contains(err.Error(), "decode") {
+		t.Fatalf("Run() parsed receipt after failure: %v", err)
+	}
+}
+
+func TestLoadReceiptValidation(t *testing.T) {
+	root := t.TempDir()
+	valid := map[string]any{
+		"schema_version": ReceiptSchemaVersion, "run_id": "run-1", "pipeline_id": "pipeline-1",
+		"output_directory": filepath.Join(root, "outputs", "run-1"), "index_file": "index.json",
+		"normalized_output_count": 1, "rejected_output_count": 0, "warning_count": 0,
+		"validation_status": "approved", "future_field": true,
+	}
+	tests := []struct {
+		name   string
+		mutate func(map[string]any)
+		raw    []byte
+		wantOK bool
+	}{
+		{name: "unknown fields tolerated", wantOK: true},
+		{name: "malformed", raw: []byte("{")},
+		{name: "unsupported version", mutate: func(v map[string]any) { v["schema_version"] = "notarius.run-result.v2" }},
+		{name: "missing field", mutate: func(v map[string]any) { delete(v, "run_id") }},
+		{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 }},
+	}
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			path := filepath.Join(root, strings.ReplaceAll(test.name, " ", "-")+".json")
+			values := cloneMap(valid)
+			if test.mutate != nil {
+				test.mutate(values)
+			}
+			if test.raw != nil {
+				if err := os.WriteFile(path, test.raw, 0o644); err != nil {
+					t.Fatalf("WriteFile() error = %v", err)
+				}
+			} else {
+				writeJSONFile(t, path, values)
+			}
+			_, err := loadReceipt(path, "pipeline-1")
+			if test.wantOK && err != nil {
+				t.Fatalf("loadReceipt() error = %v", err)
+			}
+			if !test.wantOK && err == nil {
+				t.Fatal("loadReceipt() error = nil, want validation failure")
+			}
+		})
+	}
+
+	oversized := filepath.Join(root, "oversized.json")
+	if err := os.WriteFile(oversized, []byte(strings.Repeat("x", maxReceiptBytes+1)), 0o644); err != nil {
+		t.Fatalf("WriteFile(oversized) error = %v", err)
+	}
+	if _, err := loadReceipt(oversized, "pipeline-1"); err == nil || !strings.Contains(err.Error(), "exceeds") {
+		t.Fatalf("loadReceipt(oversized) error = %v", err)
+	}
+}
+
+func TestValidateBundleRootRejectsEscapesAndSymlinks(t *testing.T) {
+	root := t.TempDir()
+	outputRoot := filepath.Join(root, "output")
+	if err := os.Mkdir(outputRoot, 0o755); err != nil {
+		t.Fatalf("Mkdir(output root) error = %v", err)
+	}
+	validBundle := filepath.Join(outputRoot, "run-1")
+	if err := os.Mkdir(validBundle, 0o755); err != nil {
+		t.Fatalf("Mkdir(bundle) error = %v", err)
+	}
+	if _, err := validateBundleRoot(outputRoot, validBundle); err != nil {
+		t.Fatalf("validateBundleRoot(valid) error = %v", err)
+	}
+
+	outside := filepath.Join(root, "output-other")
+	if err := os.Mkdir(outside, 0o755); err != nil {
+		t.Fatalf("Mkdir(outside) error = %v", err)
+	}
+	for name, candidate := range map[string]string{"equal root": outputRoot, "escape": root, "prefix confusion": outside} {
+		t.Run(name, func(t *testing.T) {
+			if _, err := validateBundleRoot(outputRoot, candidate); err == nil {
+				t.Fatalf("validateBundleRoot(%q) error = nil", candidate)
+			}
+		})
+	}
+	symlink := filepath.Join(outputRoot, "linked")
+	if err := os.Symlink(outside, symlink); err != nil {
+		t.Skipf("Symlink() unavailable: %v", err)
+	}
+	if _, err := validateBundleRoot(outputRoot, symlink); err == nil {
+		t.Fatal("validateBundleRoot(symlink) error = nil")
+	}
+}
+
+func TestLoadIndexRejectsMalformedUnsafeAndUnsupportedDocuments(t *testing.T) {
+	tests := []struct {
+		name       string
+		indexValue any
+		prepare    func(*testing.T, 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: "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"},
+		})},
+		{name: "absolute logical path", indexValue: validIndexValue([]any{map[string]any{"lane_id": "npc", "file": "/tmp/npc.json"}})},
+		{name: "lexical traversal", indexValue: validIndexValue([]any{map[string]any{"lane_id": "npc", "file": "../outside.json"}})},
+		{name: "root prefix confusion", indexValue: validIndexValue([]any{map[string]any{"lane_id": "npc", "file": "../bundle-other/npc.json"}})},
+		{name: "file symlink", indexValue: validIndexValue([]any{map[string]any{"lane_id": "npc", "file": "lanes/npc.json"}}), prepare: func(t *testing.T, bundle string) {
+			if err := os.Symlink(filepath.Join(bundle, "manifest.json"), filepath.Join(bundle, "lanes", "npc.json")); err != nil {
+				t.Skipf("Symlink() unavailable: %v", err)
+			}
+		}},
+		{name: "directory symlink", indexValue: validIndexValue([]any{map[string]any{"lane_id": "npc", "file": "linked/npc.json"}}), prepare: func(t *testing.T, bundle string) {
+			if err := os.Symlink(filepath.Join(bundle, "lanes"), filepath.Join(bundle, "linked")); err != nil {
+				t.Skipf("Symlink() unavailable: %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"}
+			return value
+		}()},
+		{name: "pipeline descriptor escape", indexValue: func() any {
+			value := validIndexValue([]any{})
+			value["evidence_context"] = map[string]any{
+				"artifact_kind": "evidence_context", "file": "../evidence.json", "media_type": "application/json",
+				"schema_id": "evidence", "schema_name": "Evidence", "schema_version": "v1",
+			}
+			return value
+		}()},
+	}
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			bundle := createBundleSkeleton(t)
+			indexPath := filepath.Join(bundle, "index.json")
+			if raw, ok := test.indexValue.(json.RawMessage); ok {
+				if err := os.WriteFile(indexPath, raw, 0o644); err != nil {
+					t.Fatalf("WriteFile(index) error = %v", err)
+				}
+			} else {
+				writeJSONFile(t, indexPath, test.indexValue)
+			}
+			if test.prepare != nil {
+				test.prepare(t, bundle)
+			}
+			if _, err := loadIndex(bundle, indexPath); err == nil {
+				t.Fatal("loadIndex() error = nil, want failure")
+			}
+		})
+	}
+
+	bundle := createBundleSkeleton(t)
+	oversizedIndex := filepath.Join(bundle, "index.json")
+	if err := os.WriteFile(oversizedIndex, []byte(strings.Repeat("x", maxIndexBytes+1)), 0o644); err != nil {
+		t.Fatalf("WriteFile(oversized index) error = %v", err)
+	}
+	if _, err := loadIndex(bundle, oversizedIndex); err == nil || !strings.Contains(err.Error(), "exceeds") {
+		t.Fatalf("loadIndex(oversized) error = %v", err)
+	}
+}
+
+func TestLoadDiagnosticSummariesValidateBoundsAndTolerateUnknownFields(t *testing.T) {
+	root := t.TempDir()
+	rejectedPath := filepath.Join(root, "rejected.json")
+	warningsPath := filepath.Join(root, "warnings.json")
+	writeJSONFile(t, rejectedPath, map[string]any{"rejected": []any{map[string]any{
+		"stage": "validate", "lane_id": "spells", "reason_code": "invalid", "message": "do not retain this", "future": true,
+	}}, "future": true})
+	writeJSONFile(t, warningsPath, map[string]any{"warnings": []any{map[string]any{
+		"scope": "lane:spells", "reason_code": "bounded", "message": "do not retain this", "future": true,
+	}}, "future": true})
+	rejections, err := loadRejections(rejectedPath)
+	if err != nil || len(rejections) != 1 || rejections[0].ReasonCode != "invalid" {
+		t.Fatalf("loadRejections() = %#v, %v", rejections, err)
+	}
+	warnings, err := loadWarnings(warningsPath)
+	if err != nil || len(warnings) != 1 || warnings[0].Scope != "lane:spells" {
+		t.Fatalf("loadWarnings() = %#v, %v", warnings, err)
+	}
+
+	for name, path := range map[string]string{"rejections": rejectedPath, "warnings": warningsPath} {
+		t.Run("malformed "+name, func(t *testing.T) {
+			if err := os.WriteFile(path, []byte("{"), 0o644); err != nil {
+				t.Fatalf("WriteFile() error = %v", err)
+			}
+			var err error
+			if name == "rejections" {
+				_, err = loadRejections(path)
+			} else {
+				_, err = loadWarnings(path)
+			}
+			if err == nil {
+				t.Fatal("summary decoder error = nil")
+			}
+		})
+	}
+	oversized := filepath.Join(root, "oversized.json")
+	if err := os.WriteFile(oversized, []byte(strings.Repeat("x", maxSummaryBytes+1)), 0o644); err != nil {
+		t.Fatalf("WriteFile(oversized) error = %v", err)
+	}
+	if _, err := loadWarnings(oversized); err == nil || !strings.Contains(err.Error(), "exceeds") {
+		t.Fatalf("loadWarnings(oversized) error = %v", err)
+	}
+	if _, err := loadRejections(oversized); err == nil || !strings.Contains(err.Error(), "exceeds") {
+		t.Fatalf("loadRejections(oversized) error = %v", err)
+	}
+}
+
+func TestFakeRunnerCapturesRequestsAndHonorsContextAndError(t *testing.T) {
+	req := RunRequest{PipelineID: "pipeline"}
+	want := RunResult{BundleRoot: "/bundle"}
+	fake := &FakeRunner{Result: want}
+	got, err := fake.Run(context.Background(), req)
+	if err != nil || !reflect.DeepEqual(got, want) || !reflect.DeepEqual(fake.Requests, []RunRequest{req}) {
+		t.Fatalf("Run() = %#v, %v; requests = %#v", got, err, fake.Requests)
+	}
+
+	wantErr := errors.New("configured failure")
+	fake.Err = wantErr
+	if _, err := fake.Run(context.Background(), req); !errors.Is(err, wantErr) {
+		t.Fatalf("Run(configured error) = %v", err)
+	}
+	canceled, cancel := context.WithCancel(context.Background())
+	cancel()
+	before := len(fake.Requests)
+	if _, err := fake.Run(canceled, req); !errors.Is(err, context.Canceled) || len(fake.Requests) != before {
+		t.Fatalf("Run(canceled) error = %v; requests = %d", err, len(fake.Requests))
+	}
+}
+
+func validRunRequest(t *testing.T) RunRequest {
+	t.Helper()
+	root := t.TempDir()
+	configPath := filepath.Join(root, "notarius.yml")
+	inputPath := filepath.Join(root, "input.json")
+	outputRoot := filepath.Join(root, "outputs")
+	workingDirectory := filepath.Join(root, "work")
+	diagnostics := filepath.Join(root, "diagnostics")
+	for _, directory := range []string{outputRoot, workingDirectory, diagnostics} {
+		if err := os.Mkdir(directory, 0o755); err != nil {
+			t.Fatalf("Mkdir(%q) error = %v", directory, err)
+		}
+	}
+	if err := os.WriteFile(configPath, []byte("pipelines: {}\n"), 0o644); err != nil {
+		t.Fatalf("WriteFile(config) error = %v", err)
+	}
+	if err := os.WriteFile(inputPath, []byte("{}\n"), 0o644); err != nil {
+		t.Fatalf("WriteFile(input) error = %v", err)
+	}
+	return RunRequest{
+		Binary: "notarius", ConfigPath: configPath, PipelineID: "dnd-session", InputPath: inputPath,
+		OutputRoot: outputRoot, WorkingDirectory: workingDirectory,
+		ReceiptPath: filepath.Join(diagnostics, "receipt.json"), LogPath: filepath.Join(diagnostics, "stderr.log"),
+		Timeout: time.Second,
+	}
+}
+
+func writeValidBundleAndReceipt(t *testing.T, req RunRequest, includeUnknown bool) {
+	t.Helper()
+	bundle := filepath.Join(req.OutputRoot, "notarius-run-1")
+	if err := os.MkdirAll(filepath.Join(bundle, "lanes"), 0o755); err != nil {
+		t.Fatalf("MkdirAll(bundle) error = %v", err)
+	}
+	for path, data := range map[string]string{
+		"manifest.json":         `{}`,
+		"lanes/npc.json":        `{}`,
+		"chunk-map.json":        `{}`,
+		"evidence-context.json": `{}`,
+	} {
+		if err := os.WriteFile(filepath.Join(bundle, filepath.FromSlash(path)), []byte(data), 0o644); err != nil {
+			t.Fatalf("WriteFile(%q) error = %v", path, err)
+		}
+	}
+	rejection := map[string]any{"stage": "validate", "lane_id": "spells", "reason_code": "invalid_spell", "message": strings.Repeat("external detail", 20)}
+	warning := map[string]any{"scope": "lane:npc-registry", "reason_code": "normalized_name", "message": strings.Repeat("external warning", 20)}
+	if includeUnknown {
+		rejection["future"] = true
+		warning["future"] = true
+	}
+	writeJSONFile(t, filepath.Join(bundle, "rejected.json"), map[string]any{"rejected": []any{rejection}, "future": true})
+	writeJSONFile(t, filepath.Join(bundle, "warnings.json"), map[string]any{"warnings": []any{warning}, "future": true})
+	index := validIndexValue([]any{map[string]any{
+		"lane_id": "npc-registry", "file": "lanes/npc.json", "media_type": "application/json",
+		"module_key": "dnd/npc-registry", "schema_id": "notarius.dnd.npc_registry",
+		"schema_name": "NPCRegistry", "schema_version": "v1", "future": true,
+	}})
+	index["chunk_map"] = map[string]any{
+		"artifact_kind": "chunk_map", "file": "chunk-map.json", "media_type": "application/json",
+		"schema_id": "notarius.chunk_map", "schema_name": "ChunkMap", "schema_version": "v1", "future": true,
+	}
+	index["evidence_context"] = map[string]any{
+		"artifact_kind": "evidence_context", "file": "evidence-context.json", "media_type": "application/json",
+		"schema_id": "notarius.evidence_context", "schema_name": "EvidenceContext", "schema_version": "v1", "future": true,
+	}
+	index["future"] = true
+	writeJSONFile(t, filepath.Join(bundle, "index.json"), index)
+	receipt := map[string]any{
+		"schema_version": ReceiptSchemaVersion, "run_id": "notarius-run-1", "pipeline_id": req.PipelineID,
+		"output_directory": bundle, "index_file": "index.json", "normalized_output_count": 1,
+		"rejected_output_count": 1, "warning_count": 1, "validation_status": "rejected",
+	}
+	if includeUnknown {
+		receipt["future"] = true
+	}
+	writeJSONFile(t, req.ReceiptPath, receipt)
+}
+
+func createBundleSkeleton(t *testing.T) string {
+	t.Helper()
+	bundle := filepath.Join(t.TempDir(), "bundle")
+	if err := os.MkdirAll(filepath.Join(bundle, "lanes"), 0o755); err != nil {
+		t.Fatalf("MkdirAll(bundle) error = %v", err)
+	}
+	for _, name := range []string{"manifest.json", "rejected.json", "warnings.json", "lanes/npc.json", "chunk-map.json"} {
+		if err := os.WriteFile(filepath.Join(bundle, filepath.FromSlash(name)), []byte("{}"), 0o644); err != nil {
+			t.Fatalf("WriteFile(%q) error = %v", name, err)
+		}
+	}
+	return bundle
+}
+
+func validIndexValue(lanes []any) map[string]any {
+	return map[string]any{
+		"manifest_file": "manifest.json", "output_files": lanes,
+		"rejected_file": "rejected.json", "warnings_file": "warnings.json",
+	}
+}
+
+func writeJSONFile(t *testing.T, path string, value any) {
+	t.Helper()
+	data, err := json.Marshal(value)
+	if err != nil {
+		t.Fatalf("json.Marshal() error = %v", err)
+	}
+	if err := os.WriteFile(path, data, 0o644); err != nil {
+		t.Fatalf("WriteFile(%q) error = %v", path, err)
+	}
+}
+
+func writeShellScript(t *testing.T, body string) string {
+	t.Helper()
+	path := filepath.Join(t.TempDir(), "notarius-helper")
+	if err := os.WriteFile(path, []byte(body), 0o755); err != nil {
+		t.Fatalf("WriteFile(script) error = %v", err)
+	}
+	return path
+}
+
+func assertTextFile(t *testing.T, path, want string) {
+	t.Helper()
+	data, err := os.ReadFile(path)
+	if err != nil {
+		t.Fatalf("ReadFile(%q) error = %v", path, err)
+	}
+	if string(data) != want {
+		t.Fatalf("ReadFile(%q) = %q, want %q", path, string(data), want)
+	}
+}
+
+func cloneMap(source map[string]any) map[string]any {
+	result := make(map[string]any, len(source))
+	for key, value := range source {
+		result[key] = value
+	}
+	return result
+}