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

Add artifact provenance and stage skip outcomes

Browse Source
This commit is contained in:
Eric Rakestraw
2026-08-09 23:20:32 +00:00
parent df58595d1e
commit 951383226c
12 changed files with 1565 additions and 305 deletions
Download Patch File Download Diff File Expand all files Collapse all files

98
internal/app/runner.go
View File

@@ -195,6 +195,9 @@ func executeStages(ctx context.Context, cfg *config.Config, stages []stage.Stage
env.Logger.Debug("manifest saved", "stage", s.Name(), "transition", "running", "path", manifestPath)
result, err := s.Run(ctx, stageEnv, m)
if err == nil {
err = validateStageResult(result)
}
if err != nil {
failedAt := nowUTC()
m.MarkStageFailed(s.Name(), failedAt, err.Error())
@@ -209,6 +212,25 @@ func executeStages(ctx context.Context, cfg *config.Config, stages []stage.Stage
env.Logger.Info("stage failed", "stage", s.Name(), "error", err)
return nil, fmt.Errorf("stage %q failed: %w", s.Name(), err)
}
if result != nil && result.Disposition == stage.StageDispositionSkipped {
skipped = append(skipped, s.Name())
skippedAt := nowUTC()
m.MarkStageSkipped(s.Name(), skippedAt, result.SkipReason)
clearStageResultDetails(m.Stages[s.Name()])
applyStageResultToManifest(m, s.Name(), result)
if err := env.ManifestStore.Save(ctx, manifestPath, m); err != nil {
return nil, fmt.Errorf("save manifest after self-skip %q: %w", s.Name(), err)
}
runManifest.MarkStageSkipped(s.Name(), skippedAt, result.SkipReason)
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 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
}
outputs := mapResultOutputs(s.Name(), result, runID)
succeededAt := nowUTC()
@@ -407,26 +429,69 @@ func mapResultOutputs(stageName string, result *stage.StageResult, runID string)
localPath = ref.RelativePath
}
kind := ref.Kind
sourceID := ""
if stageName == "analyze" {
sourceID = artifacts.ConfiguredArtifactSourceID(ref.Kind)
kind = "scriptorium_artifact"
} else {
sourceID = sourceIDForOutputKind(kind)
sourceID := strings.TrimSpace(ref.SourceID)
if sourceID == "" {
if stageName == "analyze" {
sourceID = artifacts.ConfiguredArtifactSourceID(ref.Kind)
kind = "scriptorium_artifact"
} else {
sourceID = sourceIDForOutputKind(kind)
}
}
out = append(out, manifest.ArtifactRecord{
Kind: kind,
SourceID: sourceID,
LocalPath: localPath,
ProducerRunID: runID,
RemoteKey: ref.RemoteKey,
Checksum: ref.Checksum,
Kind: kind,
SourceID: sourceID,
LocalPath: localPath,
Contract: cloneContractMetadata(ref.Contract),
ExternalProvenance: cloneExternalProvenance(ref.ExternalProvenance),
ProducerRunID: runID,
RemoteKey: ref.RemoteKey,
Checksum: ref.Checksum,
})
}
return out
}
func validateStageResult(result *stage.StageResult) error {
if result == nil {
return nil
}
switch result.Disposition {
case stage.StageDispositionSucceeded:
if strings.TrimSpace(result.SkipReason) != "" {
return fmt.Errorf("successful result contains a skip reason")
}
return nil
case stage.StageDispositionSkipped:
if strings.TrimSpace(result.SkipReason) == "" {
return fmt.Errorf("skipped result requires a skip reason")
}
if len(result.Outputs) != 0 {
return fmt.Errorf("skipped result contains %d output(s)", len(result.Outputs))
}
return nil
default:
return fmt.Errorf("unsupported stage result disposition %q", result.Disposition)
}
}
func cloneContractMetadata(value *artifactmodel.ContractMetadata) *artifactmodel.ContractMetadata {
if value == nil {
return nil
}
cloned := *value
return &cloned
}
func cloneExternalProvenance(value *artifactmodel.ExternalProvenance) *artifactmodel.ExternalProvenance {
if value == nil {
return nil
}
cloned := *value
return &cloned
}
func sourceIDForOutputKind(kind string) string {
trimmed := strings.TrimSpace(kind)
if trimmed == "" {
@@ -462,6 +527,15 @@ func applyStageResultToManifest(m *manifest.Manifest, stageName string, result *
}
}
func clearStageResultDetails(sr *manifest.StageRecord) {
if sr == nil {
return
}
sr.Logs = nil
sr.GeneratedConfigs = nil
sr.Metadata = nil
}
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

195
internal/app/runner_test.go
View File

@@ -16,6 +16,7 @@ import (
"gitea.maximumdirect.net/eric/narratio/internal/adapters/seriatim"
"gitea.maximumdirect.net/eric/narratio/internal/adapters/storage"
"gitea.maximumdirect.net/eric/narratio/internal/adapters/whisperx"
"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/manifest"
@@ -38,6 +39,25 @@ type countingStage struct {
runs *int
}
type resultStage struct {
name string
result *stage.StageResult
runs *int
order *[]string
}
func (s resultStage) Name() string { return s.name }
func (s resultStage) Declares() stage.IODecl { return stage.IODecl{} }
func (s resultStage) Run(_ context.Context, _ *stage.Env, _ *manifest.Manifest) (*stage.StageResult, error) {
if s.runs != nil {
*s.runs = *s.runs + 1
}
if s.order != nil {
*s.order = append(*s.order, s.name)
}
return s.result, nil
}
func (s countingStage) Name() string { return s.name }
func (s countingStage) Declares() stage.IODecl { return stage.IODecl{} }
func (s countingStage) Run(_ context.Context, _ *stage.Env, _ *manifest.Manifest) (*stage.StageResult, error) {
@@ -199,6 +219,61 @@ func TestExecuteStagesAnalyzeOutputsPersistAsScriptoriumArtifacts(t *testing.T)
}
}
func TestMapResultOutputsPrefersExplicitSourceAndCopiesMetadata(t *testing.T) {
contract := &artifactmodel.ContractMetadata{
MediaType: "application/json",
SchemaID: "notarius.dnd.npc_registry",
SchemaVersion: "v1",
}
provenance := &artifactmodel.ExternalProvenance{
System: "notarius",
RunID: "external-run",
PipelineID: "dnd-session",
ArtifactID: "npc-registry",
}
result := &stage.StageResult{Outputs: []artifacts.Ref{{
Kind: "structured_data",
SourceID: "narratio.example.npcs",
RelativePath: "artifacts/npcs.json",
Contract: contract,
ExternalProvenance: provenance,
}}}
got := mapResultOutputs("analyze", result, "narratio-run")
if len(got) != 1 {
t.Fatalf("outputs len = %d, want 1", len(got))
}
if got[0].SourceID != "narratio.example.npcs" {
t.Fatalf("source_id = %q, want explicit source", got[0].SourceID)
}
if got[0].Kind != "structured_data" {
t.Fatalf("kind = %q, want explicit output kind preserved", got[0].Kind)
}
if got[0].Contract == nil || *got[0].Contract != *contract {
t.Fatalf("contract = %#v, want %#v", got[0].Contract, contract)
}
if got[0].ExternalProvenance == nil || *got[0].ExternalProvenance != *provenance {
t.Fatalf("external provenance = %#v, want %#v", got[0].ExternalProvenance, provenance)
}
if got[0].Contract == contract || got[0].ExternalProvenance == provenance {
t.Fatal("mapped metadata should not alias the stage result")
}
}
func TestMapResultOutputsRetainsFallbackInference(t *testing.T) {
transcript := mapResultOutputs("trim", &stage.StageResult{Outputs: []artifacts.Ref{{
Kind: artifactmodel.TranscriptOutputKindFinalTrimmed,
}}}, "run-id")
if len(transcript) != 1 || transcript[0].SourceID != artifacts.ArtifactTranscriptFinalTrimmed {
t.Fatalf("transcript fallback = %#v, want final-trimmed source", transcript)
}
analyze := mapResultOutputs("analyze", &stage.StageResult{Outputs: []artifacts.Ref{{Kind: "session_recap"}}}, "run-id")
if len(analyze) != 1 || analyze[0].SourceID != "narratio.artifact.session_recap" || analyze[0].Kind != "scriptorium_artifact" {
t.Fatalf("analyze fallback = %#v, want configured artifact inference", analyze)
}
}
func TestNeedsObjectStoreForRunPrepareWithPreviousRequirements(t *testing.T) {
tests := []struct {
name string
@@ -738,6 +813,126 @@ func TestExecuteStagesRunManifestRecordsSkippedStage(t *testing.T) {
}
}
func TestExecuteStagesPersistsSelfSkipAndContinues(t *testing.T) {
cfg := testConfig(t)
store := &manifest.LocalStore{}
manifestPath := manifestPathFor(cfg)
seed := manifest.New(cfg.Session.SessionID, time.Date(2026, 5, 3, 1, 0, 0, 0, time.UTC))
seed.MarkStageSucceeded("optional", time.Date(2026, 5, 3, 1, 1, 0, 0, time.UTC), []manifest.ArtifactRecord{{
Kind: "old_output",
SourceID: "narratio.example.old",
LocalPath: "artifacts/old.json",
}})
seed.Stages["optional"].Logs = []string{"old.log"}
seed.Stages["optional"].GeneratedConfigs = []string{"old.yml"}
seed.Stages["optional"].Metadata = map[string]any{"old": true}
if err := os.MkdirAll(filepath.Dir(manifestPath), 0o755); err != nil {
t.Fatalf("MkdirAll() error = %v", err)
}
if err := store.Save(context.Background(), manifestPath, seed); err != nil {
t.Fatalf("Save() seed manifest error = %v", err)
}
order := []string{}
optionalRuns := 0
stages := []stage.Stage{
resultStage{
name: "optional",
runs: &optionalRuns,
order: &order,
result: &stage.StageResult{
Disposition: stage.StageDispositionSkipped,
SkipReason: "integration_disabled",
Logs: []string{"runs/current/optional.log"},
GeneratedConfigs: []string{"runs/current/optional.yml"},
Metadata: map[string]any{"enabled": false},
},
},
resultStage{name: "later", order: &order, result: &stage.StageResult{}},
}
summary, err := executeStages(context.Background(), cfg, stages, RunOptions{Force: true})
if err != nil {
t.Fatalf("executeStages() error = %v", err)
}
if strings.Join(order, ",") != "optional,later" {
t.Fatalf("execution order = %v, want optional then later", order)
}
if len(summary.Executed) != 2 || len(summary.Skipped) != 1 || summary.Skipped[0] != "optional" {
t.Fatalf("summary = %#v, want optional executed and self-skipped before later", summary)
}
sessionManifest, err := store.Load(context.Background(), summary.ManifestPath)
if err != nil {
t.Fatalf("Load() session manifest error = %v", err)
}
selfSkipped := sessionManifest.Stages["optional"]
if selfSkipped == nil || selfSkipped.Status != manifest.StatusSkipped {
t.Fatalf("optional stage = %#v, want skipped", selfSkipped)
}
if len(selfSkipped.Outputs) != 0 {
t.Fatalf("optional outputs = %#v, want old outputs cleared", selfSkipped.Outputs)
}
if selfSkipped.Error == nil || selfSkipped.Error.Message != "integration_disabled" {
t.Fatalf("optional skip reason = %#v, want integration_disabled", selfSkipped.Error)
}
if len(selfSkipped.Logs) != 1 || selfSkipped.Logs[0] != "runs/current/optional.log" ||
len(selfSkipped.GeneratedConfigs) != 1 || selfSkipped.GeneratedConfigs[0] != "runs/current/optional.yml" ||
selfSkipped.Metadata["enabled"] != false || selfSkipped.Metadata["old"] != nil {
t.Fatalf("optional result details = %#v, want current bounded diagnostics and metadata", selfSkipped)
}
if later := sessionManifest.Stages["later"]; later == nil || later.Status != manifest.StatusSucceeded {
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["optional"]
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)
}
if optionalRuns != 2 {
t.Fatalf("optional runs = %d, want self-skipped stage reconsidered", optionalRuns)
}
}
func TestExecuteStagesRejectsSkippedResultWithOutputs(t *testing.T) {
cfg := testConfig(t)
invalid := resultStage{name: "optional", result: &stage.StageResult{
Disposition: stage.StageDispositionSkipped,
SkipReason: "integration_disabled",
Outputs: []artifacts.Ref{{Kind: "unexpected"}},
}}
summary, err := executeStages(context.Background(), cfg, []stage.Stage{invalid}, RunOptions{})
if err == nil {
t.Fatal("executeStages() error = nil, want invalid skipped result failure")
}
if summary != nil {
t.Fatalf("summary = %#v, want nil", summary)
}
if !strings.Contains(err.Error(), "skipped result contains 1 output") {
t.Fatalf("error = %q, want skipped-output validation", err)
}
loaded, loadErr := (&manifest.LocalStore{}).Load(context.Background(), manifestPathFor(cfg))
if loadErr != nil {
t.Fatalf("Load() session manifest error = %v", loadErr)
}
if got := loaded.Stages["optional"]; got == nil || got.Status != manifest.StatusFailed {
t.Fatalf("optional stage = %#v, want failed", got)
}
}
func TestExecuteStagesRunLocalArtifactsAndCanonicalSync(t *testing.T) {
cfg := testConfig(t)
stages := []stage.Stage{