Updated the merge stage to normalize the per-speaker transcripts before merging them

.gitignore

@@ -22,6 +22,9 @@ AGENTS.md
 # Dependency directories (remove the comment below to include it)
 # vendor/
 
+# Go cache
+.gocache
+
 # Go workspace file
 go.work
 go.work.sum

README.md

@@ -71,7 +71,7 @@ Narratio now includes configuration and path-model foundations for archive suppo
 
 Implemented foundations:
 
-- `pipeline.storage.s3` config shape (`bucket`, `root_prefix`, `region`, `endpoint`, `force_path_style`)
+- `pipeline.storage.s3` config shape (`bucket`, `root_prefix`, `region`, `endpoint`, `force_path_style`, `access_key_id_env`, `secret_access_key_env`)
 - `pipeline.spool` config shape (`root`, `delete_audio_after_archive`)
 - `pipeline.archive` config shape (`enabled`, `upload_run`, `promote_artifacts`)
 - promotion-rule validation (`from`/`to` required, relative-only paths, traversal rejected)
@@ -83,6 +83,8 @@ Implemented foundations:
 Current defaults:
 
 - `pipeline.storage.s3.root_prefix`: `dnd`
+- `pipeline.storage.s3.access_key_id_env`: `OBJECT_STORAGE_KEY_ID`
+- `pipeline.storage.s3.secret_access_key_env`: `OBJECT_STORAGE_KEY`
 - `pipeline.workspace.cleanup_after_archive`: `false`
 - `pipeline.spool.root`: `/var/spool/narratio`
 - `pipeline.spool.delete_audio_after_archive`: `false`
@@ -110,7 +112,8 @@ Current boundaries:
   - `pipeline.workspace.cleanup_after_archive: true` removes only the run-scoped local workdir after successful archive commit
   - cleanup executes only after all selected stages for the command invocation succeed
   - cleanup does not run for failed, incomplete, skipped, or unarchived runs
-  - local development `audio_dir`/`audio_files` source inputs are never deleted by spool cleanup
+- local development `audio_dir`/`audio_files` source inputs are never deleted by spool cleanup
+- S3 credentials are resolved from configured env-var names when both are present; if either is missing, Narratio falls back to the AWS SDK default credential chain
 
 S3 input details and current boundaries are documented in [docs/s3-audio-input.md](docs/s3-audio-input.md).
 

architecture.md

@@ -151,6 +151,8 @@ Storage and archive foundations:
   - `region`
   - `endpoint`
   - `force_path_style` (default `false`)
+  - `access_key_id_env` (default `OBJECT_STORAGE_KEY_ID`)
+  - `secret_access_key_env` (default `OBJECT_STORAGE_KEY`)
 - `pipeline.spool.root` defaults to `/var/spool/narratio`
 - `pipeline.workspace.cleanup_after_archive` defaults to `false`
 - `pipeline.spool.delete_audio_after_archive` defaults to `false`
@@ -176,7 +178,9 @@ Session input foundations:
 Cross-config validation scope:
 
 - `pipeline.storage.s3.bucket` is required only when an S3-dependent feature is explicitly configured (for current foundations, that includes `session.inputs.audio_s3`, and archive upload intent when using `storage.backend: s3`)
-- no AWS credentials are stored in Narratio config; credential resolution remains an external runtime concern
+- no AWS credential values are stored in Narratio config; only env-var names are configured
+- when both configured credential env vars resolve to non-empty values, the S3 backend uses them as static credentials
+- when either configured credential value is missing, the S3 backend falls back to the AWS SDK default credential chain
 
 Remote object-store backend scope:
 

docs/s3-audio-input.md

@@ -20,6 +20,8 @@ Not implemented:
 
 - `storage.s3.bucket` must be set when S3 audio input is used.
 - `storage.s3.root_prefix` defaults to `dnd`.
+- `storage.s3.access_key_id_env` defaults to `OBJECT_STORAGE_KEY_ID`.
+- `storage.s3.secret_access_key_env` defaults to `OBJECT_STORAGE_KEY`.
 - `spool.root` defaults to `/var/spool/narratio`.
 
 `session.yml`:

docs/storage-backends.md

@@ -30,6 +30,8 @@ Construction:
   - region
   - endpoint
   - force_path_style
+  - access_key_id_env
+  - secret_access_key_env
 
 ## Key Invariant
 
@@ -42,7 +44,9 @@ S3 session/run key builders remain separate and continue to live outside backend
 ## Security Boundary
 
 - do not store AWS credentials in Narratio config
-- AWS credentials are resolved through standard AWS SDK credential chains
+- Narratio first checks configured env-var names (`access_key_id_env`, `secret_access_key_env`);
+  when both are present and non-empty, it uses static credentials from those values
+- when either configured credential value is missing, Narratio falls back to the standard AWS SDK credential chain
 - AWS SDK-specific types remain isolated to the storage adapter package
 
 ## Testing

examples/pipeline.minimal.yml

@@ -8,6 +8,9 @@ storage:
     bucket: "my-dnd-archive"
     root_prefix: "dnd"
     region: "us-east-1"
+    # Optional credential env-var names (defaulted when omitted):
+    # access_key_id_env: "OBJECT_STORAGE_KEY_ID"
+    # secret_access_key_env: "OBJECT_STORAGE_KEY"
 
 spool:
   root: "/var/spool/narratio"

