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

Make runner persistence session-only

Browse Source
This commit is contained in:
Eric Rakestraw
2026-09-03 04:09:12 +00:00
parent 6d748d65d3
commit 7997f81f47
17 changed files with 123 additions and 505 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
docs/cli.md
View File

@@ -163,7 +163,9 @@ Behavior:
- rejects repeated `--from`, `--through`, or `--force` options, including
`--name=value` spellings;
- continues interrupted or partially completed sessions by running non-succeeded stages;
- writes session and run manifests.
- writes the session manifest as the sole durable progress ledger. Each
invocation also creates its run-local directory and returns an immediate
command summary; neither produces a run manifest.
- reports the resolved profile (or `none`) and effective configuration digest.
When `--artifacts` is present, the selected range must contain `analyze` or

65
docs/internal/manifest.md
View File

@@ -121,23 +121,10 @@ The incremental executor constructs this restricted projection at each
scheduled artifact boundary. The active record is failed without output,
current transitive dependents are stale, unrelated current records survive, and
only earlier validated and materialized completions remain current in the
invocation subset. Session failure state is persisted before invocation failure
state. If either terminal save fails, its persistence error is joined with the
original adapter, validation, or filesystem cause; a failed projection save
reconciled session collection. A terminal session-save failure is joined with
the original adapter, validation, or filesystem cause; a failed projection save
does not turn incidental canonical bytes into manifest authority.
## Run Manifest
`manifest.RunManifest` is created for each invocation and records:
- required `schema_version: narratio.run-manifest.v1`
- invocation identity and `force` flag
- the selected profile (when any) and secret-free effective configuration digest
- requested stages
- per-stage action (`run` or `skip`)
- per-stage status
- overall run status (`running`, `succeeded`, `failed`)
## Remote Commit Manifest
`artifacts.RemoteCommitManifest` is a separate, versioned remote snapshot
@@ -173,7 +160,7 @@ durability guarantee. Callers reload after a returned save error before retrying
## Execution Semantics
The application runner marks an executing stage running and then succeeded or
failed in both manifests, persisting each transition. On success it records
failed in the session manifest, persisting each transition. On success it records
outputs, logs, generated configuration references, metadata, and—when the
stage implements the optional contract—a versioned semantic-configuration
fingerprint. Artifact
@@ -190,8 +177,7 @@ Failed and skipped
transitions enforce the same clearing rule directly, while success repopulates
only fields returned by the new result. Marking a record stale does not clear
those details because resume validation and diagnosis may still require them
before execution begins. Invocation run manifests remain immutable audit
records of their own outcomes.
before execution begins.
Aggregate lifecycle clearing deliberately preserves the analyze-owned
per-artifact collection. This lets later reconciliation replace only evaluated
@@ -200,24 +186,21 @@ existing aggregate-only lifecycle behavior and are forbidden from carrying the
analyze-specific fields.
A stage may explicitly return a skipped disposition and stable reason. The
runner persists that outcome in both manifests, clears older outputs for the
session-stage record along with older logs, generated configuration references,
and metadata, then applies any bounded details from the current skip and
continues. This self-skip is distinct from deciding not to execute an
runner persists that outcome in the session manifest, clears older outputs,
logs, generated configuration references, and metadata, then applies any
bounded details from the current skip and continues. This self-skip is distinct from deciding not to execute an
already-succeeded stage and is reconsidered on later runs. Skipped results
cannot contain outputs. An intentional self-skip records the current semantic
fingerprint because it is a completed, reusable stage result; failed or
interrupted work never promotes one.
When an already-succeeded stage is skipped, the invocation run manifest records
the `skip` action and reason. The session manifest deliberately retains its
existing succeeded record because it remains the cross-invocation progress
authority. If a stage supplies semantic configuration evidence, reuse first
When an already-succeeded stage is skipped, the session manifest deliberately
retains its existing succeeded record because it remains the cross-invocation
progress authority. If a stage supplies semantic configuration evidence, reuse first
requires the persisted positive schema version and lowercase SHA-256 digest to
match the current resolved stage semantics. Missing semantic evidence, malformed
evidence, or a mismatch makes the stage and its fixed transitive dependents
stale. The invocation skip copies the matched fingerprint for provenance but
does not rewrite session authority. The existing stage-specific resume
stale. The existing stage-specific resume
validator runs only after this semantic check succeeds; both checks are
required. Extraction and analyze have resume validators and may reject an
otherwise eligible skip when their selected durable evidence is obsolete; the
@@ -248,10 +231,8 @@ contents that a tool privately loads remain outside the contract when their
configured identifier is unchanged; operators must force the affected work
after such a private content change.
Session manifest is the authoritative stage-progress ledger across invocations.
Run manifest is invocation-scoped audit state.
Both manifests retain the most recently resolved invocation's bounded
The session manifest is the authoritative stage-progress ledger across
invocations. It retains the most recently resolved invocation's bounded
configuration provenance. It identifies the selected profile name and source
(`default` or `cli`) plus the effective configuration digest, but never a raw
secret or profile content. This provenance is informational: it does not
@@ -262,7 +243,7 @@ configured identifier, use `--force` for the affected work.
`session plan` computes the same current fingerprint and applies the same
comparison and invalidation rules to a cloned manifest. It predicts the runner
decision without persisting session or invocation state. The shared helper
decision without persisting session state. The shared helper
hashes deterministic JSON from stage-owned typed structs; stage providers must
exclude secrets, complete effective-configuration dumps, and operational
values that cannot affect canonical results. Concrete coverage is owned by the
@@ -272,7 +253,7 @@ Before an explicitly bounded execution starts after `prepare`, the application
reads the session manifest and accepts only `succeeded` or `skipped` for every
excluded canonical prefix stage. The first other status or absent record fails
the request before layout mutation, adapter initialization, session-manifest
writes, or run-manifest creation. Excluded prefix records are not passed to
writes, or run-directory creation. Excluded prefix records are not passed to
resume validators. Records after the selected end are not prerequisites and
may be made stale by selected work without being scheduled.
@@ -289,30 +270,28 @@ session manifest must agree on campaign and session identity before execution;
the current projection is refreshed for every invocation while stage progress,
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.
For handled failures after the session identity transition, the runner records
the failure on the session ledger and atomically attempts its save while
preserving the originating error. Abrupt process death remains an accepted case
where the last session transition can require operator interpretation.
## Invariants
- stage resume/skip decisions are session-manifest driven.
- semantic fingerprint comparison precedes stage-specific resume validation.
- only successful and intentional-skipped results promote current semantic
evidence; invocation reuse copies evidence without replacing session state.
evidence.
- running, failed, and self-skipped stages do not retain result payloads from
an earlier success.
- stale stages retain prior details until replacement execution starts.
- force reruns stale succeeded stages in the fixed dependency relation.
- run manifest does not replace session manifest as progress authority.
- the session manifest is the only durable local progress ledger.
- remote commitment is established by a verified current pointer and remote
commit relationship, never by a mutable session-manifest boolean.
## Implementation And Tests
- Models and transitions: `internal/manifest/manifest.go`,
`internal/manifest/run_manifest.go`
- Models and transitions: `internal/manifest/manifest.go`
- Remote commit model and readers: `internal/artifacts/remote_commit.go`,
`internal/artifacts/current_state_commit.go`,
- Persistence and validation: `internal/manifest/store.go`

