Make configuration truthful and clean remote session files
This commit is contained in:
@@ -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
|
||||
|
||||
|
||||
@@ -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.<name>.inputs.<input_nam
|
||||
| Field | Type | Required | Rule |
|
||||
| --- | --- | --- | --- |
|
||||
| `source` | string | Yes | built-in runtime source, prepared input source, `narratio.extraction.<name>`, `narratio.artifact.<name>`, or `narratio.previous_session.artifact.<name>` |
|
||||
| `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 |
|
||||
|
||||
@@ -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.
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -125,4 +125,4 @@ scriptorium:
|
||||
output_kind: player_handout
|
||||
|
||||
notification:
|
||||
timeout: 30s
|
||||
mode: noop
|
||||
|
||||
@@ -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")
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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) {
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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) {
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
@@ -109,7 +109,7 @@ seriatim:
|
||||
audita:
|
||||
binary: audita
|
||||
notification:
|
||||
timeout: 10s
|
||||
mode: noop
|
||||
`
|
||||
sessionYAML := `session_id: 2026-05-03
|
||||
campaign: sample-campaign
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -62,6 +62,7 @@ const (
|
||||
|
||||
DefaultArchiveEnabled = true
|
||||
DefaultArchiveUploadRun = true
|
||||
DefaultNotificationMode = "noop"
|
||||
|
||||
PathWorkDirSegment = "work"
|
||||
PathInputsDirSegment = "inputs"
|
||||
|
||||
@@ -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) {
|
||||
|
||||
68
internal/config/notification_test.go
Normal file
68
internal/config/notification_test.go
Normal file
@@ -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)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -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:
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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{}
|
||||
}
|
||||
|
||||
@@ -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,
|
||||
})
|
||||
}
|
||||
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user