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

Added run manifest scaffolding and helpers

Browse Source
This commit is contained in:
Eric Rakestraw
2026-05-17 21:15:51 +00:00
parent 550288e008
commit 622677d038
8 changed files with 665 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files

144
internal/app/runner.go
View File

@@ -26,11 +26,13 @@ type RunOptions struct {
}
type RunSummary struct {
SessionID string
ManifestPath string
StageNames []string
Executed []string
Skipped []string
SessionID string
RunID string
ManifestPath string
RunManifestPath string
StageNames []string
Executed []string
Skipped []string
}
func executeStages(ctx context.Context, cfg *config.Config, stages []stage.Stage, opts RunOptions) (*RunSummary, error) {
@@ -110,7 +112,11 @@ func executeStages(ctx context.Context, cfg *config.Config, stages []stage.Stage
if err != nil {
return nil, err
}
identityChanged, err := ensureManifestIdentity(cfg, m)
runID, err := artifacts.NewRunID()
if err != nil {
return nil, fmt.Errorf("generate run id: %w", err)
}
identityChanged, err := ensureManifestIdentity(cfg, m, runID)
if err != nil {
return nil, fmt.Errorf("initialize manifest identity: %w", err)
}
@@ -119,6 +125,29 @@ func executeStages(ctx context.Context, cfg *config.Config, stages []stage.Stage
return nil, fmt.Errorf("save manifest identity %q: %w", manifestPath, err)
}
}
runManifestPath := artifacts.SessionRunManifestPathForCampaign(
cfg.Pipeline.Workspace.Root,
cfg.Session.Campaign,
cfg.Session.SessionID,
runID,
)
runManifestStore := &manifest.LocalStore{}
runManifest, err := runManifestStore.CreateRun(
ctx,
cfg.Session.SessionID,
cfg.Session.Campaign,
runID,
opts.Force,
requestedStageNames(stages),
)
if err != nil {
return nil, fmt.Errorf("create run manifest: %w", err)
}
runManifest.SessionManifestPath = manifestPath
syncRunManifestIdentityFromSession(m, runManifest)
if err := runManifestStore.SaveRun(ctx, runManifestPath, runManifest); err != nil {
return nil, fmt.Errorf("save initial run manifest %q: %w", runManifestPath, err)
}
stageEnv := env
@@ -133,12 +162,23 @@ func executeStages(ctx context.Context, cfg *config.Config, stages []stage.Stage
if d.Action == stageActionSkip {
skipped = append(skipped, s.Name())
skipAt := nowUTC()
runManifest.SetStageAction(s.Name(), manifest.RunStageActionSkip, skipAt)
runManifest.MarkStageSkipped(s.Name(), skipAt, "already_succeeded")
if err := runManifestStore.SaveRun(ctx, runManifestPath, runManifest); err != nil {
return nil, 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
}
executed = append(executed, 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 {
return nil, fmt.Errorf("save run manifest before stage %q: %w", s.Name(), err)
}
m.MarkStageRunning(s.Name(), now)
env.Logger.Info("starting stage", "stage", s.Name())
if err := env.ManifestStore.Save(ctx, manifestPath, m); err != nil {
@@ -148,35 +188,62 @@ func executeStages(ctx context.Context, cfg *config.Config, stages []stage.Stage
result, err := s.Run(ctx, stageEnv, m)
if err != nil {
m.MarkStageFailed(s.Name(), nowUTC(), err.Error())
failedAt := nowUTC()
m.MarkStageFailed(s.Name(), failedAt, err.Error())
if saveErr := env.ManifestStore.Save(ctx, manifestPath, m); saveErr != nil {
return nil, fmt.Errorf("stage %q failed (%v) and manifest save failed (%v)", s.Name(), err, saveErr)
}
runManifest.MarkStageFailed(s.Name(), failedAt, err.Error())
syncRunManifestIdentityFromSession(m, runManifest)
if saveErr := runManifestStore.SaveRun(ctx, runManifestPath, runManifest); saveErr != nil {
return nil, fmt.Errorf("stage %q failed (%v) and run-manifest save failed (%v)", s.Name(), err, saveErr)
}
env.Logger.Info("stage failed", "stage", s.Name(), "error", err)
return nil, fmt.Errorf("stage %q failed: %w", s.Name(), err)
}
outputs := mapResultOutputs(result)
m.MarkStageSucceeded(s.Name(), nowUTC(), outputs)
succeededAt := nowUTC()
m.MarkStageSucceeded(s.Name(), succeededAt, outputs)
applyStageResultToManifest(m, s.Name(), result)
if err := env.ManifestStore.Save(ctx, manifestPath, m); err != nil {
return nil, fmt.Errorf("save manifest after stage %q: %w", s.Name(), err)
}
runManifest.MarkStageSucceeded(s.Name(), succeededAt, outputs)
applyStageResultToRunManifest(runManifest, s.Name(), result)
syncRunManifestIdentityFromSession(m, runManifest)
if err := runManifestStore.SaveRun(ctx, runManifestPath, runManifest); err != nil {
return nil, 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())
}
if err := runPostArchiveCleanup(ctx, env, manifestPath, m, executed); err != nil {
failedAt := nowUTC()
runManifest.MarkFailed(failedAt, err.Error())
syncRunManifestIdentityFromSession(m, runManifest)
if saveErr := runManifestStore.SaveRun(ctx, runManifestPath, runManifest); saveErr != nil {
return nil, fmt.Errorf("post-archive cleanup failed (%v) and run-manifest save failed (%v)", err, saveErr)
}
return nil, fmt.Errorf("post-archive cleanup: %w", err)
}
completedAt := nowUTC()
runManifest.MarkSucceeded(completedAt)
syncRunManifestIdentityFromSession(m, runManifest)
if err := runManifestStore.SaveRun(ctx, runManifestPath, runManifest); err != nil {
return nil, fmt.Errorf("save final run manifest %q: %w", runManifestPath, err)
}
return &RunSummary{
SessionID: cfg.Session.SessionID,
ManifestPath: manifestPath,
StageNames: runNames,
Executed: executed,
Skipped: skipped,
SessionID: cfg.Session.SessionID,
RunID: runID,
ManifestPath: manifestPath,
RunManifestPath: runManifestPath,
StageNames: runNames,
Executed: executed,
Skipped: skipped,
}, nil
}
@@ -357,7 +424,7 @@ func applyStageResultToManifest(m *manifest.Manifest, stageName string, result *
}
}
func ensureManifestIdentity(cfg *config.Config, m *manifest.Manifest) (bool, error) {
func ensureManifestIdentity(cfg *config.Config, m *manifest.Manifest, runID string) (bool, error) {
if cfg == nil || cfg.Pipeline == nil || cfg.Session == nil || m == nil {
return false, nil
}
@@ -373,11 +440,8 @@ func ensureManifestIdentity(cfg *config.Config, m *manifest.Manifest) (bool, err
m.Campaign = campaign
changed = true
}
if m.RunID == "" {
runID, err := artifacts.NewRunID()
if err != nil {
return false, err
}
runID = strings.TrimSpace(runID)
if runID != "" && m.RunID != runID {
m.RunID = runID
changed = true
}
@@ -409,6 +473,48 @@ func ensureManifestIdentity(cfg *config.Config, m *manifest.Manifest) (bool, err
return changed, nil
}
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 syncRunManifestIdentityFromSession(session *manifest.Manifest, run *manifest.RunManifest) {
if session == nil || run == nil {
return
}
run.Campaign = session.Campaign
run.LocalWorkDir = session.LocalWorkDir
run.LocalSpoolDir = session.LocalSpoolDir
run.S3Bucket = session.S3Bucket
run.S3SessionPrefix = session.S3SessionPrefix
run.S3RunPrefix = session.S3RunPrefix
}
func manifestPathFor(cfg *config.Config) string {
return artifacts.SessionManifestPathForCampaign(
cfg.Pipeline.Workspace.Root,

76
internal/app/runner_test.go
View File

@@ -322,6 +322,82 @@ func TestExecuteStagesLoadsExistingManifest(t *testing.T) {
}
}
func TestExecuteStagesCreatesRunManifestPerInvocation(t *testing.T) {
cfg := testConfig(t)
run1, err := executeStages(context.Background(), cfg, []stage.Stage{BuildFullPlan()[0]}, RunOptions{})
if err != nil {
t.Fatalf("first executeStages() error = %v", err)
}
run2, err := executeStages(context.Background(), cfg, []stage.Stage{BuildFullPlan()[0]}, RunOptions{Force: true})
if err != nil {
t.Fatalf("second executeStages() error = %v", err)
}
if run1.RunID == "" || run2.RunID == "" {
t.Fatalf("run ids must be set, got %q and %q", run1.RunID, run2.RunID)
}
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)
}
}
store := &manifest.LocalStore{}
sessionManifest, err := store.Load(context.Background(), run2.ManifestPath)
if err != nil {
t.Fatalf("Load session manifest error = %v", err)
}
if sessionManifest.RunID != run2.RunID {
t.Fatalf("session manifest run_id = %q, want latest run id %q", sessionManifest.RunID, run2.RunID)
}
}
func TestExecuteStagesRunManifestRecordsSkippedStage(t *testing.T) {
cfg := testConfig(t)
store := &manifest.LocalStore{}
manifestPath := manifestPathFor(cfg)
existing := manifest.New(cfg.Session.SessionID, time.Date(2026, 5, 3, 1, 0, 0, 0, time.UTC))
existing.MarkStageSucceeded("transcribe", time.Date(2026, 5, 3, 1, 1, 0, 0, time.UTC), nil)
if err := os.MkdirAll(filepath.Dir(manifestPath), 0o755); err != nil {
t.Fatalf("MkdirAll() error = %v", err)
}
if err := store.Save(context.Background(), manifestPath, existing); err != nil {
t.Fatalf("Save manifest error = %v", err)
}
summary, err := executeStages(context.Background(), cfg, []stage.Stage{BuildFullPlan()[1]}, RunOptions{})
if err != nil {
t.Fatalf("executeStages() error = %v", err)
}
if len(summary.Skipped) != 1 || summary.Skipped[0] != "transcribe" {
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 TestAdapterBackedStageFailureMarksManifestFailed(t *testing.T) {
cases := []struct {
name string

5
internal/artifacts/paths.go
View File

@@ -52,6 +52,11 @@ func SessionRunRootForCampaign(rootDir, campaign, sessionID, runID string) strin
return filepath.Join(SessionRunsDirForCampaign(rootDir, campaign, sessionID), runID)
}
// SessionRunManifestPathForCampaign returns the canonical run manifest path under runs/{run_id}/manifest.json.
func SessionRunManifestPathForCampaign(rootDir, campaign, sessionID, runID string) string {
return filepath.Join(SessionRunRootForCampaign(rootDir, campaign, sessionID, runID), config.PathManifestFile)
}
// SessionRunStageDirForCampaign returns the canonical stage directory under runs/{run_id}/{stage}.
func SessionRunStageDirForCampaign(rootDir, campaign, sessionID, runID, stageName string) string {
return filepath.Join(SessionRunRootForCampaign(rootDir, campaign, sessionID, runID), stageName)

10
internal/artifacts/paths_model_test.go
View File

@@ -48,6 +48,16 @@ func TestSessionRunRootAndStageDirForCampaign(t *testing.T) {
}
}
func TestSessionRunManifestPathForCampaign(t *testing.T) {
root := "/tmp/workspace"
runID := "20260515T031522Z-a1b2c3d4"
got := SessionRunManifestPathForCampaign(root, "forsaken", "2026-04-19", runID)
want := filepath.Join(root, "work", "forsaken", "2026-04-19", "runs", runID, "manifest.json")
if got != want {
t.Fatalf("SessionRunManifestPathForCampaign() = %q, want %q", got, want)
}
}
func TestSessionSpoolAudioDir(t *testing.T) {
root := "/var/spool/narratio"
got := SessionSpoolAudioDir(root, "forsaken", "2026-04-19", "20260515T031522Z-a1b2c3d4")

164
internal/manifest/run_manifest.go Normal file
View File

@@ -0,0 +1,164 @@
package manifest
import (
"strings"
"time"
)
type RunManifestStatus string
const (
RunManifestStatusRunning RunManifestStatus = "running"
RunManifestStatusSucceeded RunManifestStatus = "succeeded"
RunManifestStatusFailed RunManifestStatus = "failed"
)
type RunStageAction string
const (
RunStageActionRun RunStageAction = "run"
RunStageActionSkip RunStageAction = "skip"
)
// RunStageRecord tracks lifecycle and provenance for one stage within a single invocation.
type RunStageRecord struct {
Name string `json:"name"`
Action RunStageAction `json:"action"`
Status StageStatus `json:"status"`
CreatedAt time.Time `json:"created_at"`
UpdatedAt time.Time `json:"updated_at"`
StartedAt *time.Time `json:"started_at,omitempty"`
CompletedAt *time.Time `json:"completed_at,omitempty"`
Outputs []ArtifactRecord `json:"outputs,omitempty"`
Logs []string `json:"logs,omitempty"`
GeneratedConfigs []string `json:"generated_configs,omitempty"`
Error *ErrorRecord `json:"error,omitempty"`
Metadata map[string]any `json:"metadata,omitempty"`
}
// RunManifest is the invocation-scoped execution record under runs/{run_id}/manifest.json.
type RunManifest struct {
SessionID string `json:"session_id"`
Campaign string `json:"campaign,omitempty"`
RunID string `json:"run_id"`
Force bool `json:"force"`
RequestedStages []string `json:"requested_stages,omitempty"`
SessionManifestPath string `json:"session_manifest_path,omitempty"`
LocalWorkDir string `json:"local_workdir,omitempty"`
LocalSpoolDir string `json:"local_spool_dir,omitempty"`
S3Bucket string `json:"s3_bucket,omitempty"`
S3SessionPrefix string `json:"s3_session_prefix,omitempty"`
S3RunPrefix string `json:"s3_run_prefix,omitempty"`
CreatedAt time.Time `json:"created_at"`
UpdatedAt time.Time `json:"updated_at"`
StartedAt *time.Time `json:"started_at,omitempty"`
CompletedAt *time.Time `json:"completed_at,omitempty"`
Status RunManifestStatus `json:"status"`
LastError *ErrorRecord `json:"last_error,omitempty"`
Stages map[string]*RunStageRecord `json:"stages"`
Metadata map[string]any `json:"metadata,omitempty"`
}
// NewRun constructs a new run manifest with deterministic timestamps.
func NewRun(sessionID, campaign, runID string, force bool, requestedStages []string, now time.Time) *RunManifest {
return &RunManifest{
SessionID: strings.TrimSpace(sessionID),
Campaign: strings.TrimSpace(campaign),
RunID: strings.TrimSpace(runID),
Force: force,
RequestedStages: append([]string(nil), requestedStages...),
CreatedAt: now,
UpdatedAt: now,
StartedAt: timePtr(now),
Status: RunManifestStatusRunning,
Stages: map[string]*RunStageRecord{},
}
}
func (m *RunManifest) SetStageAction(name string, action RunStageAction, at time.Time) {
s := m.ensureStage(name, at)
s.Action = action
s.UpdatedAt = at
m.UpdatedAt = at
}
func (m *RunManifest) MarkStageRunning(name string, at time.Time) {
s := m.ensureStage(name, at)
s.Status = StatusRunning
s.StartedAt = timePtr(at)
s.CompletedAt = nil
s.Error = nil
s.UpdatedAt = at
m.UpdatedAt = at
}
func (m *RunManifest) MarkStageSucceeded(name string, at time.Time, outputs []ArtifactRecord) {
s := m.ensureStage(name, at)
s.Status = StatusSucceeded
s.CompletedAt = timePtr(at)
s.Error = nil
s.Outputs = append([]ArtifactRecord(nil), outputs...)
s.UpdatedAt = at
m.UpdatedAt = at
}
func (m *RunManifest) MarkStageFailed(name string, at time.Time, message string) {
s := m.ensureStage(name, at)
s.Status = StatusFailed
s.CompletedAt = timePtr(at)
s.Error = &ErrorRecord{Message: strings.TrimSpace(message), At: timePtr(at)}
s.UpdatedAt = at
m.LastError = &ErrorRecord{Message: strings.TrimSpace(message), At: timePtr(at)}
m.UpdatedAt = at
m.Status = RunManifestStatusFailed
m.CompletedAt = timePtr(at)
}
func (m *RunManifest) MarkStageSkipped(name string, at time.Time, reason string) {
s := m.ensureStage(name, at)
s.Status = StatusSkipped
s.CompletedAt = timePtr(at)
s.Error = &ErrorRecord{Message: strings.TrimSpace(reason), Code: "skipped", At: timePtr(at)}
s.UpdatedAt = at
m.UpdatedAt = at
}
func (m *RunManifest) MarkSucceeded(at time.Time) {
m.Status = RunManifestStatusSucceeded
m.CompletedAt = timePtr(at)
m.UpdatedAt = at
}
func (m *RunManifest) MarkFailed(at time.Time, message string) {
m.Status = RunManifestStatusFailed
m.CompletedAt = timePtr(at)
m.LastError = &ErrorRecord{Message: strings.TrimSpace(message), At: timePtr(at)}
m.UpdatedAt = at
}
func (m *RunManifest) ensureStage(name string, at time.Time) *RunStageRecord {
if m.Stages == nil {
m.Stages = map[string]*RunStageRecord{}
}
stageName := strings.TrimSpace(name)
s, ok := m.Stages[stageName]
if !ok || s == nil {
s = &RunStageRecord{
Name: stageName,
Action: RunStageActionRun,
Status: StatusPending,
CreatedAt: at,
UpdatedAt: at,
}
m.Stages[stageName] = s
}
if s.Name == "" {
s.Name = stageName
}
if s.CreatedAt.IsZero() {
s.CreatedAt = at
}
return s
}

50
internal/manifest/run_manifest_test.go Normal file
View File

@@ -0,0 +1,50 @@
package manifest
import (
"testing"
"time"
)
func TestRunManifestStageMarkHelpers(t *testing.T) {
rm := NewRun("2026-05-03", "forsaken", "20260517T000000Z-abcdef12", false, []string{"prepare"}, time.Date(2026, 5, 3, 10, 0, 0, 0, time.UTC))
if rm.Status != RunManifestStatusRunning {
t.Fatalf("status = %q, want %q", rm.Status, RunManifestStatusRunning)
}
runningAt := time.Date(2026, 5, 3, 10, 1, 0, 0, time.UTC)
rm.SetStageAction("prepare", RunStageActionRun, runningAt)
rm.MarkStageRunning("prepare", runningAt)
rm.MarkStageSucceeded("prepare", runningAt.Add(30*time.Second), []ArtifactRecord{
{Kind: "input", LocalPath: "inputs/session.yml"},
})
stage := rm.Stages["prepare"]
if stage == nil {
t.Fatal("prepare stage missing")
}
if stage.Action != RunStageActionRun {
t.Fatalf("action = %q, want %q", stage.Action, RunStageActionRun)
}
if stage.Status != StatusSucceeded {
t.Fatalf("status = %q, want %q", stage.Status, StatusSucceeded)
}
skippedAt := runningAt.Add(1 * time.Minute)
rm.SetStageAction("notify", RunStageActionSkip, skippedAt)
rm.MarkStageSkipped("notify", skippedAt, "already_succeeded")
skipped := rm.Stages["notify"]
if skipped == nil {
t.Fatal("notify stage missing")
}
if skipped.Action != RunStageActionSkip {
t.Fatalf("action = %q, want %q", skipped.Action, RunStageActionSkip)
}
if skipped.Status != StatusSkipped {
t.Fatalf("status = %q, want %q", skipped.Status, StatusSkipped)
}
rm.MarkSucceeded(skippedAt.Add(10 * time.Second))
if rm.Status != RunManifestStatusSucceeded {
t.Fatalf("status = %q, want %q", rm.Status, RunManifestStatusSucceeded)
}
}

165
internal/manifest/store.go
View File

@@ -129,6 +129,89 @@ func (s *LocalStore) Save(ctx context.Context, path string, m *Manifest) error {
return nil
}
// CreateRun returns a new in-memory run manifest for one invocation.
func (s *LocalStore) CreateRun(
ctx context.Context,
sessionID, campaign, runID string,
force bool,
requestedStages []string,
) (*RunManifest, error) {
if err := checkContext(ctx); err != nil {
return nil, err
}
if strings.TrimSpace(sessionID) == "" {
return nil, fmt.Errorf("create run manifest: session_id is required")
}
if strings.TrimSpace(runID) == "" {
return nil, fmt.Errorf("create run manifest: run_id is required")
}
now := time.Now().UTC()
return NewRun(sessionID, campaign, runID, force, requestedStages, now), nil
}
// LoadRun reads and validates a local JSON run manifest from path.
func (s *LocalStore) LoadRun(ctx context.Context, path string) (*RunManifest, error) {
if err := checkContext(ctx); err != nil {
return nil, err
}
if strings.TrimSpace(path) == "" {
return nil, fmt.Errorf("load run manifest: path is required")
}
data, err := os.ReadFile(path)
if err != nil {
return nil, fmt.Errorf("load run manifest %q: %w", path, err)
}
var m RunManifest
if err := json.Unmarshal(data, &m); err != nil {
return nil, fmt.Errorf("decode run manifest %q: %w", path, err)
}
if err := validateLoadedRunManifest(&m); err != nil {
return nil, fmt.Errorf("run manifest %q invalid: %w", path, err)
}
normalizeRunManifest(&m)
return &m, nil
}
// SaveRun writes the run manifest to path atomically via temp file + rename.
func (s *LocalStore) SaveRun(ctx context.Context, path string, m *RunManifest) error {
if err := checkContext(ctx); err != nil {
return err
}
if strings.TrimSpace(path) == "" {
return fmt.Errorf("save run manifest: path is required")
}
if m == nil {
return fmt.Errorf("save run manifest: manifest is nil")
}
if strings.TrimSpace(m.SessionID) == "" {
return fmt.Errorf("save run manifest: session_id is required")
}
if strings.TrimSpace(m.RunID) == "" {
return fmt.Errorf("save run manifest: run_id is required")
}
if m.CreatedAt.IsZero() {
return fmt.Errorf("save run manifest: created_at is required")
}
m.UpdatedAt = time.Now().UTC()
if m.Stages == nil {
m.Stages = map[string]*RunStageRecord{}
}
data, err := json.MarshalIndent(m, "", " ")
if err != nil {
return fmt.Errorf("save run manifest: marshal: %w", err)
}
data = append(data, '\n')
return writeJSONAtomically(ctx, path, ".run-manifest.json.tmp-*", data)
}
func validateLoadedManifest(m *Manifest) error {
if m == nil {
return fmt.Errorf("manifest is nil")
@@ -161,6 +244,88 @@ func normalizeManifest(m *Manifest) {
}
}
func validateLoadedRunManifest(m *RunManifest) error {
if m == nil {
return fmt.Errorf("manifest is nil")
}
if strings.TrimSpace(m.SessionID) == "" {
return fmt.Errorf("session_id is required")
}
if strings.TrimSpace(m.RunID) == "" {
return fmt.Errorf("run_id is required")
}
if m.CreatedAt.IsZero() {
return fmt.Errorf("created_at is required")
}
if m.UpdatedAt.IsZero() {
return fmt.Errorf("updated_at is required")
}
return nil
}
func normalizeRunManifest(m *RunManifest) {
if m.Stages == nil {
m.Stages = map[string]*RunStageRecord{}
}
for name, stage := range m.Stages {
if stage == nil {
stage = &RunStageRecord{
Name: name,
Action: RunStageActionRun,
Status: StatusPending,
CreatedAt: m.CreatedAt,
UpdatedAt: m.UpdatedAt,
}
m.Stages[name] = stage
}
if stage.Name == "" {
stage.Name = name
}
}
}
func writeJSONAtomically(ctx context.Context, path, tempPattern string, data []byte) error {
dir := filepath.Dir(path)
if err := os.MkdirAll(dir, 0o755); err != nil {
return fmt.Errorf("create directory %q: %w", dir, err)
}
tmp, err := os.CreateTemp(dir, tempPattern)
if err != nil {
return fmt.Errorf("create temp file: %w", err)
}
tmpName := tmp.Name()
removeTmp := true
defer func() {
if removeTmp {
_ = os.Remove(tmpName)
}
}()
if _, err := tmp.Write(data); err != nil {
_ = tmp.Close()
return fmt.Errorf("write temp file: %w", err)
}
if err := tmp.Sync(); err != nil {
_ = tmp.Close()
return fmt.Errorf("sync temp file: %w", err)
}
if err := tmp.Close(); err != nil {
return fmt.Errorf("close temp file: %w", err)
}
if err := checkContext(ctx); err != nil {
return err
}
if err := os.Rename(tmpName, path); err != nil {
return fmt.Errorf("rename temp file: %w", err)
}
removeTmp = false
return nil
}
func checkContext(ctx context.Context) error {
if ctx == nil {
return nil

70
internal/manifest/store_test.go
View File

@@ -150,3 +150,73 @@ func TestLoadRejectsInvalidManifest(t *testing.T) {
t.Fatalf("error = %q, want session_id validation", err.Error())
}
}
func TestLocalStoreCreateSaveLoadRunManifestRoundTrip(t *testing.T) {
store := &LocalStore{}
ctx := context.Background()
run, err := store.CreateRun(
ctx,
"2026-05-03",
"forsaken",
"20260517T000000Z-abcdef12",
true,
[]string{"prepare", "transcribe"},
)
if err != nil {
t.Fatalf("CreateRun() error = %v", err)
}
run.SessionManifestPath = "/var/lib/narratio/work/forsaken/2026-05-03/manifest.json"
run.MarkStageRunning("prepare", time.Date(2026, 5, 3, 12, 1, 0, 0, time.UTC))
run.MarkStageSucceeded("prepare", time.Date(2026, 5, 3, 12, 2, 0, 0, time.UTC), []ArtifactRecord{
{Kind: "input", LocalPath: "inputs/session.yml"},
})
run.MarkSucceeded(time.Date(2026, 5, 3, 12, 3, 0, 0, time.UTC))
path := filepath.Join(t.TempDir(), "run-manifest.json")
if err := store.SaveRun(ctx, path, run); err != nil {
t.Fatalf("SaveRun() error = %v", err)
}
loaded, err := store.LoadRun(ctx, path)
if err != nil {
t.Fatalf("LoadRun() error = %v", err)
}
if loaded.SessionID != "2026-05-03" {
t.Fatalf("SessionID = %q, want %q", loaded.SessionID, "2026-05-03")
}
if loaded.Campaign != "forsaken" {
t.Fatalf("Campaign = %q, want %q", loaded.Campaign, "forsaken")
}
if loaded.RunID != "20260517T000000Z-abcdef12" {
t.Fatalf("RunID = %q, want %q", loaded.RunID, "20260517T000000Z-abcdef12")
}
if loaded.Status != RunManifestStatusSucceeded {
t.Fatalf("Status = %q, want %q", loaded.Status, RunManifestStatusSucceeded)
}
if loaded.Stages["prepare"] == nil || loaded.Stages["prepare"].Status != StatusSucceeded {
t.Fatalf("prepare stage = %#v, want succeeded", loaded.Stages["prepare"])
}
if loaded.Stages["prepare"].Action != RunStageActionRun {
t.Fatalf("prepare action = %q, want %q", loaded.Stages["prepare"].Action, RunStageActionRun)
}
}
func TestLoadRunRejectsInvalidManifest(t *testing.T) {
store := &LocalStore{}
ctx := context.Background()
path := filepath.Join(t.TempDir(), "run-manifest.json")
if err := os.WriteFile(path, []byte(`{"session_id":"2026-05-03"}`), 0o644); err != nil {
t.Fatalf("WriteFile() error = %v", err)
}
_, err := store.LoadRun(ctx, path)
if err == nil {
t.Fatal("expected error, got nil")
}
if !strings.Contains(err.Error(), "run_id is required") {
t.Fatalf("error = %q, want run_id validation", err.Error())
}
}