4
docs/policy/architecture.md
View File

@@ -116,8 +116,8 @@ adapter and publish metadata.
Resume and skip decisions are manifest-driven. Filesystem state may be
inspected and validated, but file presence alone does not replace recorded
stage state. Invocation-scoped run records provide an audit of one execution;
they do not replace the session manifest as progress authority.
stage state. Each invocation retains an in-memory command summary and a
run-local working directory, but neither is a durable progress ledger.
Restore treats committed remote current state as its authority. It must plan
deterministically, confine remote-to-local paths, protect local conflicts, and

2
docs/roadmap/implementation.md
View File

@@ -171,7 +171,7 @@ diagnostic uploads.
## Stage 2 — Make Runner Persistence Session-Only
**Status: Pending**
**Status: Completed**
### Goal

6
internal/app/analyze_projection.go
View File

@@ -91,7 +91,6 @@ func validateAndCloneAnalyzeProjection(projection *stage.AnalyzeStateProjection)
func applyAnalyzeProjection(
sessionManifest *manifest.Manifest,
runManifest *manifest.RunManifest,
projection *validatedAnalyzeProjection,
) {
if projection == nil {
@@ -102,11 +101,6 @@ func applyAnalyzeProjection(
record.AnalyzeStateVersion = manifest.AnalyzeStateContractVersion
record.AnalyzeArtifacts = manifest.CloneAnalyzeArtifactCollection(projection.session)
}
if runManifest != nil && runManifest.Stages["analyze"] != nil {
record := runManifest.Stages["analyze"]
record.AnalyzeStateVersion = manifest.AnalyzeStateContractVersion
record.AnalyzeArtifacts = manifest.CloneAnalyzeArtifactCollection(projection.invocation)
}
}
func analyzeProjectionOutputs(records map[string]manifest.AnalyzeArtifactRecord, producerRunID string) []manifest.ArtifactRecord {

28
internal/app/analyze_projection_test.go
View File

@@ -3,8 +3,6 @@ package app
import (
"context"
"errors"
"os"
"path/filepath"
"reflect"
"strings"
"testing"
@@ -26,7 +24,7 @@ func (s projectionStage) Run(_ context.Context, env *stage.Env, m *manifest.Mani
return s.run(env, m)
}
func TestExecuteStagesProjectsSeparateSessionAndInvocationAnalyzeState(t *testing.T) {
func TestExecuteStagesProjectsReconciledSessionAnalyzeState(t *testing.T) {
cfg := testConfig(t)
oldAt := time.Date(2026, 5, 3, 9, 0, 0, 0, time.UTC)
oldRecord := appAnalyzeRecord("player_handout", manifest.AnalyzeArtifactCurrent, "old-run", oldAt)
@@ -71,17 +69,6 @@ func TestExecuteStagesProjectsSeparateSessionAndInvocationAnalyzeState(t *testin
t.Fatalf("session aggregate logs = %#v", analyze.Logs)
}
runManifest, err := store.LoadRun(context.Background(), summary.RunManifestPath)
if err != nil {
t.Fatalf("LoadRun() error = %v", err)
}
runAnalyze := runManifest.Stages["analyze"]
if len(runAnalyze.AnalyzeArtifacts) != 2 || runAnalyze.AnalyzeArtifacts["session_recap"].Status != manifest.AnalyzeArtifactCurrent {
t.Fatalf("invocation analyze state = %#v", runAnalyze.AnalyzeArtifacts)
}
if got := analyzeArtifactOutputKeys(runAnalyze.Outputs); !reflect.DeepEqual(got, []string{"session_recap"}) {
t.Fatalf("invocation outputs = %#v, want produced artifact only", got)
}
}
func TestExecuteStagesPersistsRestrictedAnalyzeStateOnPartialError(t *testing.T) {
@@ -137,19 +124,6 @@ func TestExecuteStagesPersistsRestrictedAnalyzeStateOnPartialError(t *testing.T)
t.Fatalf("publish status = %q, want stale", loaded.Stages["publish"].Status)
}
runsDir := artifacts.SessionRunsDirForCampaign(cfg.Pipeline.Workspace.Root, cfg.Session.Campaign, cfg.Session.SessionID)
entries, err := os.ReadDir(runsDir)
if err != nil || len(entries) != 1 {
t.Fatalf("run directory entries = %#v, error = %v", entries, err)
}
runManifest, err := store.LoadRun(context.Background(), filepath.Join(runsDir, entries[0].Name(), "manifest.json"))
if err != nil {
t.Fatalf("LoadRun() error = %v", err)
}
runAnalyze := runManifest.Stages["analyze"]
if runAnalyze.Status != manifest.StatusFailed || len(runAnalyze.AnalyzeArtifacts) != 2 || len(runAnalyze.Outputs) != 0 {
t.Fatalf("partial invocation projection = %#v", runAnalyze)
}
}
func TestExecuteStagesRejectsInvalidAnalyzeProjectionWithoutReplacingPriorState(t *testing.T) {

47
internal/app/assembled_workflow_test.go
View File

@@ -19,7 +19,7 @@ import (
"gitea.maximumdirect.net/eric/narratio/internal/stage"
)
func TestAssembledFullRunUsesCanonicalOrderAndBoundedRunManifests(t *testing.T) {
func TestAssembledFullRunUsesCanonicalOrder(t *testing.T) {
cfg := testConfig(t)
canonical := []string{
"prepare", "transcribe", "merge", "polish", "normalize", "trim",
@@ -37,13 +37,6 @@ func TestAssembledFullRunUsesCanonicalOrderAndBoundedRunManifests(t *testing.T)
if !reflect.DeepEqual(order, canonical) || !reflect.DeepEqual(summary.Executed, canonical) {
t.Fatalf("execution order=%#v summary=%#v, want %#v", order, summary.Executed, canonical)
}
runManifest, err := (&manifest.LocalStore{}).LoadRun(context.Background(), summary.RunManifestPath)
if err != nil {
t.Fatal(err)
}
if !reflect.DeepEqual(runManifest.RequestedStages, canonical) {
t.Fatalf("requested stages = %#v, want canonical order", runManifest.RequestedStages)
}
}
func TestAssembledCanonicalAndAliasArtifactRegenerationRequestsMatch(t *testing.T) {
@@ -103,7 +96,7 @@ func TestAssembledForcedSiblingIndependenceAndFailureBoundary(t *testing.T) {
plan := mustBoundedPlan(t, selected, selected)
runs := 0
plan.stages = []stage.Stage{countingStage{name: selected, runs: &runs}}
summary, err := executePlan(context.Background(), cfg, plan, RunOptions{Force: true})
_, err := executePlan(context.Background(), cfg, plan, RunOptions{Force: true})
if err != nil {
t.Fatal(err)
}
@@ -120,13 +113,6 @@ func TestAssembledForcedSiblingIndependenceAndFailureBoundary(t *testing.T) {
t.Fatalf("%s status = %q, want stale", dependent, loaded.Stages[dependent].Status)
}
}
runManifest, err := (&manifest.LocalStore{}).LoadRun(context.Background(), summary.RunManifestPath)
if err != nil {
t.Fatal(err)
}
if !reflect.DeepEqual(runManifest.RequestedStages, []string{selected}) || len(runManifest.Stages) != 1 {
t.Fatalf("bounded run manifest = %#v", runManifest)
}
})
}
@@ -231,20 +217,13 @@ func TestAssembledLegacyAnalyzeTransitionPublishesOnlyCurrentRecords(t *testing.
t.Fatalf("legacy unselected handout catalog entry = %#v, present=%v", entry, ok)
}
full, err := executeStages(context.Background(), cfg, []stage.Stage{analyze}, RunOptions{Env: &Env{Scriptorium: fake}})
_, err = executeStages(context.Background(), cfg, []stage.Stage{analyze}, RunOptions{Env: &Env{Scriptorium: fake}})
if err != nil {
t.Fatalf("full regeneration: %v", err)
}
if len(fake.RunRequests) != 2 || fake.RunRequests[1].PromptID != "dnd.player_handout" {
t.Fatalf("full requests = %#v, want only missing handout added", fake.RunRequests)
}
fullRun, err := (&manifest.LocalStore{}).LoadRun(context.Background(), full.RunManifestPath)
if err != nil {
t.Fatal(err)
}
if got := analyzeArtifactOutputKeys(fullRun.Stages["analyze"].Outputs); !reflect.DeepEqual(got, []string{"player_handout"}) {
t.Fatalf("full invocation outputs = %#v, want newly generated handout only", got)
}
afterFull := loadAssembledManifest(t, cfg)
for _, key := range []string{"player_handout", "session_recap"} {
if afterFull.Stages["analyze"].AnalyzeArtifacts[key].Status != manifest.AnalyzeArtifactCurrent {
@@ -257,7 +236,7 @@ func TestAssembledLegacyAnalyzeTransitionPublishesOnlyCurrentRecords(t *testing.
t.Fatal(err)
}
remote := &storage.FakeBackend{}
published, err := executeStages(context.Background(), cfg, []stage.Stage{publish}, RunOptions{Env: &Env{ObjectStore: remote}})
_, err = executeStages(context.Background(), cfg, []stage.Stage{publish}, RunOptions{Env: &Env{ObjectStore: remote}})
if err != nil {
t.Fatalf("publish current records: %v", err)
}
@@ -265,13 +244,6 @@ func TestAssembledLegacyAnalyzeTransitionPublishesOnlyCurrentRecords(t *testing.
if got := afterPublish.Stages["publish"].Metadata["published_files_uploaded"]; got != float64(2) {
t.Fatalf("published files = %#v, want 2", got)
}
publishRun, err := (&manifest.LocalStore{}).LoadRun(context.Background(), published.RunManifestPath)
if err != nil {
t.Fatal(err)
}
if !reflect.DeepEqual(publishRun.RequestedStages, []string{"publish"}) || publishRun.Stages["analyze"] != nil {
t.Fatalf("publish run manifest = %#v", publishRun)
}
}
func TestAssembledSplitBundleCommandsUseOneEffectiveFamilyConfiguration(t *testing.T) {
@@ -445,14 +417,9 @@ func assertAssembledProvenance(t *testing.T, store *manifest.LocalStore, summary
if err != nil {
t.Fatal(err)
}
run, err := store.LoadRun(context.Background(), summary.RunManifestPath)
if err != nil {
t.Fatal(err)
}
for _, value := range []*manifest.EffectiveConfigProvenance{session.EffectiveConfig, run.EffectiveConfig} {
if value == nil || value.SelectedProfile == nil || value.SelectedProfile.Name != profile || value.EffectiveConfigDigest != digest {
t.Fatalf("effective configuration provenance = %#v, want profile=%q digest=%q", value, profile, digest)
}
value := session.EffectiveConfig
if value == nil || value.SelectedProfile == nil || value.SelectedProfile.Name != profile || value.EffectiveConfigDigest != digest {
t.Fatalf("effective configuration provenance = %#v, want profile=%q digest=%q", value, profile, digest)
}
}

6
internal/app/config_provenance.go
View File

@@ -32,12 +32,6 @@ func applyEffectiveConfigProvenance(m *manifest.Manifest, cfg *config.Config) {
}
}
func applyEffectiveConfigProvenanceToRun(m *manifest.RunManifest, cfg *config.Config) {
if m != nil {
m.EffectiveConfig = effectiveConfigProvenance(cfg)
}
}
func effectiveConfigSummary(cfg *config.Config) string {
provenance := effectiveConfigProvenance(cfg)
if provenance == nil {

10
internal/app/extract_lifecycle_test.go
View File

@@ -414,7 +414,7 @@ func TestExtractLifecycleForcedFailureInvalidatesDownstream(t *testing.T) {
cfg, env, runner := extractionLifecycleFixture(t, true)
plan, _ := BuildSingleStagePlan("extract")
first, err := executeStages(context.Background(), cfg, plan, RunOptions{Env: env})
_, err := executeStages(context.Background(), cfg, plan, RunOptions{Env: env})
if err != nil {
t.Fatalf("initial executeStages() error = %v", err)
}
@@ -440,14 +440,6 @@ func TestExtractLifecycleForcedFailureInvalidatesDownstream(t *testing.T) {
if len(failed.Outputs) != 0 || len(failed.Logs) != 0 || len(failed.GeneratedConfigs) != 0 || len(failed.Metadata) != 0 {
t.Fatalf("failed replacement inherited extraction result details: %#v", failed)
}
historical, err := (&manifest.LocalStore{}).LoadRun(context.Background(), first.RunManifestPath)
if err != nil {
t.Fatalf("LoadRun(initial) error = %v", err)
}
historicalExtract := historical.Stages["extract"]
if historicalExtract == nil || historicalExtract.Status != manifest.StatusSucceeded || len(historicalExtract.Outputs) == 0 || len(historicalExtract.Logs) == 0 || len(historicalExtract.Metadata) == 0 {
t.Fatalf("historical extraction result = %#v, want preserved succeeded details", historicalExtract)
}
if _, err := os.Stat(succeeded.Outputs[0].LocalPath); err != nil {
t.Fatalf("durable extraction output was not preserved: %v", err)
}

7
internal/app/operator_helpers_test.go
View File

@@ -1455,13 +1455,6 @@ func writeValidPublishRunConfigFiles(t *testing.T, workspaceRoot string) (string
if err := store.Save(ctx, path, m); err != nil {
t.Fatalf("save manifest: %v", err)
}
runManifestPath := artifacts.SessionRunManifestPathForCampaign(cfg.Pipeline.Workspace.Root, cfg.Session.Campaign, cfg.Session.SessionID, m.RunID)
if err := os.MkdirAll(filepath.Dir(runManifestPath), 0o755); err != nil {
t.Fatalf("mkdir run manifest: %v", err)
}
if err := os.WriteFile(runManifestPath, []byte("{}\n"), 0o644); err != nil {
t.Fatalf("write run manifest: %v", err)
}
return pipelinePath, campaignPath, sessionPath
}

23
internal/app/restore_workflow_test.go
View File

@@ -131,25 +131,12 @@ func TestRestoreThenRunStageForceAnalyzeUsesRestoredDurableState(t *testing.T) {
t.Fatalf("post-analyze stage record = %#v, want succeeded", analyzeRec)
}
runManifestPaths, err := filepath.Glob(filepath.Join(sessionRoot, "runs", "*", "manifest.json"))
if err != nil {
t.Fatalf("glob run manifests: %v", err)
runRoot := artifacts.SessionRunRootForCampaign(cfg.Pipeline.Workspace.Root, cfg.Session.Campaign, cfg.Session.SessionID, afterAnalyze.RunID)
if info, statErr := os.Stat(runRoot); statErr != nil || !info.IsDir() {
t.Fatalf("run directory %q = %#v, %v; want directory", runRoot, info, statErr)
}
if len(runManifestPaths) != 1 {
t.Fatalf("run manifest count = %d, want 1; paths=%v", len(runManifestPaths), runManifestPaths)
}
runManifest, err := manifestStore.LoadRun(context.Background(), runManifestPaths[0])
if err != nil {
t.Fatalf("load run manifest %q: %v", runManifestPaths[0], err)
}
if len(runManifest.RequestedStages) != 1 || runManifest.RequestedStages[0] != "analyze" {
t.Fatalf("run manifest requested_stages = %#v, want [analyze]", runManifest.RequestedStages)
}
if runManifest.Stages["analyze"] == nil || runManifest.Stages["analyze"].Status != manifest.StatusSucceeded {
t.Fatalf("run manifest analyze stage = %#v, want succeeded", runManifest.Stages["analyze"])
}
if runManifest.Stages["prepare"] != nil {
t.Fatalf("run manifest should not include upstream prepare stage, got %#v", runManifest.Stages["prepare"])
if _, statErr := os.Stat(filepath.Join(runRoot, "manifest.json")); !os.IsNotExist(statErr) {
t.Fatalf("run manifest stat = %v, want not exist", statErr)
}
}

245
internal/app/runner.go
View File

@@ -17,6 +17,7 @@ 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"
@@ -27,17 +28,15 @@ type RunOptions struct {
SelectedArtifacts []string
EffectiveArtifacts artifacts.EffectiveArtifactSet
Env *Env
RunManifestStore manifest.RunStore
}
type RunSummary struct {
SessionID string
RunID string
ManifestPath string
RunManifestPath string
StageNames []string
Executed []string
Skipped []string
SessionID string
RunID string
ManifestPath string
StageNames []string
Executed []string
Skipped []string
}
var executeStagesFn = executePlan
@@ -151,46 +150,23 @@ func executePlan(ctx context.Context, cfg *config.Config, plan BoundedPlan, opts
if err := env.ManifestStore.Save(ctx, manifestPath, m); err != nil {
return nil, fmt.Errorf("save manifest identity %q: %w", manifestPath, err)
}
runManifestPath := artifacts.SessionRunManifestPathForCampaign(
cfg.Pipeline.Workspace.Root,
identity.Campaign,
identity.SessionID,
identity.RunID,
)
runManifestStore := opts.RunManifestStore
if runManifestStore == nil {
runManifestStore = &manifest.LocalStore{}
}
runManifest, err := runManifestStore.CreateRun(
ctx,
identity.SessionID,
identity.Campaign,
identity.RunID,
opts.Force,
requestedStageNames(stages),
)
if err != nil {
return nil, fmt.Errorf("create run manifest: %w", err)
}
identity.applyToRunManifest(runManifest, manifestPath)
applyEffectiveConfigProvenanceToRun(runManifest, cfg)
if err := runManifestStore.SaveRun(ctx, runManifestPath, runManifest); err != nil {
return nil, persistTerminalFailure(
ctx, env.ManifestStore, manifestPath, m, runManifestStore, runManifestPath, runManifest,
fmt.Errorf("save initial run manifest %q: %w", runManifestPath, 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, persistTerminalFailure(
ctx, env.ManifestStore, manifestPath, m, runManifestStore, runManifestPath, runManifest,
return nil, persistSessionTerminalFailure(
ctx, env.ManifestStore, manifestPath, m,
fmt.Errorf("load secrets from files: %w", err),
)
}
if env.WhisperX == nil && stagesContainAny(stages, "transcribe") {
client, err := buildDefaultWhisperXClient(env.Config)
if err != nil {
return nil, persistTerminalFailure(
ctx, env.ManifestStore, manifestPath, m, runManifestStore, runManifestPath, runManifest,
return nil, persistSessionTerminalFailure(
ctx, env.ManifestStore, manifestPath, m,
fmt.Errorf("initialize whisperx client: %w", err),
)
}
@@ -199,8 +175,8 @@ func executePlan(ctx context.Context, cfg *config.Config, plan BoundedPlan, opts
if env.Seriatim == nil && stagesContainAny(stages, "merge", "normalize", "trim", "render") {
runner, err := buildDefaultSeriatimRunner(env.Config)
if err != nil {
return nil, persistTerminalFailure(
ctx, env.ManifestStore, manifestPath, m, runManifestStore, runManifestPath, runManifest,
return nil, persistSessionTerminalFailure(
ctx, env.ManifestStore, manifestPath, m,
fmt.Errorf("initialize seriatim runner: %w", err),
)
}
@@ -209,8 +185,8 @@ func executePlan(ctx context.Context, cfg *config.Config, plan BoundedPlan, opts
if env.Audita == nil && stagesContainAny(stages, "polish") {
runner, err := buildDefaultAuditaRunner(env.Config)
if err != nil {
return nil, persistTerminalFailure(
ctx, env.ManifestStore, manifestPath, m, runManifestStore, runManifestPath, runManifest,
return nil, persistSessionTerminalFailure(
ctx, env.ManifestStore, manifestPath, m,
fmt.Errorf("initialize audita runner: %w", err),
)
}
@@ -225,8 +201,8 @@ func executePlan(ctx context.Context, cfg *config.Config, plan BoundedPlan, opts
if env.ObjectStore == nil && needsObjectStoreForRun(env.Config, stages, effectiveArtifacts) {
objectStore, err := newCommandObjectStore(ctx, env.Config, nil)
if err != nil {
return nil, persistTerminalFailure(
ctx, env.ManifestStore, manifestPath, m, runManifestStore, runManifestPath, runManifest, err,
return nil, persistSessionTerminalFailure(
ctx, env.ManifestStore, manifestPath, m, err,
)
}
env.ObjectStore = objectStore
@@ -234,8 +210,8 @@ func executePlan(ctx context.Context, cfg *config.Config, plan BoundedPlan, opts
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,
return nil, persistSessionTerminalFailure(
ctx, env.ManifestStore, manifestPath, m,
fmt.Errorf("load remote publish locks: %w", err),
)
}
@@ -263,8 +239,8 @@ func executePlan(ctx context.Context, cfg *config.Config, plan BoundedPlan, opts
runNames = append(runNames, s.Name())
semanticConfig, err := currentStageSemanticConfig(s, stageEnv)
if err != nil {
return nil, persistTerminalFailure(
ctx, env.ManifestStore, manifestPath, m, runManifestStore, runManifestPath, runManifest,
return nil, persistSessionTerminalFailure(
ctx, env.ManifestStore, manifestPath, m,
fmt.Errorf("fingerprint semantic configuration for stage %q: %w", s.Name(), err),
)
}
@@ -273,8 +249,8 @@ func executePlan(ctx context.Context, cfg *config.Config, plan BoundedPlan, opts
if action == stageActionSkip {
validation, err := evaluateStageResume(ctx, s, stageEnv, m, semanticConfig)
if err != nil {
return nil, persistTerminalFailure(
ctx, env.ManifestStore, manifestPath, m, runManifestStore, runManifestPath, runManifest,
return nil, persistSessionTerminalFailure(
ctx, env.ManifestStore, manifestPath, m,
fmt.Errorf("validate resume for stage %q: %w", s.Name(), err),
)
}
@@ -284,14 +260,14 @@ func executePlan(ctx context.Context, cfg *config.Config, plan BoundedPlan, opts
if _, err := invalidateDependentSucceededStagesWithReason(
m, s.Name(), staleAt, staleReasonNotResumable,
); err != nil {
return nil, persistTerminalFailure(
ctx, env.ManifestStore, manifestPath, m, runManifestStore, runManifestPath, runManifest,
return nil, persistSessionTerminalFailure(
ctx, env.ManifestStore, manifestPath, m,
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, persistTerminalFailure(
ctx, env.ManifestStore, manifestPath, m, runManifestStore, runManifestPath, runManifest,
return nil, persistSessionTerminalFailure(
ctx, env.ManifestStore, manifestPath, m,
fmt.Errorf("save manifest after resume validation for stage %q: %w", s.Name(), err),
)
}
@@ -302,16 +278,6 @@ func executePlan(ctx context.Context, cfg *config.Config, plan BoundedPlan, opts
if action == stageActionSkip {
skipped = append(skipped, s.Name())
skipAt := nowUTC()
runManifest.SetStageAction(s.Name(), manifest.RunStageActionSkip, skipAt)
runManifest.MarkStageSkipped(s.Name(), skipAt, "already_succeeded")
setRunStageSemanticConfig(runManifest, s.Name(), semanticConfig)
if err := runManifestStore.SaveRun(ctx, runManifestPath, runManifest); err != nil {
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)
continue
}
@@ -320,20 +286,11 @@ func executePlan(ctx context.Context, cfg *config.Config, plan BoundedPlan, opts
priorAnalyzeState := captureAnalyzeState(m, s.Name())
now := nowUTC()
runManifest.SetStageAction(s.Name(), manifest.RunStageActionRun, now)
runManifest.MarkStageRunning(s.Name(), now)
if err := runManifestStore.SaveRun(ctx, runManifestPath, runManifest); err != nil {
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)
if opts.Force {
if _, err := invalidateDependentSucceededStagesWithReason(m, s.Name(), now, staleReasonForcedReplacement); err != nil {
return nil, persistTerminalFailure(
ctx, env.ManifestStore, manifestPath, m, runManifestStore, runManifestPath, runManifest,
return nil, persistSessionTerminalFailure(
ctx, env.ManifestStore, manifestPath, m,
fmt.Errorf("invalidate dependents before forced stage %q: %w", s.Name(), err),
)
}
@@ -345,8 +302,8 @@ func executePlan(ctx context.Context, cfg *config.Config, plan BoundedPlan, opts
if _, invalidationErr := invalidateDependentSucceededStagesWithReason(m, s.Name(), nowUTC(), staleReasonFailure); invalidationErr != nil {
operationErr = errors.Join(operationErr, fmt.Errorf("invalidate dependents after stage %q persistence failure: %w", s.Name(), invalidationErr))
}
return nil, persistTerminalFailure(
ctx, env.ManifestStore, manifestPath, m, runManifestStore, runManifestPath, runManifest, operationErr,
return nil, persistSessionTerminalFailure(
ctx, env.ManifestStore, manifestPath, m, operationErr,
)
}
env.Logger.Debug("manifest saved", "stage", s.Name(), "transition", "running", "path", manifestPath)
@@ -367,18 +324,16 @@ func executePlan(ctx context.Context, cfg *config.Config, plan BoundedPlan, opts
}
if err != nil {
if analyzeProjection != nil {
applyAnalyzeProjection(m, runManifest, analyzeProjection)
applyAnalyzeProjection(m, analyzeProjection)
}
failedAt := nowUTC()
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))
}
runManifest.MarkStageFailed(s.Name(), failedAt, err.Error())
identity.applyToRunManifest(runManifest, manifestPath)
env.Logger.Info("stage failed", "stage", s.Name(), "error", err)
return nil, persistTerminalFailure(
ctx, env.ManifestStore, manifestPath, m, runManifestStore, runManifestPath, runManifest,
return nil, persistSessionTerminalFailure(
ctx, env.ManifestStore, manifestPath, m,
fmt.Errorf("stage %q failed: %w", s.Name(), err),
)
}
@@ -390,48 +345,36 @@ func executePlan(ctx context.Context, cfg *config.Config, plan BoundedPlan, opts
setSessionStageSemanticConfig(m, s.Name(), semanticConfig)
if !priorOutcome.isSameSelfSkip(result.SkipReason) {
if _, err := invalidateDependentSucceededStagesWithReason(m, s.Name(), skippedAt, staleReasonSelfSkip); err != nil {
return nil, persistTerminalFailure(
ctx, env.ManifestStore, manifestPath, m, runManifestStore, runManifestPath, runManifest,
return nil, persistSessionTerminalFailure(
ctx, env.ManifestStore, manifestPath, m,
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, persistTerminalFailure(
ctx, env.ManifestStore, manifestPath, m, runManifestStore, runManifestPath, runManifest,
return nil, persistSessionTerminalFailure(
ctx, env.ManifestStore, manifestPath, m,
fmt.Errorf("save manifest after self-skip %q: %w", s.Name(), err),
)
}
runManifest.MarkStageSkipped(s.Name(), skippedAt, result.SkipReason)
applyStageResultToRunManifest(runManifest, s.Name(), result)
setRunStageSemanticConfig(runManifest, s.Name(), semanticConfig)
identity.applyToRunManifest(runManifest, manifestPath)
if err := runManifestStore.SaveRun(ctx, runManifestPath, runManifest); err != nil {
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.Info("stage skipped", "stage", s.Name(), "reason", result.SkipReason)
continue
}
sessionOutputs := mapResultOutputs(s.Name(), result, runID)
runOutputs := sessionOutputs
if analyzeProjection != nil {
sessionOutputs = analyzeProjectionOutputs(analyzeProjection.session, "")
runOutputs = analyzeProjectionOutputs(analyzeProjection.invocation, runID)
}
succeededAt := nowUTC()
m.MarkStageSucceeded(s.Name(), succeededAt, sessionOutputs)
applyAnalyzeProjection(m, runManifest, analyzeProjection)
applyAnalyzeProjection(m, analyzeProjection)
applyStageResultToManifest(m, s.Name(), result)
setSessionStageSemanticConfig(m, s.Name(), semanticConfig)
if !priorOutcome.exists || priorOutcome.status != manifest.StatusSucceeded {
if _, err := invalidateDependentSucceededStagesWithReason(m, s.Name(), succeededAt, staleReasonChangedResult); err != nil {
return nil, persistTerminalFailure(
ctx, env.ManifestStore, manifestPath, m, runManifestStore, runManifestPath, runManifest,
return nil, persistSessionTerminalFailure(
ctx, env.ManifestStore, manifestPath, m,
fmt.Errorf("invalidate dependents after changed stage %q result: %w", s.Name(), err),
)
}
@@ -446,39 +389,16 @@ func executePlan(ctx context.Context, cfg *config.Config, plan BoundedPlan, opts
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))
}
runManifest.MarkStageFailed(s.Name(), failedAt, operationErr.Error())
}
return nil, persistTerminalFailure(
ctx, env.ManifestStore, manifestPath, m, runManifestStore, runManifestPath, runManifest,
return nil, persistSessionTerminalFailure(
ctx, env.ManifestStore, manifestPath, m,
operationErr,
)
}
runManifest.MarkStageSucceeded(s.Name(), succeededAt, runOutputs)
applyStageResultToRunManifest(runManifest, s.Name(), result)
setRunStageSemanticConfig(runManifest, s.Name(), semanticConfig)
identity.applyToRunManifest(runManifest, manifestPath)
if err := runManifestStore.SaveRun(ctx, runManifestPath, runManifest); err != nil {
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.Info("stage succeeded", "stage", s.Name())
}
completedAt := nowUTC()
runManifest.MarkSucceeded(completedAt)
identity.applyToRunManifest(runManifest, manifestPath)
if err := runManifestStore.SaveRun(ctx, runManifestPath, runManifest); err != nil {
return nil, persistTerminalFailure(
ctx, env.ManifestStore, manifestPath, m, runManifestStore, runManifestPath, runManifest,
fmt.Errorf("save final run manifest %q: %w", runManifestPath, err),
)
}
// The run record lives inside the run work directory, which cleanup may
// remove. Persist its completed publishing result before cleanup starts so a
// successful deletion cannot be undone by a later diagnostic write.
if containsStage(executed, "publish") {
if err := runPostPublishCleanup(ctx, env, manifestPath, m, executed); err != nil {
return nil, persistPostPublishCleanupFailure(
@@ -489,13 +409,12 @@ func executePlan(ctx context.Context, cfg *config.Config, plan BoundedPlan, opts
}
return &RunSummary{
SessionID: cfg.Session.SessionID,
RunID: runID,
ManifestPath: manifestPath,
RunManifestPath: runManifestPath,
StageNames: runNames,
Executed: executed,
Skipped: skipped,
SessionID: cfg.Session.SessionID,
RunID: runID,
ManifestPath: manifestPath,
StageNames: runNames,
Executed: executed,
Skipped: skipped,
}, nil
}
@@ -536,14 +455,11 @@ func persistPostPublishCleanupFailure(
return errors.Join(operationErr, wrapTerminalPersistenceError("save terminal session manifest", sessionErr))
}
func persistTerminalFailure(
func persistSessionTerminalFailure(
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 {
@@ -555,18 +471,10 @@ func persistTerminalFailure(
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),
)
}
@@ -898,51 +806,6 @@ func (identity invocationIdentity) applyToSessionManifest(m *manifest.Manifest)
m.S3RunPrefix = identity.S3RunPrefix
}
func (identity invocationIdentity) applyToRunManifest(m *manifest.RunManifest, sessionManifestPath string) {
if m == nil {
return
}
m.SessionID = identity.SessionID
m.Campaign = identity.Campaign
m.RunID = identity.RunID
m.SessionManifestPath = sessionManifestPath
m.LocalWorkDir = identity.LocalWorkDir
m.LocalSpoolDir = identity.LocalSpoolDir
m.S3Bucket = identity.S3Bucket
m.S3SessionPrefix = identity.S3SessionPrefix
m.S3RunPrefix = identity.S3RunPrefix
}
func requestedStageNames(stages []stage.Stage) []string {
out := make([]string, 0, len(stages))
for _, s := range stages {
if s == nil {
continue
}
out = append(out, s.Name())
}
return out
}
func applyStageResultToRunManifest(m *manifest.RunManifest, stageName string, result *stage.StageResult) {
if m == nil || result == nil {
return
}
sr := m.Stages[stageName]
if sr == nil {
return
}
if len(result.Logs) > 0 {
sr.Logs = append([]string(nil), result.Logs...)
}
if len(result.GeneratedConfigs) > 0 {
sr.GeneratedConfigs = append([]string(nil), result.GeneratedConfigs...)
}
if len(result.Metadata) > 0 {
sr.Metadata = result.Metadata
}
}
func manifestPathFor(cfg *config.Config) string {
return artifacts.SessionManifestPathForCampaign(
cfg.Pipeline.Workspace.Root,

63
internal/app/runner_terminal_failure_test.go
View File

@@ -31,54 +31,30 @@ func (s *terminalSessionStore) Save(_ context.Context, _ string, m *manifest.Man
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) {
func TestPersistSessionTerminalFailurePreservesCauseAndSessionAuthority(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 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},
{name: "already terminal session", 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,
err := persistSessionTerminalFailure(
context.Background(), sessionStore, "session.json", sessionManifest, tc.cause,
)
if !errors.Is(err, tc.cause) {
t.Fatalf("error = %v, want original cause %v", err, tc.cause)
@@ -86,35 +62,25 @@ func TestPersistTerminalFailureKeepsSessionAuthorityBeforeRunAudit(t *testing.T)
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 got := sessionStore.events; len(got) != 1 || got[0] != "session" {
t.Fatalf("save calls = %v, want [session]", 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})
}, RunOptions{})
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 {
@@ -125,7 +91,7 @@ func TestExecuteStagesTerminalizesCancelledStage(t *testing.T) {
}
}
func TestExecuteStagesResumeValidationFailureTerminalizesRunRecord(t *testing.T) {
func TestExecuteStagesResumeValidationFailureRecordsSessionFailure(t *testing.T) {
cfg := testConfig(t)
store := &manifest.LocalStore{}
seed := manifest.New(cfg.Session.SessionID, time.Now().UTC())
@@ -133,16 +99,11 @@ func TestExecuteStagesResumeValidationFailureTerminalizesRunRecord(t *testing.T)
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)
}, RunOptions{})
if err == nil || !strings.Contains(err.Error(), "inspection unavailable") {
t.Fatalf("executeStages() error = %v, want resume validation failure", err)
}
persisted, loadErr := store.Load(context.Background(), manifestPathFor(cfg))
if loadErr != nil {

91
internal/app/runner_test.go
View File

@@ -264,27 +264,6 @@ func TestExecuteStagesAnalyzeOutputsPersistAsScriptoriumArtifacts(t *testing.T)
t.Fatalf("session output local_path = %q, want %q", sessionOutput.LocalPath, outputPath)
}
runManifest, err := store.LoadRun(context.Background(), summary.RunManifestPath)
if err != nil {
t.Fatalf("load run manifest: %v", err)
}
runStage := runManifest.Stages["analyze"]
if runStage == nil {
t.Fatal("run manifest analyze stage missing")
}
if len(runStage.Outputs) != 1 {
t.Fatalf("run analyze outputs len = %d, want 1", len(runStage.Outputs))
}
runOutput := runStage.Outputs[0]
if runOutput.Kind != "scriptorium_artifact" {
t.Fatalf("run output kind = %q, want scriptorium_artifact", runOutput.Kind)
}
if runOutput.SourceID != "narratio.artifact.session_recap" {
t.Fatalf("run output source_id = %q, want narratio.artifact.session_recap", runOutput.SourceID)
}
if runOutput.LocalPath != outputPath {
t.Fatalf("run output local_path = %q, want %q", runOutput.LocalPath, outputPath)
}
}
func TestMapResultOutputsPrefersExplicitSourceAndCopiesMetadata(t *testing.T) {
@@ -836,7 +815,7 @@ func TestExecuteStagesLoadsExistingManifest(t *testing.T) {
}
}
func TestExecuteStagesCreatesRunManifestPerInvocation(t *testing.T) {
func TestExecuteStagesCreatesRunDirectoryPerInvocation(t *testing.T) {
cfg := testConfig(t)
cfg.Pipeline.Spool.Root = t.TempDir()
cfg.Pipeline.Storage.S3 = &config.StorageS3Config{
@@ -859,18 +838,6 @@ func TestExecuteStagesCreatesRunManifestPerInvocation(t *testing.T) {
if run1.RunID == run2.RunID {
t.Fatalf("expected distinct run ids, got %q", run1.RunID)
}
if run1.RunManifestPath == "" || run2.RunManifestPath == "" {
t.Fatalf("run manifest paths must be set, got %q and %q", run1.RunManifestPath, run2.RunManifestPath)
}
if run1.RunManifestPath == run2.RunManifestPath {
t.Fatalf("expected distinct run manifest paths, got %q", run1.RunManifestPath)
}
for _, path := range []string{run1.RunManifestPath, run2.RunManifestPath} {
if _, statErr := os.Stat(path); statErr != nil {
t.Fatalf("run manifest missing at %q: %v", path, statErr)
}
}
sessionManifest, err := store.Load(context.Background(), run2.ManifestPath)
if err != nil {
t.Fatalf("Load session manifest error = %v", err)
@@ -883,22 +850,12 @@ func TestExecuteStagesCreatesRunManifestPerInvocation(t *testing.T) {
}
for _, run := range []*RunSummary{run1, run2} {
runManifest, err := store.LoadRun(context.Background(), run.RunManifestPath)
if err != nil {
t.Fatalf("LoadRun(%q) error = %v", run.RunManifestPath, err)
}
wantWorkDir := artifacts.SessionRunRootForCampaign(cfg.Pipeline.Workspace.Root, cfg.Session.Campaign, cfg.Session.SessionID, run.RunID)
wantSpoolDir := artifacts.SessionSpoolAudioDir(cfg.Pipeline.Spool.Root, cfg.Session.Campaign, cfg.Session.SessionID, run.RunID)
wantSessionPrefix := artifacts.S3SessionPrefix(cfg.Pipeline.Storage.S3.RootPrefix, cfg.Session.Campaign, cfg.Session.SessionID)
wantRunPrefix := artifacts.S3RunPrefix(wantSessionPrefix, run.RunID)
if runManifest.SessionID != cfg.Session.SessionID || runManifest.Campaign != cfg.Session.Campaign || runManifest.RunID != run.RunID {
t.Fatalf("run identity = (%q, %q, %q), want (%q, %q, %q)", runManifest.SessionID, runManifest.Campaign, runManifest.RunID, cfg.Session.SessionID, cfg.Session.Campaign, run.RunID)
if info, statErr := os.Stat(wantWorkDir); statErr != nil || !info.IsDir() {
t.Fatalf("run directory %q = %#v, %v; want directory", wantWorkDir, info, statErr)
}
if runManifest.SessionManifestPath != run.ManifestPath || runManifest.LocalWorkDir != wantWorkDir || runManifest.LocalSpoolDir != wantSpoolDir {
t.Fatalf("run paths = (%q, %q, %q), want (%q, %q, %q)", runManifest.SessionManifestPath, runManifest.LocalWorkDir, runManifest.LocalSpoolDir, run.ManifestPath, wantWorkDir, wantSpoolDir)
}
if runManifest.S3Bucket != cfg.Pipeline.Storage.S3.Bucket || runManifest.S3SessionPrefix != wantSessionPrefix || runManifest.S3RunPrefix != wantRunPrefix {
t.Fatalf("run remote identity = (%q, %q, %q), want (%q, %q, %q)", runManifest.S3Bucket, runManifest.S3SessionPrefix, runManifest.S3RunPrefix, cfg.Pipeline.Storage.S3.Bucket, wantSessionPrefix, wantRunPrefix)
if _, statErr := os.Stat(filepath.Join(wantWorkDir, "manifest.json")); !os.IsNotExist(statErr) {
t.Fatalf("run manifest stat = %v, want not exist", statErr)
}
}
if sessionManifest.LocalWorkDir == artifacts.SessionRunRootForCampaign(cfg.Pipeline.Workspace.Root, cfg.Session.Campaign, cfg.Session.SessionID, run1.RunID) {
@@ -978,7 +935,7 @@ func TestExecuteStagesRejectsConfiguredCampaignDisagreement(t *testing.T) {
}
}
func TestExecuteStagesRunManifestRecordsSkippedStage(t *testing.T) {
func TestExecuteStagesReportsSkippedStage(t *testing.T) {
cfg := testConfig(t)
store := &manifest.LocalStore{}
manifestPath := manifestPathFor(cfg)
@@ -1000,20 +957,6 @@ func TestExecuteStagesRunManifestRecordsSkippedStage(t *testing.T) {
t.Fatalf("summary = %#v, want skipped transcribe", summary)
}
runManifest, err := store.LoadRun(context.Background(), summary.RunManifestPath)
if err != nil {
t.Fatalf("LoadRun() error = %v", err)
}
sr := runManifest.Stages["transcribe"]
if sr == nil {
t.Fatal("run manifest transcribe stage missing")
}
if sr.Action != manifest.RunStageActionSkip {
t.Fatalf("action = %q, want %q", sr.Action, manifest.RunStageActionSkip)
}
if sr.Status != manifest.StatusSkipped {
t.Fatalf("status = %q, want %q", sr.Status, manifest.StatusSkipped)
}
}
func TestExecuteStagesPersistsSelfSkipAndContinues(t *testing.T) {
@@ -1087,18 +1030,6 @@ func TestExecuteStagesPersistsSelfSkipAndContinues(t *testing.T) {
t.Fatalf("later stage = %#v, want succeeded", later)
}
runManifest, err := store.LoadRun(context.Background(), summary.RunManifestPath)
if err != nil {
t.Fatalf("LoadRun() error = %v", err)
}
runStage := runManifest.Stages["extract"]
if runStage == nil || runStage.Action != manifest.RunStageActionRun || runStage.Status != manifest.StatusSkipped {
t.Fatalf("run optional stage = %#v, want run action with skipped status", runStage)
}
if len(runStage.Logs) != 1 || runStage.Metadata["enabled"] != false {
t.Fatalf("run optional stage details = %#v, want result diagnostics and metadata", runStage)
}
_, err = executeStages(context.Background(), cfg, []stage.Stage{stages[0]}, RunOptions{})
if err != nil {
t.Fatalf("second executeStages() error = %v", err)
@@ -1206,17 +1137,13 @@ func TestExecuteStagesRunLocalArtifactsAndCanonicalSync(t *testing.T) {
}
}
runManifest, err := store.LoadRun(context.Background(), summary.RunManifestPath)
if err != nil {
t.Fatalf("LoadRun() error = %v", err)
}
mergeStage := runManifest.Stages["merge"]
mergeStage := sessionManifest.Stages["merge"]
if mergeStage == nil || len(mergeStage.Logs) == 0 {
t.Fatalf("merge logs missing in run manifest: %#v", mergeStage)
t.Fatalf("merge logs missing in session manifest: %#v", mergeStage)
}
for _, logPath := range mergeStage.Logs {
if !strings.Contains(logPath, filepath.Join("runs", summary.RunID, "merge", "logs")) {
t.Fatalf("run manifest merge log path = %q, want run-local merge logs path", logPath)
t.Fatalf("session manifest merge log path = %q, want run-local merge logs path", logPath)
}
}
}

7
internal/app/semantic_resume.go
View File

@@ -79,13 +79,6 @@ func setSessionStageSemanticConfig(m *manifest.Manifest, name string, value *man
m.Stages[name].SemanticConfig = cloneSemanticConfig(value)
}
func setRunStageSemanticConfig(m *manifest.RunManifest, name string, value *manifest.SemanticConfigFingerprint) {
if m == nil || m.Stages == nil || m.Stages[name] == nil {
return
}
m.Stages[name].SemanticConfig = cloneSemanticConfig(value)
}
func cloneSemanticConfig(value *manifest.SemanticConfigFingerprint) *manifest.SemanticConfigFingerprint {
if value == nil {
return nil

17
internal/app/semantic_resume_test.go
View File

@@ -129,13 +129,6 @@ func TestExecuteStagesSemanticEvidenceLifecycle(t *testing.T) {
if loaded.Stages["render"].SemanticConfig == nil || !loaded.Stages["render"].SemanticConfig.Equal(current) {
t.Fatalf("session evidence = %#v", loaded.Stages["render"].SemanticConfig)
}
run, err := store.LoadRun(context.Background(), summary.RunManifestPath)
if err != nil {
t.Fatal(err)
}
if run.Stages["render"].SemanticConfig == nil || !run.Stages["render"].SemanticConfig.Equal(current) {
t.Fatalf("run evidence = %#v", run.Stages["render"].SemanticConfig)
}
})
}
}
@@ -223,13 +216,9 @@ func TestIntentionalStageSkipPersistsSemanticEvidence(t *testing.T) {
if err != nil {
t.Fatal(err)
}
run, err := store.LoadRun(context.Background(), summary.RunManifestPath)
if err != nil {
t.Fatal(err)
}
if runs != 1 || session.Stages["render"].SemanticConfig == nil || run.Stages["render"].SemanticConfig == nil ||
!session.Stages["render"].SemanticConfig.Equal(current) || !run.Stages["render"].SemanticConfig.Equal(current) {
t.Fatalf("runs=%d session=%#v run=%#v", runs, session.Stages["render"], run.Stages["render"])
if runs != 1 || session.Stages["render"].SemanticConfig == nil ||
!session.Stages["render"].SemanticConfig.Equal(current) {
t.Fatalf("runs=%d session=%#v", runs, session.Stages["render"])
}
}

3
internal/stage/publish.go
View File

@@ -851,6 +851,9 @@ func collectPublishRunFiles(runRoot string, sessionManifest *manifest.Manifest)
continue
}
for _, output := range record.Outputs {
if output.ProducerRunID != "" && output.ProducerRunID != sessionManifest.RunID {
continue
}
if err := add(output.ArchivePath, output.Checksum); err != nil {
return nil, err
}