Add Notarius configuration and extraction source policy

2026-08-09 23:30:32 +00:00
parent 951383226c
commit 98734644d6
15 changed files with 678 additions and 30 deletions
--- a/docs/roadmap/implementation.md
+++ b/docs/roadmap/implementation.md
@@ -17,7 +17,7 @@ implementation sequence.
 | Stage | Status |
 | --- | --- |
 | Stage 1 | Complete |
-| Stage 2 | Not started |
+| Stage 2 | Complete |
 | Stage 3 | Not started |
 | Stage 4 | Not started |
 | Stage 5 | Not started |
--- a/internal/app/operator_helpers_test.go
+++ b/internal/app/operator_helpers_test.go
@@ -508,7 +508,7 @@ func TestExecuteLocksAddListAndRemoveUseRemoteLockStore(t *testing.T) {
 	if code != 0 {
 		t.Fatalf("locks remove exit code = %d, want 0; stderr=%q", code, stderr.String())
 	}
-	store, err := config.LoadPublishLockStoreBytes("locks.yml", fake.Objects[key].Data, nil)
+	store, err := config.LoadPublishLockStoreBytes("locks.yml", fake.Objects[key].Data, nil, nil)
 	if err != nil {
 		t.Fatalf("LoadPublishLockStoreBytes() error = %v", err)
 	}
--- a/internal/app/operator_locks.go
+++ b/internal/app/operator_locks.go
@@ -67,7 +67,7 @@ func LocksAdd(ctx context.Context, args []string, out io.Writer) error {
 	if err != nil {
 		return fmt.Errorf("locks add: %w", err)
 	}
-	if _, err := config.ValidatePublishLockRules([]config.PublishLockRule{{Source: source}}, cfg.Pipeline.Scriptorium, "locks add"); err != nil {
+	if _, err := config.ValidatePublishLockRules([]config.PublishLockRule{{Source: source}}, cfg.Pipeline.Scriptorium, cfg.Pipeline.Notarius, "locks add"); err != nil {
 		return fmt.Errorf("locks add: %w", err)
 	}
 	if _, ok := lockSourceSet(locks.Static)[source]; ok {
@@ -79,7 +79,7 @@ func LocksAdd(ctx context.Context, args []string, out io.Writer) error {
 	}
 	remoteSet[source] = config.PublishLockRule{Source: source, Reason: strings.TrimSpace(reason)}
 	remoteLocks := lockMapValues(remoteSet)
-	if _, err := config.ValidatePublishLockRules(remoteLocks, cfg.Pipeline.Scriptorium, "locks"); err != nil {
+	if _, err := config.ValidatePublishLockRules(remoteLocks, cfg.Pipeline.Scriptorium, cfg.Pipeline.Notarius, "locks"); err != nil {
 		return fmt.Errorf("locks add: %w", err)
 	}
 	if err := uploadRemoteLockStore(ctx, store, locks.Key, &config.PublishLockStore{Locks: remoteLocks}); err != nil {
@@ -106,7 +106,7 @@ func LocksRemove(ctx context.Context, args []string, out io.Writer) error {
 	if err != nil {
 		return fmt.Errorf("locks remove: %w", err)
 	}
-	if _, err := config.ValidatePublishLockRules([]config.PublishLockRule{{Source: source}}, cfg.Pipeline.Scriptorium, "locks remove"); err != nil {
+	if _, err := config.ValidatePublishLockRules([]config.PublishLockRule{{Source: source}}, cfg.Pipeline.Scriptorium, cfg.Pipeline.Notarius, "locks remove"); err != nil {
 		return fmt.Errorf("locks remove: %w", err)
 	}
 	remoteSet := lockSourceSet(locks.Remote)
--- a/internal/app/remote_locks.go
+++ b/internal/app/remote_locks.go
@@ -55,7 +55,7 @@ func loadRemoteLockStore(ctx context.Context, cfg *config.Config, store storage.
 	if err != nil {
 		return nil, key, fmt.Errorf("read remote locks %q: %w", key, err)
 	}
-	lockStore, err := config.LoadPublishLockStoreBytes("s3://"+s3BucketName(cfg.Pipeline)+"/"+key, data, cfg.Pipeline.Scriptorium)
+	lockStore, err := config.LoadPublishLockStoreBytes("s3://"+s3BucketName(cfg.Pipeline)+"/"+key, data, cfg.Pipeline.Scriptorium, cfg.Pipeline.Notarius)
 	if err != nil {
 		return nil, key, err
 	}
--- a/internal/app/session_oriented_cli_test.go
+++ b/internal/app/session_oriented_cli_test.go
@@ -346,7 +346,7 @@ func TestExecuteSessionLocksMutationAcceptsPositionalSessionID(t *testing.T) {
 	if code != 0 {
 		t.Fatalf("locks remove exit code = %d, want 0; stderr=%q", code, stderr.String())
 	}
-	store, err := config.LoadPublishLockStoreBytes("locks.yml", fake.Objects[key].Data, nil)
+	store, err := config.LoadPublishLockStoreBytes("locks.yml", fake.Objects[key].Data, nil, nil)
 	if err != nil {
 		t.Fatalf("LoadPublishLockStoreBytes() error = %v", err)
 	}
--- a/internal/artifactpolicy/policy.go
+++ b/internal/artifactpolicy/policy.go
@@ -18,11 +18,14 @@ const (
 	SourceInputGlossary = "narratio.input.glossary"

 	configuredSourcePrefix      = "narratio.artifact."
+	extractionSourcePrefix      = "narratio.extraction."
 	previousConfiguredSrcPrefix = "narratio.previous_session.artifact."
 )

 var configuredSourceRE = regexp.MustCompile(`^narratio\.artifact\.([a-z][a-z0-9_]*)$`)
+var extractionSourceRE = regexp.MustCompile(`^narratio\.extraction\.([a-z][a-z0-9_]*)$`)
 var previousSourceRE = regexp.MustCompile(`^narratio\.previous_session\.artifact\.([a-z][a-z0-9_]*)$`)
+var configuredKeyRE = regexp.MustCompile(`^[a-z][a-z0-9_]*$`)

 var (
 	ErrUnsupportedScriptoriumInputSource = errors.New("unsupported scriptorium input source")
@@ -34,6 +37,7 @@ type SourceKind string
 const (
 	SourceKindBuiltIn            SourceKind = "built_in"
 	SourceKindConfiguredArtifact SourceKind = "configured_artifact"
+	SourceKindExtraction         SourceKind = "extraction"
 	SourceKindPreviousArtifact   SourceKind = "previous_session_configured_artifact"
 	SourceKindStableInput        SourceKind = "stable_input"
 )
@@ -67,11 +71,30 @@ func (e *UnknownConfiguredArtifactError) Error() string {
 	return fmt.Sprintf("references unknown artifact %q", e.ConfiguredKey)
 }

+// UnknownExtractionArtifactError reports a source that references an undefined extraction key.
+type UnknownExtractionArtifactError struct {
+	ConfiguredKey string
+}
+
+func (e *UnknownExtractionArtifactError) Error() string {
+	return fmt.Sprintf("references unknown extraction output %q", e.ConfiguredKey)
+}
+
+// IsConfiguredKey reports whether a key follows the configured-artifact key grammar.
+func IsConfiguredKey(key string) bool {
+	return configuredKeyRE.MatchString(strings.TrimSpace(key))
+}
+
 // ConfiguredSourceID converts a configured artifact key into source id form.
 func ConfiguredSourceID(key string) string {
 	return configuredSourcePrefix + strings.TrimSpace(key)
 }

+// ExtractionSourceID converts an extraction output key into source id form.
+func ExtractionSourceID(key string) string {
+	return extractionSourcePrefix + strings.TrimSpace(key)
+}
+
 // PreviousSessionSourceID converts a configured artifact key into previous-session source id form.
 func PreviousSessionSourceID(key string) string {
 	return previousConfiguredSrcPrefix + strings.TrimSpace(key)
@@ -86,6 +109,15 @@ func ParseConfiguredSource(source string) (string, bool) {
 	return matches[1], true
 }

+// ParseExtractionSource extracts configured key from narratio.extraction.<key>.
+func ParseExtractionSource(source string) (string, bool) {
+	matches := extractionSourceRE.FindStringSubmatch(strings.TrimSpace(source))
+	if len(matches) != 2 {
+		return "", false
+	}
+	return matches[1], true
+}
+
 // ParsePreviousSessionSource extracts configured key from narratio.previous_session.artifact.<key>.
 func ParsePreviousSessionSource(source string) (string, bool) {
 	matches := previousSourceRE.FindStringSubmatch(strings.TrimSpace(source))
@@ -110,6 +142,9 @@ func ClassifySource(source string) (Source, error) {
 	if key, ok := ParseConfiguredSource(trimmed); ok {
 		return Source{ID: trimmed, Kind: SourceKindConfiguredArtifact, ConfiguredKey: key}, nil
 	}
+	if key, ok := ParseExtractionSource(trimmed); ok {
+		return Source{ID: trimmed, Kind: SourceKindExtraction, ConfiguredKey: key}, nil
+	}
 	if key, ok := ParsePreviousSessionSource(trimmed); ok {
 		return Source{ID: trimmed, Kind: SourceKindPreviousArtifact, ConfiguredKey: key}, nil
 	}
@@ -186,24 +221,48 @@ func PreviousSessionSourceDescriptorForConfiguredKey(configuredKey string) (Prev
 func ValidateInputConfiguredReference(
 	descriptor ScriptoriumInputSourceDescriptor,
 	configured map[string]struct{},
+) error {
+	return ValidateInputReference(descriptor, configured, nil)
+}
+
+// ValidateInputReference checks that configured and extraction sources are declared
+// by the effective pipeline configuration.
+func ValidateInputReference(
+	descriptor ScriptoriumInputSourceDescriptor,
+	configured map[string]struct{},
+	extractions map[string]struct{},
 ) error {
 	switch descriptor.Source.Kind {
 	case SourceKindConfiguredArtifact, SourceKindPreviousArtifact:
 		if _, ok := configured[descriptor.Source.ConfiguredKey]; !ok {
 			return &UnknownConfiguredArtifactError{ConfiguredKey: descriptor.Source.ConfiguredKey}
 		}
+	case SourceKindExtraction:
+		if _, ok := extractions[descriptor.Source.ConfiguredKey]; !ok {
+			return &UnknownExtractionArtifactError{ConfiguredKey: descriptor.Source.ConfiguredKey}
+		}
 	}
 	return nil
 }

 // ValidatePublishSource validates that a source is publish-compatible and references a known configured artifact.
 func ValidatePublishSource(source string, configured map[string]string) (Source, error) {
+	return ValidatePublishSourceWithExtractions(source, configured, nil)
+}
+
+// ValidatePublishSourceWithExtractions validates publish sources against the
+// configured Scriptorium artifacts and extraction outputs.
+func ValidatePublishSourceWithExtractions(
+	source string,
+	configured map[string]string,
+	extractions map[string]struct{},
+) (Source, error) {
 	classified, err := ClassifySource(source)
 	if err != nil {
-		return Source{}, fmt.Errorf("must be a built-in source id or narratio.artifact.<name>")
+		return Source{}, fmt.Errorf("must be a built-in source id, narratio.artifact.<name>, or configured narratio.extraction.<name>")
 	}
 	if classified.Kind == SourceKindPreviousArtifact {
-		return Source{}, fmt.Errorf("must be a built-in source id or narratio.artifact.<name>")
+		return Source{}, fmt.Errorf("must be a built-in source id, narratio.artifact.<name>, or configured narratio.extraction.<name>")
 	}
 	if classified.Kind == SourceKindConfiguredArtifact {
 		if configured == nil {
@@ -213,6 +272,11 @@ func ValidatePublishSource(source string, configured map[string]string) (Source,
 			return Source{}, fmt.Errorf("configured artifact %q is not defined in pipeline.scriptorium.artifacts", classified.ConfiguredKey)
 		}
 	}
+	if classified.Kind == SourceKindExtraction {
+		if _, ok := extractions[classified.ConfiguredKey]; !ok {
+			return Source{}, fmt.Errorf("extraction output %q is not defined in pipeline.notarius.outputs", classified.ConfiguredKey)
+		}
+	}
 	return classified, nil
 }

--- a/internal/artifactpolicy/policy_test.go
+++ b/internal/artifactpolicy/policy_test.go
@@ -17,6 +17,7 @@ func TestClassifySource(t *testing.T) {
 		{name: "built in transcript", source: "narratio.transcript.final_trimmed", wantKind: SourceKindBuiltIn},
 		{name: "built in bounds", source: "narratio.bounds.session", wantKind: SourceKindBuiltIn},
 		{name: "configured artifact", source: "narratio.artifact.session_recap", wantKind: SourceKindConfiguredArtifact, wantKey: "session_recap"},
+		{name: "extraction", source: "narratio.extraction.npc_registry", wantKind: SourceKindExtraction, wantKey: "npc_registry"},
 		{name: "previous session configured", source: "narratio.previous_session.artifact.session_recap", wantKind: SourceKindPreviousArtifact, wantKey: "session_recap"},
 		{name: "unsupported", source: "narratio.unknown", wantErrLike: "unsupported artifact source"},
 	}
@@ -56,6 +57,60 @@ func TestValidatePublishSource(t *testing.T) {
 	}
 }

+func TestExtractionSourcePolicy(t *testing.T) {
+	if got := ExtractionSourceID(" npc_registry "); got != "narratio.extraction.npc_registry" {
+		t.Fatalf("ExtractionSourceID() = %q, want narratio.extraction.npc_registry", got)
+	}
+	if key, ok := ParseExtractionSource(" narratio.extraction.npc_registry "); !ok || key != "npc_registry" {
+		t.Fatalf("ParseExtractionSource() = %q, %t; want npc_registry, true", key, ok)
+	}
+	for _, source := range []string{
+		"narratio.extraction.",
+		"narratio.extraction.NPC",
+		"narratio.extraction.npc-registry",
+		"narratio.extraction.npc_registry.extra",
+	} {
+		if _, ok := ParseExtractionSource(source); ok {
+			t.Fatalf("ParseExtractionSource(%q) unexpectedly matched", source)
+		}
+	}
+
+	descriptor, err := DescribeScriptoriumInputSource("narratio.extraction.npc_registry")
+	if err != nil {
+		t.Fatalf("DescribeScriptoriumInputSource(extraction) error = %v", err)
+	}
+	if descriptor.Source.Kind != SourceKindExtraction || descriptor.Source.ConfiguredKey != "npc_registry" {
+		t.Fatalf("extraction descriptor = %#v", descriptor)
+	}
+	declared := map[string]struct{}{"npc_registry": {}}
+	if err := ValidateInputReference(descriptor, nil, declared); err != nil {
+		t.Fatalf("ValidateInputReference(declared extraction) error = %v", err)
+	}
+	if err := ValidateInputReference(descriptor, nil, nil); err == nil {
+		t.Fatal("ValidateInputReference(unknown extraction) error = nil, want error")
+	}
+
+	if _, err := ValidatePublishSourceWithExtractions("narratio.extraction.npc_registry", nil, declared); err != nil {
+		t.Fatalf("ValidatePublishSourceWithExtractions(declared) error = %v", err)
+	}
+	if _, err := ValidatePublishSourceWithExtractions("narratio.extraction.unknown", nil, declared); err == nil {
+		t.Fatal("ValidatePublishSourceWithExtractions(unknown) error = nil, want error")
+	}
+
+	identities := map[string]struct{}{}
+	for _, sourceID := range []string{
+		ExtractionSourceID("npc_registry"),
+		ConfiguredSourceID("npc_registry"),
+		PreviousSessionSourceID("npc_registry"),
+		SourceBoundsSession,
+	} {
+		if _, exists := identities[sourceID]; exists {
+			t.Fatalf("source identity collision at %q", sourceID)
+		}
+		identities[sourceID] = struct{}{}
+	}
+}
+
 func TestResolvePublishedDestination(t *testing.T) {
 	configured := map[string]string{"session_recap": "artifacts/session_recap.md"}

--- a/internal/artifacts/artifact_resolver.go
+++ b/internal/artifacts/artifact_resolver.go
@@ -144,6 +144,22 @@ func ConfiguredArtifactName(source string) (string, bool) {
 	return artifactpolicy.ParseConfiguredSource(source)
 }

+// ExtractionArtifactSourceID returns narratio.extraction.<name> for a configured key.
+func ExtractionArtifactSourceID(key string) string {
+	return artifactpolicy.ExtractionSourceID(key)
+}
+
+// IsExtractionArtifactSource reports whether source is narratio.extraction.<name>.
+func IsExtractionArtifactSource(source string) bool {
+	_, ok := artifactpolicy.ParseExtractionSource(source)
+	return ok
+}
+
+// ExtractionArtifactName extracts <name> from narratio.extraction.<name>.
+func ExtractionArtifactName(source string) (string, bool) {
+	return artifactpolicy.ParseExtractionSource(source)
+}
+
 // IsPreviousSessionArtifactSource returns true when source is narratio.previous_session.artifact.<name>.
 func IsPreviousSessionArtifactSource(source string) bool {
 	_, ok := artifactpolicy.ParsePreviousSessionSource(source)
--- a/internal/artifacts/artifact_resolver_test.go
+++ b/internal/artifacts/artifact_resolver_test.go
@@ -150,6 +150,23 @@ func TestPreviousSessionArtifactSourceHelpers(t *testing.T) {
 	}
 }

+func TestExtractionArtifactSourceHelpers(t *testing.T) {
+	if got := ExtractionArtifactSourceID("npc_registry"); got != "narratio.extraction.npc_registry" {
+		t.Fatalf("ExtractionArtifactSourceID() = %q", got)
+	}
+	if !IsExtractionArtifactSource(" narratio.extraction.npc_registry ") {
+		t.Fatal("IsExtractionArtifactSource(valid) = false")
+	}
+	if key, ok := ExtractionArtifactName("narratio.extraction.npc_registry"); !ok || key != "npc_registry" {
+		t.Fatalf("ExtractionArtifactName() = %q, %t", key, ok)
+	}
+	for _, source := range []string{"narratio.extraction.", "narratio.extraction.npc-registry", "narratio.artifact.npc_registry"} {
+		if IsExtractionArtifactSource(source) {
+			t.Fatalf("IsExtractionArtifactSource(%q) = true, want false", source)
+		}
+	}
+}
+
 func TestResolveSessionArtifactPrefersManifestOutput(t *testing.T) {
 	workspace := t.TempDir()
 	paths := buildSessionPaths(workspace, "campaign", "session")
--- a/internal/config/config.go
+++ b/internal/config/config.go
@@ -30,6 +30,7 @@ type PipelineConfig struct {
 	Trim         *TrimConfig        `yaml:"trim"`
 	Render       *RenderConfig      `yaml:"render"`
 	Scriptorium  *ScriptoriumConfig `yaml:"scriptorium"`
+	Notarius     *NotariusConfig    `yaml:"notarius"`
 	Notification NotificationConfig `yaml:"notification"`
 }

@@ -252,6 +253,26 @@ type ScriptoriumInputConfig struct {
 	Required bool   `yaml:"required"`
 }

+// NotariusConfig configures structured artifact extraction by Notarius.
+type NotariusConfig struct {
+	Enabled          bool                            `yaml:"enabled"`
+	Binary           string                          `yaml:"binary"`
+	ConfigPath       string                          `yaml:"config_path"`
+	PipelineID       string                          `yaml:"pipeline_id"`
+	Timeout          string                          `yaml:"timeout"`
+	WorkingDirectory string                          `yaml:"working_directory"`
+	Outputs          map[string]NotariusOutputConfig `yaml:"outputs"`
+}
+
+// NotariusOutputConfig declares one required extraction lane contract.
+type NotariusOutputConfig struct {
+	LaneID        string `yaml:"lane_id"`
+	MediaType     string `yaml:"media_type"`
+	SchemaID      string `yaml:"schema_id"`
+	SchemaVersion string `yaml:"schema_version"`
+	ModuleKey     string `yaml:"module_key"`
+}
+
 // NotificationConfig configures notification backend settings.
 type NotificationConfig struct {
 	Backend   string `yaml:"backend"`
--- a/internal/config/defaults.go
+++ b/internal/config/defaults.go
@@ -30,9 +30,11 @@ const (
 	DefaultSeriatimCoalesceGap  = 3.0
 	DefaultSeriatimReport       = true

-	DefaultAuditaBinary  = "audita"
-	DefaultAuditaTimeout = "3h"
-	DefaultAuditaReport  = true
+	DefaultAuditaBinary    = "audita"
+	DefaultAuditaTimeout   = "3h"
+	DefaultAuditaReport    = true
+	DefaultNotariusBinary  = "notarius"
+	DefaultNotariusTimeout = "3h"

 	DefaultScriptoriumBinary                 = "scriptorium"
 	DefaultScriptoriumTimeout                = "10m"
--- a/internal/config/load.go
+++ b/internal/config/load.go
@@ -19,6 +19,9 @@ func LoadPipeline(path string) (*PipelineConfig, error) {
 		return nil, fmt.Errorf("load pipeline config: %w", err)
 	}
 	applyPipelineDefaults(&cfg)
+	if err := resolveNotariusPaths(&cfg, path); err != nil {
+		return nil, fmt.Errorf("load pipeline config: %w", err)
+	}
 	return &cfg, nil
 }

@@ -86,12 +89,17 @@ func LoadSessionBytesWithOptions(label string, data []byte, opts SessionLoadOpti

 // LoadPublishLockStoreBytes loads a mutable session lock store with strict
 // field checking and source validation.
-func LoadPublishLockStoreBytes(label string, data []byte, scriptorium *ScriptoriumConfig) (*PublishLockStore, error) {
+func LoadPublishLockStoreBytes(
+	label string,
+	data []byte,
+	scriptorium *ScriptoriumConfig,
+	notarius *NotariusConfig,
+) (*PublishLockStore, error) {
 	var store PublishLockStore
 	if err := decodeStrictYAMLFromReader("publish lock store", label, strings.NewReader(string(data)), &store); err != nil {
 		return nil, fmt.Errorf("load publish lock store: %w", err)
 	}
-	locks, err := ValidatePublishLockRules(store.Locks, scriptorium, "locks")
+	locks, err := ValidatePublishLockRules(store.Locks, scriptorium, notarius, "locks")
 	if err != nil {
 		return nil, fmt.Errorf("load publish lock store: %w", err)
 	}
@@ -359,6 +367,7 @@ func applyPipelineDefaults(cfg *PipelineConfig) {
 	}
 	applyRenderDefaults(&cfg.Render)
 	applyScriptoriumDefaults(cfg.Scriptorium)
+	applyNotariusDefaults(cfg.Notarius)
 }

 func applyCampaignsDefaults(cfg *CampaignsConfig) {
@@ -516,6 +525,57 @@ func applyScriptoriumDefaults(cfg *ScriptoriumConfig) {
 	}
 }

+func applyNotariusDefaults(cfg *NotariusConfig) {
+	if cfg == nil {
+		return
+	}
+	if cfg.Binary == "" {
+		cfg.Binary = DefaultNotariusBinary
+	}
+	if cfg.Timeout == "" {
+		cfg.Timeout = DefaultNotariusTimeout
+	}
+}
+
+func resolveNotariusPaths(cfg *PipelineConfig, pipelinePath string) error {
+	if cfg == nil || cfg.Notarius == nil || !cfg.Notarius.Enabled {
+		return nil
+	}
+	pipelineAbs, err := filepath.Abs(pipelinePath)
+	if err != nil {
+		return fmt.Errorf("resolve pipeline config path %q: %w", pipelinePath, err)
+	}
+	baseDir := filepath.Dir(pipelineAbs)
+	resolve := func(value string) (string, error) {
+		trimmed := strings.TrimSpace(value)
+		if trimmed == "" {
+			return "", nil
+		}
+		if !filepath.IsAbs(trimmed) {
+			trimmed = filepath.Join(baseDir, trimmed)
+		}
+		return filepath.Abs(trimmed)
+	}
+
+	resolvedConfigPath, err := resolve(cfg.Notarius.ConfigPath)
+	if err != nil {
+		return fmt.Errorf("resolve pipeline.notarius.config_path: %w", err)
+	}
+	cfg.Notarius.ConfigPath = resolvedConfigPath
+	if strings.TrimSpace(cfg.Notarius.WorkingDirectory) == "" {
+		if resolvedConfigPath != "" {
+			cfg.Notarius.WorkingDirectory = filepath.Dir(resolvedConfigPath)
+		}
+		return nil
+	}
+	workingDirectory, err := resolve(cfg.Notarius.WorkingDirectory)
+	if err != nil {
+		return fmt.Errorf("resolve pipeline.notarius.working_directory: %w", err)
+	}
+	cfg.Notarius.WorkingDirectory = workingDirectory
+	return nil
+}
+
 func applyTrimDefaults(cfg *TrimConfig) {
 	if cfg == nil {
 		return
--- a/internal/config/notarius_test.go
+++ b/internal/config/notarius_test.go
@@ -0,0 +1,283 @@
+package config
+
+import (
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+)
+
+func TestNotariusOmittedAndDisabledBehavior(t *testing.T) {
+	dir := t.TempDir()
+	omittedPath := filepath.Join(dir, "omitted.yml")
+	if err := os.WriteFile(omittedPath, []byte(testPipelineBaseYAML), 0o644); err != nil {
+		t.Fatalf("write omitted pipeline: %v", err)
+	}
+	omitted, err := LoadPipeline(omittedPath)
+	if err != nil {
+		t.Fatalf("LoadPipeline(omitted) error = %v", err)
+	}
+	if omitted.Notarius != nil {
+		t.Fatalf("Notarius = %#v, want nil when omitted", omitted.Notarius)
+	}
+
+	disabledPath := filepath.Join(dir, "disabled.yml")
+	disabledYAML := testPipelineBaseYAML + `
+notarius:
+  enabled: false
+  config_path: relative/notarius.yml
+`
+	if err := os.WriteFile(disabledPath, []byte(disabledYAML), 0o644); err != nil {
+		t.Fatalf("write disabled pipeline: %v", err)
+	}
+	disabled, err := LoadPipeline(disabledPath)
+	if err != nil {
+		t.Fatalf("LoadPipeline(disabled) error = %v", err)
+	}
+	if disabled.Notarius == nil {
+		t.Fatal("Notarius = nil, want configured disabled section")
+	}
+	if disabled.Notarius.Enabled {
+		t.Fatal("Notarius.Enabled = true, want false")
+	}
+	if disabled.Notarius.Binary != DefaultNotariusBinary || disabled.Notarius.Timeout != DefaultNotariusTimeout {
+		t.Fatalf("disabled defaults = %#v", disabled.Notarius)
+	}
+	if disabled.Notarius.ConfigPath != "relative/notarius.yml" || disabled.Notarius.WorkingDirectory != "" {
+		t.Fatalf("disabled paths were resolved unexpectedly: %#v", disabled.Notarius)
+	}
+}
+
+func TestNotariusEnabledDefaultsAndPathResolution(t *testing.T) {
+	dir := t.TempDir()
+	pipelinePath := filepath.Join(dir, "deployment", "pipeline.yml")
+	if err := os.MkdirAll(filepath.Dir(pipelinePath), 0o755); err != nil {
+		t.Fatalf("MkdirAll() error = %v", err)
+	}
+	pipelineYAML := testPipelineBaseYAML + `
+notarius:
+  enabled: true
+  config_path: notarius/config.yml
+  pipeline_id: dnd-session
+  outputs:
+    npc_registry:
+      lane_id: npc-registry
+      media_type: application/json
+      schema_id: notarius.dnd.npc_registry
+      schema_version: v1
+`
+	if err := os.WriteFile(pipelinePath, []byte(pipelineYAML), 0o644); err != nil {
+		t.Fatalf("write pipeline: %v", err)
+	}
+
+	cfg, err := LoadPipeline(pipelinePath)
+	if err != nil {
+		t.Fatalf("LoadPipeline() error = %v", err)
+	}
+	wantConfigPath := filepath.Join(filepath.Dir(pipelinePath), "notarius", "config.yml")
+	if cfg.Notarius.Binary != "notarius" || cfg.Notarius.Timeout != "3h" {
+		t.Fatalf("defaults = %#v", cfg.Notarius)
+	}
+	if cfg.Notarius.ConfigPath != wantConfigPath {
+		t.Fatalf("config_path = %q, want %q", cfg.Notarius.ConfigPath, wantConfigPath)
+	}
+	if cfg.Notarius.WorkingDirectory != filepath.Dir(wantConfigPath) {
+		t.Fatalf("working_directory = %q, want %q", cfg.Notarius.WorkingDirectory, filepath.Dir(wantConfigPath))
+	}
+
+	explicitPath := filepath.Join(dir, "explicit.yml")
+	explicitYAML := strings.Replace(pipelineYAML, "  pipeline_id: dnd-session\n", "  pipeline_id: dnd-session\n  working_directory: runtime\n", 1)
+	if err := os.WriteFile(explicitPath, []byte(explicitYAML), 0o644); err != nil {
+		t.Fatalf("write explicit pipeline: %v", err)
+	}
+	explicit, err := LoadPipeline(explicitPath)
+	if err != nil {
+		t.Fatalf("LoadPipeline(explicit working directory) error = %v", err)
+	}
+	if explicit.Notarius.WorkingDirectory != filepath.Join(dir, "runtime") {
+		t.Fatalf("explicit working_directory = %q, want %q", explicit.Notarius.WorkingDirectory, filepath.Join(dir, "runtime"))
+	}
+}
+
+func TestNotariusStrictYAML(t *testing.T) {
+	tests := []struct {
+		name string
+		yaml string
+	}{
+		{name: "unknown section field", yaml: "notarius:\n  unknown: true\n"},
+		{name: "unknown output field", yaml: "notarius:\n  outputs:\n    npc_registry:\n      lane_id: npc-registry\n      unknown: true\n"},
+		{name: "unsupported session id", yaml: "notarius:\n  session_id: forbidden\n"},
+		{name: "unsupported model", yaml: "notarius:\n  model: forbidden\n"},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			path := filepath.Join(t.TempDir(), "pipeline.yml")
+			if err := os.WriteFile(path, []byte(testPipelineBaseYAML+"\n"+tt.yaml), 0o644); err != nil {
+				t.Fatalf("write pipeline: %v", err)
+			}
+			if _, err := LoadPipeline(path); err == nil || !strings.Contains(err.Error(), "strict decode failed") {
+				t.Fatalf("LoadPipeline() error = %v, want strict decode failure", err)
+			}
+		})
+	}
+}
+
+func TestNotariusEnabledValidation(t *testing.T) {
+	valid := `notarius:
+  enabled: true
+  config_path: ./notarius.yml
+  pipeline_id: dnd-session
+  timeout: 45m
+  outputs:
+    npc_registry:
+      lane_id: npc-registry
+      media_type: application/json
+      schema_id: notarius.dnd.npc_registry
+      schema_version: v1
+      module_key: dnd/npc-registry
+`
+	tests := []struct {
+		name    string
+		section string
+		wantErr string
+	}{
+		{name: "valid", section: valid},
+		{name: "blank binary", section: strings.Replace(valid, "  enabled: true\n", "  enabled: true\n  binary: \"   \"\n", 1), wantErr: "pipeline.notarius.binary is required"},
+		{name: "missing config path", section: strings.Replace(valid, "  config_path: ./notarius.yml\n", "", 1), wantErr: "pipeline.notarius.config_path is required"},
+		{name: "missing pipeline id", section: strings.Replace(valid, "  pipeline_id: dnd-session\n", "", 1), wantErr: "pipeline.notarius.pipeline_id is required"},
+		{name: "missing outputs", section: strings.Split(valid, "  outputs:\n")[0], wantErr: "pipeline.notarius.outputs must contain at least one output"},
+		{name: "zero timeout", section: strings.Replace(valid, "  timeout: 45m", "  timeout: 0s", 1), wantErr: "pipeline.notarius.timeout must be positive"},
+		{name: "negative timeout", section: strings.Replace(valid, "  timeout: 45m", "  timeout: -1s", 1), wantErr: "pipeline.notarius.timeout must be positive"},
+		{name: "invalid timeout", section: strings.Replace(valid, "  timeout: 45m", "  timeout: later", 1), wantErr: "pipeline.notarius.timeout must be a valid duration"},
+		{name: "invalid output key", section: strings.Replace(valid, "    npc_registry:", "    npc-registry:", 1), wantErr: "outputs keys must match"},
+		{name: "missing lane", section: strings.Replace(valid, "      lane_id: npc-registry\n", "", 1), wantErr: "lane_id is required"},
+		{name: "missing media type", section: strings.Replace(valid, "      media_type: application/json\n", "", 1), wantErr: "media_type is required"},
+		{name: "missing schema id", section: strings.Replace(valid, "      schema_id: notarius.dnd.npc_registry\n", "", 1), wantErr: "schema_id is required"},
+		{name: "missing schema version", section: strings.Replace(valid, "      schema_version: v1\n", "", 1), wantErr: "schema_version is required"},
+		{
+			name: "normalized key collision",
+			section: strings.Replace(valid, "      module_key: dnd/npc-registry\n", `      module_key: dnd/npc-registry
+    " npc_registry ":
+      lane_id: npc-registry-two
+      media_type: application/json
+      schema_id: two
+      schema_version: v1
+`, 1),
+			wantErr: "normalize to duplicate source",
+		},
+		{
+			name: "duplicate normalized lane",
+			section: strings.Replace(valid, "      module_key: dnd/npc-registry\n", `      module_key: dnd/npc-registry
+    spells:
+      lane_id: " npc-registry "
+      media_type: application/json
+      schema_id: two
+      schema_version: v1
+`, 1),
+			wantErr: "duplicates pipeline.notarius.outputs.npc_registry.lane_id",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			pipelinePath, sessionPath := writeConfigFiles(t, testPipelineBaseYAML+"\n"+tt.section, testSessionBaseYAML)
+			cfg, err := Load(pipelinePath, sessionPath)
+			if err != nil {
+				t.Fatalf("Load() error = %v", err)
+			}
+			err = Validate(cfg)
+			if tt.wantErr == "" {
+				if err != nil {
+					t.Fatalf("Validate() error = %v", err)
+				}
+				output := cfg.Pipeline.Notarius.Outputs["npc_registry"]
+				if output.LaneID != "npc-registry" || output.ModuleKey != "dnd/npc-registry" {
+					t.Fatalf("normalized output = %#v", output)
+				}
+				return
+			}
+			if err == nil || !strings.Contains(err.Error(), tt.wantErr) {
+				t.Fatalf("Validate() error = %v, want containing %q", err, tt.wantErr)
+			}
+		})
+	}
+}
+
+func TestExtractionReferencesRequireDeclaredOutput(t *testing.T) {
+	declared := `notarius:
+  enabled: false
+  outputs:
+    npc_registry: {}
+`
+	tests := []struct {
+		name    string
+		body    string
+		wantErr string
+	}{
+		{
+			name: "scriptorium declared extraction",
+			body: declared + `scriptorium:
+  artifacts:
+    recap:
+      inputs:
+        npcs:
+          source: narratio.extraction.npc_registry
+`,
+		},
+		{
+			name: "scriptorium unknown extraction",
+			body: declared + `scriptorium:
+  artifacts:
+    recap:
+      inputs:
+        npcs:
+          source: narratio.extraction.unknown
+`,
+			wantErr: `references unknown extraction output "unknown"`,
+		},
+		{
+			name: "publish declared extraction",
+			body: declared + `publish:
+  outputs:
+    - source: narratio.extraction.npc_registry
+      dest: artifacts/npc-registry.json
+`,
+		},
+		{
+			name: "publish unknown extraction",
+			body: declared + `publish:
+  outputs:
+    - source: narratio.extraction.unknown
+      dest: artifacts/unknown.json
+`,
+			wantErr: `extraction output "unknown" is not defined`,
+		},
+		{
+			name: "publish lock declared extraction",
+			body: declared + `publish:
+  locks:
+    - source: narratio.extraction.npc_registry
+`,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			pipelinePath, sessionPath := writeConfigFiles(t, testPipelineBaseYAML+"\n"+tt.body, testSessionBaseYAML)
+			cfg, err := Load(pipelinePath, sessionPath)
+			if err != nil {
+				t.Fatalf("Load() error = %v", err)
+			}
+			err = Validate(cfg)
+			if tt.wantErr == "" {
+				if err != nil {
+					t.Fatalf("Validate() error = %v", err)
+				}
+				return
+			}
+			if err == nil || !strings.Contains(err.Error(), tt.wantErr) {
+				t.Fatalf("Validate() error = %v, want containing %q", err, tt.wantErr)
+			}
+		})
+	}
+}
--- a/internal/config/storage_publish_test.go
+++ b/internal/config/storage_publish_test.go
@@ -537,7 +537,7 @@ func TestPublishLockStoreBytesStrictDecodeAndValidation(t *testing.T) {
 	store, err := LoadPublishLockStoreBytes("locks.yml", []byte(`locks:
  - source: narratio.transcript.final_trimmed
    reason: reviewed
-`), nil)
+`), nil, nil)
 	if err != nil {
 		t.Fatalf("LoadPublishLockStoreBytes() error = %v", err)
 	}
@@ -548,7 +548,7 @@ func TestPublishLockStoreBytesStrictDecodeAndValidation(t *testing.T) {
 	_, err = LoadPublishLockStoreBytes("locks.yml", []byte(`locks:
  - source: narratio.transcript.final_trimmed
    dest: transcripts/final.trimmed.json
-`), nil)
+`), nil, nil)
 	if err == nil || !strings.Contains(err.Error(), "strict decode failed") {
 		t.Fatalf("unknown field error = %v, want strict decode failed", err)
 	}
@@ -556,7 +556,7 @@ func TestPublishLockStoreBytesStrictDecodeAndValidation(t *testing.T) {
 	_, err = LoadPublishLockStoreBytes("locks.yml", []byte(`locks:
  - source: narratio.transcript.final_trimmed
  - source: narratio.transcript.final_trimmed
-`), nil)
+`), nil, nil)
 	if err == nil || !strings.Contains(err.Error(), "duplicates another publish lock source") {
 		t.Fatalf("duplicate error = %v", err)
 	}
--- a/internal/config/validate.go
+++ b/internal/config/validate.go
@@ -6,9 +6,11 @@ import (
 	"net/url"
 	"path/filepath"
 	"regexp"
+	"sort"
 	"strings"
 	"time"

+	"gitea.maximumdirect.net/eric/narratio/internal/artifactmodel"
 	"gitea.maximumdirect.net/eric/narratio/internal/artifactpolicy"
 	"gitea.maximumdirect.net/eric/narratio/internal/pathsafe"
 )
@@ -85,7 +87,10 @@ func validatePipeline(cfg *PipelineConfig) error {
 	if err := validateCache(cfg.Cache); err != nil {
 		return err
 	}
-	if err := validatePublish(cfg.Publish, cfg.Scriptorium); err != nil {
+	if err := validateNotarius(cfg.Notarius, cfg.Scriptorium); err != nil {
+		return err
+	}
+	if err := validatePublish(cfg.Publish, cfg.Scriptorium, cfg.Notarius); err != nil {
 		return err
 	}
 	if err := validateWhisperX(cfg.WhisperX); err != nil {
@@ -106,7 +111,7 @@ func validatePipeline(cfg *PipelineConfig) error {
 	if err := validateRender(cfg.Render); err != nil {
 		return err
 	}
-	if err := validateScriptorium(cfg.Scriptorium); err != nil {
+	if err := validateScriptorium(cfg.Scriptorium, cfg.Notarius); err != nil {
 		return err
 	}
 	if err := validateDuration("pipeline.notification.timeout", cfg.Notification.Timeout); err != nil {
@@ -149,11 +154,12 @@ func validateCache(cfg CacheConfig) error {
 	return nil
 }

-func validatePublish(cfg *PublishConfig, scriptorium *ScriptoriumConfig) error {
+func validatePublish(cfg *PublishConfig, scriptorium *ScriptoriumConfig, notarius *NotariusConfig) error {
 	if cfg == nil {
 		return nil
 	}
 	configuredOutputs := scriptoriumOutputPathMap(scriptorium)
+	extractionOutputs := notariusOutputKeySet(notarius)
 	seenDest := map[string]struct{}{}
 	for i, item := range cfg.Outputs {
 		prefix := fmt.Sprintf("pipeline.publish.outputs[%d]", i)
@@ -161,7 +167,7 @@ func validatePublish(cfg *PublishConfig, scriptorium *ScriptoriumConfig) error {
 		if source == "" {
 			return fmt.Errorf("%s.source is required", prefix)
 		}
-		if _, err := artifactpolicy.ValidatePublishSource(source, configuredOutputs); err != nil {
+		if _, err := artifactpolicy.ValidatePublishSourceWithExtractions(source, configuredOutputs, extractionOutputs); err != nil {
 			return fmt.Errorf("%s.source %q is unsupported: %w", prefix, item.Source, err)
 		}
 		dest := strings.TrimSpace(item.Dest)
@@ -189,7 +195,7 @@ func validatePublish(cfg *PublishConfig, scriptorium *ScriptoriumConfig) error {
 		}
 		seenDest[normalizedDest] = struct{}{}
 	}
-	locks, err := ValidatePublishLockRules(cfg.Locks, scriptorium, "pipeline.publish.locks")
+	locks, err := ValidatePublishLockRules(cfg.Locks, scriptorium, notarius, "pipeline.publish.locks")
 	if err != nil {
 		return err
 	}
@@ -198,10 +204,11 @@ func validatePublish(cfg *PublishConfig, scriptorium *ScriptoriumConfig) error {
 }

 // ValidatePublishLockRules validates and normalizes source-based publish locks.
-func ValidatePublishLockRules(locks []PublishLockRule, scriptorium *ScriptoriumConfig, label string) ([]PublishLockRule, error) {
+func ValidatePublishLockRules(locks []PublishLockRule, scriptorium *ScriptoriumConfig, notarius *NotariusConfig, label string) ([]PublishLockRule, error) {
 	seenLocks := map[string]struct{}{}
 	out := make([]PublishLockRule, 0, len(locks))
 	configuredOutputs := scriptoriumOutputPathMap(scriptorium)
+	extractionOutputs := notariusOutputKeySet(notarius)
 	if strings.TrimSpace(label) == "" {
 		label = "publish.locks"
 	}
@@ -211,7 +218,7 @@ func ValidatePublishLockRules(locks []PublishLockRule, scriptorium *ScriptoriumC
 		if source == "" {
 			return nil, fmt.Errorf("%s.source is required", prefix)
 		}
-		if _, err := artifactpolicy.ValidatePublishSource(source, configuredOutputs); err != nil {
+		if _, err := artifactpolicy.ValidatePublishSourceWithExtractions(source, configuredOutputs, extractionOutputs); err != nil {
 			return nil, fmt.Errorf("%s.source %q is unsupported: %w", prefix, item.Source, err)
 		}
 		if _, ok := seenLocks[source]; ok {
@@ -264,6 +271,20 @@ func scriptoriumOutputPathMap(scriptorium *ScriptoriumConfig) map[string]string
 	return out
 }

+func notariusOutputKeySet(notarius *NotariusConfig) map[string]struct{} {
+	out := map[string]struct{}{}
+	if notarius == nil {
+		return out
+	}
+	for key := range notarius.Outputs {
+		trimmed := strings.TrimSpace(key)
+		if artifactpolicy.IsConfiguredKey(trimmed) {
+			out[trimmed] = struct{}{}
+		}
+	}
+	return out
+}
+
 func validateSecrets(cfg *SecretsConfig) error {
 	if cfg == nil {
 		return nil
@@ -474,7 +495,103 @@ func validateAudita(cfg AuditaConfig) error {
 	return nil
 }

-func validateScriptorium(cfg *ScriptoriumConfig) error {
+func validateNotarius(cfg *NotariusConfig, scriptorium *ScriptoriumConfig) error {
+	if cfg == nil || !cfg.Enabled {
+		return nil
+	}
+	if strings.TrimSpace(cfg.Binary) == "" {
+		return fmt.Errorf("pipeline.notarius.binary is required when pipeline.notarius.enabled is true")
+	}
+	if strings.TrimSpace(cfg.ConfigPath) == "" {
+		return fmt.Errorf("pipeline.notarius.config_path is required when pipeline.notarius.enabled is true")
+	}
+	if strings.TrimSpace(cfg.PipelineID) == "" {
+		return fmt.Errorf("pipeline.notarius.pipeline_id is required when pipeline.notarius.enabled is true")
+	}
+	if len(cfg.Outputs) == 0 {
+		return fmt.Errorf("pipeline.notarius.outputs must contain at least one output when pipeline.notarius.enabled is true")
+	}
+	duration, err := time.ParseDuration(strings.TrimSpace(cfg.Timeout))
+	if err != nil {
+		return fmt.Errorf("pipeline.notarius.timeout must be a valid duration: %w", err)
+	}
+	if duration <= 0 {
+		return fmt.Errorf("pipeline.notarius.timeout must be positive")
+	}
+	if strings.TrimSpace(cfg.WorkingDirectory) == "" {
+		return fmt.Errorf("pipeline.notarius.working_directory is required when pipeline.notarius.enabled is true")
+	}
+
+	reservedSources := map[string]string{}
+	for _, spec := range artifactmodel.RuntimeTranscriptArtifacts() {
+		reservedSources[spec.SourceID] = "built-in source"
+	}
+	reservedSources[artifactpolicy.SourceBoundsSession] = "built-in source"
+	if scriptorium != nil {
+		for key := range scriptorium.Artifacts {
+			normalizedKey := strings.TrimSpace(key)
+			reservedSources[artifactpolicy.ConfiguredSourceID(normalizedKey)] = "configured Scriptorium source"
+			reservedSources[artifactpolicy.PreviousSessionSourceID(normalizedKey)] = "previous-session source"
+		}
+	}
+
+	rawKeys := make([]string, 0, len(cfg.Outputs))
+	for key := range cfg.Outputs {
+		rawKeys = append(rawKeys, key)
+	}
+	sort.Strings(rawKeys)
+	normalizedOutputs := make(map[string]NotariusOutputConfig, len(cfg.Outputs))
+	sourceOwners := map[string]string{}
+	laneOwners := map[string]string{}
+	for _, rawKey := range rawKeys {
+		output := cfg.Outputs[rawKey]
+		key := strings.TrimSpace(rawKey)
+		if !artifactpolicy.IsConfiguredKey(key) {
+			return fmt.Errorf("pipeline.notarius.outputs keys must match ^[a-z][a-z0-9_]*$")
+		}
+		sourceID := artifactpolicy.ExtractionSourceID(key)
+		if previousKey, ok := sourceOwners[sourceID]; ok {
+			return fmt.Errorf("pipeline.notarius.outputs keys %q and %q normalize to duplicate source %q", previousKey, rawKey, sourceID)
+		}
+		if owner, ok := reservedSources[sourceID]; ok {
+			return fmt.Errorf("pipeline.notarius.outputs.%s source %q collides with %s", key, sourceID, owner)
+		}
+		sourceOwners[sourceID] = rawKey
+
+		output.LaneID = strings.TrimSpace(output.LaneID)
+		output.MediaType = strings.TrimSpace(output.MediaType)
+		output.SchemaID = strings.TrimSpace(output.SchemaID)
+		output.SchemaVersion = strings.TrimSpace(output.SchemaVersion)
+		output.ModuleKey = strings.TrimSpace(output.ModuleKey)
+		prefix := "pipeline.notarius.outputs." + key
+		if output.LaneID == "" {
+			return fmt.Errorf("%s.lane_id is required", prefix)
+		}
+		if previousKey, ok := laneOwners[output.LaneID]; ok {
+			return fmt.Errorf("%s.lane_id %q duplicates pipeline.notarius.outputs.%s.lane_id", prefix, output.LaneID, previousKey)
+		}
+		laneOwners[output.LaneID] = key
+		if output.MediaType == "" {
+			return fmt.Errorf("%s.media_type is required", prefix)
+		}
+		if output.SchemaID == "" {
+			return fmt.Errorf("%s.schema_id is required", prefix)
+		}
+		if output.SchemaVersion == "" {
+			return fmt.Errorf("%s.schema_version is required", prefix)
+		}
+		normalizedOutputs[key] = output
+	}
+	cfg.Binary = strings.TrimSpace(cfg.Binary)
+	cfg.ConfigPath = filepath.Clean(cfg.ConfigPath)
+	cfg.PipelineID = strings.TrimSpace(cfg.PipelineID)
+	cfg.Timeout = strings.TrimSpace(cfg.Timeout)
+	cfg.WorkingDirectory = filepath.Clean(cfg.WorkingDirectory)
+	cfg.Outputs = normalizedOutputs
+	return nil
+}
+
+func validateScriptorium(cfg *ScriptoriumConfig, notarius *NotariusConfig) error {
 	if cfg == nil {
 		return nil
 	}
@@ -491,7 +608,7 @@ func validateScriptorium(cfg *ScriptoriumConfig) error {
 	configuredArtifacts := make(map[string]struct{}, len(cfg.Artifacts))
 	referencedArtifacts := make(map[string]struct{})
 	for artifactName := range cfg.Artifacts {
-		if !scriptoriumArtifactKeyRE.MatchString(strings.TrimSpace(artifactName)) {
+		if !artifactpolicy.IsConfiguredKey(artifactName) {
 			return fmt.Errorf("pipeline.scriptorium.artifacts keys must match ^[a-z][a-z0-9_]*$")
 		}
 		configuredArtifacts[artifactName] = struct{}{}
@@ -544,7 +661,7 @@ func validateScriptorium(cfg *ScriptoriumConfig) error {
 				return fmt.Errorf("pipeline.scriptorium.artifacts.%s.inputs.%s.source is required", artifactName, inputName)
 			}

-			referencedArtifact, err := validateScriptoriumInputSource(artifactName, inputName, source, configuredArtifacts)
+			referencedArtifact, err := validateScriptoriumInputSource(artifactName, inputName, source, configuredArtifacts, notariusOutputKeySet(notarius))
 			if err != nil {
 				return err
 			}
@@ -684,9 +801,12 @@ func publishUploadConfiguredForS3(pipeline *PipelineConfig) bool {

 var windowsAbsPathRE = regexp.MustCompile(`^[A-Za-z]:[\\/].*`)
 var envVarNameRE = regexp.MustCompile(`^[A-Za-z_][A-Za-z0-9_]*$`)
-var scriptoriumArtifactKeyRE = regexp.MustCompile(`^[a-z][a-z0-9_]*$`)

-func validateScriptoriumInputSource(artifactName, inputName, source string, configuredArtifacts map[string]struct{}) (string, error) {
+func validateScriptoriumInputSource(
+	artifactName, inputName, source string,
+	configuredArtifacts map[string]struct{},
+	extractionOutputs map[string]struct{},
+) (string, error) {
 	trimmedSource := strings.TrimSpace(source)
 	descriptor, err := artifactpolicy.DescribeScriptoriumInputSource(trimmedSource)
 	if err != nil {
@@ -705,7 +825,7 @@ func validateScriptoriumInputSource(artifactName, inputName, source string, conf
 			source,
 		)
 	}
-	if err := artifactpolicy.ValidateInputConfiguredReference(descriptor, configuredArtifacts); err != nil {
+	if err := artifactpolicy.ValidateInputReference(descriptor, configuredArtifacts, extractionOutputs); err != nil {
 		var unknownConfigured *artifactpolicy.UnknownConfiguredArtifactError
 		if errors.As(err, &unknownConfigured) {
 			return "", fmt.Errorf(
@@ -716,6 +836,16 @@ func validateScriptoriumInputSource(artifactName, inputName, source string, conf
 				unknownConfigured.ConfiguredKey,
 			)
 		}
+		var unknownExtraction *artifactpolicy.UnknownExtractionArtifactError
+		if errors.As(err, &unknownExtraction) {
+			return "", fmt.Errorf(
+				"pipeline.scriptorium.artifacts.%s.inputs.%s.source %q references unknown extraction output %q",
+				artifactName,
+				inputName,
+				source,
+				unknownExtraction.ConfiguredKey,
+			)
+		}
 		return "", fmt.Errorf(
 			"pipeline.scriptorium.artifacts.%s.inputs.%s.source %q is unsupported",
 			artifactName,