eric/narratio
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 27 Wiki Activity

Isolate invocation collaborator composition

Browse Source
This commit is contained in:
Eric Rakestraw
2026-09-04 01:11:11 +00:00
parent e0f19833a1
commit 8afe9427ca
10 changed files with 255 additions and 130 deletions
Download Patch File Download Diff File Expand all files Collapse all files

149
internal/app/runner.go
View File

@@ -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 = &notify.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())
}