internal/adapters/storage/s3_backend.go

@@ -11,6 +11,7 @@ import (
 	"time"
 
 	awsconfig "github.com/aws/aws-sdk-go-v2/config"
+	"github.com/aws/aws-sdk-go-v2/credentials"
 	"github.com/aws/aws-sdk-go-v2/service/s3"
 	"github.com/aws/aws-sdk-go-v2/service/s3/types"
 	"github.com/aws/smithy-go"
@@ -35,6 +36,8 @@ type s3ClientOptions struct {
 	Region         string
 	Endpoint       string
 	ForcePathStyle bool
+	AccessKeyID    string
+	SecretKey      string
 }
 
 var newS3Client = func(ctx context.Context, opts s3ClientOptions) (s3API, error) {
@@ -42,6 +45,15 @@ var newS3Client = func(ctx context.Context, opts s3ClientOptions) (s3API, error)
 	if strings.TrimSpace(opts.Region) != "" {
 		loadOpts = append(loadOpts, awsconfig.WithRegion(strings.TrimSpace(opts.Region)))
 	}
+	if strings.TrimSpace(opts.AccessKeyID) != "" && strings.TrimSpace(opts.SecretKey) != "" {
+		loadOpts = append(loadOpts, awsconfig.WithCredentialsProvider(
+			credentials.NewStaticCredentialsProvider(
+				strings.TrimSpace(opts.AccessKeyID),
+				strings.TrimSpace(opts.SecretKey),
+				"",
+			),
+		))
+	}
 	awsCfg, err := awsconfig.LoadDefaultConfig(ctx, loadOpts...)
 	if err != nil {
 		return nil, fmt.Errorf("load aws config: %w", err)
@@ -67,6 +79,8 @@ func NewS3BackendFromConfig(ctx context.Context, cfg config.StorageS3Config) (*S
 		Region:         cfg.Region,
 		Endpoint:       cfg.Endpoint,
 		ForcePathStyle: cfg.ForcePathStyle,
+		AccessKeyID:    s3CredentialFromEnv(orDefaultEnvName(cfg.AccessKeyIDEnv, config.DefaultS3AccessKeyIDEnv)),
+		SecretKey:      s3CredentialFromEnv(orDefaultEnvName(cfg.SecretKeyEnv, config.DefaultS3SecretAccessKeyEnv)),
 	})
 	if err != nil {
 		return nil, fmt.Errorf("build s3 client: %w", err)
@@ -78,6 +92,26 @@ func NewS3BackendFromConfig(ctx context.Context, cfg config.StorageS3Config) (*S
 	}, nil
 }
 
+func s3CredentialFromEnv(envVarName string) string {
+	name := strings.TrimSpace(envVarName)
+	if name == "" {
+		return ""
+	}
+	value, ok := os.LookupEnv(name)
+	if !ok {
+		return ""
+	}
+	return strings.TrimSpace(value)
+}
+
+func orDefaultEnvName(name, fallback string) string {
+	trimmed := strings.TrimSpace(name)
+	if trimmed == "" {
+		return fallback
+	}
+	return trimmed
+}
+
 // List returns objects under prefix.
 func (b *S3Backend) List(ctx context.Context, prefix string) ([]ObjectInfo, error) {
 	normalizedPrefix := normalizeObjectKey(prefix)

internal/adapters/storage/s3_backend_test.go

@@ -187,6 +187,8 @@ func TestS3BackendExistsNotFound(t *testing.T) {
 func TestNewS3BackendFromConfigUsesClientOptions(t *testing.T) {
 	original := newS3Client
 	t.Cleanup(func() { newS3Client = original })
+	t.Setenv("OBJECT_STORAGE_KEY_ID", "id-123")
+	t.Setenv("OBJECT_STORAGE_KEY", "secret-abc")
 
 	var got s3ClientOptions
 	newS3Client = func(_ context.Context, opts s3ClientOptions) (s3API, error) {
@@ -209,6 +211,9 @@ func TestNewS3BackendFromConfigUsesClientOptions(t *testing.T) {
 	if got.Region != "us-east-1" || got.Endpoint != "http://localhost:9000" || !got.ForcePathStyle {
 		t.Fatalf("client options = %#v, want region/endpoint/path-style values", got)
 	}
+	if got.AccessKeyID != "id-123" || got.SecretKey != "secret-abc" {
+		t.Fatalf("client options credentials = %#v, want env-resolved static credentials", got)
+	}
 }
 
 func TestNewS3BackendFromConfigRequiresBucket(t *testing.T) {
@@ -218,6 +223,30 @@ func TestNewS3BackendFromConfigRequiresBucket(t *testing.T) {
 	}
 }
 
+func TestNewS3BackendFromConfigFallsBackWhenCredentialEnvMissing(t *testing.T) {
+	original := newS3Client
+	t.Cleanup(func() { newS3Client = original })
+
+	var got s3ClientOptions
+	newS3Client = func(_ context.Context, opts s3ClientOptions) (s3API, error) {
+		got = opts
+		return &fakeS3API{}, nil
+	}
+
+	_, err := NewS3BackendFromConfig(context.Background(), config.StorageS3Config{
+		Bucket:         "my-archive",
+		Region:         "us-east-1",
+		AccessKeyIDEnv: "MISSING_ACCESS_KEY_ID",
+		SecretKeyEnv:   "MISSING_SECRET_KEY",
+	})
+	if err != nil {
+		t.Fatalf("NewS3BackendFromConfig() error = %v", err)
+	}
+	if got.AccessKeyID != "" || got.SecretKey != "" {
+		t.Fatalf("client options credentials = %#v, want empty fallback values", got)
+	}
+}
+
 func strPtr(v string) *string { return &v }
 func int64Ptr(v int64) *int64 { return &v }
 

internal/config/config.go

@@ -60,6 +60,8 @@ type StorageS3Config struct {
 	Region         string `yaml:"region"`
 	Endpoint       string `yaml:"endpoint"`
 	ForcePathStyle bool   `yaml:"force_path_style"`
+	AccessKeyIDEnv string `yaml:"access_key_id_env"`
+	SecretKeyEnv   string `yaml:"secret_access_key_env"`
 }
 
 // SpoolConfig configures local spool storage for staged data.

internal/config/defaults.go

@@ -8,6 +8,8 @@ const (
 	DefaultSessionConfigPathLocal     = "./session.yml"
 	DefaultSessionConfigPathUsrLocal  = "/usr/local/etc/narratio/session.yml"
 	DefaultSessionConfigPathEtc       = "/etc/narratio/session.yml"
+	DefaultS3AccessKeyIDEnv           = "OBJECT_STORAGE_KEY_ID"
+	DefaultS3SecretAccessKeyEnv       = "OBJECT_STORAGE_KEY"
 )
 
 // DefaultPipelineConfigSearchPaths defines the default search order for

internal/config/load.go

@@ -176,6 +176,12 @@ func applyStorageDefaults(cfg *StorageConfig) {
 	if cfg.S3.RootPrefix == "" {
 		cfg.S3.RootPrefix = "dnd"
 	}
+	if cfg.S3.AccessKeyIDEnv == "" {
+		cfg.S3.AccessKeyIDEnv = DefaultS3AccessKeyIDEnv
+	}
+	if cfg.S3.SecretKeyEnv == "" {
+		cfg.S3.SecretKeyEnv = DefaultS3SecretAccessKeyEnv
+	}
 }
 
 func applySpoolDefaults(cfg *SpoolConfig) {

internal/config/storage_archive_test.go

@@ -24,6 +24,12 @@ storage:
 	if cfg.Pipeline.Storage.S3.RootPrefix != "dnd" {
 		t.Fatalf("storage.s3.root_prefix = %q, want dnd", cfg.Pipeline.Storage.S3.RootPrefix)
 	}
+	if cfg.Pipeline.Storage.S3.AccessKeyIDEnv != DefaultS3AccessKeyIDEnv {
+		t.Fatalf("storage.s3.access_key_id_env = %q, want %q", cfg.Pipeline.Storage.S3.AccessKeyIDEnv, DefaultS3AccessKeyIDEnv)
+	}
+	if cfg.Pipeline.Storage.S3.SecretKeyEnv != DefaultS3SecretAccessKeyEnv {
+		t.Fatalf("storage.s3.secret_access_key_env = %q, want %q", cfg.Pipeline.Storage.S3.SecretKeyEnv, DefaultS3SecretAccessKeyEnv)
+	}
 	if cfg.Pipeline.Storage.S3.ForcePathStyle {
 		t.Fatalf("storage.s3.force_path_style = true, want false default")
 	}
@@ -33,6 +39,77 @@ storage:
 	}
 }
 
+func TestStorageS3CredentialEnvNamesLoadAndValidate(t *testing.T) {
+	pipelineYAML := testPipelineBaseYAML + `
+storage:
+  backend: s3
+  s3:
+    bucket: my-dnd-archive
+    access_key_id_env: CUSTOM_KEY_ID
+    secret_access_key_env: CUSTOM_SECRET
+`
+	pipelinePath, sessionPath := writeConfigFiles(t, pipelineYAML, testSessionBaseYAML)
+
+	cfg, err := Load(pipelinePath, sessionPath)
+	if err != nil {
+		t.Fatalf("Load() error = %v", err)
+	}
+	if cfg.Pipeline.Storage.S3.AccessKeyIDEnv != "CUSTOM_KEY_ID" {
+		t.Fatalf("storage.s3.access_key_id_env = %q, want CUSTOM_KEY_ID", cfg.Pipeline.Storage.S3.AccessKeyIDEnv)
+	}
+	if cfg.Pipeline.Storage.S3.SecretKeyEnv != "CUSTOM_SECRET" {
+		t.Fatalf("storage.s3.secret_access_key_env = %q, want CUSTOM_SECRET", cfg.Pipeline.Storage.S3.SecretKeyEnv)
+	}
+	if err := Validate(cfg); err != nil {
+		t.Fatalf("Validate() error = %v", err)
+	}
+}
+
+func TestStorageS3CredentialEnvValidation(t *testing.T) {
+	tests := []struct {
+		name        string
+		pipelineYML string
+		wantErr     string
+	}{
+		{
+			name: "invalid access key env name",
+			pipelineYML: testPipelineBaseYAML + `
+storage:
+  backend: s3
+  s3:
+    bucket: my-dnd-archive
+    access_key_id_env: "123BAD"
+`,
+			wantErr: "pipeline.storage.s3.access_key_id_env must be a valid environment variable name",
+		},
+		{
+			name: "invalid secret key env name",
+			pipelineYML: testPipelineBaseYAML + `
+storage:
+  backend: s3
+  s3:
+    bucket: my-dnd-archive
+    secret_access_key_env: "bad-name"
+`,
+			wantErr: "pipeline.storage.s3.secret_access_key_env must be a valid environment variable name",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			pipelinePath, sessionPath := writeConfigFiles(t, tt.pipelineYML, testSessionBaseYAML)
+			cfg, err := Load(pipelinePath, sessionPath)
+			if err != nil {
+				t.Fatalf("Load() error = %v", err)
+			}
+			err = Validate(cfg)
+			if err == nil || !strings.Contains(err.Error(), tt.wantErr) {
+				t.Fatalf("Validate() error = %v, want to contain %q", err, tt.wantErr)
+			}
+		})
+	}
+}
+
 func TestSpoolAndArchiveDefaults(t *testing.T) {
 	pipelinePath, sessionPath := writeConfigFiles(t, testPipelineBaseYAML, testSessionBaseYAML)
 

internal/config/validate.go

@@ -91,6 +91,12 @@ func validateStorage(cfg StorageConfig) error {
 	if cfg.S3.Endpoint != "" && strings.TrimSpace(cfg.S3.Endpoint) == "" {
 		return fmt.Errorf("pipeline.storage.s3.endpoint must be non-empty when provided")
 	}
+	if err := validateEnvVarNameField("pipeline.storage.s3.access_key_id_env", cfg.S3.AccessKeyIDEnv); err != nil {
+		return err
+	}
+	if err := validateEnvVarNameField("pipeline.storage.s3.secret_access_key_env", cfg.S3.SecretKeyEnv); err != nil {
+		return err
+	}
 	return nil
 }
 
@@ -430,6 +436,18 @@ func archiveUploadConfiguredForS3(pipeline *PipelineConfig) bool {
 }
 
 var windowsAbsPathRE = regexp.MustCompile(`^[A-Za-z]:[\\/].*`)
+var envVarNameRE = regexp.MustCompile(`^[A-Za-z_][A-Za-z0-9_]*$`)
+
+func validateEnvVarNameField(fieldName, value string) error {
+	trimmed := strings.TrimSpace(value)
+	if trimmed == "" {
+		return fmt.Errorf("%s must be non-empty", fieldName)
+	}
+	if !envVarNameRE.MatchString(trimmed) {
+		return fmt.Errorf("%s must be a valid environment variable name", fieldName)
+	}
+	return nil
+}
 
 func validateRelativeSafePath(fieldName, value string) error {
 	trimmed := strings.TrimSpace(value)

internal/stage/merge.go

@@ -7,6 +7,7 @@ import (
 	"path/filepath"
 	"sort"
 	"strings"
+	"time"
 
 	"gitea.maximumdirect.net/eric/narratio/internal/adapters/seriatim"
 	"gitea.maximumdirect.net/eric/narratio/internal/artifacts"
@@ -86,10 +87,15 @@ func (mergeStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*Sta
 	stderrPath := filepath.Join(paths.LogsDir, "seriatim.stderr.log")
 	genCfgPath := filepath.Join(paths.ConfigDir, "seriatim.generated.yml")
 
+	normalizedInputs, normalizeLogs, normalizeConfigs, normalizeMeta, err := normalizeMergeInputs(ctx, env, inputs, paths)
+	if err != nil {
+		return nil, err
+	}
+
 	reportEnabled := env.Config.Pipeline.Seriatim.Report != nil && *env.Config.Pipeline.Seriatim.Report
 	req := seriatim.MergeRequest{
 		GeneratedConfigPath:        genCfgPath,
-		InputTranscriptPaths:       inputs,
+		InputTranscriptPaths:       normalizedInputs,
 		OutputMergedTranscriptPath: mergedPath,
 		ReportPath:                 "",
 		SpeakersPath:               speakersPath,
@@ -146,8 +152,11 @@ func (mergeStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*Sta
 
 	meta := map[string]any{
 		"stage":                    "merge",
-		"input_transcripts_count":  len(inputs),
+		"input_transcripts_count":  len(normalizedInputs),
 		"input_transcript_paths":   inputs,
+		"normalized_inputs_count":  len(normalizedInputs),
+		"normalized_input_paths":   normalizedInputs,
+		"normalize_inputs":         normalizeMeta,
 		"output_schema":            env.Config.Pipeline.Seriatim.OutputSchema,
 		"coalesce_gap":             coalesceGap,
 		"report_enabled":           reportEnabled,
@@ -172,12 +181,93 @@ func (mergeStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*Sta
 
 	return &StageResult{
 		Outputs:          outputs,
-		Logs:             []string{stdoutPath, stderrPath},
-		GeneratedConfigs: []string{genCfgPath},
+		Logs:             append(normalizeLogs, stdoutPath, stderrPath),
+		GeneratedConfigs: append(normalizeConfigs, genCfgPath),
 		Metadata:         meta,
 	}, nil
 }
 
+type normalizeMergeInputMeta struct {
+	InputPath         string `json:"input_path"`
+	OutputPath        string `json:"output_path"`
+	StdoutLogPath     string `json:"stdout_log_path"`
+	StderrLogPath     string `json:"stderr_log_path"`
+	GeneratedConfig   string `json:"generated_config_path"`
+	DurationMs        int64  `json:"duration_ms"`
+	ExitCode          int    `json:"exit_code"`
+	InvokedBinary     string `json:"invoked_binary"`
+	OutputSchema      string `json:"output_schema"`
+	AdapterReportPath string `json:"adapter_report_path,omitempty"`
+	AdapterOutputPath string `json:"adapter_output_path,omitempty"`
+}
+
+func normalizeMergeInputs(ctx context.Context, env *Env, rawInputs []string, paths artifacts.SessionPaths) ([]string, []string, []string, []normalizeMergeInputMeta, error) {
+	normalizedInputs := make([]string, 0, len(rawInputs))
+	logs := make([]string, 0, len(rawInputs)*2)
+	configs := make([]string, 0, len(rawInputs))
+	meta := make([]normalizeMergeInputMeta, 0, len(rawInputs))
+
+	var timeout time.Duration
+	timeoutRaw := strings.TrimSpace(env.Config.Pipeline.Seriatim.Timeout)
+	if timeoutRaw != "" {
+		parsed, err := time.ParseDuration(timeoutRaw)
+		if err != nil {
+			return nil, nil, nil, nil, fmt.Errorf("merge: parse seriatim timeout %q: %w", env.Config.Pipeline.Seriatim.Timeout, err)
+		}
+		timeout = parsed
+	}
+	for _, input := range rawInputs {
+		base := strings.TrimSuffix(filepath.Base(input), filepath.Ext(input))
+		outPath := filepath.Join(paths.TranscriptsRawDir, "normalized", base+".normalized.json")
+		stdoutPath := filepath.Join(paths.LogsDir, "seriatim.normalize."+base+".stdout.log")
+		stderrPath := filepath.Join(paths.LogsDir, "seriatim.normalize."+base+".stderr.log")
+		cfgPath := filepath.Join(paths.ConfigDir, "seriatim.normalize."+base+".generated.yml")
+
+		req := seriatim.NormalizeRequest{
+			Binary:               env.Config.Pipeline.Seriatim.Binary,
+			InputTranscriptPath:  input,
+			OutputNormalizedPath: outPath,
+			OutputSchema:         env.Config.Pipeline.Seriatim.OutputSchema,
+			ReportPath:           "",
+			StdoutLogPath:        stdoutPath,
+			StderrLogPath:        stderrPath,
+			GeneratedConfigPath:  cfgPath,
+			Timeout:              timeout,
+		}
+		res, err := env.Seriatim.Normalize(ctx, req)
+		if err != nil {
+			return nil, nil, nil, nil, fmt.Errorf("merge: normalize input %q failed: %w", input, err)
+		}
+
+		finalOutputPath := outPath
+		if strings.TrimSpace(res.OutputNormalizedPath) != "" {
+			finalOutputPath = strings.TrimSpace(res.OutputNormalizedPath)
+		}
+		if err := validateTranscriptJSONFile(finalOutputPath); err != nil {
+			return nil, nil, nil, nil, fmt.Errorf("merge: normalized transcript %q invalid (input %q): %w", finalOutputPath, input, err)
+		}
+
+		normalizedInputs = append(normalizedInputs, finalOutputPath)
+		logs = append(logs, stdoutPath, stderrPath)
+		configs = append(configs, cfgPath)
+		meta = append(meta, normalizeMergeInputMeta{
+			InputPath:         input,
+			OutputPath:        finalOutputPath,
+			StdoutLogPath:     stdoutPath,
+			StderrLogPath:     stderrPath,
+			GeneratedConfig:   cfgPath,
+			DurationMs:        res.Duration.Milliseconds(),
+			ExitCode:          res.ExitCode,
+			InvokedBinary:     res.InvokedBinary,
+			OutputSchema:      res.OutputSchema,
+			AdapterReportPath: res.ReportPath,
+			AdapterOutputPath: res.OutputNormalizedPath,
+		})
+	}
+
+	return normalizedInputs, logs, configs, meta, nil
+}
+
 func discoverRawTranscripts(m *manifest.Manifest, paths artifacts.SessionPaths) ([]string, error) {
 	fromManifest := make([]string, 0)
 	if m != nil && m.Stages != nil {

internal/stage/merge_test.go

@@ -54,6 +54,17 @@ func TestMergeStageMergesRawTranscriptsAndRecordsMetadata(t *testing.T) {
 	if len(req.InputTranscriptPaths) != 2 {
 		t.Fatalf("input transcripts = %#v, want 2", req.InputTranscriptPaths)
 	}
+	if len(fake.NormalizeRequests) != 2 {
+		t.Fatalf("normalize requests = %#v, want 2", fake.NormalizeRequests)
+	}
+	if fake.NormalizeRequests[0].InputTranscriptPath != inA || fake.NormalizeRequests[1].InputTranscriptPath != inB {
+		t.Fatalf("normalize request inputs = %#v", fake.NormalizeRequests)
+	}
+	for _, mergeIn := range req.InputTranscriptPaths {
+		if !strings.Contains(mergeIn, filepath.Join("transcripts", "raw", "normalized")) {
+			t.Fatalf("merge input path = %q, want normalized input path", mergeIn)
+		}
+	}
 
 	if len(result.Outputs) != 2 {
 		t.Fatalf("outputs len = %d, want 2", len(result.Outputs))
@@ -64,11 +75,11 @@ func TestMergeStageMergesRawTranscriptsAndRecordsMetadata(t *testing.T) {
 	if result.Outputs[1].Kind != "seriatim_report" {
 		t.Fatalf("output[1] kind = %q, want seriatim_report", result.Outputs[1].Kind)
 	}
-	if len(result.Logs) != 2 {
-		t.Fatalf("logs = %#v, want 2 paths", result.Logs)
+	if len(result.Logs) != 6 {
+		t.Fatalf("logs = %#v, want 6 paths (4 normalize + 2 merge)", result.Logs)
 	}
-	if len(result.GeneratedConfigs) != 1 {
-		t.Fatalf("generated configs = %#v, want 1 path", result.GeneratedConfigs)
+	if len(result.GeneratedConfigs) != 3 {
+		t.Fatalf("generated configs = %#v, want 3 paths (2 normalize + 1 merge)", result.GeneratedConfigs)
 	}
 
 	meta := result.Metadata
@@ -84,6 +95,12 @@ func TestMergeStageMergesRawTranscriptsAndRecordsMetadata(t *testing.T) {
 	if meta["input_transcripts_count"] != 2 {
 		t.Fatalf("metadata input_transcripts_count = %#v, want 2", meta["input_transcripts_count"])
 	}
+	if meta["normalized_inputs_count"] != 2 {
+		t.Fatalf("metadata normalized_inputs_count = %#v, want 2", meta["normalized_inputs_count"])
+	}
+	if _, ok := meta["normalize_inputs"]; !ok {
+		t.Fatalf("metadata normalize_inputs missing: %#v", meta)
+	}
 }
 
 func TestMergeStageFailsWhenNoRawTranscripts(t *testing.T) {
@@ -152,6 +169,9 @@ func TestMergeStageFallsBackToRawDirectoryWhenTranscribeOutputsMissing(t *testin
 	if len(fake.Requests) != 1 || len(fake.Requests[0].InputTranscriptPaths) != 1 {
 		t.Fatalf("fallback inputs = %#v", fake.Requests)
 	}
+	if len(fake.NormalizeRequests) != 1 {
+		t.Fatalf("normalize requests = %#v, want 1", fake.NormalizeRequests)
+	}
 }
 
 func TestMergeStageResolvesWorkspaceQualifiedManifestOutputsWithoutDuplication(t *testing.T) {
@@ -180,8 +200,51 @@ func TestMergeStageResolvesWorkspaceQualifiedManifestOutputsWithoutDuplication(t
 	if len(got) != 1 {
 		t.Fatalf("input transcript paths = %#v, want len 1", got)
 	}
-	if got[0] != filepath.Clean(rawPath) {
-		t.Fatalf("resolved transcript path = %q, want %q", got[0], filepath.Clean(rawPath))
+	if !strings.Contains(got[0], filepath.Join("transcripts", "raw", "normalized")) {
+		t.Fatalf("resolved transcript path = %q, want normalized path under transcripts/raw/normalized", got[0])
+	}
+}
+
+func TestMergeStageFailsWhenNormalizeAdapterFails(t *testing.T) {
+	env, m := setupMergeEnv(t)
+	paths := env.ArtifactStore.SessionPaths(m.SessionID)
+	in := filepath.Join(paths.TranscriptsRawDir, "alice.json")
+	writeFile(t, in, `{"segments":[]}`)
+	writeFile(t, filepath.Join(paths.InputsDir, "speakers.yml"), "match: []\n")
+	writeFile(t, filepath.Join(paths.InputsDir, "autocorrect.yml"), "rules: []\n")
+	env.Seriatim = &seriatim.FakeRunner{NormalizeErr: context.DeadlineExceeded}
+
+	_, err := (mergeStage{}).Run(context.Background(), env, m)
+	if err == nil {
+		t.Fatal("expected error, got nil")
+	}
+	if !strings.Contains(err.Error(), "normalize input") {
+		t.Fatalf("error = %q", err.Error())
+	}
+}
+
+func TestMergeStageFailsWhenNormalizedOutputInvalid(t *testing.T) {
+	env, m := setupMergeEnv(t)
+	paths := env.ArtifactStore.SessionPaths(m.SessionID)
+	in := filepath.Join(paths.TranscriptsRawDir, "alice.json")
+	writeFile(t, in, `{"segments":[]}`)
+	writeFile(t, filepath.Join(paths.InputsDir, "speakers.yml"), "match: []\n")
+	writeFile(t, filepath.Join(paths.InputsDir, "autocorrect.yml"), "rules: []\n")
+
+	badNormalized := filepath.Join(paths.ArtifactsDir, "bad.normalized.json")
+	writeFile(t, badNormalized, "not-json")
+	env.Seriatim = &seriatim.FakeRunner{
+		NormalizeResult: seriatim.NormalizeResult{
+			OutputNormalizedPath: badNormalized,
+		},
+	}
+
+	_, err := (mergeStage{}).Run(context.Background(), env, m)
+	if err == nil {
+		t.Fatal("expected error, got nil")
+	}
+	if !strings.Contains(err.Error(), "normalized transcript") {
+		t.Fatalf("error = %q", err.Error())
 	}
 }
 
@@ -211,8 +274,8 @@ func TestMergeStageResolvesSessionRelativeManifestOutputs(t *testing.T) {
 	if len(got) != 1 {
 		t.Fatalf("input transcript paths = %#v, want len 1", got)
 	}
-	if got[0] != filepath.Clean(rawPath) {
-		t.Fatalf("resolved transcript path = %q, want %q", got[0], filepath.Clean(rawPath))
+	if !strings.Contains(got[0], filepath.Join("transcripts", "raw", "normalized")) {
+		t.Fatalf("resolved transcript path = %q, want normalized path under transcripts/raw/normalized", got[0])
 	}
 }
 

report-gemma4.md

@@ -1,31 +0,0 @@
-# Narratio UX Evaluation Report
-
-## 1. Executive Summary
-Narratio has a functional core pipeline with robust S3 integration for input and output, but it currently falls short of the intended "minimalist" operator UX. The primary gaps are the lack of session configuration discovery, the absence of session template support (and the `--session-id` flag), and the missing local cleanup logic. While the pipeline runs successfully, the operator must currently provide explicit session file paths for every run.
-
-## 2. Feature Matrix
-
-| Feature | Status | Evidence | Tests | Documentation | Notes |
-| :--- | :--- | :--- | :--- | :--- | :--- |
-| **Pipeline Config Discovery** | Implemented | `internal/app/pipeline_config_path_test.go` | Yes | Accurate | Checks `/usr/local/etc` and `/etc`. |
-| **Session Config Discovery** | Missing | `internal/app/run.go:30` | N/A | Stale | `--session` is mandatory. |
-| **Session Templates** | Missing | `internal/config/load.go` | N/A | Missing | No variable interpolation in `session.yml`. |
-| **`--session-id` CLI Flag** | Missing | `cmd/narratio` | N/A | Missing | Not implemented in CLI. |
-| **Minimal Seriatim Config** | Implemented | `internal/config/load.go` | Yes | Accurate | Defaults for timeout/schema provided. |
-| **Minimal Audita Config** | Implemented | `internal/config/load.go` | Yes | Accurate | Defaults for base_url/model provided. |
-| **S3 Audio Input** | Implemented | `internal/stage/prepare.go` | Yes | Accurate | Supports `.flac` downloads from S3. |
-| **S3 Archive & Promotion** | Implemented | `internal/stage/archive.go` | Yes | Accurate | Correct paths and commit markers. |
-| **Local Cleanup** | Missing | `architecture.md:136` | No | Stale | Config exists, logic is not implemented. |
-
-## 3. Current Happy Path
-The shortest command that works today is:
-`narratio run --session <path_to_session.yml>`
-*(Assuming `pipeline.yml` is present in `/etc/narratio/` or `/usr/local/etc/narratio/`)*.
-
-## 4. Gaps to Intended UX
-1. **Session Discovery & Templates (High):** The requirement to pass `--session` and the inability to use `--session-id` with a template is the largest friction point for operators.
-2. **Local Cleanup (Medium):** Spool and work directories are not cleaned up after successful archival, leading to local disk growth.
-3. **Local Pipeline Config (Low):** Narratio does not check `./pipeline.yml`, requiring users to use `--config` or move files to system directories.
-
-## 5. Recommended Next Implementation Prompt
-"Implement session configuration discovery and template support. Specifically: 1) Add a search order for `session.yml` (e.g., `./session.yml`, `/etc/narratio/session.yml`) if `--session` is omitted. 2) Implement the `--session-id` CLI flag. 3) Add variable interpolation to `session.yml` so that `{{session_id}}` can be replaced by the value from the flag or the discovered session config before YAML decoding."

report.md

@@ -1,54 +0,0 @@
-## 1. Executive Summary
-Narratio is close on S3 input/archive mechanics but not yet close on the intended minimal operator UX.
-Core S3 workflow is implemented (prepare S3 audio download, archive run upload, promotions, current pointers), but key UX items are missing: no `--session-id` flag, no session auto-discovery, and no session template variable injection. Cleanup/retention for spool/workdirs after archive is also still future work.
-
-## 2. Feature Matrix
-
-| Feature | Status | Evidence | Tests | Documentation status | Notes |
-|---|---|---|---|---|---|
-| Pipeline config auto-discovery when `--config` omitted | Implemented | `internal/app/pipeline_config_path.go`, `internal/config/defaults.go` | `internal/app/pipeline_config_path_test.go`, `internal/app/commands_test.go` | Accurate in `README.md`, `architecture.md` | Order: `/usr/local/etc/narratio/pipeline.yml`, then `/etc/narratio/pipeline.yml`; no `./pipeline.yml` default |
-| Session config auto-discovery when `--session` omitted | Missing | `--session` required in `internal/app/run.go`, `plan.go`, `resume.go`, `run_stage.go` | Covered by missing-flag tests in `internal/app/commands_test.go` | Accurate (docs do not claim auto-discovery) | No precedence order exists for session file search |
-| Session template variables in `session.yml` | Missing | Strict decode path in `internal/config/load.go` + strict YAML behavior | No template tests found | Not documented as implemented | No render-before-decode templating mechanism found |
-| `--session-id` CLI injection | Missing | No `--session-id` flag in command parsers (`run/plan/resume/run-stage`) | No tests for `--session-id` | Not documented as implemented | Intended minimal UX command not currently supported |
-| Campaign/run-aware work+spool paths | Implemented | `internal/artifacts/paths.go`, usage in prepare/archive | Path/helper tests in `internal/artifacts` + stage tests | Documented in README/architecture/roadmap | Layout includes `{campaign}/{session_id}/{run_id}` |
-| Run ID generation format | Implemented | `internal/artifacts/run_id.go` | Run ID tests in `internal/artifacts` | Documented | UTC timestamp + random suffix format present |
-| Storage backend abstraction | Implemented | `internal/adapters/storage/object_store.go` | Storage backend tests in `internal/adapters/storage` | Documented in README/architecture | Narrow interface (`List/Download/Upload/Exists`) |
-| S3 backend + fake backend | Implemented | `internal/adapters/storage/s3_backend.go`, `fake.go` | Adapter tests pass without live S3 | Documented | No AWS creds in config schema/examples |
-| Prepare S3 audio input (`inputs.audio_s3`) | Implemented | `internal/stage/prepare.go` | `internal/stage/prepare_test.go` | Documented in `docs/s3-audio-input.md`, README, architecture | Lists prefix, filters `.flac`, downloads/materializes, fails on none |
-| Local audio workflow | Implemented | Prepare logic still supports `audio_dir`/`audio_files` | Prepare tests cover local behavior and conflict with `audio_s3` | Documented | Local+S3 conflict is enforced |
-| Manifest provenance for S3 audio | Implemented | S3 source metadata assignment in prepare stage | Covered by S3 prepare tests | Documented | ETag recorded as metadata, not checksum |
-| Archive run upload under `runs/{run_id}` | Implemented | `internal/stage/archive.go` | `internal/stage/archive_test.go` | Documented in `docs/archive-storage.md`, README, architecture | Successful/completed runs only |
-| Archive promotion rules | Implemented | Archive stage promotion handling | Archive tests cover required/optional/mapping behavior | Documented | Default promoted outputs: `transcripts/trimmed.json`, `artifacts/session_recap.md` |
-| `current/manifest.json` + `current/run_id.txt` last | Implemented | Archive stage upload order logic | Archive tests verify ordering and pointer content | Documented | `current/run_id.txt` is commit marker; written last |
-| Avoid upload of failed/incomplete runs | Implemented | Archive prerequisite checks | Archive tests cover prerequisite failure path | Documented | Failed runs stay local |
-| Spool/workdir cleanup after successful archive | Missing | `spool.delete_audio_after_archive` exists but no cleanup behavior in stages/app | No cleanup behavior tests found | Docs accurately call cleanup future work | Gap vs intended UX item 12 |
-| Minimal Seriatim config | Partial | Validation requires `seriatim.binary`; defaults fill timeout/schema/gap | Config load/validate tests | Docs mostly accurate | “Binary-only” works after defaults, but still validated post-defaults |
-| Minimal Audita config | Partial | Validation requires `audita.binary` and `audita.model`; defaults for timeout/base_url/etc in loader | Config tests in `internal/config` | Docs currently list `timeout`/`base_url` as required in README section | UX expectation “binary + llm_api_key_env only” does not hold because model is required |
-
-## 3. Current Happy Path
-Shortest realistic command today is:
-
-`narratio run --session /path/to/session.yml`
-
-That works only if pipeline config is discoverable at `/usr/local/etc/narratio/pipeline.yml` or `/etc/narratio/pipeline.yml`.
-Otherwise minimum is:
-
-`narratio run --config /path/to/pipeline.yml --session /path/to/session.yml`
-
-`narratio run --session-id 2026-04-04` does not work today (flag not implemented).
-
-## 4. Gaps to Intended UX
-1. Missing `--session-id` flow with session template injection (largest UX gap).
-2. No session config auto-discovery order when `--session` is omitted.
-3. No session template rendering engine / unresolved-variable handling.
-4. Cleanup policy not implemented (`spool.delete_audio_after_archive` is modeled only).
-5. Audita minimal config UX still stricter than intended (model required).
-6. Optional doc refinement: explicitly call out that `./pipeline.yml` is not in current default search order.
-
-## 5. Recommended Next Implementation Prompt
-Implement session template and `--session-id` UX only:
-
-> Add session discovery and template rendering support so `narratio run --session-id <id>` works with no `--session` in normal setups.
-> Requirements: define deterministic session discovery order; support rendering template variables in `session.yml` before strict YAML decode; inject CLI `--session-id` into template variables; fail clearly on unresolved variables; preserve strict field validation after render; keep existing `--session` explicit path behavior; add tests for discovery precedence, render success/failure, and CLI integration; update README/architecture/examples accordingly; do not change archive/prepare storage behavior.
-
-Validation note: `go test ./...` passes for the inspected state.