From 7997f81f47dfaa79f7673427fa19dad2bb7fed67 Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Thu, 3 Sep 2026 04:09:12 +0000 Subject: [PATCH] Make runner persistence session-only --- docs/cli.md | 4 +- docs/internal/manifest.md | 65 ++--- docs/policy/architecture.md | 4 +- docs/roadmap/implementation.md | 2 +- internal/app/analyze_projection.go | 6 - internal/app/analyze_projection_test.go | 28 +-- internal/app/assembled_workflow_test.go | 47 +--- internal/app/config_provenance.go | 6 - internal/app/extract_lifecycle_test.go | 10 +- internal/app/operator_helpers_test.go | 7 - internal/app/restore_workflow_test.go | 23 +- internal/app/runner.go | 245 ++++--------------- internal/app/runner_terminal_failure_test.go | 63 +---- internal/app/runner_test.go | 91 +------ internal/app/semantic_resume.go | 7 - internal/app/semantic_resume_test.go | 17 +- internal/stage/publish.go | 3 + 17 files changed, 123 insertions(+), 505 deletions(-) diff --git a/docs/cli.md b/docs/cli.md index 49231fa..afc641c 100644 --- a/docs/cli.md +++ b/docs/cli.md @@ -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 diff --git a/docs/internal/manifest.md b/docs/internal/manifest.md index 75cb635..bbeb7b4 100644 --- a/docs/internal/manifest.md +++ b/docs/internal/manifest.md @@ -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` diff --git a/docs/policy/architecture.md b/docs/policy/architecture.md index 99964b7..a64bf31 100644 --- a/docs/policy/architecture.md +++ b/docs/policy/architecture.md @@ -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 diff --git a/docs/roadmap/implementation.md b/docs/roadmap/implementation.md index 31f9c76..78d3b11 100644 --- a/docs/roadmap/implementation.md +++ b/docs/roadmap/implementation.md @@ -171,7 +171,7 @@ diagnostic uploads. ## Stage 2 — Make Runner Persistence Session-Only -**Status: Pending** +**Status: Completed** ### Goal diff --git a/internal/app/analyze_projection.go b/internal/app/analyze_projection.go index 8584bac..3ff19c8 100644 --- a/internal/app/analyze_projection.go +++ b/internal/app/analyze_projection.go @@ -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 { diff --git a/internal/app/analyze_projection_test.go b/internal/app/analyze_projection_test.go index 24f0cc0..fa93b94 100644 --- a/internal/app/analyze_projection_test.go +++ b/internal/app/analyze_projection_test.go @@ -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) { diff --git a/internal/app/assembled_workflow_test.go b/internal/app/assembled_workflow_test.go index a4076c5..e1aad5e 100644 --- a/internal/app/assembled_workflow_test.go +++ b/internal/app/assembled_workflow_test.go @@ -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) } } diff --git a/internal/app/config_provenance.go b/internal/app/config_provenance.go index 159d9ef..f75a66a 100644 --- a/internal/app/config_provenance.go +++ b/internal/app/config_provenance.go @@ -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 { diff --git a/internal/app/extract_lifecycle_test.go b/internal/app/extract_lifecycle_test.go index 2e7707b..b7774c2 100644 --- a/internal/app/extract_lifecycle_test.go +++ b/internal/app/extract_lifecycle_test.go @@ -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) } diff --git a/internal/app/operator_helpers_test.go b/internal/app/operator_helpers_test.go index 56d8e77..51cb4de 100644 --- a/internal/app/operator_helpers_test.go +++ b/internal/app/operator_helpers_test.go @@ -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 } diff --git a/internal/app/restore_workflow_test.go b/internal/app/restore_workflow_test.go index 15a4efe..9619b8d 100644 --- a/internal/app/restore_workflow_test.go +++ b/internal/app/restore_workflow_test.go @@ -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) } } diff --git a/internal/app/runner.go b/internal/app/runner.go index 88e9cfe..64c7fcd 100644 --- a/internal/app/runner.go +++ b/internal/app/runner.go @@ -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, diff --git a/internal/app/runner_terminal_failure_test.go b/internal/app/runner_terminal_failure_test.go index c741e8b..3548e72 100644 --- a/internal/app/runner_terminal_failure_test.go +++ b/internal/app/runner_terminal_failure_test.go @@ -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 { diff --git a/internal/app/runner_test.go b/internal/app/runner_test.go index 4c790fe..5f7fdf8 100644 --- a/internal/app/runner_test.go +++ b/internal/app/runner_test.go @@ -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) } } } diff --git a/internal/app/semantic_resume.go b/internal/app/semantic_resume.go index b9e9f7d..44635d9 100644 --- a/internal/app/semantic_resume.go +++ b/internal/app/semantic_resume.go @@ -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 diff --git a/internal/app/semantic_resume_test.go b/internal/app/semantic_resume_test.go index a97fc6e..b1895aa 100644 --- a/internal/app/semantic_resume_test.go +++ b/internal/app/semantic_resume_test.go @@ -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"]) } } diff --git a/internal/stage/publish.go b/internal/stage/publish.go index fc57819..ec9c104 100644 --- a/internal/stage/publish.go +++ b/internal/stage/publish.go @@ -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 }