Complete cleanup stage workstream

This commit is contained in:
2026-09-05 15:42:00 +00:00
parent 078f661e3a
commit c2806a1615
6 changed files with 87 additions and 13 deletions

View File

@@ -183,7 +183,7 @@ manifests remain compatible when those fields are absent. A successful forced
rerun marks only succeeded transitive dependent session-stage records stale. rerun marks only succeeded transitive dependent session-stage records stale.
The application owns a fixed dependency relation distinct from execution order; The application owns a fixed dependency relation distinct from execution order;
dependents are returned in canonical order. Render and extract therefore never dependents are returned in canonical order. Render and extract therefore never
stale one another, while either can stale analyze, publish, and notify. stale one another, while either can stale analyze, publish, notify, and cleanup.
Starting an execution clears the current session-stage record's prior outputs, Starting an execution clears the current session-stage record's prior outputs,
logs, generated configuration references, metadata, and semantic fingerprint. logs, generated configuration references, metadata, and semantic fingerprint.
@@ -216,11 +216,11 @@ match the current resolved stage semantics. Missing semantic evidence, malformed
evidence, or a mismatch makes the stage and its fixed transitive dependents evidence, or a mismatch makes the stage and its fixed transitive dependents
stale. The existing stage-specific resume stale. The existing stage-specific resume
validator runs only after this semantic check succeeds; both checks are validator runs only after this semantic check succeeds; both checks are
required. Extraction and analyze have resume validators and may reject an required. Extraction, analyze, and cleanup have resume validators and may
otherwise eligible skip when their selected durable evidence is obsolete; the reject an otherwise eligible skip when their selected durable evidence is
runner marks the aggregate record stale and executes it. Analyze's validator obsolete; the runner marks the aggregate record stale and executes it.
can still accept a partial selection when only unrelated artifact records are Analyze's validator can still accept a partial selection when only unrelated
stale. artifact records are stale.
Implemented reuse coverage is deliberately split between aggregate semantic Implemented reuse coverage is deliberately split between aggregate semantic
evidence and focused durable validators: evidence and focused durable validators:
@@ -238,6 +238,7 @@ evidence and focused durable validators:
| analyze artifacts | per-artifact fingerprint, reconciliation, and output validator | [analyze](stage-analyze.md), [Scriptorium](../integrations/scriptorium.md) | | analyze artifacts | per-artifact fingerprint, reconciliation, and output validator | [analyze](stage-analyze.md), [Scriptorium](../integrations/scriptorium.md) |
| publish | aggregate semantic fingerprint plus immediate lock/commit checks | [publish](stage-publish.md), [storage adapter](adapters.md) | | publish | aggregate semantic fingerprint plus immediate lock/commit checks | [publish](stage-publish.md), [storage adapter](adapters.md) |
| notify | aggregate delivery-mode fingerprint | [pipeline overview](overview.md), [configuration](../config.md#notifications) | | notify | aggregate delivery-mode fingerprint | [pipeline overview](overview.md), [configuration](../config.md#notifications) |
| cleanup | aggregate semantic fingerprint plus publish-evidence and target-absence validator | [cleanup](stage-cleanup.md), [workspace](workspace.md) |
These contracts record resolved choices Narratio can observe, not operational These contracts record resolved choices Narratio can observe, not operational
runner tuning. External model, module, prompt, profile, and configuration-file runner tuning. External model, module, prompt, profile, and configuration-file

View File

@@ -158,9 +158,10 @@ Supported source families:
artifact failed without output authority, and stales its current transitive artifact failed without output authority, and stales its current transitive
dependents. dependents.
- the runner's ordinary error path marks aggregate analyze failed, clears its - the runner's ordinary error path marks aggregate analyze failed, clears its
aggregate outputs, invalidates publish and notify through the application aggregate outputs, invalidates publish, notify, and cleanup through the
dependency relation, and persists the live retry collection. Terminal-save application dependency relation, and persists the live retry collection.
errors are joined with the original artifact or checkpoint failure. Terminal-save errors are joined with the original artifact or checkpoint
failure.
## Invariants ## Invariants

View File

@@ -285,7 +285,7 @@ documents fail strict decoding rather than entering a compatibility path.
## Stage 5 — Consolidate, Validate, And Close Workstream 9 ## Stage 5 — Consolidate, Validate, And Close Workstream 9
**Status:** Pending **Status:** Completed
### Goal ### Goal

View File

@@ -420,7 +420,7 @@ version of a locked artifact while unlocked artifacts advance.
### Workstream 9 — Cleanup as the final pipeline stage ### Workstream 9 — Cleanup as the final pipeline stage
**Status:** Pending **Status:** Complete
**Audit sources:** H20, H21, C02, and C04. **Audit sources:** H20, H21, C02, and C04.

View File

@@ -1,11 +1,13 @@
package app package app
import ( import (
"bytes"
"context" "context"
"errors" "errors"
"os" "os"
"path/filepath" "path/filepath"
"reflect" "reflect"
"strings"
"testing" "testing"
"time" "time"
@@ -162,6 +164,76 @@ func TestCleanupFailurePreservesPublishAndRetriesWithoutRepublishing(t *testing.
} }
} }
func TestRunStageCleanupUsesPublishEvidenceAndFailsClosedWithoutIt(t *testing.T) {
workspaceRoot := t.TempDir()
pipelinePath, campaignPath, sessionPath := writeValidConfigFiles(t, workspaceRoot)
enableCleanupForCommandTest(t, pipelinePath)
cfg, err := config.Load(pipelinePath, campaignPath, sessionPath)
if err != nil {
t.Fatalf("load cleanup config: %v", err)
}
publishedRunID := "20260904T010203Z-a1b2c3d4"
m := manifest.New(cfg.Session.SessionID, time.Date(2026, 9, 4, 1, 2, 3, 0, time.UTC))
m.Campaign = cfg.Session.Campaign
m.RunID = publishedRunID
m.S3SessionPrefix = artifacts.S3SessionPrefix(cfg.Pipeline.Storage.S3.RootPrefix, m.Campaign, m.SessionID)
m.MarkStageSucceeded("publish", time.Date(2026, 9, 4, 1, 3, 0, 0, time.UTC), nil)
m.Stages["publish"].Metadata = cleanupPublishMetadata(m)
manifestPath := artifacts.SessionManifestPathForCampaign(cfg.Pipeline.Workspace.Root, m.Campaign, m.SessionID)
if err := (&manifest.LocalStore{}).Save(context.Background(), manifestPath, m); err != nil {
t.Fatalf("save publication evidence: %v", err)
}
workspaceTarget := artifacts.SessionRunRootForCampaign(cfg.Pipeline.Workspace.Root, m.Campaign, m.SessionID, publishedRunID)
spoolTarget := artifacts.SessionSpoolAudioDir(cfg.Pipeline.Spool.Root, m.Campaign, m.SessionID, publishedRunID)
for _, target := range []string{workspaceTarget, spoolTarget} {
if err := os.MkdirAll(target, 0o755); err != nil {
t.Fatalf("create cleanup target: %v", err)
}
}
var output bytes.Buffer
if err := RunStage(context.Background(), []string{
"cleanup", cfg.Session.SessionID, "--config", pipelinePath,
"--campaign-file", campaignPath, "--session", sessionPath,
}, &output); err != nil {
t.Fatalf("RunStage(cleanup): %v", err)
}
if !strings.Contains(output.String(), "stage=cleanup executed=1 skipped=0") {
t.Fatalf("output = %q, want direct cleanup execution", output.String())
}
for _, target := range []string{workspaceTarget, spoolTarget} {
if _, statErr := os.Lstat(target); !errors.Is(statErr, os.ErrNotExist) {
t.Fatalf("cleanup target %q stat error = %v, want absent", target, statErr)
}
}
missingRoot := t.TempDir()
missingPipeline, missingCampaign, missingSession := writeValidConfigFiles(t, missingRoot)
enableCleanupForCommandTest(t, missingPipeline)
err = RunStage(context.Background(), []string{
"cleanup", "2026-05-03", "--config", missingPipeline,
"--campaign-file", missingCampaign, "--session", missingSession,
}, &bytes.Buffer{})
if err == nil || !strings.Contains(err.Error(), "cleanup requires a recorded publish stage") {
t.Fatalf("RunStage(cleanup) without evidence error = %v", err)
}
}
func enableCleanupForCommandTest(t *testing.T, pipelinePath string) {
t.Helper()
data, err := os.ReadFile(pipelinePath)
if err != nil {
t.Fatalf("read pipeline config: %v", err)
}
content := strings.Replace(string(data), "workspace:\n root: ", "workspace:\n cleanup_after_publish: true\n root: ", 1)
content = strings.Replace(content, "spool:\n root: ", "spool:\n delete_audio_after_publish: true\n root: ", 1)
content = strings.Replace(content, " upload_run: false", " upload_run: true", 1)
if err := os.WriteFile(pipelinePath, []byte(content), 0o644); err != nil {
t.Fatalf("write cleanup config: %v", err)
}
}
func cleanupOrchestrationConfig(t *testing.T) (*config.Config, *storage.FakeBackend) { func cleanupOrchestrationConfig(t *testing.T) (*config.Config, *storage.FakeBackend) {
t.Helper() t.Helper()
cfg := testConfig(t) cfg := testConfig(t)

View File

@@ -91,10 +91,10 @@ func validateCleanupPublishPolicy(cfg *PipelineConfig) error {
return nil return nil
} }
if cfg.Publish == nil || (cfg.Publish.Enabled != nil && !*cfg.Publish.Enabled) { if cfg.Publish == nil || (cfg.Publish.Enabled != nil && !*cfg.Publish.Enabled) {
return fmt.Errorf("post-publish cleanup policies require pipeline.publish.enabled=true") return fmt.Errorf("cleanup stage policies require pipeline.publish.enabled=true")
} }
if cfg.Publish.UploadRun != nil && !*cfg.Publish.UploadRun { if cfg.Publish.UploadRun != nil && !*cfg.Publish.UploadRun {
return fmt.Errorf("post-publish cleanup policies require pipeline.publish.upload_run=true") return fmt.Errorf("cleanup stage policies require pipeline.publish.upload_run=true")
} }
return nil return nil
} }