diff --git a/docs/roadmap/implementation.md b/docs/roadmap/implementation.md index b9b7c5e..049b405 100644 --- a/docs/roadmap/implementation.md +++ b/docs/roadmap/implementation.md @@ -299,7 +299,7 @@ selective, testable per invocation, and independent of stage transitions. ## Stage 4 — Extract One-Stage Execution -**Status: Pending** +**Status: Completed** ### Goal diff --git a/internal/app/runner.go b/internal/app/runner.go index 15ec97c..fd1fea8 100644 --- a/internal/app/runner.go +++ b/internal/app/runner.go @@ -71,144 +71,22 @@ func executePlanWithDependencies(ctx context.Context, cfg *config.Config, plan B return nil, err } - stages := runtime.stages env := runtime.env manifestPath := runtime.manifestPath m := runtime.manifest 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) + for _, selected := range runtime.stages { + runtime.stageNames = append(runtime.stageNames, selected.Name()) + outcome, err := executeSelectedStage(runtime, selected) if err != nil { - return nil, runtime.persistTerminalFailure( - fmt.Errorf("fingerprint semantic configuration for stage %q: %w", s.Name(), err), - ) + return nil, err } - decision := decideStageExecution(recordedStageStatus(m, s.Name()), opts.Force, nil) - - if decision.action == stageActionSkip { - validation, err := evaluateStageResume(ctx, s, stageEnv, m, semanticConfig) - if err != nil { - 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 := runtime.dependencies.now() - if _, err := applyPreExecutionDecision(m, s.Name(), decision, staleAt); err != nil { - 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, runtime.persistTerminalFailure( - fmt.Errorf("save manifest after resume validation for stage %q: %w", s.Name(), err), - ) - } - env.Logger.Info("stage result is not resumable", "stage", s.Name(), "reason", decision.selectedStaleReason) - } + if outcome.executed { + runtime.executed = append(runtime.executed, selected.Name()) } - - if decision.action == stageActionSkip { - runtime.skipped = append(runtime.skipped, s.Name()) - env.Logger.Info("skipping stage", "stage", s.Name(), "reason", "already_succeeded", "force", opts.Force) - continue + if outcome.skipped { + runtime.skipped = append(runtime.skipped, selected.Name()) } - runtime.executed = append(runtime.executed, s.Name()) - priorOutcome := capturePriorStageOutcome(m, s.Name()) - priorAnalyzeState := captureAnalyzeState(m, s.Name()) - - now := runtime.dependencies.now() - if _, err := beginStageAttempt(m, s.Name(), decision, now); err != nil { - 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(), 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, runtime.persistTerminalFailure(operationErr) - } - env.Logger.Debug("manifest saved", "stage", s.Name(), "transition", "running", "path", manifestPath) - - result, err := s.Run(ctx, stageEnv, m) - var analyzeProjection *validatedAnalyzeProjection - if err == nil { - err = validateStageResult(result) - if err == nil { - analyzeProjection, err = validateSuccessfulAnalyzeProjection(s.Name(), result) - } - } else if result != nil && result.AnalyzeState != nil { - var projectionErr error - analyzeProjection, projectionErr = validateFailedAnalyzeProjection(s.Name(), result) - if projectionErr != nil { - err = errors.Join(err, projectionErr) - } - } - if err != nil { - if analyzeProjection != nil { - applyAnalyzeProjection(m, analyzeProjection) - } - 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, 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 := runtime.dependencies.now() - if _, err := applyCompletedStageOutcome(m, s.Name(), result, nil, semanticConfig, priorOutcome, skippedAt); err != nil { - 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, runtime.persistTerminalFailure( - fmt.Errorf("save manifest after self-skip %q: %w", s.Name(), err), - ) - } - env.Logger.Debug("manifest saved", "stage", s.Name(), "transition", "skipped", "path", manifestPath) - env.Logger.Info("stage skipped", "stage", s.Name(), "reason", result.SkipReason) - continue - } - - sessionOutputs := mapResultOutputs(s.Name(), result, runID) - if analyzeProjection != nil { - sessionOutputs = analyzeProjectionOutputs(analyzeProjection.session) - } - succeededAt := runtime.dependencies.now() - if _, err := applyCompletedStageOutcome(m, s.Name(), result, sessionOutputs, semanticConfig, priorOutcome, succeededAt); err != nil { - return nil, runtime.persistTerminalFailure( - fmt.Errorf("invalidate dependents after changed stage %q result: %w", s.Name(), err), - ) - } - applyAnalyzeProjection(m, analyzeProjection) - - if err := env.ManifestStore.Save(ctx, manifestPath, m); err != nil { - operationErr := fmt.Errorf("save manifest after stage %q: %w", s.Name(), err) - if analyzeProjection != nil { - restoreAnalyzeState(m, priorAnalyzeState) - 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, 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") { diff --git a/internal/app/runner_stage_execution.go b/internal/app/runner_stage_execution.go new file mode 100644 index 0000000..4d04977 --- /dev/null +++ b/internal/app/runner_stage_execution.go @@ -0,0 +1,148 @@ +package app + +import ( + "errors" + "fmt" + + "gitea.maximumdirect.net/eric/narratio/internal/stage" +) + +type stageExecutionOutcome struct { + executed bool + skipped bool +} + +func executeSelectedStage(runtime *invocationRuntime, selected stage.Stage) (stageExecutionOutcome, error) { + env := runtime.env + m := runtime.manifest + env.Force = runtime.opts.Force + + semanticConfig, err := currentStageSemanticConfig(selected, env) + if err != nil { + return stageExecutionOutcome{}, runtime.persistTerminalFailure( + fmt.Errorf("fingerprint semantic configuration for stage %q: %w", selected.Name(), err), + ) + } + decision := decideStageExecution(recordedStageStatus(m, selected.Name()), runtime.opts.Force, nil) + + if decision.action == stageActionSkip { + validation, err := evaluateStageResume(runtime.ctx, selected, env, m, semanticConfig) + if err != nil { + return stageExecutionOutcome{}, runtime.persistTerminalFailure( + fmt.Errorf("validate resume for stage %q: %w", selected.Name(), err), + ) + } + decision = decideStageExecution(recordedStageStatus(m, selected.Name()), runtime.opts.Force, validation) + if decision.markSelectedStale { + staleAt := runtime.dependencies.now() + if _, err := applyPreExecutionDecision(m, selected.Name(), decision, staleAt); err != nil { + return stageExecutionOutcome{}, runtime.persistTerminalFailure( + fmt.Errorf("invalidate dependents after resume validation for stage %q: %w", selected.Name(), err), + ) + } + if err := runtime.manifestStore.Save(runtime.ctx, runtime.manifestPath, m); err != nil { + return stageExecutionOutcome{}, runtime.persistTerminalFailure( + fmt.Errorf("save manifest after resume validation for stage %q: %w", selected.Name(), err), + ) + } + env.Logger.Info("stage result is not resumable", "stage", selected.Name(), "reason", decision.selectedStaleReason) + } + } + + if decision.action == stageActionSkip { + env.Logger.Info("skipping stage", "stage", selected.Name(), "reason", "already_succeeded", "force", runtime.opts.Force) + return stageExecutionOutcome{skipped: true}, nil + } + + priorOutcome := capturePriorStageOutcome(m, selected.Name()) + priorAnalyzeState := captureAnalyzeState(m, selected.Name()) + attempted := stageExecutionOutcome{executed: true} + startedAt := runtime.dependencies.now() + if _, err := beginStageAttempt(m, selected.Name(), decision, startedAt); err != nil { + return stageExecutionOutcome{}, runtime.persistTerminalFailure( + fmt.Errorf("invalidate dependents before forced stage %q: %w", selected.Name(), err), + ) + } + env.Logger.Info("starting stage", "stage", selected.Name()) + if err := runtime.manifestStore.Save(runtime.ctx, runtime.manifestPath, m); err != nil { + operationErr := fmt.Errorf("save manifest before stage %q: %w", selected.Name(), err) + m.MarkStageFailed(selected.Name(), runtime.dependencies.now(), operationErr.Error()) + if _, invalidationErr := invalidateDependentSucceededStagesWithReason(m, selected.Name(), runtime.dependencies.now(), staleReasonFailure); invalidationErr != nil { + operationErr = errors.Join(operationErr, fmt.Errorf("invalidate dependents after stage %q persistence failure: %w", selected.Name(), invalidationErr)) + } + return stageExecutionOutcome{}, runtime.persistTerminalFailure(operationErr) + } + env.Logger.Debug("manifest saved", "stage", selected.Name(), "transition", "running", "path", runtime.manifestPath) + + result, err := selected.Run(runtime.ctx, env, m) + var analyzeProjection *validatedAnalyzeProjection + if err == nil { + err = validateStageResult(result) + if err == nil { + analyzeProjection, err = validateSuccessfulAnalyzeProjection(selected.Name(), result) + } + } else if result != nil && result.AnalyzeState != nil { + var projectionErr error + analyzeProjection, projectionErr = validateFailedAnalyzeProjection(selected.Name(), result) + if projectionErr != nil { + err = errors.Join(err, projectionErr) + } + } + if err != nil { + if analyzeProjection != nil { + applyAnalyzeProjection(m, analyzeProjection) + } + failedAt := runtime.dependencies.now() + m.MarkStageFailed(selected.Name(), failedAt, err.Error()) + if _, invalidationErr := invalidateDependentSucceededStagesWithReason(m, selected.Name(), failedAt, staleReasonFailure); invalidationErr != nil { + err = errors.Join(err, fmt.Errorf("invalidate dependents after stage %q failure: %w", selected.Name(), invalidationErr)) + } + env.Logger.Info("stage failed", "stage", selected.Name(), "error", err) + return stageExecutionOutcome{}, runtime.persistTerminalFailure(fmt.Errorf("stage %q failed: %w", selected.Name(), err)) + } + if result != nil && result.Disposition == stage.StageDispositionSkipped { + skippedAt := runtime.dependencies.now() + if _, err := applyCompletedStageOutcome(m, selected.Name(), result, nil, semanticConfig, priorOutcome, skippedAt); err != nil { + return stageExecutionOutcome{}, runtime.persistTerminalFailure( + fmt.Errorf("invalidate dependents after stage %q self-skip: %w", selected.Name(), err), + ) + } + if err := runtime.manifestStore.Save(runtime.ctx, runtime.manifestPath, m); err != nil { + return stageExecutionOutcome{}, runtime.persistTerminalFailure( + fmt.Errorf("save manifest after self-skip %q: %w", selected.Name(), err), + ) + } + env.Logger.Debug("manifest saved", "stage", selected.Name(), "transition", "skipped", "path", runtime.manifestPath) + env.Logger.Info("stage skipped", "stage", selected.Name(), "reason", result.SkipReason) + attempted.skipped = true + return attempted, nil + } + + outputs := mapResultOutputs(selected.Name(), result, runtime.identity.RunID) + if analyzeProjection != nil { + outputs = analyzeProjectionOutputs(analyzeProjection.session) + } + succeededAt := runtime.dependencies.now() + if _, err := applyCompletedStageOutcome(m, selected.Name(), result, outputs, semanticConfig, priorOutcome, succeededAt); err != nil { + return stageExecutionOutcome{}, runtime.persistTerminalFailure( + fmt.Errorf("invalidate dependents after changed stage %q result: %w", selected.Name(), err), + ) + } + applyAnalyzeProjection(m, analyzeProjection) + + if err := runtime.manifestStore.Save(runtime.ctx, runtime.manifestPath, m); err != nil { + operationErr := fmt.Errorf("save manifest after stage %q: %w", selected.Name(), err) + if analyzeProjection != nil { + restoreAnalyzeState(m, priorAnalyzeState) + failedAt := runtime.dependencies.now() + m.MarkStageFailed(selected.Name(), failedAt, operationErr.Error()) + if _, invalidationErr := invalidateDependentSucceededStagesWithReason(m, selected.Name(), failedAt, staleReasonFailure); invalidationErr != nil { + operationErr = errors.Join(operationErr, fmt.Errorf("invalidate dependents after analyze projection persistence failure: %w", invalidationErr)) + } + } + return stageExecutionOutcome{}, runtime.persistTerminalFailure(operationErr) + } + env.Logger.Debug("manifest saved", "stage", selected.Name(), "transition", "succeeded", "path", runtime.manifestPath) + env.Logger.Info("stage succeeded", "stage", selected.Name()) + return attempted, nil +}