Add artifact-aware analyze resume planning
This commit is contained in:
@@ -2,19 +2,21 @@ package app
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"flag"
|
||||
"fmt"
|
||||
"io"
|
||||
"log/slog"
|
||||
"os"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"gitea.maximumdirect.net/eric/narratio/internal/artifacts"
|
||||
"gitea.maximumdirect.net/eric/narratio/internal/config"
|
||||
"gitea.maximumdirect.net/eric/narratio/internal/logging"
|
||||
"gitea.maximumdirect.net/eric/narratio/internal/manifest"
|
||||
"gitea.maximumdirect.net/eric/narratio/internal/stage"
|
||||
)
|
||||
|
||||
// Plan validates configuration, prepares the local workdir, and prints stage order.
|
||||
// Plan validates configuration and prints a read-only execution preview.
|
||||
func Plan(ctx context.Context, args []string, out io.Writer) error {
|
||||
request, err := parseBoundedRunRequest("plan", args, out)
|
||||
if err != nil {
|
||||
@@ -33,7 +35,8 @@ func Plan(ctx context.Context, args []string, out io.Writer) error {
|
||||
if err := config.Validate(cfg); err != nil {
|
||||
return fmt.Errorf("plan: %w", err)
|
||||
}
|
||||
if _, err := resolveEffectiveArtifacts(cfg, request.SelectedArtifacts); err != nil {
|
||||
effective, err := resolveEffectiveArtifacts(cfg, request.SelectedArtifacts)
|
||||
if err != nil {
|
||||
return fmt.Errorf("plan: %w", err)
|
||||
}
|
||||
m, err := loadManifestIfPresent(ctx, cfg)
|
||||
@@ -43,33 +46,64 @@ func Plan(ctx context.Context, args []string, out io.Writer) error {
|
||||
if err := validateBoundedPrerequisites(request.Plan, m); err != nil {
|
||||
return fmt.Errorf("plan: %w", err)
|
||||
}
|
||||
if _, err := loadSecretsFromConfig(cfg, logging.NewLogger(os.Stderr, slog.LevelInfo)); err != nil {
|
||||
return fmt.Errorf("plan: %w", err)
|
||||
}
|
||||
|
||||
store := artifacts.NewLocalStore(cfg.Pipeline.Workspace.Root)
|
||||
paths, err := store.EnsureLayoutFor(cfg.Session.Campaign, cfg.Session.SessionID)
|
||||
paths := store.SessionPathsFor(cfg.Session.Campaign, cfg.Session.SessionID)
|
||||
model, err := cloneManifestForPlan(m, cfg)
|
||||
if err != nil {
|
||||
return fmt.Errorf("plan: prepare workdir: %w", err)
|
||||
return fmt.Errorf("plan: clone session state: %w", err)
|
||||
}
|
||||
|
||||
stages := request.Plan.Stages()
|
||||
decisions := decideStageActions(stages, m, request.Force)
|
||||
stageEnv := &stage.Env{
|
||||
Config: cfg, SelectedArtifactKeys: append([]string(nil), request.SelectedArtifacts...),
|
||||
EffectiveArtifacts: effective, ArtifactStore: store, Force: request.Force,
|
||||
}
|
||||
|
||||
runCount := 0
|
||||
skipCount := 0
|
||||
if _, err := fmt.Fprintf(out, "narratio session plan: workdir prepared at %s\n", paths.Root); err != nil {
|
||||
if _, err := fmt.Fprintf(out, "narratio session plan: read-only workdir at %s\n", paths.Root); err != nil {
|
||||
return err
|
||||
}
|
||||
for _, d := range decisions {
|
||||
if d.Action == stageActionRun {
|
||||
for _, selectedStage := range stages {
|
||||
action := decideStageAction(selectedStage, model, request.Force)
|
||||
var validation *stage.ResumeValidation
|
||||
if validator, ok := selectedStage.(stage.ResumeValidator); ok &&
|
||||
(action == stageActionSkip || selectedStage.Name() == "analyze") {
|
||||
checked, validationErr := validator.ValidateResume(ctx, stageEnv, model)
|
||||
if validationErr != nil {
|
||||
return fmt.Errorf("plan: validate resume for stage %q: %w", selectedStage.Name(), validationErr)
|
||||
}
|
||||
checked = checked.Normalized()
|
||||
validation = &checked
|
||||
if action == stageActionSkip && !checked.Resumable {
|
||||
at := time.Now().UTC()
|
||||
model.MarkStageStale(selectedStage.Name(), at, checked.Reason)
|
||||
if _, invalidationErr := invalidateDependentSucceededStagesWithReason(
|
||||
model, selectedStage.Name(), at, staleReasonNotResumable,
|
||||
); invalidationErr != nil {
|
||||
return fmt.Errorf("plan: model resume invalidation for stage %q: %w", selectedStage.Name(), invalidationErr)
|
||||
}
|
||||
action = stageActionRun
|
||||
}
|
||||
}
|
||||
if action == stageActionRun {
|
||||
runCount++
|
||||
} else {
|
||||
skipCount++
|
||||
}
|
||||
if _, err := fmt.Fprintf(out, "%s: %s\n", d.Stage.Name(), d.Action); err != nil {
|
||||
if _, err := fmt.Fprintf(out, "%s: %s\n", selectedStage.Name(), action); err != nil {
|
||||
return err
|
||||
}
|
||||
if validation != nil && validation.Analyze != nil {
|
||||
if err := writeAnalyzePlanDetails(out, validation.Analyze); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
if action == stageActionRun {
|
||||
if err := modelPlannedStageRun(model, selectedStage, cfg, request.Force); err != nil {
|
||||
return fmt.Errorf("plan: model stage %q: %w", selectedStage.Name(), err)
|
||||
}
|
||||
}
|
||||
}
|
||||
if _, err := fmt.Fprintf(out, "totals: run=%d skip=%d\n", runCount, skipCount); err != nil {
|
||||
return err
|
||||
@@ -77,3 +111,102 @@ func Plan(ctx context.Context, args []string, out io.Writer) error {
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func cloneManifestForPlan(source *manifest.Manifest, cfg *config.Config) (*manifest.Manifest, error) {
|
||||
if source == nil {
|
||||
created := manifest.New(cfg.Session.SessionID, time.Now().UTC())
|
||||
created.Campaign = cfg.Session.Campaign
|
||||
return created, nil
|
||||
}
|
||||
data, err := json.Marshal(source)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var cloned manifest.Manifest
|
||||
if err := json.Unmarshal(data, &cloned); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &cloned, nil
|
||||
}
|
||||
|
||||
func modelPlannedStageRun(model *manifest.Manifest, selectedStage stage.Stage, cfg *config.Config, force bool) error {
|
||||
prior := capturePriorStageOutcome(model, selectedStage.Name())
|
||||
at := time.Now().UTC()
|
||||
model.MarkStageRunning(selectedStage.Name(), at)
|
||||
if force {
|
||||
if _, err := invalidateDependentSucceededStagesWithReason(
|
||||
model, selectedStage.Name(), at, staleReasonForcedReplacement,
|
||||
); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
if reason := plannedSelfSkipReason(selectedStage.Name(), cfg); reason != "" {
|
||||
model.MarkStageSkipped(selectedStage.Name(), at, reason)
|
||||
if !prior.isSameSelfSkip(reason) {
|
||||
_, err := invalidateDependentSucceededStagesWithReason(
|
||||
model, selectedStage.Name(), at, staleReasonSelfSkip,
|
||||
)
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
}
|
||||
model.MarkStageSucceeded(selectedStage.Name(), at, nil)
|
||||
if !prior.exists || prior.status != manifest.StatusSucceeded {
|
||||
if _, err := invalidateDependentSucceededStagesWithReason(
|
||||
model, selectedStage.Name(), at, staleReasonChangedResult,
|
||||
); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func plannedSelfSkipReason(stageName string, cfg *config.Config) string {
|
||||
if stageName == "extract" && cfg != nil && cfg.Pipeline != nil &&
|
||||
(cfg.Pipeline.Notarius == nil || !cfg.Pipeline.Notarius.Enabled) {
|
||||
return "notarius_disabled"
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
func writeAnalyzePlanDetails(out io.Writer, summary *stage.AnalyzeResumeSummary) error {
|
||||
if summary == nil {
|
||||
return nil
|
||||
}
|
||||
if _, err := fmt.Fprintf(out, " targets: %s\n", planStringList(summary.ExplicitTargets)); err != nil {
|
||||
return err
|
||||
}
|
||||
if _, err := fmt.Fprintf(out, " prerequisites: %s\n", planArtifactList(summary.PrerequisiteWork)); err != nil {
|
||||
return err
|
||||
}
|
||||
if _, err := fmt.Fprintf(out, " execute: %s\n", planArtifactList(summary.ExecutionOrder)); err != nil {
|
||||
return err
|
||||
}
|
||||
_, err := fmt.Fprintf(out, " reuse: %s\n", planArtifactList(summary.ReusedCurrent))
|
||||
return err
|
||||
}
|
||||
|
||||
func planStringList(values []string) string {
|
||||
if len(values) == 0 {
|
||||
return "none"
|
||||
}
|
||||
return strings.Join(values, ", ")
|
||||
}
|
||||
|
||||
func planArtifactList(values []stage.AnalyzeResumeArtifact) string {
|
||||
if len(values) == 0 {
|
||||
return "none"
|
||||
}
|
||||
parts := make([]string, 0, len(values))
|
||||
for _, value := range values {
|
||||
detail := value.Role
|
||||
if value.Reason != "" {
|
||||
detail += ":" + value.Reason
|
||||
}
|
||||
if value.Forced {
|
||||
detail += ":forced"
|
||||
}
|
||||
parts = append(parts, fmt.Sprintf("%s(%s)", value.Key, detail))
|
||||
}
|
||||
return strings.Join(parts, ", ")
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user