Terminalize handled invocation failures
This commit is contained in:
@@ -98,6 +98,12 @@ session manifest must agree on campaign and session identity before execution;
|
|||||||
the current projection is refreshed for every invocation while stage progress,
|
the current projection is refreshed for every invocation while stage progress,
|
||||||
inputs, and durable artifacts remain session history.
|
inputs, and durable artifacts remain session history.
|
||||||
|
|
||||||
|
For handled failures after an invocation record is created, the runner records
|
||||||
|
the failure on the session ledger and persists it before persisting the failed
|
||||||
|
run audit record. This preserves the resume authority while making a partial
|
||||||
|
persistence disagreement visible. Abrupt process death remains an accepted case
|
||||||
|
where a durable running record can require operator interpretation.
|
||||||
|
|
||||||
## Invariants
|
## Invariants
|
||||||
|
|
||||||
- stage resume/skip decisions are session-manifest driven.
|
- stage resume/skip decisions are session-manifest driven.
|
||||||
|
|||||||
@@ -27,7 +27,7 @@ All stages are pending when this plan is created.
|
|||||||
| 9 | Redact and cap subprocess diagnostics | RSK-012 | Completed |
|
| 9 | Redact and cap subprocess diagnostics | RSK-012 | Completed |
|
||||||
| 10 | Confine publish archive reads | COR-005 | Completed |
|
| 10 | Confine publish archive reads | COR-005 | Completed |
|
||||||
| 11 | Make manifest and run identity singular | COR-001, TST-006 | Completed |
|
| 11 | Make manifest and run identity singular | COR-001, TST-006 | Completed |
|
||||||
| 12 | Centralize handled terminal-failure persistence | RSK-001, TST-002, SIM-001, COM-001 | Pending |
|
| 12 | Centralize handled terminal-failure persistence | RSK-001, TST-002, SIM-001, COM-001 | Completed |
|
||||||
| 13 | Introduce the immutable remote-commit model and legacy boundary | ARC-003 | Pending |
|
| 13 | Introduce the immutable remote-commit model and legacy boundary | ARC-003 | Pending |
|
||||||
| 14 | Publish through immutable commits and canonical mappings | COR-004, COR-011, DUP-002, TST-004 | Pending |
|
| 14 | Publish through immutable commits and canonical mappings | COR-004, COR-011, DUP-002, TST-004 | Pending |
|
||||||
| 15 | Make remote locks generation-safe and harden pagination | RSK-005, RSK-014 | Pending |
|
| 15 | Make remote locks generation-safe and harden pagination | RSK-005, RSK-014 | Pending |
|
||||||
|
|||||||
@@ -26,6 +26,7 @@ type RunOptions struct {
|
|||||||
Force bool
|
Force bool
|
||||||
SelectedArtifacts []string
|
SelectedArtifacts []string
|
||||||
Env *Env
|
Env *Env
|
||||||
|
RunManifestStore manifest.RunStore
|
||||||
}
|
}
|
||||||
|
|
||||||
type RunSummary struct {
|
type RunSummary struct {
|
||||||
@@ -79,53 +80,6 @@ func executeStages(ctx context.Context, cfg *config.Config, stages []stage.Stage
|
|||||||
if env.Logger == nil {
|
if env.Logger == nil {
|
||||||
env.Logger = logging.NewLogger(os.Stderr, slog.LevelInfo)
|
env.Logger = logging.NewLogger(os.Stderr, slog.LevelInfo)
|
||||||
}
|
}
|
||||||
if _, err := loadSecretsFromConfig(env.Config, env.Logger); err != nil {
|
|
||||||
return nil, fmt.Errorf("load secrets from files: %w", err)
|
|
||||||
}
|
|
||||||
if env.WhisperX == nil {
|
|
||||||
client, err := buildDefaultWhisperXClient(env.Config)
|
|
||||||
if err != nil {
|
|
||||||
return nil, fmt.Errorf("initialize whisperx client: %w", err)
|
|
||||||
}
|
|
||||||
env.WhisperX = client
|
|
||||||
}
|
|
||||||
if env.Seriatim == nil {
|
|
||||||
runner, err := buildDefaultSeriatimRunner(env.Config)
|
|
||||||
if err != nil {
|
|
||||||
return nil, fmt.Errorf("initialize seriatim runner: %w", err)
|
|
||||||
}
|
|
||||||
env.Seriatim = runner
|
|
||||||
}
|
|
||||||
if env.Audita == nil {
|
|
||||||
runner, err := buildDefaultAuditaRunner(env.Config)
|
|
||||||
if err != nil {
|
|
||||||
return nil, 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 {
|
|
||||||
env.Scriptorium = scriptorium.NewSubprocessRunner()
|
|
||||||
}
|
|
||||||
if env.ObjectStore == nil && needsObjectStoreForRun(env.Config, stages) {
|
|
||||||
objectStore, err := newCommandObjectStore(ctx, env.Config, nil)
|
|
||||||
if err != nil {
|
|
||||||
return nil, err
|
|
||||||
}
|
|
||||||
env.ObjectStore = objectStore
|
|
||||||
}
|
|
||||||
if needsRemoteLocksForRun(env.Config, stages) {
|
|
||||||
locks, err := loadEffectiveLocks(ctx, env.Config, env.ObjectStore)
|
|
||||||
if err != nil {
|
|
||||||
return nil, fmt.Errorf("load remote publish locks: %w", err)
|
|
||||||
}
|
|
||||||
applyEffectiveLocks(env.Config, locks.All)
|
|
||||||
}
|
|
||||||
if env.Notifier == nil {
|
|
||||||
env.Notifier = ¬ify.NoopSender{}
|
|
||||||
}
|
|
||||||
|
|
||||||
artifactStore := env.ArtifactStore
|
artifactStore := env.ArtifactStore
|
||||||
paths, err := artifactStore.EnsureLayoutFor(identity.Campaign, identity.SessionID)
|
paths, err := artifactStore.EnsureLayoutFor(identity.Campaign, identity.SessionID)
|
||||||
@@ -174,7 +128,10 @@ func executeStages(ctx context.Context, cfg *config.Config, stages []stage.Stage
|
|||||||
identity.SessionID,
|
identity.SessionID,
|
||||||
identity.RunID,
|
identity.RunID,
|
||||||
)
|
)
|
||||||
runManifestStore := &manifest.LocalStore{}
|
runManifestStore := opts.RunManifestStore
|
||||||
|
if runManifestStore == nil {
|
||||||
|
runManifestStore = &manifest.LocalStore{}
|
||||||
|
}
|
||||||
runManifest, err := runManifestStore.CreateRun(
|
runManifest, err := runManifestStore.CreateRun(
|
||||||
ctx,
|
ctx,
|
||||||
identity.SessionID,
|
identity.SessionID,
|
||||||
@@ -188,7 +145,74 @@ func executeStages(ctx context.Context, cfg *config.Config, stages []stage.Stage
|
|||||||
}
|
}
|
||||||
identity.applyToRunManifest(runManifest, manifestPath)
|
identity.applyToRunManifest(runManifest, manifestPath)
|
||||||
if err := runManifestStore.SaveRun(ctx, runManifestPath, runManifest); err != nil {
|
if err := runManifestStore.SaveRun(ctx, runManifestPath, runManifest); err != nil {
|
||||||
return nil, fmt.Errorf("save initial run manifest %q: %w", runManifestPath, err)
|
return nil, persistTerminalFailure(
|
||||||
|
ctx, env.ManifestStore, manifestPath, m, runManifestStore, runManifestPath, runManifest,
|
||||||
|
fmt.Errorf("save initial run manifest %q: %w", runManifestPath, err),
|
||||||
|
)
|
||||||
|
}
|
||||||
|
if _, err := loadSecretsFromConfig(env.Config, env.Logger); err != nil {
|
||||||
|
return nil, persistTerminalFailure(
|
||||||
|
ctx, env.ManifestStore, manifestPath, m, runManifestStore, runManifestPath, runManifest,
|
||||||
|
fmt.Errorf("load secrets from files: %w", err),
|
||||||
|
)
|
||||||
|
}
|
||||||
|
if env.WhisperX == nil {
|
||||||
|
client, err := buildDefaultWhisperXClient(env.Config)
|
||||||
|
if err != nil {
|
||||||
|
return nil, persistTerminalFailure(
|
||||||
|
ctx, env.ManifestStore, manifestPath, m, runManifestStore, runManifestPath, runManifest,
|
||||||
|
fmt.Errorf("initialize whisperx client: %w", err),
|
||||||
|
)
|
||||||
|
}
|
||||||
|
env.WhisperX = client
|
||||||
|
}
|
||||||
|
if env.Seriatim == nil {
|
||||||
|
runner, err := buildDefaultSeriatimRunner(env.Config)
|
||||||
|
if err != nil {
|
||||||
|
return nil, persistTerminalFailure(
|
||||||
|
ctx, env.ManifestStore, manifestPath, m, runManifestStore, runManifestPath, runManifest,
|
||||||
|
fmt.Errorf("initialize seriatim runner: %w", err),
|
||||||
|
)
|
||||||
|
}
|
||||||
|
env.Seriatim = runner
|
||||||
|
}
|
||||||
|
if env.Audita == nil {
|
||||||
|
runner, err := buildDefaultAuditaRunner(env.Config)
|
||||||
|
if err != nil {
|
||||||
|
return nil, persistTerminalFailure(
|
||||||
|
ctx, env.ManifestStore, manifestPath, m, runManifestStore, runManifestPath, runManifest,
|
||||||
|
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 {
|
||||||
|
env.Scriptorium = scriptorium.NewSubprocessRunner()
|
||||||
|
}
|
||||||
|
if env.ObjectStore == nil && needsObjectStoreForRun(env.Config, stages) {
|
||||||
|
objectStore, err := newCommandObjectStore(ctx, env.Config, nil)
|
||||||
|
if err != nil {
|
||||||
|
return nil, persistTerminalFailure(
|
||||||
|
ctx, env.ManifestStore, manifestPath, m, runManifestStore, runManifestPath, runManifest, err,
|
||||||
|
)
|
||||||
|
}
|
||||||
|
env.ObjectStore = objectStore
|
||||||
|
}
|
||||||
|
if needsRemoteLocksForRun(env.Config, stages) {
|
||||||
|
locks, err := loadEffectiveLocks(ctx, env.Config, env.ObjectStore)
|
||||||
|
if err != nil {
|
||||||
|
return nil, persistTerminalFailure(
|
||||||
|
ctx, env.ManifestStore, manifestPath, m, runManifestStore, runManifestPath, runManifest,
|
||||||
|
fmt.Errorf("load remote publish locks: %w", err),
|
||||||
|
)
|
||||||
|
}
|
||||||
|
applyEffectiveLocks(env.Config, locks.All)
|
||||||
|
}
|
||||||
|
if env.Notifier == nil {
|
||||||
|
env.Notifier = ¬ify.NoopSender{}
|
||||||
}
|
}
|
||||||
|
|
||||||
stageEnv := env
|
stageEnv := env
|
||||||
@@ -207,7 +231,10 @@ func executeStages(ctx context.Context, cfg *config.Config, stages []stage.Stage
|
|||||||
if validator, ok := s.(stage.ResumeValidator); ok {
|
if validator, ok := s.(stage.ResumeValidator); ok {
|
||||||
validation, err := validator.ValidateResume(ctx, stageEnv, m)
|
validation, err := validator.ValidateResume(ctx, stageEnv, m)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, fmt.Errorf("validate resume for stage %q: %w", s.Name(), err)
|
return nil, persistTerminalFailure(
|
||||||
|
ctx, env.ManifestStore, manifestPath, m, runManifestStore, runManifestPath, runManifest,
|
||||||
|
fmt.Errorf("validate resume for stage %q: %w", s.Name(), err),
|
||||||
|
)
|
||||||
}
|
}
|
||||||
validation = validation.Normalized()
|
validation = validation.Normalized()
|
||||||
if !validation.Resumable {
|
if !validation.Resumable {
|
||||||
@@ -217,7 +244,10 @@ func executeStages(ctx context.Context, cfg *config.Config, stages []stage.Stage
|
|||||||
m, s.Name(), staleAt, staleReasonNotResumable,
|
m, s.Name(), staleAt, staleReasonNotResumable,
|
||||||
)
|
)
|
||||||
if err := env.ManifestStore.Save(ctx, manifestPath, m); err != nil {
|
if err := env.ManifestStore.Save(ctx, manifestPath, m); err != nil {
|
||||||
return nil, fmt.Errorf("save manifest after resume validation for stage %q: %w", s.Name(), err)
|
return nil, persistTerminalFailure(
|
||||||
|
ctx, env.ManifestStore, manifestPath, m, runManifestStore, runManifestPath, runManifest,
|
||||||
|
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", validation.Reason)
|
env.Logger.Info("stage result is not resumable", "stage", s.Name(), "reason", validation.Reason)
|
||||||
d.Action = stageActionRun
|
d.Action = stageActionRun
|
||||||
@@ -231,7 +261,10 @@ func executeStages(ctx context.Context, cfg *config.Config, stages []stage.Stage
|
|||||||
runManifest.SetStageAction(s.Name(), manifest.RunStageActionSkip, skipAt)
|
runManifest.SetStageAction(s.Name(), manifest.RunStageActionSkip, skipAt)
|
||||||
runManifest.MarkStageSkipped(s.Name(), skipAt, "already_succeeded")
|
runManifest.MarkStageSkipped(s.Name(), skipAt, "already_succeeded")
|
||||||
if err := runManifestStore.SaveRun(ctx, runManifestPath, runManifest); err != nil {
|
if err := runManifestStore.SaveRun(ctx, runManifestPath, runManifest); err != nil {
|
||||||
return nil, fmt.Errorf("save run manifest after skip %q: %w", s.Name(), err)
|
return nil, persistTerminalFailure(
|
||||||
|
ctx, env.ManifestStore, manifestPath, m, runManifestStore, runManifestPath, runManifest,
|
||||||
|
fmt.Errorf("save run manifest after skip %q: %w", s.Name(), err),
|
||||||
|
)
|
||||||
}
|
}
|
||||||
env.Logger.Info("skipping stage", "stage", s.Name(), "reason", "already_succeeded", "force", opts.Force)
|
env.Logger.Info("skipping stage", "stage", s.Name(), "reason", "already_succeeded", "force", opts.Force)
|
||||||
continue
|
continue
|
||||||
@@ -243,7 +276,11 @@ func executeStages(ctx context.Context, cfg *config.Config, stages []stage.Stage
|
|||||||
runManifest.SetStageAction(s.Name(), manifest.RunStageActionRun, now)
|
runManifest.SetStageAction(s.Name(), manifest.RunStageActionRun, now)
|
||||||
runManifest.MarkStageRunning(s.Name(), now)
|
runManifest.MarkStageRunning(s.Name(), now)
|
||||||
if err := runManifestStore.SaveRun(ctx, runManifestPath, runManifest); err != nil {
|
if err := runManifestStore.SaveRun(ctx, runManifestPath, runManifest); err != nil {
|
||||||
return nil, fmt.Errorf("save run manifest before stage %q: %w", s.Name(), err)
|
operationErr := fmt.Errorf("save run manifest before stage %q: %w", s.Name(), err)
|
||||||
|
runManifest.MarkStageFailed(s.Name(), nowUTC(), operationErr.Error())
|
||||||
|
return nil, persistTerminalFailure(
|
||||||
|
ctx, env.ManifestStore, manifestPath, m, runManifestStore, runManifestPath, runManifest, operationErr,
|
||||||
|
)
|
||||||
}
|
}
|
||||||
m.MarkStageRunning(s.Name(), now)
|
m.MarkStageRunning(s.Name(), now)
|
||||||
if opts.Force {
|
if opts.Force {
|
||||||
@@ -251,7 +288,12 @@ func executeStages(ctx context.Context, cfg *config.Config, stages []stage.Stage
|
|||||||
}
|
}
|
||||||
env.Logger.Info("starting stage", "stage", s.Name())
|
env.Logger.Info("starting stage", "stage", s.Name())
|
||||||
if err := env.ManifestStore.Save(ctx, manifestPath, m); err != nil {
|
if err := env.ManifestStore.Save(ctx, manifestPath, m); err != nil {
|
||||||
return nil, fmt.Errorf("save manifest before stage %q: %w", s.Name(), err)
|
operationErr := fmt.Errorf("save manifest before stage %q: %w", s.Name(), err)
|
||||||
|
m.MarkStageFailed(s.Name(), nowUTC(), operationErr.Error())
|
||||||
|
invalidateDownstreamSucceededStagesWithReason(m, s.Name(), nowUTC(), staleReasonFailure)
|
||||||
|
return nil, persistTerminalFailure(
|
||||||
|
ctx, env.ManifestStore, manifestPath, m, runManifestStore, runManifestPath, runManifest, operationErr,
|
||||||
|
)
|
||||||
}
|
}
|
||||||
env.Logger.Debug("manifest saved", "stage", s.Name(), "transition", "running", "path", manifestPath)
|
env.Logger.Debug("manifest saved", "stage", s.Name(), "transition", "running", "path", manifestPath)
|
||||||
|
|
||||||
@@ -263,16 +305,13 @@ func executeStages(ctx context.Context, cfg *config.Config, stages []stage.Stage
|
|||||||
failedAt := nowUTC()
|
failedAt := nowUTC()
|
||||||
m.MarkStageFailed(s.Name(), failedAt, err.Error())
|
m.MarkStageFailed(s.Name(), failedAt, err.Error())
|
||||||
invalidateDownstreamSucceededStagesWithReason(m, s.Name(), failedAt, staleReasonFailure)
|
invalidateDownstreamSucceededStagesWithReason(m, s.Name(), failedAt, staleReasonFailure)
|
||||||
if saveErr := env.ManifestStore.Save(ctx, manifestPath, m); saveErr != nil {
|
|
||||||
return nil, fmt.Errorf("stage %q failed (%v) and manifest save failed (%v)", s.Name(), err, saveErr)
|
|
||||||
}
|
|
||||||
runManifest.MarkStageFailed(s.Name(), failedAt, err.Error())
|
runManifest.MarkStageFailed(s.Name(), failedAt, err.Error())
|
||||||
identity.applyToRunManifest(runManifest, manifestPath)
|
identity.applyToRunManifest(runManifest, manifestPath)
|
||||||
if saveErr := runManifestStore.SaveRun(ctx, runManifestPath, runManifest); saveErr != nil {
|
|
||||||
return nil, fmt.Errorf("stage %q failed (%v) and run-manifest save failed (%v)", s.Name(), err, saveErr)
|
|
||||||
}
|
|
||||||
env.Logger.Info("stage failed", "stage", s.Name(), "error", err)
|
env.Logger.Info("stage failed", "stage", s.Name(), "error", err)
|
||||||
return nil, fmt.Errorf("stage %q failed: %w", s.Name(), err)
|
return nil, persistTerminalFailure(
|
||||||
|
ctx, env.ManifestStore, manifestPath, m, runManifestStore, runManifestPath, runManifest,
|
||||||
|
fmt.Errorf("stage %q failed: %w", s.Name(), err),
|
||||||
|
)
|
||||||
}
|
}
|
||||||
if result != nil && result.Disposition == stage.StageDispositionSkipped {
|
if result != nil && result.Disposition == stage.StageDispositionSkipped {
|
||||||
skipped = append(skipped, s.Name())
|
skipped = append(skipped, s.Name())
|
||||||
@@ -283,13 +322,19 @@ func executeStages(ctx context.Context, cfg *config.Config, stages []stage.Stage
|
|||||||
invalidateDownstreamSucceededStagesWithReason(m, s.Name(), skippedAt, staleReasonSelfSkip)
|
invalidateDownstreamSucceededStagesWithReason(m, s.Name(), skippedAt, staleReasonSelfSkip)
|
||||||
}
|
}
|
||||||
if err := env.ManifestStore.Save(ctx, manifestPath, m); err != nil {
|
if err := env.ManifestStore.Save(ctx, manifestPath, m); err != nil {
|
||||||
return nil, fmt.Errorf("save manifest after self-skip %q: %w", s.Name(), err)
|
return nil, persistTerminalFailure(
|
||||||
|
ctx, env.ManifestStore, manifestPath, m, runManifestStore, runManifestPath, runManifest,
|
||||||
|
fmt.Errorf("save manifest after self-skip %q: %w", s.Name(), err),
|
||||||
|
)
|
||||||
}
|
}
|
||||||
runManifest.MarkStageSkipped(s.Name(), skippedAt, result.SkipReason)
|
runManifest.MarkStageSkipped(s.Name(), skippedAt, result.SkipReason)
|
||||||
applyStageResultToRunManifest(runManifest, s.Name(), result)
|
applyStageResultToRunManifest(runManifest, s.Name(), result)
|
||||||
identity.applyToRunManifest(runManifest, manifestPath)
|
identity.applyToRunManifest(runManifest, manifestPath)
|
||||||
if err := runManifestStore.SaveRun(ctx, runManifestPath, runManifest); err != nil {
|
if err := runManifestStore.SaveRun(ctx, runManifestPath, runManifest); err != nil {
|
||||||
return nil, fmt.Errorf("save run manifest after self-skip %q: %w", s.Name(), err)
|
return nil, persistTerminalFailure(
|
||||||
|
ctx, env.ManifestStore, manifestPath, m, runManifestStore, runManifestPath, runManifest,
|
||||||
|
fmt.Errorf("save run manifest after self-skip %q: %w", s.Name(), err),
|
||||||
|
)
|
||||||
}
|
}
|
||||||
env.Logger.Debug("manifest saved", "stage", s.Name(), "transition", "skipped", "path", manifestPath)
|
env.Logger.Debug("manifest saved", "stage", s.Name(), "transition", "skipped", "path", manifestPath)
|
||||||
env.Logger.Info("stage skipped", "stage", s.Name(), "reason", result.SkipReason)
|
env.Logger.Info("stage skipped", "stage", s.Name(), "reason", result.SkipReason)
|
||||||
@@ -305,32 +350,38 @@ func executeStages(ctx context.Context, cfg *config.Config, stages []stage.Stage
|
|||||||
}
|
}
|
||||||
|
|
||||||
if err := env.ManifestStore.Save(ctx, manifestPath, m); err != nil {
|
if err := env.ManifestStore.Save(ctx, manifestPath, m); err != nil {
|
||||||
return nil, fmt.Errorf("save manifest after stage %q: %w", s.Name(), err)
|
return nil, persistTerminalFailure(
|
||||||
|
ctx, env.ManifestStore, manifestPath, m, runManifestStore, runManifestPath, runManifest,
|
||||||
|
fmt.Errorf("save manifest after stage %q: %w", s.Name(), err),
|
||||||
|
)
|
||||||
}
|
}
|
||||||
runManifest.MarkStageSucceeded(s.Name(), succeededAt, outputs)
|
runManifest.MarkStageSucceeded(s.Name(), succeededAt, outputs)
|
||||||
applyStageResultToRunManifest(runManifest, s.Name(), result)
|
applyStageResultToRunManifest(runManifest, s.Name(), result)
|
||||||
identity.applyToRunManifest(runManifest, manifestPath)
|
identity.applyToRunManifest(runManifest, manifestPath)
|
||||||
if err := runManifestStore.SaveRun(ctx, runManifestPath, runManifest); err != nil {
|
if err := runManifestStore.SaveRun(ctx, runManifestPath, runManifest); err != nil {
|
||||||
return nil, fmt.Errorf("save run manifest after stage %q: %w", s.Name(), err)
|
return nil, persistTerminalFailure(
|
||||||
|
ctx, env.ManifestStore, manifestPath, m, runManifestStore, runManifestPath, runManifest,
|
||||||
|
fmt.Errorf("save run manifest after stage %q: %w", s.Name(), err),
|
||||||
|
)
|
||||||
}
|
}
|
||||||
env.Logger.Debug("manifest saved", "stage", s.Name(), "transition", "succeeded", "path", manifestPath)
|
env.Logger.Debug("manifest saved", "stage", s.Name(), "transition", "succeeded", "path", manifestPath)
|
||||||
env.Logger.Info("stage succeeded", "stage", s.Name())
|
env.Logger.Info("stage succeeded", "stage", s.Name())
|
||||||
}
|
}
|
||||||
|
|
||||||
if err := runPostPublishCleanup(ctx, env, manifestPath, m, executed); err != nil {
|
if err := runPostPublishCleanup(ctx, env, manifestPath, m, executed); err != nil {
|
||||||
failedAt := nowUTC()
|
return nil, persistTerminalFailure(
|
||||||
runManifest.MarkFailed(failedAt, err.Error())
|
ctx, env.ManifestStore, manifestPath, m, runManifestStore, runManifestPath, runManifest,
|
||||||
identity.applyToRunManifest(runManifest, manifestPath)
|
fmt.Errorf("post-publish cleanup: %w", err),
|
||||||
if saveErr := runManifestStore.SaveRun(ctx, runManifestPath, runManifest); saveErr != nil {
|
)
|
||||||
return nil, fmt.Errorf("post-publish cleanup failed (%v) and run-manifest save failed (%v)", err, saveErr)
|
|
||||||
}
|
|
||||||
return nil, fmt.Errorf("post-publish cleanup: %w", err)
|
|
||||||
}
|
}
|
||||||
completedAt := nowUTC()
|
completedAt := nowUTC()
|
||||||
runManifest.MarkSucceeded(completedAt)
|
runManifest.MarkSucceeded(completedAt)
|
||||||
identity.applyToRunManifest(runManifest, manifestPath)
|
identity.applyToRunManifest(runManifest, manifestPath)
|
||||||
if err := runManifestStore.SaveRun(ctx, runManifestPath, runManifest); err != nil {
|
if err := runManifestStore.SaveRun(ctx, runManifestPath, runManifest); err != nil {
|
||||||
return nil, fmt.Errorf("save final run manifest %q: %w", runManifestPath, err)
|
return nil, persistTerminalFailure(
|
||||||
|
ctx, env.ManifestStore, manifestPath, m, runManifestStore, runManifestPath, runManifest,
|
||||||
|
fmt.Errorf("save final run manifest %q: %w", runManifestPath, err),
|
||||||
|
)
|
||||||
}
|
}
|
||||||
|
|
||||||
return &RunSummary{
|
return &RunSummary{
|
||||||
@@ -344,6 +395,47 @@ func executeStages(ctx context.Context, cfg *config.Config, stages []stage.Stage
|
|||||||
}, nil
|
}, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func persistTerminalFailure(
|
||||||
|
ctx context.Context,
|
||||||
|
sessionStore manifest.Store,
|
||||||
|
sessionManifestPath string,
|
||||||
|
sessionManifest *manifest.Manifest,
|
||||||
|
runStore manifest.RunStore,
|
||||||
|
runManifestPath string,
|
||||||
|
runManifest *manifest.RunManifest,
|
||||||
|
operationErr error,
|
||||||
|
) error {
|
||||||
|
if operationErr == nil {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
at := nowUTC()
|
||||||
|
if sessionManifest != nil {
|
||||||
|
sessionManifest.RecordFailure(at, operationErr.Error())
|
||||||
|
}
|
||||||
|
|
||||||
|
// A running run record is durable before execution. At termination, session
|
||||||
|
// state controls resume while the run record only diagnoses this invocation,
|
||||||
|
// so session persistence comes first and any disagreement remains visible.
|
||||||
|
sessionErr := sessionStore.Save(ctx, sessionManifestPath, sessionManifest)
|
||||||
|
if runManifest != nil {
|
||||||
|
runManifest.MarkFailed(at, operationErr.Error())
|
||||||
|
}
|
||||||
|
runErr := runStore.SaveRun(ctx, runManifestPath, runManifest)
|
||||||
|
return errors.Join(
|
||||||
|
operationErr,
|
||||||
|
wrapTerminalPersistenceError("save terminal session manifest", sessionErr),
|
||||||
|
wrapTerminalPersistenceError("save terminal run manifest", runErr),
|
||||||
|
)
|
||||||
|
}
|
||||||
|
|
||||||
|
func wrapTerminalPersistenceError(operation string, err error) error {
|
||||||
|
if err == nil {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
return fmt.Errorf("%s: %w", operation, err)
|
||||||
|
}
|
||||||
|
|
||||||
func buildDefaultWhisperXClient(cfg *config.Config) (whisperx.Client, error) {
|
func buildDefaultWhisperXClient(cfg *config.Config) (whisperx.Client, error) {
|
||||||
if cfg == nil || cfg.Pipeline == nil {
|
if cfg == nil || cfg.Pipeline == nil {
|
||||||
return &whisperx.NoopClient{}, nil
|
return &whisperx.NoopClient{}, nil
|
||||||
|
|||||||
154
internal/app/runner_terminal_failure_test.go
Normal file
154
internal/app/runner_terminal_failure_test.go
Normal file
@@ -0,0 +1,154 @@
|
|||||||
|
package app
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"errors"
|
||||||
|
"strings"
|
||||||
|
"testing"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"gitea.maximumdirect.net/eric/narratio/internal/manifest"
|
||||||
|
"gitea.maximumdirect.net/eric/narratio/internal/stage"
|
||||||
|
)
|
||||||
|
|
||||||
|
type terminalSessionStore struct {
|
||||||
|
events []string
|
||||||
|
saveErr error
|
||||||
|
saved *manifest.Manifest
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *terminalSessionStore) Create(_ context.Context, sessionID string) (*manifest.Manifest, error) {
|
||||||
|
return manifest.New(sessionID, time.Now().UTC()), nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *terminalSessionStore) Load(context.Context, string) (*manifest.Manifest, error) {
|
||||||
|
return nil, errors.New("unexpected manifest load")
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *terminalSessionStore) Save(_ context.Context, _ string, m *manifest.Manifest) error {
|
||||||
|
s.events = append(s.events, "session")
|
||||||
|
s.saved = m
|
||||||
|
return s.saveErr
|
||||||
|
}
|
||||||
|
|
||||||
|
type terminalRunStore struct {
|
||||||
|
events []string
|
||||||
|
saveErr error
|
||||||
|
saved *manifest.RunManifest
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *terminalRunStore) CreateRun(_ context.Context, sessionID, campaign, runID string, force bool, requestedStages []string) (*manifest.RunManifest, error) {
|
||||||
|
return manifest.NewRun(sessionID, campaign, runID, force, requestedStages, time.Now().UTC()), nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *terminalRunStore) SaveRun(_ context.Context, _ string, m *manifest.RunManifest) error {
|
||||||
|
s.events = append(s.events, "run")
|
||||||
|
s.saved = m
|
||||||
|
return s.saveErr
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestPersistTerminalFailureKeepsSessionAuthorityBeforeRunAudit(t *testing.T) {
|
||||||
|
cause := errors.New("stage failed")
|
||||||
|
sessionFailure := errors.New("session unavailable")
|
||||||
|
runFailure := errors.New("run unavailable")
|
||||||
|
|
||||||
|
for _, tc := range []struct {
|
||||||
|
name string
|
||||||
|
sessionErr error
|
||||||
|
runErr error
|
||||||
|
cause error
|
||||||
|
alreadyTerminal bool
|
||||||
|
}{
|
||||||
|
{name: "both ledgers saved", cause: cause},
|
||||||
|
{name: "session save fails", sessionErr: sessionFailure, cause: cause},
|
||||||
|
{name: "run save fails", runErr: runFailure, cause: cause},
|
||||||
|
{name: "both saves fail", sessionErr: sessionFailure, runErr: runFailure, cause: cause},
|
||||||
|
{name: "cancellation", cause: context.Canceled},
|
||||||
|
{name: "already terminal run", cause: cause, alreadyTerminal: true},
|
||||||
|
} {
|
||||||
|
t.Run(tc.name, func(t *testing.T) {
|
||||||
|
sessionStore := &terminalSessionStore{saveErr: tc.sessionErr}
|
||||||
|
runStore := &terminalRunStore{saveErr: tc.runErr}
|
||||||
|
sessionManifest := manifest.New("2026-05-03", time.Now().UTC())
|
||||||
|
runManifest := manifest.NewRun("2026-05-03", "sample-campaign", "20260503T010203Z-a1b2c3d4", false, nil, time.Now().UTC())
|
||||||
|
if tc.alreadyTerminal {
|
||||||
|
sessionManifest.RecordFailure(time.Now().UTC(), "earlier failure")
|
||||||
|
runManifest.MarkFailed(time.Now().UTC(), "earlier failure")
|
||||||
|
}
|
||||||
|
|
||||||
|
err := persistTerminalFailure(
|
||||||
|
context.Background(), sessionStore, "session.json", sessionManifest,
|
||||||
|
runStore, "run.json", runManifest, tc.cause,
|
||||||
|
)
|
||||||
|
if !errors.Is(err, tc.cause) {
|
||||||
|
t.Fatalf("error = %v, want original cause %v", err, tc.cause)
|
||||||
|
}
|
||||||
|
if tc.sessionErr != nil && !errors.Is(err, tc.sessionErr) {
|
||||||
|
t.Fatalf("error = %v, want session failure %v", err, tc.sessionErr)
|
||||||
|
}
|
||||||
|
if tc.runErr != nil && !errors.Is(err, tc.runErr) {
|
||||||
|
t.Fatalf("error = %v, want run failure %v", err, tc.runErr)
|
||||||
|
}
|
||||||
|
if got := append(sessionStore.events, runStore.events...); len(got) != 2 || got[0] != "session" || got[1] != "run" {
|
||||||
|
t.Fatalf("save order = %v, want [session run]", got)
|
||||||
|
}
|
||||||
|
if sessionStore.saved == nil || sessionStore.saved.LastError == nil || sessionStore.saved.LastError.Message != tc.cause.Error() {
|
||||||
|
t.Fatalf("session terminal error = %#v, want %q", sessionStore.saved, tc.cause)
|
||||||
|
}
|
||||||
|
if runStore.saved == nil || runStore.saved.Status != manifest.RunManifestStatusFailed || runStore.saved.LastError == nil || runStore.saved.LastError.Message != tc.cause.Error() {
|
||||||
|
t.Fatalf("run terminal record = %#v, want failed with %q", runStore.saved, tc.cause)
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestExecuteStagesTerminalizesCancelledStage(t *testing.T) {
|
||||||
|
cfg := testConfig(t)
|
||||||
|
runStore := &terminalRunStore{}
|
||||||
|
|
||||||
|
_, err := executeStages(context.Background(), cfg, []stage.Stage{
|
||||||
|
failingStage{name: "prepare", err: context.Canceled},
|
||||||
|
}, RunOptions{RunManifestStore: runStore})
|
||||||
|
if !errors.Is(err, context.Canceled) {
|
||||||
|
t.Fatalf("executeStages() error = %v, want context cancellation", err)
|
||||||
|
}
|
||||||
|
if runStore.saved == nil || runStore.saved.Status != manifest.RunManifestStatusFailed {
|
||||||
|
t.Fatalf("run terminal record = %#v, want failed", runStore.saved)
|
||||||
|
}
|
||||||
|
store := &manifest.LocalStore{}
|
||||||
|
sessionManifest, loadErr := store.Load(context.Background(), manifestPathFor(cfg))
|
||||||
|
if loadErr != nil {
|
||||||
|
t.Fatalf("load session manifest: %v", loadErr)
|
||||||
|
}
|
||||||
|
if sessionManifest.Stages["prepare"] == nil || sessionManifest.Stages["prepare"].Status != manifest.StatusFailed || sessionManifest.LastError == nil {
|
||||||
|
t.Fatalf("session terminal record = %#v, want failed prepare stage and last error", sessionManifest)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestExecuteStagesResumeValidationFailureTerminalizesRunRecord(t *testing.T) {
|
||||||
|
cfg := testConfig(t)
|
||||||
|
store := &manifest.LocalStore{}
|
||||||
|
seed := manifest.New(cfg.Session.SessionID, time.Now().UTC())
|
||||||
|
seed.MarkStageSucceeded("checked", time.Now().UTC(), nil)
|
||||||
|
if err := store.Save(context.Background(), manifestPathFor(cfg), seed); err != nil {
|
||||||
|
t.Fatalf("save session manifest: %v", err)
|
||||||
|
}
|
||||||
|
runStore := &terminalRunStore{}
|
||||||
|
|
||||||
|
_, err := executeStages(context.Background(), cfg, []stage.Stage{
|
||||||
|
resumeCheckingStage{name: "checked", validateErr: errors.New("inspection unavailable"), runs: new(int)},
|
||||||
|
}, RunOptions{RunManifestStore: runStore})
|
||||||
|
if err == nil || !strings.Contains(err.Error(), "inspection unavailable") || runStore.saved == nil {
|
||||||
|
t.Fatalf("executeStages() error = %v, want resume validation failure and terminal run", err)
|
||||||
|
}
|
||||||
|
if runStore.saved.Status != manifest.RunManifestStatusFailed {
|
||||||
|
t.Fatalf("run manifest status = %q, want failed", runStore.saved.Status)
|
||||||
|
}
|
||||||
|
persisted, loadErr := store.Load(context.Background(), manifestPathFor(cfg))
|
||||||
|
if loadErr != nil {
|
||||||
|
t.Fatalf("load session manifest: %v", loadErr)
|
||||||
|
}
|
||||||
|
if persisted.Stages["checked"].Status != manifest.StatusSucceeded || persisted.LastError == nil {
|
||||||
|
t.Fatalf("session manifest = %#v, want preserved success with terminal error", persisted)
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -121,6 +121,15 @@ func (m *Manifest) MarkStageFailed(name string, at time.Time, message string) {
|
|||||||
m.UpdatedAt = at
|
m.UpdatedAt = at
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// RecordFailure records an invocation failure without changing session progress.
|
||||||
|
func (m *Manifest) RecordFailure(at time.Time, message string) {
|
||||||
|
if m == nil {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
m.LastError = &ErrorRecord{Message: strings.TrimSpace(message), At: timePtr(at)}
|
||||||
|
m.UpdatedAt = at
|
||||||
|
}
|
||||||
|
|
||||||
// MarkStageSkipped marks a stage as skipped and records the skip reason.
|
// MarkStageSkipped marks a stage as skipped and records the skip reason.
|
||||||
func (m *Manifest) MarkStageSkipped(name string, at time.Time, reason string) {
|
func (m *Manifest) MarkStageSkipped(name string, at time.Time, reason string) {
|
||||||
s := m.ensureStage(name, at)
|
s := m.ensureStage(name, at)
|
||||||
|
|||||||
@@ -21,6 +21,12 @@ type Store interface {
|
|||||||
Save(ctx context.Context, path string, m *Manifest) error
|
Save(ctx context.Context, path string, m *Manifest) error
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// RunStore persists invocation-scoped run manifests.
|
||||||
|
type RunStore interface {
|
||||||
|
CreateRun(ctx context.Context, sessionID, campaign, runID string, force bool, requestedStages []string) (*RunManifest, error)
|
||||||
|
SaveRun(ctx context.Context, path string, m *RunManifest) error
|
||||||
|
}
|
||||||
|
|
||||||
// LocalStore stores manifests as JSON on the local filesystem.
|
// LocalStore stores manifests as JSON on the local filesystem.
|
||||||
type LocalStore struct{}
|
type LocalStore struct{}
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user