Carry dependencies through pipeline commands

This commit is contained in:
2026-09-05 17:53:56 +00:00
parent c3307ab5fb
commit dd905c4daa
13 changed files with 153 additions and 89 deletions

View File

@@ -152,7 +152,7 @@ local dependency values.
## Stage 2 — Carry Dependencies Through Pipeline Commands ## Stage 2 — Carry Dependencies Through Pipeline Commands
**Status:** Pending **Status:** Completed
### Goal ### Goal

View File

@@ -58,11 +58,12 @@ func TestAssembledCanonicalAndAliasArtifactRegenerationRequestsMatch(t *testing.
"2026-05-03", "--force", "--from", "extract", "--through", "analyze", "2026-05-03", "--force", "--from", "extract", "--through", "analyze",
"--config", pipelinePath, "--campaign-file", campaignPath, "--session", sessionPath, "--config", pipelinePath, "--campaign-file", campaignPath, "--session", sessionPath,
} }
if err := runWithPipelineExecutor(context.Background(), base, &bytes.Buffer{}, executor); err != nil { deps := fakeCommandDependencies(&storage.FakeBackend{}, nil)
if err := runWithDependencies(context.Background(), base, &bytes.Buffer{}, executor, deps); err != nil {
t.Fatalf("canonical unselected Run() error = %v", err) t.Fatalf("canonical unselected Run() error = %v", err)
} }
selected := append(append([]string(nil), base...), "--artifacts", "session_recap") selected := append(append([]string(nil), base...), "--artifacts", "session_recap")
if err := runWithPipelineExecutor(context.Background(), selected, &bytes.Buffer{}, executor); err != nil { if err := runWithDependencies(context.Background(), selected, &bytes.Buffer{}, executor, deps); err != nil {
t.Fatalf("canonical selected Run() error = %v", err) t.Fatalf("canonical selected Run() error = %v", err)
} }
var stdout, stderr bytes.Buffer var stdout, stderr bytes.Buffer
@@ -70,7 +71,7 @@ func TestAssembledCanonicalAndAliasArtifactRegenerationRequestsMatch(t *testing.
"regenerate-artifacts", "2026-05-03", "--artifacts", "session_recap", "regenerate-artifacts", "2026-05-03", "--artifacts", "session_recap",
"--config", pipelinePath, "--campaign-file", campaignPath, "--session", sessionPath, "--config", pipelinePath, "--campaign-file", campaignPath, "--session", sessionPath,
} }
if code := executeWithPipelineExecutor(alias, &stdout, &stderr, executor); code != 0 { if code := executeWithDependencies(alias, &stdout, &stderr, deps, executor); code != 0 {
t.Fatalf("alias exit=%d stderr=%q", code, stderr.String()) t.Fatalf("alias exit=%d stderr=%q", code, stderr.String())
} }
if len(captured) != 3 { if len(captured) != 3 {

View File

@@ -11,10 +11,15 @@ var supportedCommands = []string{"version", "run", "regenerate-artifacts", "run-
// Execute dispatches CLI commands and returns a process exit code. // Execute dispatches CLI commands and returns a process exit code.
func Execute(args []string, stdout, stderr io.Writer) int { func Execute(args []string, stdout, stderr io.Writer) int {
return executeWithPipelineExecutor(args, stdout, stderr, executePlan) return executeWithDependencies(args, stdout, stderr, commandDependencies{}, executePlan)
} }
func executeWithPipelineExecutor(args []string, stdout, stderr io.Writer, executor pipelineExecutor) int { func executeWithPipelineExecutor(args []string, stdout, stderr io.Writer, executor pipelineExecutor) int {
return executeWithDependencies(args, stdout, stderr, commandDependencies{}, executor)
}
func executeWithDependencies(args []string, stdout, stderr io.Writer, deps commandDependencies, executor pipelineExecutor) int {
deps = normalizeCommandDependencies(deps)
if len(args) == 0 { if len(args) == 0 {
printUsage(stderr) printUsage(stderr)
return 1 return 1
@@ -29,17 +34,17 @@ func executeWithPipelineExecutor(args []string, stdout, stderr io.Writer, execut
case "version": case "version":
err = Version(cmdArgs, stdout) err = Version(cmdArgs, stdout)
case "run": case "run":
err = runWithPipelineExecutor(ctx, cmdArgs, stdout, executor) err = runWithDependencies(ctx, cmdArgs, stdout, executor, deps)
case "regenerate-artifacts": case "regenerate-artifacts":
err = regenerateArtifactsWithPipelineExecutor(ctx, cmdArgs, stdout, executor) err = regenerateArtifactsWithDependencies(ctx, cmdArgs, stdout, executor, deps)
case "run-stage": case "run-stage":
err = runStageWithPipelineExecutor(ctx, cmdArgs, stdout, executor) err = runStageWithDependencies(ctx, cmdArgs, stdout, executor, deps)
case "analyze": case "analyze":
err = analyzeWithPipelineExecutor(ctx, cmdArgs, stdout, executor) err = analyzeWithDependencies(ctx, cmdArgs, stdout, executor, deps)
case "publish": case "publish":
err = publishWithPipelineExecutor(ctx, cmdArgs, stdout, executor) err = publishWithDependencies(ctx, cmdArgs, stdout, executor, deps)
case "session": case "session":
err = Session(ctx, cmdArgs, stdout) err = sessionWithDependencies(ctx, cmdArgs, stdout, deps)
case "clean": case "clean":
err = Clean(ctx, cmdArgs, stdout) err = Clean(ctx, cmdArgs, stdout)
case "config": case "config":

View File

@@ -47,6 +47,12 @@ func (f commonConfigFlags) sessionOptions() config.SessionLoadOptions {
// Session dispatches session helper subcommands. // Session dispatches session helper subcommands.
func Session(ctx context.Context, args []string, out io.Writer) error { func Session(ctx context.Context, args []string, out io.Writer) error {
return sessionWithDependencies(ctx, args, out, commandDependencies{
newObjectStore: newObjectStoreFromConfigFn,
})
}
func sessionWithDependencies(ctx context.Context, args []string, out io.Writer, deps commandDependencies) error {
if len(args) == 0 { if len(args) == 0 {
return fmt.Errorf("session: expected subcommand: init|validate|status|plan|restore|artifacts|locks") return fmt.Errorf("session: expected subcommand: init|validate|status|plan|restore|artifacts|locks")
} }
@@ -56,9 +62,9 @@ func Session(ctx context.Context, args []string, out io.Writer) error {
case "validate": case "validate":
return SessionValidate(ctx, args[1:], out) return SessionValidate(ctx, args[1:], out)
case "status": case "status":
return Status(ctx, args[1:], out) return statusWithDependencies(ctx, args[1:], out, deps)
case "plan": case "plan":
return Plan(ctx, args[1:], out) return planWithDependencies(ctx, args[1:], out, deps)
case "restore": case "restore":
return Restore(ctx, args[1:], out) return Restore(ctx, args[1:], out)
case "artifacts": case "artifacts":
@@ -142,10 +148,16 @@ func loadHelperContext(ctx context.Context, flags commonConfigFlags, needStore b
} }
func objectStoreIfConfigured(ctx context.Context, cfg *config.Config) (storage.ObjectStore, error) { func objectStoreIfConfigured(ctx context.Context, cfg *config.Config) (storage.ObjectStore, error) {
return objectStoreIfConfiguredWithDependencies(ctx, cfg, commandDependencies{
newObjectStore: newObjectStoreFromConfigFn,
})
}
func objectStoreIfConfiguredWithDependencies(ctx context.Context, cfg *config.Config, deps commandDependencies) (storage.ObjectStore, error) {
if cfg == nil || cfg.Pipeline == nil || cfg.Pipeline.Storage.S3 == nil || strings.TrimSpace(cfg.Pipeline.Storage.S3.Bucket) == "" { if cfg == nil || cfg.Pipeline == nil || cfg.Pipeline.Storage.S3 == nil || strings.TrimSpace(cfg.Pipeline.Storage.S3.Bucket) == "" {
return nil, nil return nil, nil
} }
store, err := newCommandObjectStore(ctx, cfg, nil) store, err := newCommandObjectStoreWithDependencies(ctx, cfg, nil, deps)
if err != nil { if err != nil {
return nil, err return nil, err
} }

View File

@@ -493,7 +493,7 @@ func TestOperatorCommandsReportConfiguredSpellCatalog(t *testing.T) {
stdout.Reset() stdout.Reset()
stderr.Reset() stderr.Reset()
statusArgs := append([]string{"session", "status"}, commonArgs...) statusArgs := append([]string{"session", "status"}, commonArgs...)
if code := Execute(statusArgs, &stdout, &stderr); code != 0 { if code := executeWithDependencies(statusArgs, &stdout, &stderr, fakeCommandDependencies(fake, &storeInitCalls), executePlan); code != 0 {
t.Fatalf("status exit code = %d, want 0; stdout=%q stderr=%q", code, stdout.String(), stderr.String()) t.Fatalf("status exit code = %d, want 0; stdout=%q stderr=%q", code, stdout.String(), stderr.String())
} }
if !strings.Contains(stdout.String(), "Stable input spell_catalog: "+spellPath) { if !strings.Contains(stdout.String(), "Stable input spell_catalog: "+spellPath) {
@@ -1057,15 +1057,16 @@ func TestExecuteStatusReportsRemoteArtifactCatalog(t *testing.T) {
seedOperatorRemoteLock(t, fake, sessionPrefix, "narratio.transcript.final_trimmed", "remote review") seedOperatorRemoteLock(t, fake, sessionPrefix, "narratio.transcript.final_trimmed", "remote review")
var storeInitCalls int var storeInitCalls int
restoreAppConfigTestGlobals(t, fake, &storeInitCalls, []string{sessionPath}) restoreAppConfigTestGlobals(t, fake, &storeInitCalls, []string{sessionPath})
deps := fakeCommandDependencies(fake, &storeInitCalls)
var stdout bytes.Buffer var stdout bytes.Buffer
var stderr bytes.Buffer var stderr bytes.Buffer
code := Execute([]string{ code := executeWithDependencies([]string{
"session", "status", "2026-05-03", "session", "status", "2026-05-03",
"--config", pipelinePath, "--config", pipelinePath,
"--campaign-file", campaignPath, "--campaign-file", campaignPath,
"--session", sessionPath, "--session", sessionPath,
}, &stdout, &stderr) }, &stdout, &stderr, deps, executePlan)
if code != 0 { if code != 0 {
t.Fatalf("exit code = %d, want 0; stderr=%q", code, stderr.String()) t.Fatalf("exit code = %d, want 0; stderr=%q", code, stderr.String())
} }
@@ -1103,15 +1104,16 @@ func TestExecuteStatusReportsRemoteArtifactCatalogErrorsWithoutFailing(t *testin
fake := &storage.FakeBackend{ExistsErr: fmt.Errorf("exists failed")} fake := &storage.FakeBackend{ExistsErr: fmt.Errorf("exists failed")}
var storeInitCalls int var storeInitCalls int
restoreAppConfigTestGlobals(t, fake, &storeInitCalls, []string{sessionPath}) restoreAppConfigTestGlobals(t, fake, &storeInitCalls, []string{sessionPath})
deps := fakeCommandDependencies(fake, &storeInitCalls)
var stdout bytes.Buffer var stdout bytes.Buffer
var stderr bytes.Buffer var stderr bytes.Buffer
code := Execute([]string{ code := executeWithDependencies([]string{
"session", "status", "2026-05-03", "session", "status", "2026-05-03",
"--config", pipelinePath, "--config", pipelinePath,
"--campaign-file", campaignPath, "--campaign-file", campaignPath,
"--session", sessionPath, "--session", sessionPath,
}, &stdout, &stderr) }, &stdout, &stderr, deps, executePlan)
if code != 0 { if code != 0 {
t.Fatalf("exit code = %d, want 0; stderr=%q", code, stderr.String()) t.Fatalf("exit code = %d, want 0; stderr=%q", code, stderr.String())
} }
@@ -1130,15 +1132,16 @@ func TestExecuteStatusReportsMissingRemoteCurrentStateWithoutFailing(t *testing.
fake := &storage.FakeBackend{} fake := &storage.FakeBackend{}
var storeInitCalls int var storeInitCalls int
restoreAppConfigTestGlobals(t, fake, &storeInitCalls, []string{sessionPath}) restoreAppConfigTestGlobals(t, fake, &storeInitCalls, []string{sessionPath})
deps := fakeCommandDependencies(fake, &storeInitCalls)
var stdout bytes.Buffer var stdout bytes.Buffer
var stderr bytes.Buffer var stderr bytes.Buffer
code := Execute([]string{ code := executeWithDependencies([]string{
"session", "status", "2026-05-03", "session", "status", "2026-05-03",
"--config", pipelinePath, "--config", pipelinePath,
"--campaign-file", campaignPath, "--campaign-file", campaignPath,
"--session", sessionPath, "--session", sessionPath,
}, &stdout, &stderr) }, &stdout, &stderr, deps, executePlan)
if code != 0 { if code != 0 {
t.Fatalf("exit code = %d, want 0; stderr=%q", code, stderr.String()) t.Fatalf("exit code = %d, want 0; stderr=%q", code, stderr.String())
} }
@@ -1155,15 +1158,16 @@ func TestExecuteStatusReportsPreviousStateReadinessWithoutFailing(t *testing.T)
fake := &storage.FakeBackend{} fake := &storage.FakeBackend{}
var storeInitCalls int var storeInitCalls int
restoreAppConfigTestGlobals(t, fake, &storeInitCalls, []string{sessionPath}) restoreAppConfigTestGlobals(t, fake, &storeInitCalls, []string{sessionPath})
deps := fakeCommandDependencies(fake, &storeInitCalls)
var stdout bytes.Buffer var stdout bytes.Buffer
var stderr bytes.Buffer var stderr bytes.Buffer
code := Execute([]string{ code := executeWithDependencies([]string{
"session", "status", "2026-05-03", "session", "status", "2026-05-03",
"--config", pipelinePath, "--config", pipelinePath,
"--campaign-file", campaignPath, "--campaign-file", campaignPath,
"--session", sessionPath, "--session", sessionPath,
}, &stdout, &stderr) }, &stdout, &stderr, deps, executePlan)
if code != 0 { if code != 0 {
t.Fatalf("exit code = %d, want 0; stderr=%q", code, stderr.String()) t.Fatalf("exit code = %d, want 0; stderr=%q", code, stderr.String())
} }
@@ -1185,15 +1189,16 @@ func TestExecuteStatusDetectsMissingRequiredPreviousArtifact(t *testing.T) {
seedRestorePreviousCurrentManifestOnly(t, fake, cfg) seedRestorePreviousCurrentManifestOnly(t, fake, cfg)
var storeInitCalls int var storeInitCalls int
restoreAppConfigTestGlobals(t, fake, &storeInitCalls, []string{sessionPath}) restoreAppConfigTestGlobals(t, fake, &storeInitCalls, []string{sessionPath})
deps := fakeCommandDependencies(fake, &storeInitCalls)
var stdout bytes.Buffer var stdout bytes.Buffer
var stderr bytes.Buffer var stderr bytes.Buffer
code := Execute([]string{ code := executeWithDependencies([]string{
"session", "status", "2026-05-03", "session", "status", "2026-05-03",
"--config", pipelinePath, "--config", pipelinePath,
"--campaign-file", campaignPath, "--campaign-file", campaignPath,
"--session", sessionPath, "--session", sessionPath,
}, &stdout, &stderr) }, &stdout, &stderr, deps, executePlan)
if code != 0 { if code != 0 {
t.Fatalf("exit code = %d, want 0; stderr=%q", code, stderr.String()) t.Fatalf("exit code = %d, want 0; stderr=%q", code, stderr.String())
} }

View File

@@ -184,10 +184,17 @@ func inspectPreviousArtifactReadiness(
} }
func inspectRemoteCurrentState(ctx context.Context, cfg *config.Config, store storage.ObjectStore) remoteCurrentStateCheck { func inspectRemoteCurrentState(ctx context.Context, cfg *config.Config, store storage.ObjectStore) remoteCurrentStateCheck {
return inspectRemoteCurrentStateWithDependencies(ctx, cfg, store, commandDependencies{
discoverRemoteCurrentState: discoverRemoteCurrentStateFn,
})
}
func inspectRemoteCurrentStateWithDependencies(ctx context.Context, cfg *config.Config, store storage.ObjectStore, deps commandDependencies) remoteCurrentStateCheck {
if store == nil { if store == nil {
return remoteCurrentStateCheck{} return remoteCurrentStateCheck{}
} }
current, err := discoverRemoteCurrentStateFn(ctx, cfg, store) deps = normalizeCommandDependencies(deps)
current, err := deps.discoverRemoteCurrentState(ctx, cfg, store)
if err != nil { if err != nil {
return remoteCurrentStateCheck{Err: err} return remoteCurrentStateCheck{Err: err}
} }

View File

@@ -16,6 +16,11 @@ import (
// Status reports effective local/remote session state. // Status reports effective local/remote session state.
func Status(ctx context.Context, args []string, out io.Writer) error { func Status(ctx context.Context, args []string, out io.Writer) error {
return statusWithDependencies(ctx, args, out, commandDependencies{})
}
func statusWithDependencies(ctx context.Context, args []string, out io.Writer, deps commandDependencies) error {
deps = normalizeCommandDependencies(deps)
fs := flag.NewFlagSet("status", flag.ContinueOnError) fs := flag.NewFlagSet("status", flag.ContinueOnError)
fs.SetOutput(io.Discard) fs.SetOutput(io.Discard)
var flags commonConfigFlags var flags commonConfigFlags
@@ -26,7 +31,7 @@ func Status(ctx context.Context, args []string, out io.Writer) error {
if strings.TrimSpace(flags.sessionID) == "" { if strings.TrimSpace(flags.sessionID) == "" {
return fmt.Errorf("status: session_id is required") return fmt.Errorf("status: session_id is required")
} }
loaded, err := loadCommandConfig(ctx, flags.pipelinePath, flags.campaignPath, flags.campaignFilePath, flags.sessionPath, flags.sessionOptions()) loaded, err := loadCommandConfigWithDependencies(ctx, flags.pipelinePath, flags.campaignPath, flags.campaignFilePath, flags.sessionPath, flags.sessionOptions(), deps)
if err != nil { if err != nil {
return fmt.Errorf("status: %w", err) return fmt.Errorf("status: %w", err)
} }
@@ -56,13 +61,13 @@ func Status(ctx context.Context, args []string, out io.Writer) error {
writeStageStatuses(out, m) writeStageStatuses(out, m)
} }
store, storeErr := objectStoreIfConfigured(ctx, cfg) store, storeErr := objectStoreIfConfiguredWithDependencies(ctx, cfg, deps)
var remoteCurrent *RemoteCurrentState var remoteCurrent *RemoteCurrentState
var remoteCurrentErr error var remoteCurrentErr error
if storeErr != nil { if storeErr != nil {
fmt.Fprintf(out, "Remote publish: unavailable: %v\n", storeErr) fmt.Fprintf(out, "Remote publish: unavailable: %v\n", storeErr)
} else if store != nil { } else if store != nil {
current := inspectRemoteCurrentState(ctx, cfg, store) current := inspectRemoteCurrentStateWithDependencies(ctx, cfg, store, deps)
if current.Err != nil { if current.Err != nil {
remoteCurrentErr = current.Err remoteCurrentErr = current.Err
fmt.Fprintf(out, "Remote publish: missing or unavailable: %v\n", current.Err) fmt.Fprintf(out, "Remote publish: missing or unavailable: %v\n", current.Err)

View File

@@ -18,6 +18,11 @@ import (
// Plan validates configuration and prints a read-only execution preview. // Plan validates configuration and prints a read-only execution preview.
func Plan(ctx context.Context, args []string, out io.Writer) error { func Plan(ctx context.Context, args []string, out io.Writer) error {
return planWithDependencies(ctx, args, out, commandDependencies{})
}
func planWithDependencies(ctx context.Context, args []string, out io.Writer, deps commandDependencies) error {
deps = normalizeCommandDependencies(deps)
request, err := parseBoundedRunRequest("plan", args, out) request, err := parseBoundedRunRequest("plan", args, out)
if err != nil { if err != nil {
if errors.Is(err, flag.ErrHelp) { if errors.Is(err, flag.ErrHelp) {
@@ -26,7 +31,7 @@ func Plan(ctx context.Context, args []string, out io.Writer) error {
return err return err
} }
flags := request.Config flags := request.Config
loaded, err := loadCommandConfig(ctx, flags.pipelinePath, flags.campaignPath, flags.campaignFilePath, flags.sessionPath, flags.sessionOptions()) loaded, err := loadCommandConfigWithDependencies(ctx, flags.pipelinePath, flags.campaignPath, flags.campaignFilePath, flags.sessionPath, flags.sessionOptions(), deps)
if err != nil { if err != nil {
return fmt.Errorf("plan: %w", err) return fmt.Errorf("plan: %w", err)
} }

View File

@@ -9,10 +9,14 @@ import (
// RegenerateArtifacts expands the convenience command into its canonical run // RegenerateArtifacts expands the convenience command into its canonical run
// invocation. The run command remains the sole owner of parsing and execution. // invocation. The run command remains the sole owner of parsing and execution.
func RegenerateArtifacts(ctx context.Context, args []string, out io.Writer) error { func RegenerateArtifacts(ctx context.Context, args []string, out io.Writer) error {
return regenerateArtifactsWithPipelineExecutor(ctx, args, out, executePlan) return regenerateArtifactsWithDependencies(ctx, args, out, executePlan, commandDependencies{})
} }
func regenerateArtifactsWithPipelineExecutor(ctx context.Context, args []string, out io.Writer, executor pipelineExecutor) error { func regenerateArtifactsWithPipelineExecutor(ctx context.Context, args []string, out io.Writer, executor pipelineExecutor) error {
return regenerateArtifactsWithDependencies(ctx, args, out, executor, commandDependencies{})
}
func regenerateArtifactsWithDependencies(ctx context.Context, args []string, out io.Writer, executor pipelineExecutor, deps commandDependencies) error {
if containsHelpOption(args) { if containsHelpOption(args) {
printRegenerateArtifactsHelp(out) printRegenerateArtifactsHelp(out)
return nil return nil
@@ -25,7 +29,7 @@ func regenerateArtifactsWithPipelineExecutor(ctx context.Context, args []string,
} }
expanded = append(expanded, "--force", "--from", "extract", "--through", "analyze") expanded = append(expanded, "--force", "--from", "extract", "--through", "analyze")
expanded = append(expanded, args...) expanded = append(expanded, args...)
return runWithPipelineExecutor(ctx, expanded, out, executor) return runWithDependencies(ctx, expanded, out, executor, deps)
} }
func containsHelpOption(args []string) bool { func containsHelpOption(args []string) bool {

View File

@@ -25,11 +25,12 @@ inputs:
prefix: audio/ prefix: audio/
`) `)
var storeInitCalls int var storeInitCalls int
restoreAppConfigTestGlobals(t, fake, &storeInitCalls, []string{filepath.Join(t.TempDir(), "session.yml")}) setSessionConfigSearchPaths(t, []string{filepath.Join(t.TempDir(), "session.yml")})
deps := fakeCommandDependencies(fake, &storeInitCalls)
var stdout bytes.Buffer var stdout bytes.Buffer
var stderr bytes.Buffer var stderr bytes.Buffer
code := Execute([]string{"session", "plan", "2026-05-03", "--config", pipelinePath, "--campaign-file", campaignPath}, &stdout, &stderr) code := executeWithDependencies([]string{"session", "plan", "2026-05-03", "--config", pipelinePath, "--campaign-file", campaignPath}, &stdout, &stderr, deps, executePlan)
if code != 0 { if code != 0 {
t.Fatalf("exit code = %d, want 0; stderr=%q", code, stderr.String()) t.Fatalf("exit code = %d, want 0; stderr=%q", code, stderr.String())
} }
@@ -154,7 +155,8 @@ inputs:
fake := &storage.FakeBackend{} fake := &storage.FakeBackend{}
seedRemoteSessionConfig(t, fake, "2026-05-03", tt.sessionYAML) seedRemoteSessionConfig(t, fake, "2026-05-03", tt.sessionYAML)
var storeInitCalls int var storeInitCalls int
restoreAppConfigTestGlobals(t, fake, &storeInitCalls, []string{filepath.Join(t.TempDir(), "session.yml")}) setSessionConfigSearchPaths(t, []string{filepath.Join(t.TempDir(), "session.yml")})
deps := fakeCommandDependencies(fake, &storeInitCalls)
tempRoot := t.TempDir() tempRoot := t.TempDir()
t.Setenv("TMPDIR", tempRoot) t.Setenv("TMPDIR", tempRoot)
@@ -166,7 +168,7 @@ inputs:
args := append(append([]string(nil), tt.command...), "--config", pipelinePath, "--campaign-file", campaignPath) args := append(append([]string(nil), tt.command...), "--config", pipelinePath, "--campaign-file", campaignPath)
var stdout bytes.Buffer var stdout bytes.Buffer
var stderr bytes.Buffer var stderr bytes.Buffer
code := executeWithPipelineExecutor(args, &stdout, &stderr, executor) code := executeWithDependencies(args, &stdout, &stderr, deps, executor)
if tt.wantSuccessful && code != 0 { if tt.wantSuccessful && code != 0 {
t.Fatalf("exit code = %d, want 0; stderr=%q", code, stderr.String()) t.Fatalf("exit code = %d, want 0; stderr=%q", code, stderr.String())
} }
@@ -243,23 +245,17 @@ inputs:
audio_s3: audio_s3:
prefix: audio/ prefix: audio/
`) `)
origStoreFn := newObjectStoreFromConfigFn setSessionConfigSearchPaths(t, []string{filepath.Join(t.TempDir(), "session.yml")})
origSessionDefaults := append([]string(nil), config.DefaultSessionConfigSearchPaths...) deps := commandDependencies{newObjectStore: func(context.Context, *config.Config) (storage.ObjectStore, error) {
config.DefaultSessionConfigSearchPaths = []string{filepath.Join(t.TempDir(), "session.yml")}
newObjectStoreFromConfigFn = func(context.Context, *config.Config) (storage.ObjectStore, error) {
if os.Getenv(accessKeyEnv) != "remote-session-key-id" || os.Getenv(secretKeyEnv) != "remote-session-secret" { if os.Getenv(accessKeyEnv) != "remote-session-key-id" || os.Getenv(secretKeyEnv) != "remote-session-secret" {
return nil, fmt.Errorf("secrets were not loaded before remote session object store init") return nil, fmt.Errorf("secrets were not loaded before remote session object store init")
} }
return fake, nil return fake, nil
} }}
t.Cleanup(func() {
newObjectStoreFromConfigFn = origStoreFn
config.DefaultSessionConfigSearchPaths = origSessionDefaults
})
var stdout bytes.Buffer var stdout bytes.Buffer
var stderr bytes.Buffer var stderr bytes.Buffer
code := Execute([]string{"session", "plan", "2026-05-03", "--config", pipelinePath, "--campaign-file", campaignPath}, &stdout, &stderr) code := executeWithDependencies([]string{"session", "plan", "2026-05-03", "--config", pipelinePath, "--campaign-file", campaignPath}, &stdout, &stderr, deps, executePlan)
if code != 0 { if code != 0 {
t.Fatalf("exit code = %d, want 0; stdout=%q stderr=%q", code, stdout.String(), stderr.String()) t.Fatalf("exit code = %d, want 0; stdout=%q stderr=%q", code, stdout.String(), stderr.String())
} }
@@ -270,11 +266,12 @@ func TestExecuteExplicitLocalSessionPrecedenceSkipsRemote(t *testing.T) {
pipelinePath, campaignPath, sessionPath := writeValidConfigFiles(t, workspaceRoot) pipelinePath, campaignPath, sessionPath := writeValidConfigFiles(t, workspaceRoot)
fake := &storage.FakeBackend{} fake := &storage.FakeBackend{}
var storeInitCalls int var storeInitCalls int
restoreAppConfigTestGlobals(t, fake, &storeInitCalls, []string{filepath.Join(t.TempDir(), "session.yml")}) setSessionConfigSearchPaths(t, []string{filepath.Join(t.TempDir(), "session.yml")})
deps := fakeCommandDependencies(fake, &storeInitCalls)
var stdout bytes.Buffer var stdout bytes.Buffer
var stderr bytes.Buffer var stderr bytes.Buffer
code := Execute([]string{"session", "plan", "2026-05-03", "--config", pipelinePath, "--campaign-file", campaignPath, "--session", sessionPath}, &stdout, &stderr) code := executeWithDependencies([]string{"session", "plan", "2026-05-03", "--config", pipelinePath, "--campaign-file", campaignPath, "--session", sessionPath}, &stdout, &stderr, deps, executePlan)
if code != 0 { if code != 0 {
t.Fatalf("exit code = %d, want 0; stderr=%q", code, stderr.String()) t.Fatalf("exit code = %d, want 0; stderr=%q", code, stderr.String())
} }
@@ -288,11 +285,12 @@ func TestExecuteLocalSessionDiscoveryPrecedenceSkipsRemote(t *testing.T) {
pipelinePath, campaignPath, sessionPath := writeValidConfigFiles(t, workspaceRoot) pipelinePath, campaignPath, sessionPath := writeValidConfigFiles(t, workspaceRoot)
fake := &storage.FakeBackend{} fake := &storage.FakeBackend{}
var storeInitCalls int var storeInitCalls int
restoreAppConfigTestGlobals(t, fake, &storeInitCalls, []string{sessionPath}) setSessionConfigSearchPaths(t, []string{sessionPath})
deps := fakeCommandDependencies(fake, &storeInitCalls)
var stdout bytes.Buffer var stdout bytes.Buffer
var stderr bytes.Buffer var stderr bytes.Buffer
code := Execute([]string{"session", "plan", "2026-05-03", "--config", pipelinePath, "--campaign-file", campaignPath}, &stdout, &stderr) code := executeWithDependencies([]string{"session", "plan", "2026-05-03", "--config", pipelinePath, "--campaign-file", campaignPath}, &stdout, &stderr, deps, executePlan)
if code != 0 { if code != 0 {
t.Fatalf("exit code = %d, want 0; stderr=%q", code, stderr.String()) t.Fatalf("exit code = %d, want 0; stderr=%q", code, stderr.String())
} }
@@ -307,11 +305,12 @@ func TestExecuteRemoteSessionMissingObjectFailsClearly(t *testing.T) {
fake := &storage.FakeBackend{} fake := &storage.FakeBackend{}
var storeInitCalls int var storeInitCalls int
missingSessionPath := filepath.Join(t.TempDir(), "session.yml") missingSessionPath := filepath.Join(t.TempDir(), "session.yml")
restoreAppConfigTestGlobals(t, fake, &storeInitCalls, []string{missingSessionPath}) setSessionConfigSearchPaths(t, []string{missingSessionPath})
deps := fakeCommandDependencies(fake, &storeInitCalls)
var stdout bytes.Buffer var stdout bytes.Buffer
var stderr bytes.Buffer var stderr bytes.Buffer
code := Execute([]string{"session", "plan", "2026-05-03", "--config", pipelinePath, "--campaign-file", campaignPath}, &stdout, &stderr) code := executeWithDependencies([]string{"session", "plan", "2026-05-03", "--config", pipelinePath, "--campaign-file", campaignPath}, &stdout, &stderr, deps, executePlan)
if code == 0 { if code == 0 {
t.Fatal("exit code = 0, want non-zero") t.Fatal("exit code = 0, want non-zero")
} }
@@ -327,11 +326,12 @@ func TestExecuteRemoteSessionRequiresSessionID(t *testing.T) {
workspaceRoot := t.TempDir() workspaceRoot := t.TempDir()
pipelinePath, campaignPath, _ := writeValidConfigFiles(t, workspaceRoot) pipelinePath, campaignPath, _ := writeValidConfigFiles(t, workspaceRoot)
var storeInitCalls int var storeInitCalls int
restoreAppConfigTestGlobals(t, &storage.FakeBackend{}, &storeInitCalls, []string{filepath.Join(t.TempDir(), "session.yml")}) setSessionConfigSearchPaths(t, []string{filepath.Join(t.TempDir(), "session.yml")})
deps := fakeCommandDependencies(&storage.FakeBackend{}, &storeInitCalls)
var stdout bytes.Buffer var stdout bytes.Buffer
var stderr bytes.Buffer var stderr bytes.Buffer
code := Execute([]string{"session", "plan", "--config", pipelinePath, "--campaign-file", campaignPath}, &stdout, &stderr) code := executeWithDependencies([]string{"session", "plan", "--config", pipelinePath, "--campaign-file", campaignPath}, &stdout, &stderr, deps, executePlan)
if code == 0 { if code == 0 {
t.Fatal("exit code = 0, want non-zero") t.Fatal("exit code = 0, want non-zero")
} }
@@ -346,20 +346,14 @@ func TestExecuteRemoteSessionRequiresSessionID(t *testing.T) {
func TestExecuteRemoteSessionStorageInitErrorFailsClearly(t *testing.T) { func TestExecuteRemoteSessionStorageInitErrorFailsClearly(t *testing.T) {
workspaceRoot := t.TempDir() workspaceRoot := t.TempDir()
pipelinePath, campaignPath, _ := writeValidConfigFiles(t, workspaceRoot) pipelinePath, campaignPath, _ := writeValidConfigFiles(t, workspaceRoot)
origStoreFn := newObjectStoreFromConfigFn setSessionConfigSearchPaths(t, []string{filepath.Join(t.TempDir(), "session.yml")})
origSessionDefaults := append([]string(nil), config.DefaultSessionConfigSearchPaths...) deps := commandDependencies{newObjectStore: func(context.Context, *config.Config) (storage.ObjectStore, error) {
config.DefaultSessionConfigSearchPaths = []string{filepath.Join(t.TempDir(), "session.yml")}
newObjectStoreFromConfigFn = func(context.Context, *config.Config) (storage.ObjectStore, error) {
return nil, errors.New("storage unavailable") return nil, errors.New("storage unavailable")
} }}
t.Cleanup(func() {
newObjectStoreFromConfigFn = origStoreFn
config.DefaultSessionConfigSearchPaths = origSessionDefaults
})
var stdout bytes.Buffer var stdout bytes.Buffer
var stderr bytes.Buffer var stderr bytes.Buffer
code := Execute([]string{"session", "plan", "2026-05-03", "--config", pipelinePath, "--campaign-file", campaignPath}, &stdout, &stderr) code := executeWithDependencies([]string{"session", "plan", "2026-05-03", "--config", pipelinePath, "--campaign-file", campaignPath}, &stdout, &stderr, deps, executePlan)
if code == 0 { if code == 0 {
t.Fatal("exit code = 0, want non-zero") t.Fatal("exit code = 0, want non-zero")
} }
@@ -374,11 +368,12 @@ func TestExecuteRemoteSessionMalformedYAMLFailsStrictDecode(t *testing.T) {
fake := &storage.FakeBackend{} fake := &storage.FakeBackend{}
seedRemoteSessionConfig(t, fake, "2026-05-03", "session_id: 2026-05-03\nunknown: true\n") seedRemoteSessionConfig(t, fake, "2026-05-03", "session_id: 2026-05-03\nunknown: true\n")
var storeInitCalls int var storeInitCalls int
restoreAppConfigTestGlobals(t, fake, &storeInitCalls, []string{filepath.Join(t.TempDir(), "session.yml")}) setSessionConfigSearchPaths(t, []string{filepath.Join(t.TempDir(), "session.yml")})
deps := fakeCommandDependencies(fake, &storeInitCalls)
var stdout bytes.Buffer var stdout bytes.Buffer
var stderr bytes.Buffer var stderr bytes.Buffer
code := Execute([]string{"session", "plan", "2026-05-03", "--config", pipelinePath, "--campaign-file", campaignPath}, &stdout, &stderr) code := executeWithDependencies([]string{"session", "plan", "2026-05-03", "--config", pipelinePath, "--campaign-file", campaignPath}, &stdout, &stderr, deps, executePlan)
if code == 0 { if code == 0 {
t.Fatal("exit code = 0, want non-zero") t.Fatal("exit code = 0, want non-zero")
} }
@@ -397,11 +392,12 @@ inputs:
prefix: audio/ prefix: audio/
`) `)
var storeInitCalls int var storeInitCalls int
restoreAppConfigTestGlobals(t, fake, &storeInitCalls, []string{filepath.Join(t.TempDir(), "session.yml")}) setSessionConfigSearchPaths(t, []string{filepath.Join(t.TempDir(), "session.yml")})
deps := fakeCommandDependencies(fake, &storeInitCalls)
var stdout bytes.Buffer var stdout bytes.Buffer
var stderr bytes.Buffer var stderr bytes.Buffer
code := Execute([]string{"session", "plan", "2026-05-03", "--config", pipelinePath, "--campaign-file", campaignPath}, &stdout, &stderr) code := executeWithDependencies([]string{"session", "plan", "2026-05-03", "--config", pipelinePath, "--campaign-file", campaignPath}, &stdout, &stderr, deps, executePlan)
if code == 0 { if code == 0 {
t.Fatal("exit code = 0, want non-zero") t.Fatal("exit code = 0, want non-zero")
} }
@@ -416,11 +412,12 @@ func TestExecuteRemoteSessionMismatchFails(t *testing.T) {
fake := &storage.FakeBackend{} fake := &storage.FakeBackend{}
seedRemoteSessionConfig(t, fake, "2026-05-03", "session_id: 2026-05-04\ninputs:\n audio_s3:\n prefix: audio/\n") seedRemoteSessionConfig(t, fake, "2026-05-03", "session_id: 2026-05-04\ninputs:\n audio_s3:\n prefix: audio/\n")
var storeInitCalls int var storeInitCalls int
restoreAppConfigTestGlobals(t, fake, &storeInitCalls, []string{filepath.Join(t.TempDir(), "session.yml")}) setSessionConfigSearchPaths(t, []string{filepath.Join(t.TempDir(), "session.yml")})
deps := fakeCommandDependencies(fake, &storeInitCalls)
var stdout bytes.Buffer var stdout bytes.Buffer
var stderr bytes.Buffer var stderr bytes.Buffer
code := Execute([]string{"session", "plan", "2026-05-03", "--config", pipelinePath, "--campaign-file", campaignPath}, &stdout, &stderr) code := executeWithDependencies([]string{"session", "plan", "2026-05-03", "--config", pipelinePath, "--campaign-file", campaignPath}, &stdout, &stderr, deps, executePlan)
if code == 0 { if code == 0 {
t.Fatal("exit code = 0, want non-zero") t.Fatal("exit code = 0, want non-zero")
} }
@@ -448,12 +445,16 @@ func restoreAppConfigTestGlobals(t *testing.T, fake *storage.FakeBackend, storeI
func remoteSessionLoaderDependencies(fake *storage.FakeBackend) (commandDependencies, *int) { func remoteSessionLoaderDependencies(fake *storage.FakeBackend) (commandDependencies, *int) {
storeInitCalls := new(int) storeInitCalls := new(int)
return commandDependencies{ return fakeCommandDependencies(fake, storeInitCalls), storeInitCalls
newObjectStore: func(context.Context, *config.Config) (storage.ObjectStore, error) { }
func fakeCommandDependencies(fake *storage.FakeBackend, storeInitCalls *int) commandDependencies {
return commandDependencies{newObjectStore: func(context.Context, *config.Config) (storage.ObjectStore, error) {
if storeInitCalls != nil {
*storeInitCalls++ *storeInitCalls++
return fake, nil }
}, return fake, nil
}, storeInitCalls }}
} }
func setSessionConfigSearchPaths(t *testing.T, paths []string) { func setSessionConfigSearchPaths(t *testing.T, paths []string) {

View File

@@ -12,10 +12,15 @@ import (
// Run executes the pipeline plan and persists manifest state. // Run executes the pipeline plan and persists manifest state.
func Run(ctx context.Context, args []string, out io.Writer) error { func Run(ctx context.Context, args []string, out io.Writer) error {
return runWithPipelineExecutor(ctx, args, out, executePlan) return runWithDependencies(ctx, args, out, executePlan, commandDependencies{})
} }
func runWithPipelineExecutor(ctx context.Context, args []string, out io.Writer, executor pipelineExecutor) error { func runWithPipelineExecutor(ctx context.Context, args []string, out io.Writer, executor pipelineExecutor) error {
return runWithDependencies(ctx, args, out, executor, commandDependencies{})
}
func runWithDependencies(ctx context.Context, args []string, out io.Writer, executor pipelineExecutor, deps commandDependencies) error {
deps = normalizeCommandDependencies(deps)
request, err := parseBoundedRunRequest("run", args, out) request, err := parseBoundedRunRequest("run", args, out)
if err != nil { if err != nil {
if errors.Is(err, flag.ErrHelp) { if errors.Is(err, flag.ErrHelp) {
@@ -24,7 +29,7 @@ func runWithPipelineExecutor(ctx context.Context, args []string, out io.Writer,
return err return err
} }
flags := request.Config flags := request.Config
loaded, err := loadCommandConfig(ctx, flags.pipelinePath, flags.campaignPath, flags.campaignFilePath, flags.sessionPath, flags.sessionOptions()) loaded, err := loadCommandConfigWithDependencies(ctx, flags.pipelinePath, flags.campaignPath, flags.campaignFilePath, flags.sessionPath, flags.sessionOptions(), deps)
if err != nil { if err != nil {
return fmt.Errorf("run: %w", err) return fmt.Errorf("run: %w", err)
} }

View File

@@ -12,10 +12,14 @@ import (
// RunStage executes exactly one selected stage. // RunStage executes exactly one selected stage.
func RunStage(ctx context.Context, args []string, out io.Writer) error { func RunStage(ctx context.Context, args []string, out io.Writer) error {
return runStageWithPipelineExecutor(ctx, args, out, executePlan) return runStageWithDependencies(ctx, args, out, executePlan, commandDependencies{})
} }
func runStageWithPipelineExecutor(ctx context.Context, args []string, out io.Writer, executor pipelineExecutor) error { func runStageWithPipelineExecutor(ctx context.Context, args []string, out io.Writer, executor pipelineExecutor) error {
return runStageWithDependencies(ctx, args, out, executor, commandDependencies{})
}
func runStageWithDependencies(ctx context.Context, args []string, out io.Writer, executor pipelineExecutor, deps commandDependencies) error {
var stageName string var stageName string
var positionalSessionID string var positionalSessionID string
if len(args) >= 2 && !isCLIFlagToken(args[0]) && !isCLIFlagToken(args[1]) { if len(args) >= 2 && !isCLIFlagToken(args[0]) && !isCLIFlagToken(args[1]) {
@@ -78,7 +82,7 @@ func runStageWithPipelineExecutor(ctx context.Context, args []string, out io.Wri
PreviousSessionID: flags.previousSessionID, PreviousSessionID: flags.previousSessionID,
Force: force, Force: force,
SelectedArtifacts: normalizedArtifacts, SelectedArtifacts: normalizedArtifacts,
}, executor) }, executor, deps)
if err != nil { if err != nil {
return err return err
} }
@@ -97,10 +101,14 @@ func runStageWithPipelineExecutor(ctx context.Context, args []string, out io.Wri
// Analyze force-runs the analyze stage. // Analyze force-runs the analyze stage.
func Analyze(ctx context.Context, args []string, out io.Writer) error { func Analyze(ctx context.Context, args []string, out io.Writer) error {
return analyzeWithPipelineExecutor(ctx, args, out, executePlan) return analyzeWithDependencies(ctx, args, out, executePlan, commandDependencies{})
} }
func analyzeWithPipelineExecutor(ctx context.Context, args []string, out io.Writer, executor pipelineExecutor) error { func analyzeWithPipelineExecutor(ctx context.Context, args []string, out io.Writer, executor pipelineExecutor) error {
return analyzeWithDependencies(ctx, args, out, executor, commandDependencies{})
}
func analyzeWithDependencies(ctx context.Context, args []string, out io.Writer, executor pipelineExecutor, deps commandDependencies) error {
fs := flag.NewFlagSet("analyze", flag.ContinueOnError) fs := flag.NewFlagSet("analyze", flag.ContinueOnError)
fs.SetOutput(io.Discard) fs.SetOutput(io.Discard)
@@ -131,7 +139,7 @@ func analyzeWithPipelineExecutor(ctx context.Context, args []string, out io.Writ
PreviousSessionID: flags.previousSessionID, PreviousSessionID: flags.previousSessionID,
Force: true, Force: true,
SelectedArtifacts: normalizedArtifacts, SelectedArtifacts: normalizedArtifacts,
}, executor) }, executor, deps)
if err != nil { if err != nil {
return err return err
} }
@@ -148,10 +156,14 @@ func analyzeWithPipelineExecutor(ctx context.Context, args []string, out io.Writ
// Publish force-runs the publish stage. // Publish force-runs the publish stage.
func Publish(ctx context.Context, args []string, out io.Writer) error { func Publish(ctx context.Context, args []string, out io.Writer) error {
return publishWithPipelineExecutor(ctx, args, out, executePlan) return publishWithDependencies(ctx, args, out, executePlan, commandDependencies{})
} }
func publishWithPipelineExecutor(ctx context.Context, args []string, out io.Writer, executor pipelineExecutor) error { func publishWithPipelineExecutor(ctx context.Context, args []string, out io.Writer, executor pipelineExecutor) error {
return publishWithDependencies(ctx, args, out, executor, commandDependencies{})
}
func publishWithDependencies(ctx context.Context, args []string, out io.Writer, executor pipelineExecutor, deps commandDependencies) error {
fs := flag.NewFlagSet("publish", flag.ContinueOnError) fs := flag.NewFlagSet("publish", flag.ContinueOnError)
fs.SetOutput(io.Discard) fs.SetOutput(io.Discard)
@@ -182,7 +194,7 @@ func publishWithPipelineExecutor(ctx context.Context, args []string, out io.Writ
PreviousSessionID: flags.previousSessionID, PreviousSessionID: flags.previousSessionID,
Force: true, Force: true,
SelectedArtifacts: normalizedArtifacts, SelectedArtifacts: normalizedArtifacts,
}, executor) }, executor, deps)
if err != nil { if err != nil {
return err return err
} }
@@ -210,16 +222,17 @@ type singleStageCommand struct {
SelectedArtifacts []string SelectedArtifacts []string
} }
func runSingleStageCommand(ctx context.Context, req singleStageCommand, executor pipelineExecutor) (*RunSummary, error) { func runSingleStageCommand(ctx context.Context, req singleStageCommand, executor pipelineExecutor, deps commandDependencies) (*RunSummary, error) {
deps = normalizeCommandDependencies(deps)
plan, err := buildSingleStageExecutionPlan(req.StageName) plan, err := buildSingleStageExecutionPlan(req.StageName)
if err != nil { if err != nil {
return nil, fmt.Errorf("%s: %w", req.CommandName, err) return nil, fmt.Errorf("%s: %w", req.CommandName, err)
} }
loaded, err := loadCommandConfig(ctx, req.PipelinePath, req.CampaignPath, req.CampaignFilePath, req.SessionPath, config.SessionLoadOptions{ loaded, err := loadCommandConfigWithDependencies(ctx, req.PipelinePath, req.CampaignPath, req.CampaignFilePath, req.SessionPath, config.SessionLoadOptions{
SessionID: req.SessionID, SessionID: req.SessionID,
PreviousSessionID: req.PreviousSessionID, PreviousSessionID: req.PreviousSessionID,
}) }, deps)
if err != nil { if err != nil {
return nil, fmt.Errorf("%s: %w", req.CommandName, err) return nil, fmt.Errorf("%s: %w", req.CommandName, err)
} }

View File

@@ -106,7 +106,8 @@ func TestExecuteRemoteSessionFallbackUsesPositionalSessionID(t *testing.T) {
pipelinePath, campaignPath, _ := writeValidConfigFiles(t, workspaceRoot) pipelinePath, campaignPath, _ := writeValidConfigFiles(t, workspaceRoot)
fake := &storage.FakeBackend{} fake := &storage.FakeBackend{}
var storeInitCalls int var storeInitCalls int
restoreAppConfigTestGlobals(t, fake, &storeInitCalls, []string{filepath.Join(t.TempDir(), "session.yml")}) setSessionConfigSearchPaths(t, []string{filepath.Join(t.TempDir(), "session.yml")})
deps := fakeCommandDependencies(fake, &storeInitCalls)
remoteKey := seedRemoteSessionConfig(t, fake, "2026-06-07", `session_id: 2026-06-07 remoteKey := seedRemoteSessionConfig(t, fake, "2026-06-07", `session_id: 2026-06-07
inputs: inputs:
audio_s3: audio_s3:
@@ -122,12 +123,12 @@ inputs:
var stdout bytes.Buffer var stdout bytes.Buffer
var stderr bytes.Buffer var stderr bytes.Buffer
code := executeWithPipelineExecutor([]string{ code := executeWithDependencies([]string{
"run", "run",
"2026-06-07", "2026-06-07",
"--config", pipelinePath, "--config", pipelinePath,
"--campaign-file", campaignPath, "--campaign-file", campaignPath,
}, &stdout, &stderr, executor) }, &stdout, &stderr, deps, executor)
if code != 0 { if code != 0 {
t.Fatalf("exit code = %d, want 0; stderr=%q", code, stderr.String()) t.Fatalf("exit code = %d, want 0; stderr=%q", code, stderr.String())
} }