diff --git a/docs/roadmap/implementation.md b/docs/roadmap/implementation.md index cc3f53f..b9b7c5e 100644 --- a/docs/roadmap/implementation.md +++ b/docs/roadmap/implementation.md @@ -243,7 +243,7 @@ without hiding session-lock ownership. ## Stage 3 — Isolate Collaborator Composition -**Status: Pending** +**Status: Completed** ### Goal diff --git a/internal/app/post_publish_cleanup.go b/internal/app/post_publish_cleanup.go index dd0c324..17fe3c8 100644 --- a/internal/app/post_publish_cleanup.go +++ b/internal/app/post_publish_cleanup.go @@ -12,9 +12,14 @@ import ( "gitea.maximumdirect.net/eric/narratio/internal/manifest" ) -var removeRunScopedDirFn = removeRunScopedDir - -func runPostPublishCleanup(ctx context.Context, env *Env, manifestPath string, m *manifest.Manifest, executed []string) error { +func runPostPublishCleanup( + ctx context.Context, + env *Env, + manifestPath string, + m *manifest.Manifest, + executed []string, + removeScopedDir func(root, target, policy string) error, +) error { if env == nil || env.Config == nil || env.Config.Pipeline == nil || m == nil { return nil } @@ -40,7 +45,7 @@ func runPostPublishCleanup(ctx context.Context, env *Env, manifestPath string, m if target.Completed { continue } - if err := removeRunScopedDirFn(target.Root, target.Path, target.Policy); err != nil { + if err := removeScopedDir(target.Root, target.Path, target.Policy); err != nil { return fmt.Errorf("complete post-publish cleanup for %q: %w", target.Path, err) } target.Completed = true diff --git a/internal/app/post_publish_cleanup_test.go b/internal/app/post_publish_cleanup_test.go index 2189f66..c8f3390 100644 --- a/internal/app/post_publish_cleanup_test.go +++ b/internal/app/post_publish_cleanup_test.go @@ -162,16 +162,20 @@ func TestPostPublishCleanupRetriesFailedDeletionWithoutTouchingOtherRuns(t *test cfg.Pipeline.Spool.DeleteAudioAfterPublish = true cfg.Pipeline.Workspace.CleanupAfterPublish = true - originalRemove := removeRunScopedDirFn - removeRunScopedDirFn = func(root, target, policy string) error { + remove := func(root, target, policy string) error { if policy == "pipeline.workspace.cleanup_after_publish" { return errors.New("injected deletion failure") } - return originalRemove(root, target, policy) + return removeRunScopedDir(root, target, policy) } - t.Cleanup(func() { removeRunScopedDirFn = originalRemove }) - _, err := executeStages(context.Background(), cfg, []stage.Stage{publishSuccessStage{targets: &seed}}, RunOptions{Env: &Env{ObjectStore: &storage.FakeBackend{}}}) + _, err := executePlanWithDependencies( + context.Background(), + cfg, + BoundedPlan{stages: []stage.Stage{publishSuccessStage{targets: &seed}}}, + RunOptions{Env: &Env{ObjectStore: &storage.FakeBackend{}}}, + invocationDependencies{removeScopedDir: remove}, + ) if err == nil || !strings.Contains(err.Error(), "post-publish cleanup incomplete") { t.Fatalf("executeStages() error = %v, want incomplete cleanup", err) } @@ -181,7 +185,6 @@ func TestPostPublishCleanupRetriesFailedDeletionWithoutTouchingOtherRuns(t *test assertExists(t, seed.otherRunDir) assertExists(t, seed.previousCachePath) - removeRunScopedDirFn = originalRemove if _, err := executeStages(context.Background(), cfg, nil, RunOptions{Env: &Env{ObjectStore: &storage.FakeBackend{}}}); err != nil { t.Fatalf("non-publish executeStages() error = %v", err) } diff --git a/internal/app/runner.go b/internal/app/runner.go index 6c774ae..15ec97c 100644 --- a/internal/app/runner.go +++ b/internal/app/runner.go @@ -6,11 +6,9 @@ import ( "fmt" "os" "strings" + "time" "gitea.maximumdirect.net/eric/narratio/internal/adapters/audita" - "gitea.maximumdirect.net/eric/narratio/internal/adapters/notarius" - "gitea.maximumdirect.net/eric/narratio/internal/adapters/notify" - "gitea.maximumdirect.net/eric/narratio/internal/adapters/scriptorium" "gitea.maximumdirect.net/eric/narratio/internal/adapters/seriatim" "gitea.maximumdirect.net/eric/narratio/internal/adapters/whisperx" "gitea.maximumdirect.net/eric/narratio/internal/artifactmodel" @@ -39,7 +37,12 @@ type RunSummary struct { var executeStagesFn = executePlan func executePlan(ctx context.Context, cfg *config.Config, plan BoundedPlan, opts RunOptions) (summary *RunSummary, resultErr error) { - runtime, err := prepareInvocation(ctx, cfg, plan, opts) + return executePlanWithDependencies(ctx, cfg, plan, opts, invocationDependencies{}) +} + +func executePlanWithDependencies(ctx context.Context, cfg *config.Config, plan BoundedPlan, opts RunOptions, dependencies invocationDependencies) (summary *RunSummary, resultErr error) { + dependencies = dependencies.normalized() + runtime, err := prepareInvocation(ctx, cfg, plan, opts, dependencies) if err != nil { return nil, err } @@ -64,90 +67,23 @@ func executePlan(ctx context.Context, cfg *config.Config, plan BoundedPlan, opts if err := initializeLockedInvocation(runtime); err != nil { return nil, err } + if err := composeInvocation(runtime); err != nil { + return nil, err + } stages := runtime.stages - effectiveArtifacts := runtime.effectiveArtifacts - identity := runtime.identity env := runtime.env manifestPath := runtime.manifestPath m := runtime.manifest - runID := identity.RunID - if env.WhisperX == nil && stagesContainAny(stages, "transcribe") { - client, err := buildDefaultWhisperXClient(env.Config) - if err != nil { - return nil, persistSessionTerminalFailure( - ctx, env.ManifestStore, manifestPath, m, - fmt.Errorf("initialize whisperx client: %w", err), - ) - } - env.WhisperX = client - } - if env.Seriatim == nil && stagesContainAny(stages, "merge", "normalize", "trim", "render") { - runner, err := buildDefaultSeriatimRunner(env.Config) - if err != nil { - return nil, persistSessionTerminalFailure( - ctx, env.ManifestStore, manifestPath, m, - fmt.Errorf("initialize seriatim runner: %w", err), - ) - } - env.Seriatim = runner - } - if env.Audita == nil && stagesContainAny(stages, "polish") { - runner, err := buildDefaultAuditaRunner(env.Config) - if err != nil { - return nil, persistSessionTerminalFailure( - ctx, env.ManifestStore, manifestPath, m, - fmt.Errorf("initialize audita runner: %w", err), - ) - } - env.Audita = runner - } - if env.Notarius == nil && needsNotariusForRun(env.Config, stages) { - env.Notarius = notarius.NewSubprocessRunner() - } - if env.Scriptorium == nil && stagesContainAny(stages, "trim", "analyze") { - env.Scriptorium = scriptorium.NewSubprocessRunner() - } - if env.ObjectStore == nil && needsObjectStoreForRun(env.Config, stages, effectiveArtifacts) { - objectStore, err := newCommandObjectStore(ctx, env.Config, nil) - if err != nil { - return nil, persistSessionTerminalFailure( - ctx, env.ManifestStore, manifestPath, m, err, - ) - } - env.ObjectStore = objectStore - } - if needsRemoteLocksForRun(env.Config, stages) { - locks, err := loadEffectiveLocks(ctx, env.Config, env.ObjectStore) - if err != nil { - return nil, persistSessionTerminalFailure( - ctx, env.ManifestStore, manifestPath, m, - fmt.Errorf("load remote publish locks: %w", err), - ) - } - applyEffectiveLocks(env.Config, locks.All) - staticLocks := append([]config.PublishLockRule(nil), locks.Static...) - env.RevalidatePublishLocks = func(recheckCtx context.Context) ([]config.PublishLockRule, error) { - remote, _, _, err := loadRemoteLockStore(recheckCtx, env.Config, env.ObjectStore) - if err != nil { - return nil, err - } - return config.MergePublishLockRules(staticLocks, remote.Locks), nil - } - } - if env.Notifier == nil && stagesContainAny(stages, "notify") { - env.Notifier = ¬ify.NoopSender{} - } - - stageEnv := env + runID := runtime.identity.RunID + stageEnv := runtime.env for _, s := range stages { stageEnv.Force = opts.Force runtime.stageNames = append(runtime.stageNames, s.Name()) semanticConfig, err := currentStageSemanticConfig(s, stageEnv) if err != nil { - return nil, persistSessionTerminalFailure( - ctx, env.ManifestStore, manifestPath, m, + return nil, runtime.persistTerminalFailure( fmt.Errorf("fingerprint semantic configuration for stage %q: %w", s.Name(), err), ) } @@ -156,23 +92,20 @@ func executePlan(ctx context.Context, cfg *config.Config, plan BoundedPlan, opts if decision.action == stageActionSkip { validation, err := evaluateStageResume(ctx, s, stageEnv, m, semanticConfig) if err != nil { - return nil, persistSessionTerminalFailure( - ctx, env.ManifestStore, manifestPath, m, + return nil, runtime.persistTerminalFailure( fmt.Errorf("validate resume for stage %q: %w", s.Name(), err), ) } decision = decideStageExecution(recordedStageStatus(m, s.Name()), opts.Force, validation) if decision.markSelectedStale { - staleAt := nowUTC() + staleAt := runtime.dependencies.now() if _, err := applyPreExecutionDecision(m, s.Name(), decision, staleAt); err != nil { - return nil, persistSessionTerminalFailure( - ctx, env.ManifestStore, manifestPath, m, + return nil, runtime.persistTerminalFailure( fmt.Errorf("invalidate dependents after resume validation for stage %q: %w", s.Name(), err), ) } if err := env.ManifestStore.Save(ctx, manifestPath, m); err != nil { - return nil, persistSessionTerminalFailure( - ctx, env.ManifestStore, manifestPath, m, + return nil, runtime.persistTerminalFailure( fmt.Errorf("save manifest after resume validation for stage %q: %w", s.Name(), err), ) } @@ -189,23 +122,20 @@ func executePlan(ctx context.Context, cfg *config.Config, plan BoundedPlan, opts priorOutcome := capturePriorStageOutcome(m, s.Name()) priorAnalyzeState := captureAnalyzeState(m, s.Name()) - now := nowUTC() + now := runtime.dependencies.now() if _, err := beginStageAttempt(m, s.Name(), decision, now); err != nil { - return nil, persistSessionTerminalFailure( - ctx, env.ManifestStore, manifestPath, m, + return nil, runtime.persistTerminalFailure( fmt.Errorf("invalidate dependents before forced stage %q: %w", s.Name(), err), ) } env.Logger.Info("starting stage", "stage", s.Name()) if err := env.ManifestStore.Save(ctx, manifestPath, m); err != nil { operationErr := fmt.Errorf("save manifest before stage %q: %w", s.Name(), err) - m.MarkStageFailed(s.Name(), nowUTC(), operationErr.Error()) - if _, invalidationErr := invalidateDependentSucceededStagesWithReason(m, s.Name(), nowUTC(), staleReasonFailure); invalidationErr != nil { + m.MarkStageFailed(s.Name(), runtime.dependencies.now(), operationErr.Error()) + if _, invalidationErr := invalidateDependentSucceededStagesWithReason(m, s.Name(), runtime.dependencies.now(), staleReasonFailure); invalidationErr != nil { operationErr = errors.Join(operationErr, fmt.Errorf("invalidate dependents after stage %q persistence failure: %w", s.Name(), invalidationErr)) } - return nil, persistSessionTerminalFailure( - ctx, env.ManifestStore, manifestPath, m, operationErr, - ) + return nil, runtime.persistTerminalFailure(operationErr) } env.Logger.Debug("manifest saved", "stage", s.Name(), "transition", "running", "path", manifestPath) @@ -227,29 +157,24 @@ func executePlan(ctx context.Context, cfg *config.Config, plan BoundedPlan, opts if analyzeProjection != nil { applyAnalyzeProjection(m, analyzeProjection) } - failedAt := nowUTC() + failedAt := runtime.dependencies.now() m.MarkStageFailed(s.Name(), failedAt, err.Error()) if _, invalidationErr := invalidateDependentSucceededStagesWithReason(m, s.Name(), failedAt, staleReasonFailure); invalidationErr != nil { err = errors.Join(err, fmt.Errorf("invalidate dependents after stage %q failure: %w", s.Name(), invalidationErr)) } env.Logger.Info("stage failed", "stage", s.Name(), "error", err) - return nil, persistSessionTerminalFailure( - ctx, env.ManifestStore, manifestPath, m, - fmt.Errorf("stage %q failed: %w", s.Name(), err), - ) + return nil, runtime.persistTerminalFailure(fmt.Errorf("stage %q failed: %w", s.Name(), err)) } if result != nil && result.Disposition == stage.StageDispositionSkipped { runtime.skipped = append(runtime.skipped, s.Name()) - skippedAt := nowUTC() + skippedAt := runtime.dependencies.now() if _, err := applyCompletedStageOutcome(m, s.Name(), result, nil, semanticConfig, priorOutcome, skippedAt); err != nil { - return nil, persistSessionTerminalFailure( - ctx, env.ManifestStore, manifestPath, m, + return nil, runtime.persistTerminalFailure( fmt.Errorf("invalidate dependents after stage %q self-skip: %w", s.Name(), err), ) } if err := env.ManifestStore.Save(ctx, manifestPath, m); err != nil { - return nil, persistSessionTerminalFailure( - ctx, env.ManifestStore, manifestPath, m, + return nil, runtime.persistTerminalFailure( fmt.Errorf("save manifest after self-skip %q: %w", s.Name(), err), ) } @@ -262,10 +187,9 @@ func executePlan(ctx context.Context, cfg *config.Config, plan BoundedPlan, opts if analyzeProjection != nil { sessionOutputs = analyzeProjectionOutputs(analyzeProjection.session) } - succeededAt := nowUTC() + succeededAt := runtime.dependencies.now() if _, err := applyCompletedStageOutcome(m, s.Name(), result, sessionOutputs, semanticConfig, priorOutcome, succeededAt); err != nil { - return nil, persistSessionTerminalFailure( - ctx, env.ManifestStore, manifestPath, m, + return nil, runtime.persistTerminalFailure( fmt.Errorf("invalidate dependents after changed stage %q result: %w", s.Name(), err), ) } @@ -275,26 +199,24 @@ func executePlan(ctx context.Context, cfg *config.Config, plan BoundedPlan, opts operationErr := fmt.Errorf("save manifest after stage %q: %w", s.Name(), err) if analyzeProjection != nil { restoreAnalyzeState(m, priorAnalyzeState) - failedAt := nowUTC() + failedAt := runtime.dependencies.now() m.MarkStageFailed(s.Name(), failedAt, operationErr.Error()) if _, invalidationErr := invalidateDependentSucceededStagesWithReason(m, s.Name(), failedAt, staleReasonFailure); invalidationErr != nil { operationErr = errors.Join(operationErr, fmt.Errorf("invalidate dependents after analyze projection persistence failure: %w", invalidationErr)) } } - return nil, persistSessionTerminalFailure( - ctx, env.ManifestStore, manifestPath, m, - operationErr, - ) + return nil, runtime.persistTerminalFailure(operationErr) } env.Logger.Debug("manifest saved", "stage", s.Name(), "transition", "succeeded", "path", manifestPath) env.Logger.Info("stage succeeded", "stage", s.Name()) } if containsStage(runtime.executed, "publish") { - if err := runPostPublishCleanup(ctx, env, manifestPath, m, runtime.executed); err != nil { + if err := runPostPublishCleanup(ctx, env, manifestPath, m, runtime.executed, runtime.dependencies.removeScopedDir); err != nil { return nil, persistPostPublishCleanupFailure( ctx, env.ManifestStore, manifestPath, m, fmt.Errorf("post-publish cleanup incomplete: %w", err), + runtime.dependencies.now(), ) } } @@ -331,12 +253,13 @@ func persistPostPublishCleanupFailure( sessionManifestPath string, sessionManifest *manifest.Manifest, operationErr error, + at time.Time, ) error { if operationErr == nil { return nil } if sessionManifest != nil { - sessionManifest.RecordFailure(nowUTC(), operationErr.Error()) + sessionManifest.RecordFailure(at, operationErr.Error()) } // Cleanup may have deliberately removed the selected run directory. The // committed remote snapshot remains authoritative, while the session @@ -351,12 +274,12 @@ func persistSessionTerminalFailure( sessionManifestPath string, sessionManifest *manifest.Manifest, operationErr error, + at time.Time, ) error { if operationErr == nil { return nil } - at := nowUTC() if sessionManifest != nil { sessionManifest.RecordFailure(at, operationErr.Error()) } diff --git a/internal/app/runner_composition.go b/internal/app/runner_composition.go new file mode 100644 index 0000000..446f85f --- /dev/null +++ b/internal/app/runner_composition.go @@ -0,0 +1,69 @@ +package app + +import ( + "context" + "fmt" + + "gitea.maximumdirect.net/eric/narratio/internal/adapters/notify" + "gitea.maximumdirect.net/eric/narratio/internal/config" +) + +func composeInvocation(runtime *invocationRuntime) error { + env := runtime.env + stages := runtime.stages + dependencies := runtime.dependencies + + if env.WhisperX == nil && stagesContainAny(stages, "transcribe") { + client, err := dependencies.newWhisperX(env.Config) + if err != nil { + return runtime.persistTerminalFailure(fmt.Errorf("initialize whisperx client: %w", err)) + } + env.WhisperX = client + } + if env.Seriatim == nil && stagesContainAny(stages, "merge", "normalize", "trim", "render") { + runner, err := dependencies.newSeriatim(env.Config) + if err != nil { + return runtime.persistTerminalFailure(fmt.Errorf("initialize seriatim runner: %w", err)) + } + env.Seriatim = runner + } + if env.Audita == nil && stagesContainAny(stages, "polish") { + runner, err := dependencies.newAudita(env.Config) + if err != nil { + return runtime.persistTerminalFailure(fmt.Errorf("initialize audita runner: %w", err)) + } + env.Audita = runner + } + if env.Notarius == nil && needsNotariusForRun(env.Config, stages) { + env.Notarius = dependencies.newNotarius() + } + if env.Scriptorium == nil && stagesContainAny(stages, "trim", "analyze") { + env.Scriptorium = dependencies.newScriptorium() + } + if env.ObjectStore == nil && needsObjectStoreForRun(env.Config, stages, runtime.effectiveArtifacts) { + objectStore, err := dependencies.newObjectStore(runtime.ctx, env.Config) + if err != nil { + return runtime.persistTerminalFailure(err) + } + env.ObjectStore = objectStore + } + if needsRemoteLocksForRun(env.Config, stages) { + locks, err := dependencies.loadEffectiveLocks(runtime.ctx, env.Config, env.ObjectStore) + if err != nil { + return runtime.persistTerminalFailure(fmt.Errorf("load remote publish locks: %w", err)) + } + applyEffectiveLocks(env.Config, locks.All) + staticLocks := append([]config.PublishLockRule(nil), locks.Static...) + env.RevalidatePublishLocks = func(recheckCtx context.Context) ([]config.PublishLockRule, error) { + remote, _, _, err := dependencies.loadRemoteLockStore(recheckCtx, env.Config, env.ObjectStore) + if err != nil { + return nil, err + } + return config.MergePublishLockRules(staticLocks, remote.Locks), nil + } + } + if env.Notifier == nil && stagesContainAny(stages, "notify") { + env.Notifier = ¬ify.NoopSender{} + } + return nil +} diff --git a/internal/app/runner_dependencies.go b/internal/app/runner_dependencies.go new file mode 100644 index 0000000..4eafad7 --- /dev/null +++ b/internal/app/runner_dependencies.go @@ -0,0 +1,68 @@ +package app + +import ( + "context" + "time" + + "gitea.maximumdirect.net/eric/narratio/internal/adapters/audita" + "gitea.maximumdirect.net/eric/narratio/internal/adapters/notarius" + "gitea.maximumdirect.net/eric/narratio/internal/adapters/scriptorium" + "gitea.maximumdirect.net/eric/narratio/internal/adapters/seriatim" + "gitea.maximumdirect.net/eric/narratio/internal/adapters/storage" + "gitea.maximumdirect.net/eric/narratio/internal/adapters/whisperx" + "gitea.maximumdirect.net/eric/narratio/internal/artifacts" + "gitea.maximumdirect.net/eric/narratio/internal/config" +) + +type invocationDependencies struct { + now func() time.Time + newRunID func() (string, error) + newWhisperX func(*config.Config) (whisperx.Client, error) + newSeriatim func(*config.Config) (seriatim.Runner, error) + newAudita func(*config.Config) (audita.Runner, error) + newNotarius func() notarius.Runner + newScriptorium func() scriptorium.Runner + newObjectStore func(context.Context, *config.Config) (storage.ObjectStore, error) + loadEffectiveLocks func(context.Context, *config.Config, storage.ObjectStore) (*effectiveLocks, error) + loadRemoteLockStore func(context.Context, *config.Config, storage.ObjectStore) (*config.PublishLockStore, string, string, error) + removeScopedDir func(root, target, policy string) error +} + +func (dependencies invocationDependencies) normalized() invocationDependencies { + if dependencies.now == nil { + dependencies.now = nowUTC + } + if dependencies.newRunID == nil { + dependencies.newRunID = artifacts.NewRunID + } + if dependencies.newWhisperX == nil { + dependencies.newWhisperX = buildDefaultWhisperXClient + } + if dependencies.newSeriatim == nil { + dependencies.newSeriatim = buildDefaultSeriatimRunner + } + if dependencies.newAudita == nil { + dependencies.newAudita = buildDefaultAuditaRunner + } + if dependencies.newNotarius == nil { + dependencies.newNotarius = func() notarius.Runner { return notarius.NewSubprocessRunner() } + } + if dependencies.newScriptorium == nil { + dependencies.newScriptorium = func() scriptorium.Runner { return scriptorium.NewSubprocessRunner() } + } + if dependencies.newObjectStore == nil { + dependencies.newObjectStore = func(ctx context.Context, cfg *config.Config) (storage.ObjectStore, error) { + return newCommandObjectStore(ctx, cfg, nil) + } + } + if dependencies.loadEffectiveLocks == nil { + dependencies.loadEffectiveLocks = loadEffectiveLocks + } + if dependencies.loadRemoteLockStore == nil { + dependencies.loadRemoteLockStore = loadRemoteLockStore + } + if dependencies.removeScopedDir == nil { + dependencies.removeScopedDir = removeRunScopedDir + } + return dependencies +} diff --git a/internal/app/runner_runtime.go b/internal/app/runner_runtime.go index 00244c8..39f6d74 100644 --- a/internal/app/runner_runtime.go +++ b/internal/app/runner_runtime.go @@ -31,9 +31,10 @@ type invocationRuntime struct { stageNames []string executed []string skipped []string + dependencies invocationDependencies } -func prepareInvocation(ctx context.Context, cfg *config.Config, plan BoundedPlan, opts RunOptions) (*invocationRuntime, error) { +func prepareInvocation(ctx context.Context, cfg *config.Config, plan BoundedPlan, opts RunOptions, dependencies invocationDependencies) (*invocationRuntime, error) { var prerequisiteStore manifest.Store if opts.Env != nil { prerequisiteStore = opts.Env.ManifestStore @@ -58,7 +59,7 @@ func prepareInvocation(ctx context.Context, cfg *config.Config, plan BoundedPlan } } - runID, err := artifacts.NewRunID() + runID, err := dependencies.newRunID() if err != nil { return nil, fmt.Errorf("generate run id: %w", err) } @@ -107,9 +108,10 @@ func prepareInvocation(ctx context.Context, cfg *config.Config, plan BoundedPlan identity.Campaign, identity.SessionID, ), - stageNames: make([]string, 0, len(stages)), - executed: make([]string, 0, len(stages)), - skipped: make([]string, 0, len(stages)), + stageNames: make([]string, 0, len(stages)), + executed: make([]string, 0, len(stages)), + skipped: make([]string, 0, len(stages)), + dependencies: dependencies, }, nil } @@ -171,5 +173,6 @@ func (runtime *invocationRuntime) persistTerminalFailure(err error) error { runtime.manifestPath, runtime.manifest, err, + runtime.dependencies.now(), ) } diff --git a/internal/app/runner_terminal_failure_test.go b/internal/app/runner_terminal_failure_test.go index 3548e72..ed9a3a1 100644 --- a/internal/app/runner_terminal_failure_test.go +++ b/internal/app/runner_terminal_failure_test.go @@ -55,6 +55,7 @@ func TestPersistSessionTerminalFailurePreservesCauseAndSessionAuthority(t *testi err := persistSessionTerminalFailure( context.Background(), sessionStore, "session.json", sessionManifest, tc.cause, + time.Now().UTC(), ) if !errors.Is(err, tc.cause) { t.Fatalf("error = %v, want original cause %v", err, tc.cause) diff --git a/internal/app/runner_test.go b/internal/app/runner_test.go index e030cba..d7af60c 100644 --- a/internal/app/runner_test.go +++ b/internal/app/runner_test.go @@ -71,6 +71,59 @@ func TestExecutePlanPreservesSessionLockReleaseErrors(t *testing.T) { } } +func TestExecutePlanUsesInvocationClockAndRunID(t *testing.T) { + cfg := testConfig(t) + fixed := time.Date(2026, time.September, 4, 1, 2, 3, 0, time.UTC) + runID := "20260904T010203Z-a1b2c3d4" + summary, err := executePlanWithDependencies( + context.Background(), + cfg, + BoundedPlan{stages: []stage.Stage{resultStage{name: "prepare"}}}, + RunOptions{}, + invocationDependencies{ + now: func() time.Time { return fixed }, + newRunID: func() (string, error) { return runID, nil }, + }, + ) + if err != nil { + t.Fatalf("executePlanWithDependencies() error = %v", err) + } + if summary.RunID != runID { + t.Fatalf("summary run id = %q, want %q", summary.RunID, runID) + } + persisted, err := (&manifest.LocalStore{}).Load(context.Background(), summary.ManifestPath) + if err != nil { + t.Fatalf("load manifest: %v", err) + } + record := persisted.Stages["prepare"] + if record == nil || record.StartedAt == nil || !record.StartedAt.Equal(fixed) || record.CompletedAt == nil || !record.CompletedAt.Equal(fixed) { + t.Fatalf("prepare timestamps = %#v, want invocation clock %v", record, fixed) + } +} + +func TestExecutePlanDoesNotConstructUnselectedObjectStore(t *testing.T) { + cfg := testConfig(t) + constructed := false + _, err := executePlanWithDependencies( + context.Background(), + cfg, + BoundedPlan{stages: []stage.Stage{resultStage{name: "prepare"}}}, + RunOptions{}, + invocationDependencies{ + newObjectStore: func(context.Context, *config.Config) (storage.ObjectStore, error) { + constructed = true + return nil, errors.New("must not construct object store") + }, + }, + ) + if err != nil { + t.Fatalf("executePlanWithDependencies() error = %v", err) + } + if constructed { + t.Fatal("object store was constructed for an invocation that does not require it") + } +} + type countingStage struct { name string runs *int diff --git a/internal/app/time.go b/internal/app/time.go index ac4aa5c..ab62b44 100644 --- a/internal/app/time.go +++ b/internal/app/time.go @@ -2,6 +2,6 @@ package app import "time" -var nowUTC = func() time.Time { +func nowUTC() time.Time { return time.Now().UTC() }