From e0f19833a120a04b54c40c4f3554340af7e93a76 Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Fri, 4 Sep 2026 01:05:44 +0000 Subject: [PATCH] Establish explicit runner lifecycle state --- docs/roadmap/implementation.md | 2 +- internal/app/runner.go | 141 +++++--------------------- internal/app/runner_runtime.go | 175 +++++++++++++++++++++++++++++++++ internal/app/runner_test.go | 40 ++++++++ 4 files changed, 240 insertions(+), 118 deletions(-) create mode 100644 internal/app/runner_runtime.go diff --git a/docs/roadmap/implementation.md b/docs/roadmap/implementation.md index 3376435..cc3f53f 100644 --- a/docs/roadmap/implementation.md +++ b/docs/roadmap/implementation.md @@ -187,7 +187,7 @@ runtime outcomes. ## Stage 2 — Establish The Invocation Runtime And Lock Lifecycle -**Status: Pending** +**Status: Completed** ### Goal diff --git a/internal/app/runner.go b/internal/app/runner.go index bc189e1..6c774ae 100644 --- a/internal/app/runner.go +++ b/internal/app/runner.go @@ -4,7 +4,6 @@ import ( "context" "errors" "fmt" - "log/slog" "os" "strings" @@ -17,8 +16,6 @@ import ( "gitea.maximumdirect.net/eric/narratio/internal/artifactmodel" "gitea.maximumdirect.net/eric/narratio/internal/artifacts" "gitea.maximumdirect.net/eric/narratio/internal/config" - "gitea.maximumdirect.net/eric/narratio/internal/fileops" - "gitea.maximumdirect.net/eric/narratio/internal/logging" "gitea.maximumdirect.net/eric/narratio/internal/manifest" "gitea.maximumdirect.net/eric/narratio/internal/stage" ) @@ -42,74 +39,20 @@ type RunSummary struct { var executeStagesFn = executePlan func executePlan(ctx context.Context, cfg *config.Config, plan BoundedPlan, opts RunOptions) (summary *RunSummary, resultErr error) { - stages := plan.Stages() - var prerequisiteStore manifest.Store - if opts.Env != nil { - prerequisiteStore = opts.Env.ManifestStore - } - if err := inspectBoundedPrerequisites(ctx, cfg, plan, prerequisiteStore); err != nil { - return nil, fmt.Errorf("validate bounded run prerequisites: %w", err) - } - effectiveArtifacts := opts.EffectiveArtifacts - if !effectiveArtifacts.Resolved() && cfg != nil && cfg.Pipeline != nil && cfg.Pipeline.Scriptorium != nil { - var err error - effectiveArtifacts, err = resolveEffectiveArtifacts(cfg, nil) - if err != nil { - return nil, fmt.Errorf("resolve effective artifacts: %w", err) - } - } - if !effectiveArtifacts.Resolved() { - var err error - effectiveArtifacts, err = artifacts.ResolveEffectiveArtifactSet(nil, nil) - if err != nil { - return nil, fmt.Errorf("resolve default effective artifacts: %w", err) - } - } - runID, err := artifacts.NewRunID() + runtime, err := prepareInvocation(ctx, cfg, plan, opts) if err != nil { - return nil, fmt.Errorf("generate run id: %w", err) + return nil, err } - identity, err := resolveInvocationIdentity(cfg, runID) - if err != nil { - return nil, fmt.Errorf("resolve invocation identity: %w", err) + if err := prepareInvocationLayout(runtime); err != nil { + return nil, err } - env := opts.Env - if env == nil { - env = &Env{} - } - // The resolved configuration is the invocation's single source of truth. - // Injected environments supply collaborators, never an alternate config. - env.Config = cfg - env.SelectedArtifactKeys = append([]string(nil), opts.SelectedArtifacts...) - env.EffectiveArtifacts = effectiveArtifacts - if env.ManifestStore == nil { - env.ManifestStore = &manifest.LocalStore{} - } - manifestPath := artifacts.SessionManifestPathForCampaign( - cfg.Pipeline.Workspace.Root, - identity.Campaign, - identity.SessionID, - ) - if env.ArtifactStore == nil { - env.ArtifactStore = artifacts.NewLocalStore(cfg.Pipeline.Workspace.Root) - } - if env.Logger == nil { - env.Logger = logging.NewLogger(os.Stderr, slog.LevelInfo) - } - - artifactStore := env.ArtifactStore - paths, err := artifactStore.EnsureLayoutFor(identity.Campaign, identity.SessionID) - if err != nil { - return nil, fmt.Errorf("prepare workdir: %w", err) - } - - lock, err := artifactStore.AcquireSessionLockForContext(ctx, identity.Campaign, identity.SessionID) + lock, err := runtime.artifactStore.AcquireSessionLockForContext(ctx, runtime.identity.Campaign, runtime.identity.SessionID) if err != nil { return nil, fmt.Errorf("acquire session lock: %w", err) } defer func() { - if releaseErr := artifactStore.ReleaseSessionLock(lock); releaseErr != nil { + if releaseErr := runtime.artifactStore.ReleaseSessionLock(lock); releaseErr != nil { if resultErr == nil { resultErr = fmt.Errorf("release session lock: %w", releaseErr) } else { @@ -118,50 +61,17 @@ func executePlan(ctx context.Context, cfg *config.Config, plan BoundedPlan, opts } }() - if paths.ManifestPath != manifestPath { - return nil, fmt.Errorf("prepared manifest path %q does not match resolved manifest path %q", paths.ManifestPath, manifestPath) - } - if err := requireCompleteRestore(paths); err != nil { + if err := initializeLockedInvocation(runtime); err != nil { return nil, err } - m, present, err := loadManifestAtPathIfPresent(ctx, env.ManifestStore, manifestPath) - if err != nil { - return nil, err - } - if present { - if err := identity.validateSessionManifest(m); err != nil { - return nil, err - } - } else { - m, err = env.ManifestStore.Create(ctx, identity.SessionID) - if err != nil { - return nil, fmt.Errorf("create manifest: %w", err) - } - } - // The preflight prerequisite inspection avoids creating run state for an - // already-invalid request. Recheck the manifest protected by the session - // lock because another invocation may have changed prerequisite state while - // this invocation waited to acquire the lock. - if err := validateBoundedPrerequisites(plan, m); err != nil { - return nil, fmt.Errorf("validate bounded run prerequisites under session lock: %w", err) - } - identity.applyToSessionManifest(m) - applyEffectiveConfigProvenance(m, cfg) - if err := env.ManifestStore.Save(ctx, manifestPath, m); err != nil { - return nil, fmt.Errorf("save manifest identity %q: %w", manifestPath, err) - } - if err := fileops.EnsureWorkspaceDirectory(identity.LocalWorkDir); err != nil { - return nil, persistSessionTerminalFailure( - ctx, env.ManifestStore, manifestPath, m, - fmt.Errorf("create run directory %q: %w", identity.LocalWorkDir, err), - ) - } - if _, err := loadSecretsFromConfig(env.Config, env.Logger); err != nil { - return nil, persistSessionTerminalFailure( - ctx, env.ManifestStore, manifestPath, m, - fmt.Errorf("load secrets from files: %w", 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 { @@ -231,12 +141,9 @@ func executePlan(ctx context.Context, cfg *config.Config, plan BoundedPlan, opts stageEnv := env - runNames := make([]string, 0, len(stages)) - executed := make([]string, 0, len(stages)) - skipped := make([]string, 0, len(stages)) for _, s := range stages { stageEnv.Force = opts.Force - runNames = append(runNames, s.Name()) + runtime.stageNames = append(runtime.stageNames, s.Name()) semanticConfig, err := currentStageSemanticConfig(s, stageEnv) if err != nil { return nil, persistSessionTerminalFailure( @@ -274,11 +181,11 @@ func executePlan(ctx context.Context, cfg *config.Config, plan BoundedPlan, opts } if decision.action == stageActionSkip { - skipped = append(skipped, s.Name()) + runtime.skipped = append(runtime.skipped, s.Name()) env.Logger.Info("skipping stage", "stage", s.Name(), "reason", "already_succeeded", "force", opts.Force) continue } - executed = append(executed, s.Name()) + runtime.executed = append(runtime.executed, s.Name()) priorOutcome := capturePriorStageOutcome(m, s.Name()) priorAnalyzeState := captureAnalyzeState(m, s.Name()) @@ -332,7 +239,7 @@ func executePlan(ctx context.Context, cfg *config.Config, plan BoundedPlan, opts ) } if result != nil && result.Disposition == stage.StageDispositionSkipped { - skipped = append(skipped, s.Name()) + runtime.skipped = append(runtime.skipped, s.Name()) skippedAt := nowUTC() if _, err := applyCompletedStageOutcome(m, s.Name(), result, nil, semanticConfig, priorOutcome, skippedAt); err != nil { return nil, persistSessionTerminalFailure( @@ -383,8 +290,8 @@ func executePlan(ctx context.Context, cfg *config.Config, plan BoundedPlan, opts env.Logger.Info("stage succeeded", "stage", s.Name()) } - if containsStage(executed, "publish") { - if err := runPostPublishCleanup(ctx, env, manifestPath, m, executed); err != nil { + if containsStage(runtime.executed, "publish") { + if err := runPostPublishCleanup(ctx, env, manifestPath, m, runtime.executed); err != nil { return nil, persistPostPublishCleanupFailure( ctx, env.ManifestStore, manifestPath, m, fmt.Errorf("post-publish cleanup incomplete: %w", err), @@ -396,9 +303,9 @@ func executePlan(ctx context.Context, cfg *config.Config, plan BoundedPlan, opts SessionID: cfg.Session.SessionID, RunID: runID, ManifestPath: manifestPath, - StageNames: runNames, - Executed: executed, - Skipped: skipped, + StageNames: runtime.stageNames, + Executed: runtime.executed, + Skipped: runtime.skipped, }, nil } diff --git a/internal/app/runner_runtime.go b/internal/app/runner_runtime.go new file mode 100644 index 0000000..00244c8 --- /dev/null +++ b/internal/app/runner_runtime.go @@ -0,0 +1,175 @@ +package app + +import ( + "context" + "fmt" + "log/slog" + "os" + + "gitea.maximumdirect.net/eric/narratio/internal/artifacts" + "gitea.maximumdirect.net/eric/narratio/internal/config" + "gitea.maximumdirect.net/eric/narratio/internal/fileops" + "gitea.maximumdirect.net/eric/narratio/internal/logging" + "gitea.maximumdirect.net/eric/narratio/internal/manifest" + "gitea.maximumdirect.net/eric/narratio/internal/stage" +) + +type invocationRuntime struct { + ctx context.Context + cfg *config.Config + plan BoundedPlan + stages []stage.Stage + opts RunOptions + effectiveArtifacts artifacts.EffectiveArtifactSet + identity invocationIdentity + paths artifacts.SessionPaths + env *stage.Env + manifestStore manifest.Store + artifactStore artifacts.Store + manifestPath string + manifest *manifest.Manifest + stageNames []string + executed []string + skipped []string +} + +func prepareInvocation(ctx context.Context, cfg *config.Config, plan BoundedPlan, opts RunOptions) (*invocationRuntime, error) { + var prerequisiteStore manifest.Store + if opts.Env != nil { + prerequisiteStore = opts.Env.ManifestStore + } + if err := inspectBoundedPrerequisites(ctx, cfg, plan, prerequisiteStore); err != nil { + return nil, fmt.Errorf("validate bounded run prerequisites: %w", err) + } + + effectiveArtifacts := opts.EffectiveArtifacts + if !effectiveArtifacts.Resolved() && cfg != nil && cfg.Pipeline != nil && cfg.Pipeline.Scriptorium != nil { + var err error + effectiveArtifacts, err = resolveEffectiveArtifacts(cfg, nil) + if err != nil { + return nil, fmt.Errorf("resolve effective artifacts: %w", err) + } + } + if !effectiveArtifacts.Resolved() { + var err error + effectiveArtifacts, err = artifacts.ResolveEffectiveArtifactSet(nil, nil) + if err != nil { + return nil, fmt.Errorf("resolve default effective artifacts: %w", err) + } + } + + runID, err := artifacts.NewRunID() + if err != nil { + return nil, fmt.Errorf("generate run id: %w", err) + } + identity, err := resolveInvocationIdentity(cfg, runID) + if err != nil { + return nil, fmt.Errorf("resolve invocation identity: %w", err) + } + + env := opts.Env + if env == nil { + env = &stage.Env{} + } + // The resolved configuration is the invocation's single source of truth. + // Injected environments supply collaborators, never alternate invocation data. + env.Config = cfg + env.SelectedArtifactKeys = append([]string(nil), opts.SelectedArtifacts...) + env.EffectiveArtifacts = effectiveArtifacts + if env.ManifestStore == nil { + env.ManifestStore = &manifest.LocalStore{} + } + if env.ArtifactStore == nil { + env.ArtifactStore = artifacts.NewLocalStore(cfg.Pipeline.Workspace.Root) + } + if env.Logger == nil { + env.Logger = logging.NewLogger(os.Stderr, slog.LevelInfo) + } + + normalizedOpts := opts + normalizedOpts.SelectedArtifacts = append([]string(nil), opts.SelectedArtifacts...) + normalizedOpts.EffectiveArtifacts = effectiveArtifacts + normalizedOpts.Env = env + stages := plan.Stages() + return &invocationRuntime{ + ctx: ctx, + cfg: cfg, + plan: plan, + stages: stages, + opts: normalizedOpts, + effectiveArtifacts: effectiveArtifacts, + identity: identity, + env: env, + manifestStore: env.ManifestStore, + artifactStore: env.ArtifactStore, + manifestPath: artifacts.SessionManifestPathForCampaign( + cfg.Pipeline.Workspace.Root, + identity.Campaign, + identity.SessionID, + ), + stageNames: make([]string, 0, len(stages)), + executed: make([]string, 0, len(stages)), + skipped: make([]string, 0, len(stages)), + }, nil +} + +func prepareInvocationLayout(runtime *invocationRuntime) error { + paths, err := runtime.artifactStore.EnsureLayoutFor(runtime.identity.Campaign, runtime.identity.SessionID) + if err != nil { + return fmt.Errorf("prepare workdir: %w", err) + } + if paths.ManifestPath != runtime.manifestPath { + return fmt.Errorf("prepared manifest path %q does not match resolved manifest path %q", paths.ManifestPath, runtime.manifestPath) + } + runtime.paths = paths + return nil +} + +func initializeLockedInvocation(runtime *invocationRuntime) error { + if err := requireCompleteRestore(runtime.paths); err != nil { + return err + } + m, present, err := loadManifestAtPathIfPresent(runtime.ctx, runtime.manifestStore, runtime.manifestPath) + if err != nil { + return err + } + if present { + if err := runtime.identity.validateSessionManifest(m); err != nil { + return err + } + } else { + m, err = runtime.manifestStore.Create(runtime.ctx, runtime.identity.SessionID) + if err != nil { + return fmt.Errorf("create manifest: %w", err) + } + } + runtime.manifest = m + + // The unlocked inspection avoids creating invocation state for an invalid + // request. Recheck while holding the session lock in case prior state changed. + if err := validateBoundedPrerequisites(runtime.plan, m); err != nil { + return fmt.Errorf("validate bounded run prerequisites under session lock: %w", err) + } + runtime.identity.applyToSessionManifest(m) + applyEffectiveConfigProvenance(m, runtime.cfg) + if err := runtime.manifestStore.Save(runtime.ctx, runtime.manifestPath, m); err != nil { + return fmt.Errorf("save manifest identity %q: %w", runtime.manifestPath, err) + } + if err := fileops.EnsureWorkspaceDirectory(runtime.identity.LocalWorkDir); err != nil { + return runtime.persistTerminalFailure(fmt.Errorf("create run directory %q: %w", runtime.identity.LocalWorkDir, err)) + } + if _, err := loadSecretsFromConfig(runtime.env.Config, runtime.env.Logger); err != nil { + return runtime.persistTerminalFailure(fmt.Errorf("load secrets from files: %w", err)) + } + return nil +} + +func (runtime *invocationRuntime) persistTerminalFailure(err error) error { + return persistSessionTerminalFailure( + runtime.ctx, + runtime.manifestStore, + runtime.manifestPath, + runtime.manifest, + err, + ) +} diff --git a/internal/app/runner_test.go b/internal/app/runner_test.go index f73beb1..e030cba 100644 --- a/internal/app/runner_test.go +++ b/internal/app/runner_test.go @@ -26,11 +26,51 @@ type failingStage struct { err error } +type releaseErrorArtifactStore struct { + artifacts.Store + err error +} + +func (s *releaseErrorArtifactStore) ReleaseSessionLock(lock *artifacts.LockHandle) error { + return errors.Join(s.Store.ReleaseSessionLock(lock), s.err) +} + func (s failingStage) Name() string { return s.name } func (s failingStage) Run(_ context.Context, _ *stage.Env, _ *manifest.Manifest) (*stage.StageResult, error) { return nil, s.err } +func TestExecutePlanPreservesSessionLockReleaseErrors(t *testing.T) { + releaseErr := errors.New("release failed") + executionErr := errors.New("execution failed") + tests := []struct { + name string + selectedStage stage.Stage + wantExecution bool + }{ + {name: "release is final error", selectedStage: resultStage{name: "prepare"}}, + {name: "release joins execution error", selectedStage: failingStage{name: "prepare", err: executionErr}, wantExecution: true}, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + cfg := testConfig(t) + store := &releaseErrorArtifactStore{ + Store: artifacts.NewLocalStore(cfg.Pipeline.Workspace.Root), + err: releaseErr, + } + _, err := executePlan(context.Background(), cfg, BoundedPlan{stages: []stage.Stage{test.selectedStage}}, RunOptions{ + Env: &Env{ArtifactStore: store}, + }) + if !errors.Is(err, releaseErr) { + t.Fatalf("executePlan() error = %v, want release error", err) + } + if got := errors.Is(err, executionErr); got != test.wantExecution { + t.Fatalf("executePlan() contains execution error = %t, want %t", got, test.wantExecution) + } + }) + } +} + type countingStage struct { name string runs *int