From 32653f54f9ab87db872979dee8abfd45679055b1 Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Mon, 10 Aug 2026 21:41:00 +0000 Subject: [PATCH] Make configuration truthful and clean remote session files --- docs/cli.md | 2 + docs/config.md | 22 ++- docs/internal/overview.md | 8 +- docs/roadmap/implementation.md | 2 + examples/pipeline.full.annotated.yml | 6 +- examples/pipeline.production.yml | 2 +- internal/app/clean.go | 4 +- internal/app/commands_test.go | 6 +- internal/app/config_loader.go | 66 ++++++++- internal/app/operator_artifacts_list.go | 3 +- internal/app/operator_helpers.go | 24 ++-- internal/app/operator_locks.go | 9 +- internal/app/operator_session_validate.go | 4 +- internal/app/operator_status.go | 4 +- internal/app/plan.go | 4 +- internal/app/plan_test.go | 2 +- internal/app/remote_session_test.go | 141 ++++++++++++++++++++ internal/app/restore.go | 4 +- internal/app/run.go | 4 +- internal/app/run_stage.go | 4 +- internal/app/runner_test.go | 2 +- internal/config/cache_config_test.go | 2 +- internal/config/campaign_config_test.go | 4 +- internal/config/config.go | 9 +- internal/config/defaults.go | 1 + internal/config/load.go | 10 ++ internal/config/notification_test.go | 68 ++++++++++ internal/config/scriptorium_test.go | 19 ++- internal/config/validate.go | 9 +- internal/config/validation_duration_test.go | 2 - internal/stage/prepare.go | 17 ++- internal/stage/prepare_test.go | 3 +- 32 files changed, 395 insertions(+), 72 deletions(-) create mode 100644 internal/config/notification_test.go diff --git a/docs/cli.md b/docs/cli.md index 4c27461..29222ae 100644 --- a/docs/cli.md +++ b/docs/cli.md @@ -50,6 +50,8 @@ Rules: - `--previous-session-id` is a strict expectation: the selected session file must contain the same `previous_session_id`. - `clean --all` cannot be combined with campaign/session selectors. +- notification delivery is currently limited to the configured `noop` mode; see + the [configuration reference](./config.md#notifications). ## Session ID Input Rules diff --git a/docs/config.md b/docs/config.md index a7aa622..98f0b98 100644 --- a/docs/config.md +++ b/docs/config.md @@ -38,6 +38,10 @@ If local session discovery fails and a `session_id` is known, Narratio attempts using configured object storage. +The downloaded remote session file is command-scoped: Narratio removes it after +the command finishes and records only the remote object provenance alongside +the durable copied session input. + ### Identity segments Campaign IDs (`campaign_id` and `default_campaign_id`), session IDs, previous @@ -235,9 +239,7 @@ Rules: | `pipeline.scriptorium.timeout` | duration | No | `10m` | | `pipeline.scriptorium.render_debug` | bool | No | `false` | | `pipeline.scriptorium.artifacts` | map | No | empty | -| `pipeline.notification.backend` | string | No | empty | -| `pipeline.notification.recipient` | string | No | empty | -| `pipeline.notification.timeout` | duration | No | empty | +| `pipeline.notification.mode` | string | No | `noop`; the only supported notification mode until a provider is implemented | ### Notarius Output Entries @@ -282,10 +284,20 @@ For each artifact input `pipeline.scriptorium.artifacts..inputs.`, `narratio.artifact.`, or `narratio.previous_session.artifact.` | -| `artifact` | string | No | optional passthrough adapter field | -| `path` | string | No | optional passthrough adapter field | | `required` | bool | No | optional input requirement | +`artifact` and `path` are obsolete and rejected by strict configuration +loading. Use the canonical `source` identifier to select the input; Narratio +does not provide adapter-specific input passthrough fields. + +### Notifications + +Narratio currently supports only `notification.mode: noop`, which is also the +default when the section is omitted. The notify stage performs no delivery in +this mode. Backend, recipient, timeout, and other provider settings are +rejected by strict configuration loading until Narratio has a provider +integration. + ### Campaign | Field | Type | Required | Notes | diff --git a/docs/internal/overview.md b/docs/internal/overview.md index 860fa87..d1f89c2 100644 --- a/docs/internal/overview.md +++ b/docs/internal/overview.md @@ -57,11 +57,11 @@ The implemented canonical order is: 8. [`render`](stage-render.md) 9. [`analyze`](stage-analyze.md) 10. [`publish`](stage-publish.md) -11. `notify` (placeholder) +11. `notify` (no-op) -`notify` currently has optional notifier call behavior and no persisted pipeline -outputs; its default collaborator is a no-op sender. The focused stage -documents own implementation mechanics. The +`notify` currently has no persisted pipeline outputs and uses the explicit +`noop` notification mode. The focused stage documents own implementation +mechanics. The [CLI](../cli.md) and [Operations](../operations.md) own user-visible invocation and execution semantics. diff --git a/docs/roadmap/implementation.md b/docs/roadmap/implementation.md index f9802f9..9834a96 100644 --- a/docs/roadmap/implementation.md +++ b/docs/roadmap/implementation.md @@ -776,6 +776,8 @@ temporary paths within a command-scoped lifecycle. backend/recipient rejection, every remote-config exit path, cancellation, stale cleanup, and manifest inspection for ephemeral paths. +**Status:** Completed. + ## Stage 23 — Stream WhisperX uploads and make the adapter race-safe **Read first:** `audit-findings.md` lines 2254–2281 (COR-016), 3113–3156 diff --git a/examples/pipeline.full.annotated.yml b/examples/pipeline.full.annotated.yml index 0bd4293..147ee64 100644 --- a/examples/pipeline.full.annotated.yml +++ b/examples/pipeline.full.annotated.yml @@ -260,7 +260,5 @@ scriptorium: output_kind: player_handout notification: - # Optional notification settings. - backend: "" - recipient: "" - timeout: 30s + # No delivery provider is currently implemented. + mode: noop diff --git a/examples/pipeline.production.yml b/examples/pipeline.production.yml index b9fcf50..92ff115 100644 --- a/examples/pipeline.production.yml +++ b/examples/pipeline.production.yml @@ -125,4 +125,4 @@ scriptorium: output_kind: player_handout notification: - timeout: 30s + mode: noop diff --git a/internal/app/clean.go b/internal/app/clean.go index ec34fb4..5e4f2aa 100644 --- a/internal/app/clean.go +++ b/internal/app/clean.go @@ -40,10 +40,12 @@ func cleanSession(ctx context.Context, flags commonConfigFlags, dryRun, clearCac if strings.TrimSpace(flags.sessionID) == "" { return fmt.Errorf("clean: session_id is required unless --all is set") } - cfg, err := loadCommandConfig(ctx, flags.pipelinePath, flags.campaignPath, flags.campaignFilePath, flags.sessionPath, flags.sessionOptions()) + loaded, err := loadCommandConfig(ctx, flags.pipelinePath, flags.campaignPath, flags.campaignFilePath, flags.sessionPath, flags.sessionOptions()) if err != nil { return fmt.Errorf("clean: %w", err) } + defer func() { _ = loaded.Close() }() + cfg := loaded.Config if cfg == nil || cfg.Pipeline == nil || cfg.Session == nil { return fmt.Errorf("clean: resolved pipeline and session config are required") } diff --git a/internal/app/commands_test.go b/internal/app/commands_test.go index 4382545..24bcbc0 100644 --- a/internal/app/commands_test.go +++ b/internal/app/commands_test.go @@ -218,7 +218,7 @@ audita: binary: ` + auditaBinary + ` llm_api_key_env: OPENROUTER_API_KEY notification: - timeout: 10s + mode: noop ` sessionYAML := `session_id: ` + sessionID + ` campaign: sample-campaign @@ -283,7 +283,7 @@ seriatim: audita: binary: audita notification: - timeout: 10s + mode: noop ` sessionYAML := `session_id: 2026-05-03 campaign: sample-campaign @@ -510,7 +510,7 @@ seriatim: audita: binary: ` + auditaBinary + ` notification: - timeout: 10s + mode: noop ` sessionYAML := `session_id: 2026-05-03 diff --git a/internal/app/config_loader.go b/internal/app/config_loader.go index ab516cb..d15fd87 100644 --- a/internal/app/config_loader.go +++ b/internal/app/config_loader.go @@ -2,13 +2,16 @@ package app import ( "context" + "errors" "fmt" "os" + "path/filepath" "strings" "gitea.maximumdirect.net/eric/narratio/internal/adapters/storage" "gitea.maximumdirect.net/eric/narratio/internal/artifacts" "gitea.maximumdirect.net/eric/narratio/internal/config" + "gitea.maximumdirect.net/eric/narratio/internal/fileops" ) type pipelineCampaignConfig struct { @@ -18,14 +21,48 @@ type pipelineCampaignConfig struct { Campaign *config.CampaignConfig } -func loadCommandConfig(ctx context.Context, pipelineFlag, campaignFlag, campaignFileFlag, sessionFlag string, sessionOpts config.SessionLoadOptions) (*config.Config, error) { +var downloadObjectToTempFn = storage.DownloadObjectToTemp + +type commandConfig struct { + Config *config.Config + cleanup func() error +} + +func (c *commandConfig) Close() error { + if c == nil || c.cleanup == nil { + return nil + } + cleanup := c.cleanup + c.cleanup = nil + return cleanup() +} + +func retainedCommandConfig(cfg *config.Config) *commandConfig { + return &commandConfig{Config: cfg} +} + +func loadCommandConfig(ctx context.Context, pipelineFlag, campaignFlag, campaignFileFlag, sessionFlag string, sessionOpts config.SessionLoadOptions) (loaded *commandConfig, err error) { + var cleanup func() error + defer func() { + if err == nil || cleanup == nil { + return + } + if cleanupErr := cleanup(); cleanupErr != nil { + err = errors.Join(err, cleanupErr) + } + }() + base, err := loadPipelineCampaignConfig(pipelineFlag, campaignFlag, campaignFileFlag) if err != nil { return nil, err } if explicitSession := strings.TrimSpace(sessionFlag); explicitSession != "" { - return config.LoadWithSessionOptions(base.PipelinePath, base.CampaignPath, explicitSession, sessionOpts) + cfg, err := config.LoadWithSessionOptions(base.PipelinePath, base.CampaignPath, explicitSession, sessionOpts) + if err != nil { + return nil, err + } + return retainedCommandConfig(cfg), nil } discoveredSession, err := discoverSessionConfigPathWithCandidates(config.DefaultSessionConfigSearchPaths) @@ -33,7 +70,11 @@ func loadCommandConfig(ctx context.Context, pipelineFlag, campaignFlag, campaign return nil, err } if discoveredSession.Path != "" { - return config.LoadWithSessionOptions(base.PipelinePath, base.CampaignPath, discoveredSession.Path, sessionOpts) + cfg, err := config.LoadWithSessionOptions(base.PipelinePath, base.CampaignPath, discoveredSession.Path, sessionOpts) + if err != nil { + return nil, err + } + return retainedCommandConfig(cfg), nil } sessionID := strings.TrimSpace(sessionOpts.SessionID) @@ -62,20 +103,26 @@ func loadCommandConfig(ctx context.Context, pipelineFlag, campaignFlag, campaign if err != nil { return nil, missingSessionConfigError(discoveredSession.Searched, err.Error()) } - sessionTempPath, err := storage.DownloadObjectToTemp(ctx, store, remoteKey, "narratio-session-*.yml") + sessionTempPath, err := downloadObjectToTempFn(ctx, store, remoteKey, "narratio-session-*.yml") if err != nil { return nil, missingSessionConfigError(discoveredSession.Searched, fmt.Sprintf("remote session %q download failed: %v", remoteKey, err)) } + cleanup = func() error { + if err := fileops.RemoveAllUnderRoot(filepath.Dir(sessionTempPath), sessionTempPath); err != nil { + return fmt.Errorf("remove downloaded remote session config: %w", err) + } + return nil + } sessionBytes, err := os.ReadFile(sessionTempPath) if err != nil { - return nil, fmt.Errorf("read downloaded remote session %q: %w", sessionTempPath, err) + return nil, fmt.Errorf("read downloaded remote session config: %w", err) } sessionCfg, err := config.LoadSessionBytesWithOptions("s3://"+s3BucketName(base.Pipeline)+"/"+remoteKey, sessionBytes, sessionOpts) if err != nil { return nil, err } - return config.Resolve( + cfg, err := config.Resolve( base.PipelinePath, base.Pipeline, base.CampaignPath, @@ -89,9 +136,14 @@ func loadCommandConfig(ctx context.Context, pipelineFlag, campaignFlag, campaign S3Key: remoteKey, S3Size: sessionInfo.Size, S3ETag: sessionInfo.ETag, - SpoolPath: sessionTempPath, }, ) + if err != nil { + return nil, err + } + loaded = &commandConfig{Config: cfg, cleanup: cleanup} + cleanup = nil + return loaded, nil } func loadPipelineCampaignConfig(pipelineFlag, campaignFlag, campaignFileFlag string) (*pipelineCampaignConfig, error) { diff --git a/internal/app/operator_artifacts_list.go b/internal/app/operator_artifacts_list.go index 9ee6894..5ebd502 100644 --- a/internal/app/operator_artifacts_list.go +++ b/internal/app/operator_artifacts_list.go @@ -22,10 +22,11 @@ func ArtifactsList(ctx context.Context, args []string, out io.Writer) error { if strings.TrimSpace(flags.sessionID) == "" { return fmt.Errorf("artifacts list: session_id is required") } - cfg, store, locks, m, err := loadHelperContext(ctx, flags, remote) + cfg, store, locks, m, cleanup, err := loadHelperContext(ctx, flags, remote) if err != nil { return fmt.Errorf("artifacts list: %w", err) } + defer cleanup() catalog, err := buildHelperArtifactCatalog(cfg, m) if err != nil { return fmt.Errorf("artifacts list: %w", err) diff --git a/internal/app/operator_helpers.go b/internal/app/operator_helpers.go index f6725a6..50b0b70 100644 --- a/internal/app/operator_helpers.go +++ b/internal/app/operator_helpers.go @@ -90,33 +90,41 @@ func Artifacts(ctx context.Context, args []string, out io.Writer) error { } } -func loadHelperContext(ctx context.Context, flags commonConfigFlags, needStore bool) (*config.Config, storage.ObjectStore, *effectiveLocks, *manifest.Manifest, error) { - cfg, err := loadCommandConfig(ctx, flags.pipelinePath, flags.campaignPath, flags.campaignFilePath, flags.sessionPath, flags.sessionOptions()) +func loadHelperContext(ctx context.Context, flags commonConfigFlags, needStore bool) (*config.Config, storage.ObjectStore, *effectiveLocks, *manifest.Manifest, func(), error) { + loaded, err := loadCommandConfig(ctx, flags.pipelinePath, flags.campaignPath, flags.campaignFilePath, flags.sessionPath, flags.sessionOptions()) if err != nil { - return nil, nil, nil, nil, err + return nil, nil, nil, nil, nil, err } + release := true + defer func() { + if release { + _ = loaded.Close() + } + }() + cfg := loaded.Config if err := config.Validate(cfg); err != nil { - return nil, nil, nil, nil, err + return nil, nil, nil, nil, nil, err } var store storage.ObjectStore if needStore { store, err = newCommandObjectStore(ctx, cfg, nil) if err != nil { - return nil, nil, nil, nil, err + return nil, nil, nil, nil, nil, err } } else { store, _ = objectStoreIfConfigured(ctx, cfg) } locks, err := loadEffectiveLocks(ctx, cfg, store) if err != nil { - return nil, nil, nil, nil, err + return nil, nil, nil, nil, nil, err } paths := artifacts.NewLocalStore(cfg.Pipeline.Workspace.Root).SessionPathsFor(cfg.Session.Campaign, cfg.Session.SessionID) m, err := loadLocalManifest(ctx, paths.ManifestPath) if err != nil { - return nil, nil, nil, nil, err + return nil, nil, nil, nil, nil, err } - return cfg, store, locks, m, nil + release = false + return cfg, store, locks, m, func() { _ = loaded.Close() }, nil } func objectStoreIfConfigured(ctx context.Context, cfg *config.Config) (storage.ObjectStore, error) { diff --git a/internal/app/operator_locks.go b/internal/app/operator_locks.go index a149597..f4426d5 100644 --- a/internal/app/operator_locks.go +++ b/internal/app/operator_locks.go @@ -38,10 +38,11 @@ func LocksList(ctx context.Context, args []string, out io.Writer) error { if strings.TrimSpace(flags.sessionID) == "" { return fmt.Errorf("locks: session_id is required") } - cfg, _, locks, _, err := loadHelperContext(ctx, flags, true) + cfg, _, locks, _, cleanup, err := loadHelperContext(ctx, flags, true) if err != nil { return fmt.Errorf("locks: %w", err) } + defer cleanup() writeLocks(out, cfg, locks) return nil } @@ -63,10 +64,11 @@ func LocksAdd(ctx context.Context, args []string, out io.Writer) error { if strings.TrimSpace(flags.sessionID) == "" { return fmt.Errorf("locks add: session_id is required") } - cfg, store, locks, _, err := loadHelperContext(ctx, flags, true) + cfg, store, locks, _, cleanup, err := loadHelperContext(ctx, flags, true) if err != nil { return fmt.Errorf("locks add: %w", err) } + defer cleanup() 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) } @@ -106,10 +108,11 @@ func LocksRemove(ctx context.Context, args []string, out io.Writer) error { if strings.TrimSpace(flags.sessionID) == "" { return fmt.Errorf("locks remove: session_id is required") } - cfg, store, locks, _, err := loadHelperContext(ctx, flags, true) + cfg, store, locks, _, cleanup, err := loadHelperContext(ctx, flags, true) if err != nil { return fmt.Errorf("locks remove: %w", err) } + defer cleanup() 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) } diff --git a/internal/app/operator_session_validate.go b/internal/app/operator_session_validate.go index ce53309..7575386 100644 --- a/internal/app/operator_session_validate.go +++ b/internal/app/operator_session_validate.go @@ -25,11 +25,13 @@ func SessionValidate(ctx context.Context, args []string, out io.Writer) error { } findings := []finding{} - cfg, err := loadCommandConfig(ctx, flags.pipelinePath, flags.campaignPath, flags.campaignFilePath, flags.sessionPath, flags.sessionOptions()) + loaded, err := loadCommandConfig(ctx, flags.pipelinePath, flags.campaignPath, flags.campaignFilePath, flags.sessionPath, flags.sessionOptions()) if err != nil { findings = append(findings, errorFinding("config", err.Error())) return renderFindings(out, "", "", findings) } + defer func() { _ = loaded.Close() }() + cfg := loaded.Config if err := config.Validate(cfg); err != nil { findings = append(findings, errorFinding("config", err.Error())) } else { diff --git a/internal/app/operator_status.go b/internal/app/operator_status.go index 11bb74a..dd07f2b 100644 --- a/internal/app/operator_status.go +++ b/internal/app/operator_status.go @@ -26,10 +26,12 @@ func Status(ctx context.Context, args []string, out io.Writer) error { if strings.TrimSpace(flags.sessionID) == "" { return fmt.Errorf("status: session_id is required") } - cfg, err := loadCommandConfig(ctx, flags.pipelinePath, flags.campaignPath, flags.campaignFilePath, flags.sessionPath, flags.sessionOptions()) + loaded, err := loadCommandConfig(ctx, flags.pipelinePath, flags.campaignPath, flags.campaignFilePath, flags.sessionPath, flags.sessionOptions()) if err != nil { return fmt.Errorf("status: %w", err) } + defer func() { _ = loaded.Close() }() + cfg := loaded.Config if err := config.Validate(cfg); err != nil { return fmt.Errorf("status: %w", err) } diff --git a/internal/app/plan.go b/internal/app/plan.go index 8b734a1..536a4e5 100644 --- a/internal/app/plan.go +++ b/internal/app/plan.go @@ -30,10 +30,12 @@ func Plan(ctx context.Context, args []string, out io.Writer) error { if flags.sessionID == "" { return fmt.Errorf("plan: session_id is required") } - cfg, err := loadCommandConfig(ctx, flags.pipelinePath, flags.campaignPath, flags.campaignFilePath, flags.sessionPath, flags.sessionOptions()) + loaded, err := loadCommandConfig(ctx, flags.pipelinePath, flags.campaignPath, flags.campaignFilePath, flags.sessionPath, flags.sessionOptions()) if err != nil { return fmt.Errorf("plan: %w", err) } + defer func() { _ = loaded.Close() }() + cfg := loaded.Config if err := config.Validate(cfg); err != nil { return fmt.Errorf("plan: %w", err) } diff --git a/internal/app/plan_test.go b/internal/app/plan_test.go index 6a694db..5ec8eb5 100644 --- a/internal/app/plan_test.go +++ b/internal/app/plan_test.go @@ -109,7 +109,7 @@ seriatim: audita: binary: audita notification: - timeout: 10s + mode: noop ` sessionYAML := `session_id: 2026-05-03 campaign: sample-campaign diff --git a/internal/app/remote_session_test.go b/internal/app/remote_session_test.go index 89e6bbc..8431160 100644 --- a/internal/app/remote_session_test.go +++ b/internal/app/remote_session_test.go @@ -13,6 +13,7 @@ import ( "gitea.maximumdirect.net/eric/narratio/internal/adapters/storage" "gitea.maximumdirect.net/eric/narratio/internal/artifacts" "gitea.maximumdirect.net/eric/narratio/internal/config" + "gitea.maximumdirect.net/eric/narratio/internal/stage" ) func TestExecuteRemoteSessionFallbackLoadsFromObjectStore(t *testing.T) { @@ -44,6 +45,146 @@ inputs: } } +func TestRemoteSessionConfigIsRemovedAfterEveryCommandExit(t *testing.T) { + tests := []struct { + name string + sessionYAML string + command []string + configureRun func() + wantSuccessful bool + }{ + { + name: "success", + sessionYAML: `session_id: 2026-05-03 +inputs: + audio_s3: + prefix: audio/ +`, + command: []string{"session", "plan", "2026-05-03"}, + wantSuccessful: true, + }, + { + name: "validation failure", + sessionYAML: `session_id: 2026-05-03 +`, + command: []string{"session", "plan", "2026-05-03"}, + }, + { + name: "load failure", + sessionYAML: `session_id: 2026-05-03 +unknown: true +`, + command: []string{"session", "plan", "2026-05-03"}, + }, + { + name: "adapter failure", + sessionYAML: `session_id: 2026-05-03 +inputs: + audio_s3: + prefix: audio/ +`, + command: []string{"run", "2026-05-03"}, + configureRun: func() { + executeStagesFn = func(context.Context, *config.Config, []stage.Stage, RunOptions) (*RunSummary, error) { + return nil, errors.New("adapter failed") + } + }, + }, + { + name: "cancellation", + sessionYAML: `session_id: 2026-05-03 +inputs: + audio_s3: + prefix: audio/ +`, + command: []string{"run", "2026-05-03"}, + configureRun: func() { + executeStagesFn = func(context.Context, *config.Config, []stage.Stage, RunOptions) (*RunSummary, error) { + return nil, context.Canceled + } + }, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + workspaceRoot := t.TempDir() + pipelinePath, campaignPath, _ := writeValidConfigFiles(t, workspaceRoot) + fake := &storage.FakeBackend{} + seedRemoteSessionConfig(t, fake, "2026-05-03", tt.sessionYAML) + var storeInitCalls int + restoreAppConfigTestGlobals(t, fake, &storeInitCalls, []string{filepath.Join(t.TempDir(), "session.yml")}) + + var downloadedPath string + captureRemoteSessionTempPath(t, &downloadedPath) + if tt.configureRun != nil { + origExecuteStagesFn := executeStagesFn + t.Cleanup(func() { executeStagesFn = origExecuteStagesFn }) + tt.configureRun() + } + + args := append(append([]string(nil), tt.command...), "--config", pipelinePath, "--campaign-file", campaignPath) + var stdout bytes.Buffer + var stderr bytes.Buffer + code := Execute(args, &stdout, &stderr) + if tt.wantSuccessful && code != 0 { + t.Fatalf("exit code = %d, want 0; stderr=%q", code, stderr.String()) + } + if !tt.wantSuccessful && code == 0 { + t.Fatal("exit code = 0, want non-zero") + } + if downloadedPath == "" { + t.Fatal("remote session download path was not captured") + } + if _, err := os.Stat(downloadedPath); !errors.Is(err, os.ErrNotExist) { + t.Fatalf("downloaded remote session path still exists or could not be inspected: %q, err=%v", downloadedPath, err) + } + }) + } +} + +func TestRemoteSessionConfigCloseIsIdempotent(t *testing.T) { + workspaceRoot := t.TempDir() + pipelinePath, campaignPath, _ := writeValidConfigFiles(t, workspaceRoot) + fake := &storage.FakeBackend{} + seedRemoteSessionConfig(t, fake, "2026-05-03", `session_id: 2026-05-03 +inputs: + audio_s3: + prefix: audio/ +`) + var storeInitCalls int + restoreAppConfigTestGlobals(t, fake, &storeInitCalls, []string{filepath.Join(t.TempDir(), "session.yml")}) + + var downloadedPath string + captureRemoteSessionTempPath(t, &downloadedPath) + loaded, err := loadCommandConfig(context.Background(), pipelinePath, "", campaignPath, "", config.SessionLoadOptions{SessionID: "2026-05-03"}) + if err != nil { + t.Fatalf("loadCommandConfig() error = %v", err) + } + if err := loaded.Close(); err != nil { + t.Fatalf("first Close() error = %v", err) + } + if err := loaded.Close(); err != nil { + t.Fatalf("second Close() error = %v", err) + } + if _, err := os.Stat(downloadedPath); !errors.Is(err, os.ErrNotExist) { + t.Fatalf("downloaded remote session path still exists or could not be inspected: %q, err=%v", downloadedPath, err) + } +} + +func captureRemoteSessionTempPath(t *testing.T, destination *string) { + t.Helper() + original := downloadObjectToTempFn + downloadObjectToTempFn = func(ctx context.Context, store storage.ObjectStore, key, pattern string) (string, error) { + path, err := original(ctx, store, key, pattern) + if err == nil { + *destination = path + } + return path, err + } + t.Cleanup(func() { downloadObjectToTempFn = original }) +} + func TestExecuteRemoteSessionFallbackLoadsSecretsBeforeObjectStoreInit(t *testing.T) { workspaceRoot := t.TempDir() pipelinePath, campaignPath, _ := writeValidConfigFiles(t, workspaceRoot) diff --git a/internal/app/restore.go b/internal/app/restore.go index 96c55f1..502aaa9 100644 --- a/internal/app/restore.go +++ b/internal/app/restore.go @@ -54,10 +54,12 @@ func Restore(ctx context.Context, args []string, out io.Writer) (resultErr error if strings.TrimSpace(flags.sessionID) == "" { return fmt.Errorf("restore: session_id is required") } - cfg, err := loadCommandConfig(ctx, flags.pipelinePath, flags.campaignPath, flags.campaignFilePath, flags.sessionPath, flags.sessionOptions()) + loaded, err := loadCommandConfig(ctx, flags.pipelinePath, flags.campaignPath, flags.campaignFilePath, flags.sessionPath, flags.sessionOptions()) if err != nil { return fmt.Errorf("restore: %w", err) } + defer func() { _ = loaded.Close() }() + cfg := loaded.Config if err := config.Validate(cfg); err != nil { return fmt.Errorf("restore: %w", err) } diff --git a/internal/app/run.go b/internal/app/run.go index f4f0274..a3e9990 100644 --- a/internal/app/run.go +++ b/internal/app/run.go @@ -27,10 +27,12 @@ func Run(ctx context.Context, args []string, out io.Writer) error { if flags.sessionID == "" { return fmt.Errorf("run: session_id is required") } - cfg, err := loadCommandConfig(ctx, flags.pipelinePath, flags.campaignPath, flags.campaignFilePath, flags.sessionPath, flags.sessionOptions()) + loaded, err := loadCommandConfig(ctx, flags.pipelinePath, flags.campaignPath, flags.campaignFilePath, flags.sessionPath, flags.sessionOptions()) if err != nil { return fmt.Errorf("run: %w", err) } + defer func() { _ = loaded.Close() }() + cfg := loaded.Config if err := config.Validate(cfg); err != nil { return fmt.Errorf("run: %w", err) } diff --git a/internal/app/run_stage.go b/internal/app/run_stage.go index a92b55f..ccec48a 100644 --- a/internal/app/run_stage.go +++ b/internal/app/run_stage.go @@ -204,13 +204,15 @@ func runSingleStageCommand(ctx context.Context, req singleStageCommand) (*RunSum return nil, fmt.Errorf("%s: %w", req.CommandName, err) } - cfg, err := loadCommandConfig(ctx, req.PipelinePath, req.CampaignPath, req.CampaignFilePath, req.SessionPath, config.SessionLoadOptions{ + loaded, err := loadCommandConfig(ctx, req.PipelinePath, req.CampaignPath, req.CampaignFilePath, req.SessionPath, config.SessionLoadOptions{ SessionID: req.SessionID, PreviousSessionID: req.PreviousSessionID, }) if err != nil { return nil, fmt.Errorf("%s: %w", req.CommandName, err) } + defer func() { _ = loaded.Close() }() + cfg := loaded.Config if err := config.Validate(cfg); err != nil { return nil, fmt.Errorf("%s: %w", req.CommandName, err) } diff --git a/internal/app/runner_test.go b/internal/app/runner_test.go index 770c307..e5fe636 100644 --- a/internal/app/runner_test.go +++ b/internal/app/runner_test.go @@ -1603,7 +1603,7 @@ func TestBuildDefaultRunnersWithOmittedToolSections(t *testing.T) { whisperx: transcribe_url: https://example.com/transcribe notification: - timeout: 10s + mode: noop ` campaignYAML := `campaign_id: sample-campaign inputs: diff --git a/internal/config/cache_config_test.go b/internal/config/cache_config_test.go index 83f2a3b..fda30f9 100644 --- a/internal/config/cache_config_test.go +++ b/internal/config/cache_config_test.go @@ -11,7 +11,7 @@ func TestCacheDefaults(t *testing.T) { whisperx: transcribe_url: https://example.com/transcribe notification: - timeout: 10s + mode: noop `, `session_id: 2026-05-03 inputs: audio_dir: ./audio diff --git a/internal/config/campaign_config_test.go b/internal/config/campaign_config_test.go index f71ea57..6843128 100644 --- a/internal/config/campaign_config_test.go +++ b/internal/config/campaign_config_test.go @@ -18,7 +18,7 @@ campaigns: whisperx: transcribe_url: https://example.com/transcribe notification: - timeout: 10s + mode: noop ` if err := os.WriteFile(pipelinePath, []byte(pipelineYAML), 0o644); err != nil { t.Fatalf("write pipeline.yml: %v", err) @@ -196,7 +196,7 @@ func writeCampaignConfigTestFiles(t *testing.T, campaignYAML, sessionYAML string campaignPath := filepath.Join(dir, "campaign.yml") sessionPath := filepath.Join(dir, "session.yml") - pipelineYAML := "workspace:\n root: " + filepath.ToSlash(filepath.Join(dir, "work")) + "\nwhisperx:\n transcribe_url: https://example.com/transcribe\nnotification:\n timeout: 10s\n" + pipelineYAML := "workspace:\n root: " + filepath.ToSlash(filepath.Join(dir, "work")) + "\nwhisperx:\n transcribe_url: https://example.com/transcribe\nnotification:\n mode: noop\n" if err := os.WriteFile(pipelinePath, []byte(pipelineYAML), 0o644); err != nil { t.Fatalf("write pipeline.yml: %v", err) } diff --git a/internal/config/config.go b/internal/config/config.go index 8005166..94febfe 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -255,8 +255,6 @@ type ScriptoriumArtifactConfig struct { // ScriptoriumInputConfig configures one named prompt input source. type ScriptoriumInputConfig struct { Source string `yaml:"source"` - Artifact string `yaml:"artifact"` - Path string `yaml:"path"` Required bool `yaml:"required"` } @@ -280,11 +278,9 @@ type NotariusOutputConfig struct { ModuleKey string `yaml:"module_key"` } -// NotificationConfig configures notification backend settings. +// NotificationConfig configures the supported notification behavior. type NotificationConfig struct { - Backend string `yaml:"backend"` - Recipient string `yaml:"recipient"` - Timeout string `yaml:"timeout"` + Mode string `yaml:"mode"` } // SessionInputsConfig contains per-session input references. @@ -329,5 +325,4 @@ type SessionSource struct { S3Key string S3Size int64 S3ETag string - SpoolPath string } diff --git a/internal/config/defaults.go b/internal/config/defaults.go index 143be10..1dfea5b 100644 --- a/internal/config/defaults.go +++ b/internal/config/defaults.go @@ -62,6 +62,7 @@ const ( DefaultArchiveEnabled = true DefaultArchiveUploadRun = true + DefaultNotificationMode = "noop" PathWorkDirSegment = "work" PathInputsDirSegment = "inputs" diff --git a/internal/config/load.go b/internal/config/load.go index b03c349..2f179b8 100644 --- a/internal/config/load.go +++ b/internal/config/load.go @@ -378,6 +378,16 @@ func applyPipelineDefaults(cfg *PipelineConfig) { applyRenderDefaults(&cfg.Render) applyScriptoriumDefaults(cfg.Scriptorium) applyNotariusDefaults(cfg.Notarius) + applyNotificationDefaults(&cfg.Notification) +} + +func applyNotificationDefaults(cfg *NotificationConfig) { + if cfg == nil { + return + } + if strings.TrimSpace(cfg.Mode) == "" { + cfg.Mode = DefaultNotificationMode + } } func applyCampaignsDefaults(cfg *CampaignsConfig) { diff --git a/internal/config/notification_test.go b/internal/config/notification_test.go new file mode 100644 index 0000000..c139b1d --- /dev/null +++ b/internal/config/notification_test.go @@ -0,0 +1,68 @@ +package config + +import ( + "strings" + "testing" +) + +func TestNotificationConfigSupportsOnlyNoopMode(t *testing.T) { + tests := []struct { + name string + section string + wantLoad string + wantValidate string + }{ + { + name: "default noop mode", + section: "", + }, + { + name: "explicit noop mode", + section: "notification:\n mode: noop\n", + }, + { + name: "backend is rejected", + section: "notification:\n backend: email\n", + wantLoad: "field backend not found", + }, + { + name: "recipient is rejected", + section: "notification:\n recipient: party@example.com\n", + wantLoad: "field recipient not found", + }, + { + name: "provider mode is rejected", + section: "notification:\n mode: email\n", + wantValidate: "pipeline.notification.mode must be \"noop\"", + }, + } + + 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 tt.wantLoad != "" { + if err == nil || !strings.Contains(err.Error(), tt.wantLoad) { + t.Fatalf("Load() error = %v, want %q", err, tt.wantLoad) + } + return + } + if err != nil { + t.Fatalf("Load() error = %v", err) + } + if got := cfg.Pipeline.Notification.Mode; tt.wantValidate == "" && got != DefaultNotificationMode { + t.Fatalf("notification.mode = %q, want %q", got, DefaultNotificationMode) + } + err = Validate(cfg) + if tt.wantValidate != "" { + if err == nil || !strings.Contains(err.Error(), tt.wantValidate) { + t.Fatalf("Validate() error = %v, want %q", err, tt.wantValidate) + } + return + } + if err != nil { + t.Fatalf("Validate() error = %v", err) + } + }) + } +} diff --git a/internal/config/scriptorium_test.go b/internal/config/scriptorium_test.go index d192fc5..9a7be2c 100644 --- a/internal/config/scriptorium_test.go +++ b/internal/config/scriptorium_test.go @@ -95,7 +95,7 @@ func TestScriptoriumLoadAndValidate(t *testing.T) { wantValidateErr: "pipeline.scriptorium.timeout must be a valid duration", }, { - name: "legacy previous session artifact source fails validation", + name: "legacy previous session source fails validation", scriptoriumYAML: `scriptorium: binary: scriptorium artifacts: @@ -109,8 +109,6 @@ func TestScriptoriumLoadAndValidate(t *testing.T) { required: true previous_recap: source: ` + legacyPreviousSource + ` - artifact: session_recap - path: "" required: false vars: session_id: true @@ -118,6 +116,21 @@ func TestScriptoriumLoadAndValidate(t *testing.T) { `, wantValidateErr: `pipeline.scriptorium.artifacts.session_recap.inputs.previous_recap.source "` + legacyPreviousSource + `" is unsupported`, }, + { + name: "obsolete input passthrough fields fail strict decoding", + scriptoriumYAML: `scriptorium: + artifacts: + session_recap: + enabled: true + prompt_id: dnd.session_recap + output_path: artifacts/session_recap.md + inputs: + transcript: + source: narratio.transcript.polished + artifact: session_recap +`, + wantLoadErr: "field artifact not found", + }, { name: "canonical previous-session source is accepted", scriptoriumYAML: `scriptorium: diff --git a/internal/config/validate.go b/internal/config/validate.go index 1f77341..74bb514 100644 --- a/internal/config/validate.go +++ b/internal/config/validate.go @@ -122,13 +122,20 @@ func validatePipeline(cfg *PipelineConfig) error { if err := validateScriptorium(cfg.Scriptorium, cfg.Notarius); err != nil { return err } - if err := validateDuration("pipeline.notification.timeout", cfg.Notification.Timeout); err != nil { + if err := validateNotification(cfg.Notification); err != nil { return err } return nil } +func validateNotification(cfg NotificationConfig) error { + if strings.EqualFold(strings.TrimSpace(cfg.Mode), DefaultNotificationMode) { + return nil + } + return fmt.Errorf("pipeline.notification.mode must be %q until a notification provider is configured", DefaultNotificationMode) +} + func validateStorage(cfg StorageConfig) error { backend := strings.ToLower(strings.TrimSpace(cfg.Backend)) switch backend { diff --git a/internal/config/validation_duration_test.go b/internal/config/validation_duration_test.go index a45a1b2..43dd60d 100644 --- a/internal/config/validation_duration_test.go +++ b/internal/config/validation_duration_test.go @@ -23,7 +23,6 @@ func TestValidateDurationsRequirePositiveValues(t *testing.T) { p.Scriptorium.Artifacts["session_recap"] = ScriptoriumArtifactConfig{Timeout: value} }, want: "pipeline.scriptorium.artifacts.session_recap.timeout"}, {name: "trim bounds timeout", set: func(p *PipelineConfig, value string) { p.Trim.Bounds.Timeout = value }, want: "pipeline.trim.bounds.timeout"}, - {name: "notification timeout", set: func(p *PipelineConfig, value string) { p.Notification.Timeout = value }, want: "pipeline.notification.timeout"}, } for _, value := range []string{"0s", "-1ms"} { @@ -82,7 +81,6 @@ func TestValidateDurationRejectsOverflowAndAcceptsPositiveSubsecondValues(t *tes cfg.Pipeline.Audita.Timeout = "1ms" cfg.Pipeline.Scriptorium.Timeout = "1ms" cfg.Pipeline.Trim.Bounds.Timeout = "1ms" - cfg.Pipeline.Notification.Timeout = "1ms" if cfg.Pipeline.Scriptorium.Artifacts == nil { cfg.Pipeline.Scriptorium.Artifacts = map[string]ScriptoriumArtifactConfig{} } diff --git a/internal/stage/prepare.go b/internal/stage/prepare.go index 5984709..31f5c41 100644 --- a/internal/stage/prepare.go +++ b/internal/stage/prepare.go @@ -135,15 +135,14 @@ func (prepareStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*S source.Source = "session_config" } inputs = append(inputs, manifest.InputRecord{ - Kind: "session_config", - Path: path, - Checksum: checksum, - Source: source.Source, - S3Bucket: source.S3Bucket, - S3Key: source.S3Key, - S3Size: source.S3Size, - S3ETag: source.S3ETag, - SpoolPath: source.SpoolPath, + Kind: "session_config", + Path: path, + Checksum: checksum, + Source: source.Source, + S3Bucket: source.S3Bucket, + S3Key: source.S3Key, + S3Size: source.S3Size, + S3ETag: source.S3ETag, }) } diff --git a/internal/stage/prepare_test.go b/internal/stage/prepare_test.go index c01fcb3..b0c3565 100644 --- a/internal/stage/prepare_test.go +++ b/internal/stage/prepare_test.go @@ -186,7 +186,6 @@ func TestPrepareStageRecordsRemoteSessionProvenance(t *testing.T) { S3Key: "dnd/campaigns/sample-campaign/sessions/2026-05-03/session.yml", S3Size: 58, S3ETag: "session-etag", - SpoolPath: remoteSessionPath, } env.Config.Pipeline.Spool = config.SpoolConfig{Root: filepath.Join(t.TempDir(), "spool")} env.Config.Pipeline.Storage.S3 = &config.StorageS3Config{Bucket: "my-dnd-archive", RootPrefix: "dnd"} @@ -207,7 +206,7 @@ func TestPrepareStageRecordsRemoteSessionProvenance(t *testing.T) { if sessionInput.S3Bucket != "my-dnd-archive" || sessionInput.S3Key != "dnd/campaigns/sample-campaign/sessions/2026-05-03/session.yml" { t.Fatalf("remote session input missing bucket/key: %#v", sessionInput) } - if sessionInput.S3Size != 58 || sessionInput.S3ETag != "session-etag" || sessionInput.SpoolPath != remoteSessionPath { + if sessionInput.S3Size != 58 || sessionInput.S3ETag != "session-etag" || sessionInput.SpoolPath != "" { t.Fatalf("remote session input missing metadata: %#v", sessionInput) } }