Persist partial analysis artifact results
This commit is contained in:
@@ -66,6 +66,23 @@ audit of only the artifacts evaluated or attempted by that run. These records
|
|||||||
remain analyze-owned data inside the fixed stage; they are not dynamic stages
|
remain analyze-owned data inside the fixed stage; they are not dynamic stages
|
||||||
or generic subtasks.
|
or generic subtasks.
|
||||||
|
|
||||||
|
The stage result contract has one analyze-specific projection boundary. On
|
||||||
|
success, the runner validates and deep-copies the complete reconciled session
|
||||||
|
collection and the invocation subset. Aggregate session outputs are rebuilt in
|
||||||
|
configured-key order from current session records only; invocation outputs are
|
||||||
|
limited to current records produced by that invocation's run ID. Ordinary
|
||||||
|
stage outputs cannot accompany this projection, so there is one source of
|
||||||
|
artifact authority.
|
||||||
|
|
||||||
|
Analyze may return a projection together with an error. That restricted result
|
||||||
|
cannot carry ordinary outputs, skip state, aggregate logs, generated configs,
|
||||||
|
or metadata. The runner persists only the validated per-artifact collections,
|
||||||
|
then marks the aggregate analyze and run state failed and invalidates delivery
|
||||||
|
dependents conservatively. Unrelated current records survive because the
|
||||||
|
session projection is complete. A malformed projection is not applied, and a
|
||||||
|
failed session projection save restores the prior per-artifact authority before
|
||||||
|
terminal failure persistence.
|
||||||
|
|
||||||
## Run Manifest
|
## Run Manifest
|
||||||
|
|
||||||
`manifest.RunManifest` is created for each invocation and records:
|
`manifest.RunManifest` is created for each invocation and records:
|
||||||
|
|||||||
@@ -362,6 +362,8 @@ can represent independently current artifacts without treating them as stages.
|
|||||||
|
|
||||||
## Stage 7 — Runner Projection And Partial-Error State Boundary
|
## Stage 7 — Runner Projection And Partial-Error State Boundary
|
||||||
|
|
||||||
|
**Status: Completed**
|
||||||
|
|
||||||
### Goal
|
### Goal
|
||||||
|
|
||||||
Give analyze one safe way to promote reconciled session state and invocation
|
Give analyze one safe way to promote reconciled session state and invocation
|
||||||
|
|||||||
134
internal/app/analyze_projection.go
Normal file
134
internal/app/analyze_projection.go
Normal file
@@ -0,0 +1,134 @@
|
|||||||
|
package app
|
||||||
|
|
||||||
|
import (
|
||||||
|
"fmt"
|
||||||
|
"reflect"
|
||||||
|
"sort"
|
||||||
|
|
||||||
|
"gitea.maximumdirect.net/eric/narratio/internal/manifest"
|
||||||
|
"gitea.maximumdirect.net/eric/narratio/internal/stage"
|
||||||
|
)
|
||||||
|
|
||||||
|
type validatedAnalyzeProjection struct {
|
||||||
|
session map[string]manifest.AnalyzeArtifactRecord
|
||||||
|
invocation map[string]manifest.AnalyzeArtifactRecord
|
||||||
|
}
|
||||||
|
|
||||||
|
type analyzeStateSnapshot struct {
|
||||||
|
version int
|
||||||
|
records map[string]manifest.AnalyzeArtifactRecord
|
||||||
|
}
|
||||||
|
|
||||||
|
func captureAnalyzeState(manifestValue *manifest.Manifest, stageName string) analyzeStateSnapshot {
|
||||||
|
if manifestValue == nil || stageName != "analyze" || manifestValue.Stages["analyze"] == nil {
|
||||||
|
return analyzeStateSnapshot{}
|
||||||
|
}
|
||||||
|
record := manifestValue.Stages["analyze"]
|
||||||
|
return analyzeStateSnapshot{
|
||||||
|
version: record.AnalyzeStateVersion,
|
||||||
|
records: manifest.CloneAnalyzeArtifactCollection(record.AnalyzeArtifacts),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func restoreAnalyzeState(manifestValue *manifest.Manifest, snapshot analyzeStateSnapshot) {
|
||||||
|
if manifestValue == nil || manifestValue.Stages["analyze"] == nil {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
record := manifestValue.Stages["analyze"]
|
||||||
|
record.AnalyzeStateVersion = snapshot.version
|
||||||
|
record.AnalyzeArtifacts = manifest.CloneAnalyzeArtifactCollection(snapshot.records)
|
||||||
|
}
|
||||||
|
|
||||||
|
func validateSuccessfulAnalyzeProjection(stageName string, result *stage.StageResult) (*validatedAnalyzeProjection, error) {
|
||||||
|
if result == nil || result.AnalyzeState == nil {
|
||||||
|
return nil, nil
|
||||||
|
}
|
||||||
|
if stageName != "analyze" {
|
||||||
|
return nil, fmt.Errorf("stage %q returned analyze-owned state projection", stageName)
|
||||||
|
}
|
||||||
|
if result.Disposition == stage.StageDispositionSkipped {
|
||||||
|
return nil, fmt.Errorf("skipped analyze result cannot contain analyze-owned state projection")
|
||||||
|
}
|
||||||
|
if len(result.Outputs) != 0 {
|
||||||
|
return nil, fmt.Errorf("analyze result with state projection cannot contain ordinary outputs")
|
||||||
|
}
|
||||||
|
return validateAndCloneAnalyzeProjection(result.AnalyzeState)
|
||||||
|
}
|
||||||
|
|
||||||
|
func validateFailedAnalyzeProjection(stageName string, result *stage.StageResult) (*validatedAnalyzeProjection, error) {
|
||||||
|
if result == nil || result.AnalyzeState == nil {
|
||||||
|
return nil, nil
|
||||||
|
}
|
||||||
|
if stageName != "analyze" {
|
||||||
|
return nil, fmt.Errorf("stage %q returned analyze-owned state projection with an error", stageName)
|
||||||
|
}
|
||||||
|
if result.Disposition != stage.StageDispositionSucceeded || result.SkipReason != "" || len(result.Outputs) != 0 || len(result.Logs) != 0 || len(result.GeneratedConfigs) != 0 || len(result.Metadata) != 0 {
|
||||||
|
return nil, fmt.Errorf("analyze result with an error may contain only analyze-owned state projection")
|
||||||
|
}
|
||||||
|
return validateAndCloneAnalyzeProjection(result.AnalyzeState)
|
||||||
|
}
|
||||||
|
|
||||||
|
func validateAndCloneAnalyzeProjection(projection *stage.AnalyzeStateProjection) (*validatedAnalyzeProjection, error) {
|
||||||
|
session := manifest.CloneAnalyzeArtifactCollection(projection.Session)
|
||||||
|
invocation := manifest.CloneAnalyzeArtifactCollection(projection.Invocation)
|
||||||
|
if err := manifest.ValidateAnalyzeArtifactCollection(manifest.AnalyzeStateContractVersion, session); err != nil {
|
||||||
|
return nil, fmt.Errorf("validate reconciled session analyze state: %w", err)
|
||||||
|
}
|
||||||
|
if err := manifest.ValidateAnalyzeArtifactCollection(manifest.AnalyzeStateContractVersion, invocation); err != nil {
|
||||||
|
return nil, fmt.Errorf("validate invocation analyze state: %w", err)
|
||||||
|
}
|
||||||
|
for key, invocationRecord := range invocation {
|
||||||
|
sessionRecord, ok := session[key]
|
||||||
|
if !ok {
|
||||||
|
return nil, fmt.Errorf("invocation analyze artifact %q is absent from reconciled session state", key)
|
||||||
|
}
|
||||||
|
if !reflect.DeepEqual(invocationRecord, sessionRecord) {
|
||||||
|
return nil, fmt.Errorf("invocation analyze artifact %q contradicts reconciled session state", key)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return &validatedAnalyzeProjection{session: session, invocation: invocation}, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func applyAnalyzeProjection(
|
||||||
|
sessionManifest *manifest.Manifest,
|
||||||
|
runManifest *manifest.RunManifest,
|
||||||
|
projection *validatedAnalyzeProjection,
|
||||||
|
) {
|
||||||
|
if projection == nil {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
if sessionManifest != nil && sessionManifest.Stages["analyze"] != nil {
|
||||||
|
record := sessionManifest.Stages["analyze"]
|
||||||
|
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 {
|
||||||
|
keys := make([]string, 0, len(records))
|
||||||
|
for key, record := range records {
|
||||||
|
if record.Status != manifest.AnalyzeArtifactCurrent || record.Output == nil {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
if producerRunID != "" && record.ProducerRunID != producerRunID {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
keys = append(keys, key)
|
||||||
|
}
|
||||||
|
sort.Strings(keys)
|
||||||
|
outputs := make([]manifest.ArtifactRecord, 0, len(keys))
|
||||||
|
for _, key := range keys {
|
||||||
|
record := manifest.CloneAnalyzeArtifactCollection(map[string]manifest.AnalyzeArtifactRecord{key: records[key]})[key]
|
||||||
|
output := *record.Output
|
||||||
|
if output.ProducerRunID == "" {
|
||||||
|
output.ProducerRunID = record.ProducerRunID
|
||||||
|
}
|
||||||
|
outputs = append(outputs, output)
|
||||||
|
}
|
||||||
|
return outputs
|
||||||
|
}
|
||||||
346
internal/app/analyze_projection_test.go
Normal file
346
internal/app/analyze_projection_test.go
Normal file
@@ -0,0 +1,346 @@
|
|||||||
|
package app
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"errors"
|
||||||
|
"os"
|
||||||
|
"path/filepath"
|
||||||
|
"reflect"
|
||||||
|
"strings"
|
||||||
|
"testing"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"gitea.maximumdirect.net/eric/narratio/internal/artifactmodel"
|
||||||
|
"gitea.maximumdirect.net/eric/narratio/internal/artifacts"
|
||||||
|
"gitea.maximumdirect.net/eric/narratio/internal/manifest"
|
||||||
|
"gitea.maximumdirect.net/eric/narratio/internal/stage"
|
||||||
|
)
|
||||||
|
|
||||||
|
type projectionStage struct {
|
||||||
|
name string
|
||||||
|
run func(*stage.Env, *manifest.Manifest) (*stage.StageResult, error)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s projectionStage) Name() string { return s.name }
|
||||||
|
func (s projectionStage) Run(_ context.Context, env *stage.Env, m *manifest.Manifest) (*stage.StageResult, error) {
|
||||||
|
return s.run(env, m)
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestExecuteStagesProjectsSeparateSessionAndInvocationAnalyzeState(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)
|
||||||
|
|
||||||
|
stageToRun := projectionStage{name: "analyze", run: func(_ *stage.Env, m *manifest.Manifest) (*stage.StageResult, error) {
|
||||||
|
newRecord := appAnalyzeRecord("session_recap", manifest.AnalyzeArtifactCurrent, m.RunID, time.Now().UTC())
|
||||||
|
staleRecord := appAnalyzeRecord("quest_log", manifest.AnalyzeArtifactStale, m.RunID, time.Now().UTC())
|
||||||
|
session := map[string]manifest.AnalyzeArtifactRecord{
|
||||||
|
"player_handout": oldRecord,
|
||||||
|
"quest_log": staleRecord,
|
||||||
|
"session_recap": newRecord,
|
||||||
|
}
|
||||||
|
return &stage.StageResult{
|
||||||
|
Logs: []string{"aggregate-analyze.log"},
|
||||||
|
AnalyzeState: &stage.AnalyzeStateProjection{
|
||||||
|
Session: session,
|
||||||
|
Invocation: map[string]manifest.AnalyzeArtifactRecord{
|
||||||
|
"player_handout": oldRecord,
|
||||||
|
"session_recap": newRecord,
|
||||||
|
},
|
||||||
|
},
|
||||||
|
}, nil
|
||||||
|
}}
|
||||||
|
|
||||||
|
summary, err := executeStages(context.Background(), cfg, []stage.Stage{stageToRun}, RunOptions{})
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("executeStages() error = %v", err)
|
||||||
|
}
|
||||||
|
store := &manifest.LocalStore{}
|
||||||
|
sessionManifest, err := store.Load(context.Background(), summary.ManifestPath)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("Load(session) error = %v", err)
|
||||||
|
}
|
||||||
|
analyze := sessionManifest.Stages["analyze"]
|
||||||
|
if analyze.AnalyzeStateVersion != manifest.AnalyzeStateContractVersion || len(analyze.AnalyzeArtifacts) != 3 {
|
||||||
|
t.Fatalf("session analyze state = %#v", analyze)
|
||||||
|
}
|
||||||
|
if got := analyzeArtifactOutputKeys(analyze.Outputs); !reflect.DeepEqual(got, []string{"player_handout", "session_recap"}) {
|
||||||
|
t.Fatalf("session aggregate outputs = %#v, want current records only", got)
|
||||||
|
}
|
||||||
|
if len(analyze.Logs) != 1 || analyze.Logs[0] != "aggregate-analyze.log" {
|
||||||
|
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) {
|
||||||
|
cfg := testConfig(t)
|
||||||
|
now := time.Date(2026, 5, 3, 9, 0, 0, 0, time.UTC)
|
||||||
|
seed := manifest.New(cfg.Session.SessionID, now)
|
||||||
|
seed.Campaign = cfg.Session.Campaign
|
||||||
|
seed.MarkStageSucceeded("analyze", now, nil)
|
||||||
|
seed.Stages["analyze"].AnalyzeStateVersion = manifest.AnalyzeStateContractVersion
|
||||||
|
seed.Stages["analyze"].AnalyzeArtifacts = map[string]manifest.AnalyzeArtifactRecord{
|
||||||
|
"player_handout": appAnalyzeRecord("player_handout", manifest.AnalyzeArtifactCurrent, "old-run", now),
|
||||||
|
}
|
||||||
|
seed.MarkStageSucceeded("publish", now, nil)
|
||||||
|
saveBoundedManifest(t, cfg, seed)
|
||||||
|
|
||||||
|
stageToRun := projectionStage{name: "analyze", run: func(_ *stage.Env, m *manifest.Manifest) (*stage.StageResult, error) {
|
||||||
|
unrelated := seed.Stages["analyze"].AnalyzeArtifacts["player_handout"]
|
||||||
|
completed := appAnalyzeRecord("session_recap", manifest.AnalyzeArtifactCurrent, m.RunID, time.Now().UTC())
|
||||||
|
failed := appAnalyzeRecord("quest_log", manifest.AnalyzeArtifactFailed, m.RunID, time.Now().UTC())
|
||||||
|
session := map[string]manifest.AnalyzeArtifactRecord{
|
||||||
|
"player_handout": unrelated,
|
||||||
|
"quest_log": failed,
|
||||||
|
"session_recap": completed,
|
||||||
|
}
|
||||||
|
return &stage.StageResult{AnalyzeState: &stage.AnalyzeStateProjection{
|
||||||
|
Session: session,
|
||||||
|
Invocation: map[string]manifest.AnalyzeArtifactRecord{
|
||||||
|
"quest_log": failed,
|
||||||
|
"session_recap": completed,
|
||||||
|
},
|
||||||
|
}}, errors.New("quest log failed")
|
||||||
|
}}
|
||||||
|
|
||||||
|
_, err := executeStages(context.Background(), cfg, []stage.Stage{stageToRun}, RunOptions{Force: true})
|
||||||
|
if err == nil || !strings.Contains(err.Error(), "quest log failed") {
|
||||||
|
t.Fatalf("executeStages() error = %v", err)
|
||||||
|
}
|
||||||
|
store := &manifest.LocalStore{}
|
||||||
|
loaded, err := store.Load(context.Background(), manifestPathFor(cfg))
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("Load(session) error = %v", err)
|
||||||
|
}
|
||||||
|
analyze := loaded.Stages["analyze"]
|
||||||
|
if analyze.Status != manifest.StatusFailed || len(analyze.Outputs) != 0 {
|
||||||
|
t.Fatalf("aggregate analyze state = %#v, want failed without outputs", analyze)
|
||||||
|
}
|
||||||
|
if analyze.AnalyzeArtifacts["player_handout"].Status != manifest.AnalyzeArtifactCurrent ||
|
||||||
|
analyze.AnalyzeArtifacts["session_recap"].Status != manifest.AnalyzeArtifactCurrent ||
|
||||||
|
analyze.AnalyzeArtifacts["quest_log"].Status != manifest.AnalyzeArtifactFailed {
|
||||||
|
t.Fatalf("partial session projection = %#v", analyze.AnalyzeArtifacts)
|
||||||
|
}
|
||||||
|
if loaded.Stages["publish"].Status != manifest.StatusStale {
|
||||||
|
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) {
|
||||||
|
cfg := testConfig(t)
|
||||||
|
now := time.Date(2026, 5, 3, 9, 0, 0, 0, time.UTC)
|
||||||
|
prior := appAnalyzeRecord("player_handout", manifest.AnalyzeArtifactCurrent, "old-run", now)
|
||||||
|
seed := manifest.New(cfg.Session.SessionID, now)
|
||||||
|
seed.Campaign = cfg.Session.Campaign
|
||||||
|
seed.MarkStageSucceeded("analyze", now, nil)
|
||||||
|
seed.Stages["analyze"].AnalyzeStateVersion = manifest.AnalyzeStateContractVersion
|
||||||
|
seed.Stages["analyze"].AnalyzeArtifacts = map[string]manifest.AnalyzeArtifactRecord{"player_handout": prior}
|
||||||
|
saveBoundedManifest(t, cfg, seed)
|
||||||
|
|
||||||
|
stageToRun := projectionStage{name: "analyze", run: func(_ *stage.Env, m *manifest.Manifest) (*stage.StageResult, error) {
|
||||||
|
invalid := appAnalyzeRecord("session_recap", manifest.AnalyzeArtifactCurrent, m.RunID, time.Now().UTC())
|
||||||
|
invalid.Output.Checksum = "invalid"
|
||||||
|
return &stage.StageResult{AnalyzeState: &stage.AnalyzeStateProjection{
|
||||||
|
Session: map[string]manifest.AnalyzeArtifactRecord{"session_recap": invalid},
|
||||||
|
Invocation: map[string]manifest.AnalyzeArtifactRecord{"session_recap": invalid},
|
||||||
|
}}, errors.New("analysis failed")
|
||||||
|
}}
|
||||||
|
|
||||||
|
_, err := executeStages(context.Background(), cfg, []stage.Stage{stageToRun}, RunOptions{Force: true})
|
||||||
|
if err == nil || !strings.Contains(err.Error(), "checksum") {
|
||||||
|
t.Fatalf("executeStages() error = %v, want projection validation failure", err)
|
||||||
|
}
|
||||||
|
loaded, loadErr := (&manifest.LocalStore{}).Load(context.Background(), manifestPathFor(cfg))
|
||||||
|
if loadErr != nil {
|
||||||
|
t.Fatalf("Load() error = %v", loadErr)
|
||||||
|
}
|
||||||
|
if len(loaded.Stages["analyze"].AnalyzeArtifacts) != 1 || !reflect.DeepEqual(loaded.Stages["analyze"].AnalyzeArtifacts["player_handout"], prior) {
|
||||||
|
t.Fatalf("prior state replaced by invalid projection: %#v", loaded.Stages["analyze"].AnalyzeArtifacts)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestExecuteStagesRollsBackAnalyzeAuthorityWhenProjectionSaveFails(t *testing.T) {
|
||||||
|
cfg := testConfig(t)
|
||||||
|
now := time.Date(2026, 5, 3, 9, 0, 0, 0, time.UTC)
|
||||||
|
prior := appAnalyzeRecord("player_handout", manifest.AnalyzeArtifactCurrent, "old-run", now)
|
||||||
|
seed := manifest.New(cfg.Session.SessionID, now)
|
||||||
|
seed.Campaign = cfg.Session.Campaign
|
||||||
|
seed.MarkStageSucceeded("analyze", now, nil)
|
||||||
|
seed.Stages["analyze"].AnalyzeStateVersion = manifest.AnalyzeStateContractVersion
|
||||||
|
seed.Stages["analyze"].AnalyzeArtifacts = map[string]manifest.AnalyzeArtifactRecord{"player_handout": prior}
|
||||||
|
saveBoundedManifest(t, cfg, seed)
|
||||||
|
|
||||||
|
stageReturned := false
|
||||||
|
store := &analyzeProjectionFailingStore{delegate: &manifest.LocalStore{}, shouldFail: func(m *manifest.Manifest) bool {
|
||||||
|
return stageReturned && m.Stages["analyze"] != nil && m.Stages["analyze"].Status == manifest.StatusSucceeded && m.Stages["analyze"].AnalyzeArtifacts["session_recap"].Status == manifest.AnalyzeArtifactCurrent
|
||||||
|
}}
|
||||||
|
stageToRun := projectionStage{name: "analyze", run: func(_ *stage.Env, m *manifest.Manifest) (*stage.StageResult, error) {
|
||||||
|
stageReturned = true
|
||||||
|
newRecord := appAnalyzeRecord("session_recap", manifest.AnalyzeArtifactCurrent, m.RunID, time.Now().UTC())
|
||||||
|
return &stage.StageResult{AnalyzeState: &stage.AnalyzeStateProjection{
|
||||||
|
Session: map[string]manifest.AnalyzeArtifactRecord{
|
||||||
|
"player_handout": prior,
|
||||||
|
"session_recap": newRecord,
|
||||||
|
},
|
||||||
|
Invocation: map[string]manifest.AnalyzeArtifactRecord{"session_recap": newRecord},
|
||||||
|
}}, nil
|
||||||
|
}}
|
||||||
|
|
||||||
|
_, err := executeStages(context.Background(), cfg, []stage.Stage{stageToRun}, RunOptions{
|
||||||
|
Force: true,
|
||||||
|
Env: &Env{ManifestStore: store},
|
||||||
|
})
|
||||||
|
if err == nil || !strings.Contains(err.Error(), "injected analyze projection save failure") {
|
||||||
|
t.Fatalf("executeStages() error = %v", err)
|
||||||
|
}
|
||||||
|
if !store.failed {
|
||||||
|
t.Fatal("projection persistence failure was not injected")
|
||||||
|
}
|
||||||
|
loaded, loadErr := store.delegate.Load(context.Background(), manifestPathFor(cfg))
|
||||||
|
if loadErr != nil {
|
||||||
|
t.Fatalf("Load() error = %v", loadErr)
|
||||||
|
}
|
||||||
|
analyze := loaded.Stages["analyze"]
|
||||||
|
if analyze.Status != manifest.StatusFailed || len(analyze.AnalyzeArtifacts) != 1 || !reflect.DeepEqual(analyze.AnalyzeArtifacts["player_handout"], prior) {
|
||||||
|
t.Fatalf("durable analyze state after rollback = %#v", analyze)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestExecuteStagesRejectsAnalyzeProjectionFromOtherStage(t *testing.T) {
|
||||||
|
cfg := testConfig(t)
|
||||||
|
stageToRun := projectionStage{name: "prepare", run: func(_ *stage.Env, m *manifest.Manifest) (*stage.StageResult, error) {
|
||||||
|
record := appAnalyzeRecord("session_recap", manifest.AnalyzeArtifactCurrent, m.RunID, time.Now().UTC())
|
||||||
|
return &stage.StageResult{AnalyzeState: &stage.AnalyzeStateProjection{
|
||||||
|
Session: map[string]manifest.AnalyzeArtifactRecord{"session_recap": record},
|
||||||
|
}}, nil
|
||||||
|
}}
|
||||||
|
_, err := executeStages(context.Background(), cfg, []stage.Stage{stageToRun}, RunOptions{})
|
||||||
|
if err == nil || !strings.Contains(err.Error(), "returned analyze-owned state projection") {
|
||||||
|
t.Fatalf("executeStages() error = %v", err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestExecuteStagesRejectsContradictoryAnalyzeResultWithError(t *testing.T) {
|
||||||
|
cfg := testConfig(t)
|
||||||
|
stageToRun := projectionStage{name: "analyze", run: func(_ *stage.Env, m *manifest.Manifest) (*stage.StageResult, error) {
|
||||||
|
record := appAnalyzeRecord("session_recap", manifest.AnalyzeArtifactCurrent, m.RunID, time.Now().UTC())
|
||||||
|
return &stage.StageResult{
|
||||||
|
Outputs: []artifacts.Ref{{Kind: "session_recap", RelativePath: "artifacts/session-recap.md"}},
|
||||||
|
AnalyzeState: &stage.AnalyzeStateProjection{
|
||||||
|
Session: map[string]manifest.AnalyzeArtifactRecord{"session_recap": record},
|
||||||
|
Invocation: map[string]manifest.AnalyzeArtifactRecord{"session_recap": record},
|
||||||
|
},
|
||||||
|
}, errors.New("analysis failed")
|
||||||
|
}}
|
||||||
|
_, err := executeStages(context.Background(), cfg, []stage.Stage{stageToRun}, RunOptions{})
|
||||||
|
if err == nil || !strings.Contains(err.Error(), "may contain only analyze-owned state projection") {
|
||||||
|
t.Fatalf("executeStages() error = %v", err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestExecuteStagesExposesSelectedForceDecisionToStage(t *testing.T) {
|
||||||
|
for _, force := range []bool{false, true} {
|
||||||
|
t.Run(strings.ToLower(strings.TrimSpace(map[bool]string{false: "ordinary", true: "forced"}[force])), func(t *testing.T) {
|
||||||
|
cfg := testConfig(t)
|
||||||
|
captured := !force
|
||||||
|
stageToRun := projectionStage{name: "prepare", run: func(env *stage.Env, _ *manifest.Manifest) (*stage.StageResult, error) {
|
||||||
|
captured = env.Force
|
||||||
|
return &stage.StageResult{}, nil
|
||||||
|
}}
|
||||||
|
if _, err := executeStages(context.Background(), cfg, []stage.Stage{stageToRun}, RunOptions{Force: force}); err != nil {
|
||||||
|
t.Fatalf("executeStages() error = %v", err)
|
||||||
|
}
|
||||||
|
if captured != force {
|
||||||
|
t.Fatalf("stage env force = %v, want %v", captured, force)
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func appAnalyzeRecord(key string, status manifest.AnalyzeArtifactStatus, producerRunID string, at time.Time) manifest.AnalyzeArtifactRecord {
|
||||||
|
record := manifest.AnalyzeArtifactRecord{
|
||||||
|
Key: key,
|
||||||
|
Status: status,
|
||||||
|
ProducerRunID: producerRunID,
|
||||||
|
UpdatedAt: at,
|
||||||
|
}
|
||||||
|
if status == manifest.AnalyzeArtifactFailed {
|
||||||
|
record.Error = "scriptorium failed"
|
||||||
|
return record
|
||||||
|
}
|
||||||
|
if status != manifest.AnalyzeArtifactCurrent {
|
||||||
|
record.FingerprintVersion = manifest.AnalyzeFingerprintContractVersion
|
||||||
|
record.Fingerprint = strings.Repeat("b", 64)
|
||||||
|
return record
|
||||||
|
}
|
||||||
|
record.FingerprintVersion = manifest.AnalyzeFingerprintContractVersion
|
||||||
|
record.Fingerprint = strings.Repeat("a", 64)
|
||||||
|
record.OutputSize = 42
|
||||||
|
record.Output = &manifest.ArtifactRecord{
|
||||||
|
Kind: key,
|
||||||
|
SourceID: artifacts.ConfiguredArtifactSourceID(key),
|
||||||
|
LocalPath: "artifacts/" + strings.ReplaceAll(key, "_", "-") + ".md",
|
||||||
|
ProducerRunID: producerRunID,
|
||||||
|
Checksum: strings.Repeat("c", 64),
|
||||||
|
Contract: &artifactmodel.ContractMetadata{
|
||||||
|
MediaType: "text/markdown", SchemaID: "narratio." + key, SchemaVersion: "1",
|
||||||
|
},
|
||||||
|
}
|
||||||
|
return record
|
||||||
|
}
|
||||||
|
|
||||||
|
func analyzeArtifactOutputKeys(outputs []manifest.ArtifactRecord) []string {
|
||||||
|
keys := make([]string, 0, len(outputs))
|
||||||
|
for _, output := range outputs {
|
||||||
|
keys = append(keys, strings.TrimPrefix(output.SourceID, "narratio.artifact."))
|
||||||
|
}
|
||||||
|
return keys
|
||||||
|
}
|
||||||
|
|
||||||
|
type analyzeProjectionFailingStore struct {
|
||||||
|
delegate *manifest.LocalStore
|
||||||
|
shouldFail func(*manifest.Manifest) bool
|
||||||
|
failed bool
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *analyzeProjectionFailingStore) Create(ctx context.Context, sessionID string) (*manifest.Manifest, error) {
|
||||||
|
return s.delegate.Create(ctx, sessionID)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *analyzeProjectionFailingStore) Load(ctx context.Context, path string) (*manifest.Manifest, error) {
|
||||||
|
return s.delegate.Load(ctx, path)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *analyzeProjectionFailingStore) Save(ctx context.Context, path string, m *manifest.Manifest) error {
|
||||||
|
if !s.failed && s.shouldFail != nil && s.shouldFail(m) {
|
||||||
|
s.failed = true
|
||||||
|
return errors.New("injected analyze projection save failure")
|
||||||
|
}
|
||||||
|
return s.delegate.Save(ctx, path, m)
|
||||||
|
}
|
||||||
@@ -253,6 +253,7 @@ func executeStages(ctx context.Context, cfg *config.Config, stages []stage.Stage
|
|||||||
skipped := make([]string, 0, len(decisions))
|
skipped := make([]string, 0, len(decisions))
|
||||||
for _, d := range decisions {
|
for _, d := range decisions {
|
||||||
s := d.Stage
|
s := d.Stage
|
||||||
|
stageEnv.Force = opts.Force
|
||||||
runNames = append(runNames, s.Name())
|
runNames = append(runNames, s.Name())
|
||||||
d.Action = decideStageAction(s, m, opts.Force)
|
d.Action = decideStageAction(s, m, opts.Force)
|
||||||
|
|
||||||
@@ -305,6 +306,7 @@ func executeStages(ctx context.Context, cfg *config.Config, stages []stage.Stage
|
|||||||
}
|
}
|
||||||
executed = append(executed, s.Name())
|
executed = append(executed, s.Name())
|
||||||
priorOutcome := capturePriorStageOutcome(m, s.Name())
|
priorOutcome := capturePriorStageOutcome(m, s.Name())
|
||||||
|
priorAnalyzeState := captureAnalyzeState(m, s.Name())
|
||||||
|
|
||||||
now := nowUTC()
|
now := nowUTC()
|
||||||
runManifest.SetStageAction(s.Name(), manifest.RunStageActionRun, now)
|
runManifest.SetStageAction(s.Name(), manifest.RunStageActionRun, now)
|
||||||
@@ -339,10 +341,23 @@ func executeStages(ctx context.Context, cfg *config.Config, stages []stage.Stage
|
|||||||
env.Logger.Debug("manifest saved", "stage", s.Name(), "transition", "running", "path", manifestPath)
|
env.Logger.Debug("manifest saved", "stage", s.Name(), "transition", "running", "path", manifestPath)
|
||||||
|
|
||||||
result, err := s.Run(ctx, stageEnv, m)
|
result, err := s.Run(ctx, stageEnv, m)
|
||||||
|
var analyzeProjection *validatedAnalyzeProjection
|
||||||
if err == nil {
|
if err == nil {
|
||||||
err = validateStageResult(result)
|
err = validateStageResult(result)
|
||||||
|
if err == nil {
|
||||||
|
analyzeProjection, err = validateSuccessfulAnalyzeProjection(s.Name(), result)
|
||||||
|
}
|
||||||
|
} else if result != nil && result.AnalyzeState != nil {
|
||||||
|
var projectionErr error
|
||||||
|
analyzeProjection, projectionErr = validateFailedAnalyzeProjection(s.Name(), result)
|
||||||
|
if projectionErr != nil {
|
||||||
|
err = errors.Join(err, projectionErr)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
if analyzeProjection != nil {
|
||||||
|
applyAnalyzeProjection(m, runManifest, analyzeProjection)
|
||||||
|
}
|
||||||
failedAt := nowUTC()
|
failedAt := nowUTC()
|
||||||
m.MarkStageFailed(s.Name(), failedAt, err.Error())
|
m.MarkStageFailed(s.Name(), failedAt, err.Error())
|
||||||
if _, invalidationErr := invalidateDependentSucceededStagesWithReason(m, s.Name(), failedAt, staleReasonFailure); invalidationErr != nil {
|
if _, invalidationErr := invalidateDependentSucceededStagesWithReason(m, s.Name(), failedAt, staleReasonFailure); invalidationErr != nil {
|
||||||
@@ -389,9 +404,15 @@ func executeStages(ctx context.Context, cfg *config.Config, stages []stage.Stage
|
|||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
|
|
||||||
outputs := mapResultOutputs(s.Name(), result, runID)
|
sessionOutputs := mapResultOutputs(s.Name(), result, runID)
|
||||||
|
runOutputs := sessionOutputs
|
||||||
|
if analyzeProjection != nil {
|
||||||
|
sessionOutputs = analyzeProjectionOutputs(analyzeProjection.session, "")
|
||||||
|
runOutputs = analyzeProjectionOutputs(analyzeProjection.invocation, runID)
|
||||||
|
}
|
||||||
succeededAt := nowUTC()
|
succeededAt := nowUTC()
|
||||||
m.MarkStageSucceeded(s.Name(), succeededAt, outputs)
|
m.MarkStageSucceeded(s.Name(), succeededAt, sessionOutputs)
|
||||||
|
applyAnalyzeProjection(m, runManifest, analyzeProjection)
|
||||||
applyStageResultToManifest(m, s.Name(), result)
|
applyStageResultToManifest(m, s.Name(), result)
|
||||||
if !priorOutcome.exists || priorOutcome.status != manifest.StatusSucceeded {
|
if !priorOutcome.exists || priorOutcome.status != manifest.StatusSucceeded {
|
||||||
if _, err := invalidateDependentSucceededStagesWithReason(m, s.Name(), succeededAt, staleReasonChangedResult); err != nil {
|
if _, err := invalidateDependentSucceededStagesWithReason(m, s.Name(), succeededAt, staleReasonChangedResult); err != nil {
|
||||||
@@ -403,12 +424,22 @@ func executeStages(ctx context.Context, cfg *config.Config, stages []stage.Stage
|
|||||||
}
|
}
|
||||||
|
|
||||||
if err := env.ManifestStore.Save(ctx, manifestPath, m); err != nil {
|
if err := env.ManifestStore.Save(ctx, manifestPath, m); err != nil {
|
||||||
|
operationErr := fmt.Errorf("save manifest after stage %q: %w", s.Name(), err)
|
||||||
|
if analyzeProjection != nil {
|
||||||
|
restoreAnalyzeState(m, priorAnalyzeState)
|
||||||
|
failedAt := nowUTC()
|
||||||
|
m.MarkStageFailed(s.Name(), failedAt, operationErr.Error())
|
||||||
|
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(
|
return nil, persistTerminalFailure(
|
||||||
ctx, env.ManifestStore, manifestPath, m, runManifestStore, runManifestPath, runManifest,
|
ctx, env.ManifestStore, manifestPath, m, runManifestStore, runManifestPath, runManifest,
|
||||||
fmt.Errorf("save manifest after stage %q: %w", s.Name(), err),
|
operationErr,
|
||||||
)
|
)
|
||||||
}
|
}
|
||||||
runManifest.MarkStageSucceeded(s.Name(), succeededAt, outputs)
|
runManifest.MarkStageSucceeded(s.Name(), succeededAt, runOutputs)
|
||||||
applyStageResultToRunManifest(runManifest, s.Name(), result)
|
applyStageResultToRunManifest(runManifest, s.Name(), result)
|
||||||
identity.applyToRunManifest(runManifest, manifestPath)
|
identity.applyToRunManifest(runManifest, manifestPath)
|
||||||
if err := runManifestStore.SaveRun(ctx, runManifestPath, runManifest); err != nil {
|
if err := runManifestStore.SaveRun(ctx, runManifestPath, runManifest); err != nil {
|
||||||
|
|||||||
@@ -84,6 +84,25 @@ func ValidateAnalyzeArtifactCollection(version int, records map[string]AnalyzeAr
|
|||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// CloneAnalyzeArtifactCollection returns a deep copy in canonical dependency
|
||||||
|
// order so projections cannot be mutated after application.
|
||||||
|
func CloneAnalyzeArtifactCollection(records map[string]AnalyzeArtifactRecord) map[string]AnalyzeArtifactRecord {
|
||||||
|
if records == nil {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
cloned := make(map[string]AnalyzeArtifactRecord, len(records))
|
||||||
|
for key, record := range records {
|
||||||
|
record.Dependencies = cloneStrings(record.Dependencies)
|
||||||
|
record.Logs = cloneStrings(record.Logs)
|
||||||
|
record.GeneratedConfigs = cloneStrings(record.GeneratedConfigs)
|
||||||
|
record.Output = cloneArtifactRecord(record.Output)
|
||||||
|
record.Scriptorium = cloneAnalyzeProvenance(record.Scriptorium)
|
||||||
|
cloned[key] = record
|
||||||
|
}
|
||||||
|
normalizeAnalyzeArtifactCollection(cloned)
|
||||||
|
return cloned
|
||||||
|
}
|
||||||
|
|
||||||
func validateAnalyzeStageState(stageName string, version int, records map[string]AnalyzeArtifactRecord) error {
|
func validateAnalyzeStageState(stageName string, version int, records map[string]AnalyzeArtifactRecord) error {
|
||||||
if stageName != "analyze" {
|
if stageName != "analyze" {
|
||||||
if version != 0 || records != nil {
|
if version != 0 || records != nil {
|
||||||
|
|||||||
@@ -26,6 +26,7 @@ type Env struct {
|
|||||||
ArtifactStore artifacts.Store
|
ArtifactStore artifacts.Store
|
||||||
ManifestStore manifest.Store
|
ManifestStore manifest.Store
|
||||||
Logger *slog.Logger
|
Logger *slog.Logger
|
||||||
|
Force bool
|
||||||
|
|
||||||
WhisperX whisperx.Client
|
WhisperX whisperx.Client
|
||||||
Seriatim seriatim.Runner
|
Seriatim seriatim.Runner
|
||||||
@@ -105,4 +106,12 @@ type StageResult struct {
|
|||||||
Logs []string
|
Logs []string
|
||||||
GeneratedConfigs []string
|
GeneratedConfigs []string
|
||||||
Metadata map[string]any
|
Metadata map[string]any
|
||||||
|
AnalyzeState *AnalyzeStateProjection
|
||||||
|
}
|
||||||
|
|
||||||
|
// AnalyzeStateProjection carries analyze-owned reconciled session authority and
|
||||||
|
// the invocation subset evaluated by the current run.
|
||||||
|
type AnalyzeStateProjection struct {
|
||||||
|
Session map[string]manifest.AnalyzeArtifactRecord
|
||||||
|
Invocation map[string]manifest.AnalyzeArtifactRecord
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user