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

Establish explicit runner lifecycle state

Browse Source
This commit is contained in:
Eric Rakestraw
2026-09-04 01:05:44 +00:00
parent ca1b694aa9
commit e0f19833a1
4 changed files with 240 additions and 118 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
docs/roadmap/implementation.md
View File

@@ -187,7 +187,7 @@ runtime outcomes.
## Stage 2 — Establish The Invocation Runtime And Lock Lifecycle
**Status: Pending**
**Status: Completed**
### Goal

141
internal/app/runner.go
View File

@@ -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
}

175
internal/app/runner_runtime.go Normal file
View File

@@ -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,
)
}

40
internal/app/runner_test.go
View File

@@ -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