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

Make analyze input resolution deterministic

Browse Source
This commit is contained in:
Eric Rakestraw
2026-08-10 22:36:21 +00:00
parent af492c9e97
commit 131ffd9887
5 changed files with 260 additions and 139 deletions
Download Patch File Download Diff File Expand all files Collapse all files

258
internal/stage/analyze.go
View File

@@ -5,7 +5,6 @@ import (
"encoding/json"
"errors"
"fmt"
"os"
"path/filepath"
"sort"
"strings"
@@ -46,6 +45,31 @@ type analyzeArtifactExecutionResult struct {
ReusedArtifacts []map[string]any
}
type analyzeExecutionContext struct {
Env *Env
Manifest *manifest.Manifest
Paths artifacts.SessionPaths
RunLayout runStageLayout
SessionID string
TranscriptRefs analyzeTranscriptInputs
Catalog *artifacts.ArtifactCatalog
}
type analyzeInputResolutionState uint8
const (
analyzeInputPresent analyzeInputResolutionState = iota
analyzeInputAbsent
analyzeInputError
)
type analyzeInputResolution struct {
State analyzeInputResolutionState
Path string
Artifact *artifacts.ResolvedSessionArtifact
Err error
}
func (analyzeStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*StageResult, error) {
if env == nil || env.Config == nil {
return nil, fmt.Errorf("analyze: stage environment config is required")
@@ -118,8 +142,15 @@ func (analyzeStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*S
}}, nil
}
transcriptRefs := discoverAnalyzeTranscriptRefs(m, paths)
sessionDir := filepath.Dir(strings.TrimSpace(env.Config.SessionPath))
execution := analyzeExecutionContext{
Env: env,
Manifest: m,
Paths: paths,
RunLayout: runLayout,
SessionID: sessionID,
TranscriptRefs: discoverAnalyzeTranscriptRefs(m, paths),
Catalog: runtimeCatalog,
}
outputs := make([]artifacts.Ref, 0, len(plans))
logs := []string{}
@@ -129,18 +160,7 @@ func (analyzeStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*S
reusedSeen := map[string]struct{}{}
for _, plan := range plans {
artifactResult, err := executeAnalyzeArtifact(
ctx,
env,
m,
paths,
runLayout,
sessionID,
sessionDir,
plan,
transcriptRefs,
runtimeCatalog,
)
artifactResult, err := executeAnalyzeArtifact(ctx, execution, plan)
if err != nil {
return nil, err
}
@@ -229,14 +249,17 @@ func orderSelectedScriptoriumArtifacts(
catalog *artifacts.ArtifactCatalog,
) ([]string, error) {
selectedSet := map[string]struct{}{}
for _, key := range effective.Keys() {
selected := effective.Keys()
for _, key := range selected {
selectedSet[key] = struct{}{}
}
for selectedKey := range selectedSet {
dependencyErrors := []string{}
for _, selectedKey := range selected {
cfg, ok := artifactsCfg[selectedKey]
if !ok {
return nil, fmt.Errorf("selected artifact %q is not configured", selectedKey)
dependencyErrors = append(dependencyErrors, fmt.Sprintf("selected artifact %q is not configured", selectedKey))
continue
}
for _, dep := range cfg.DependsOn {
trimmedDep := strings.TrimSpace(dep)
@@ -248,21 +271,25 @@ func orderSelectedScriptoriumArtifacts(
}
sourceID, ok := catalog.SourceIDForConfiguredKey(trimmedDep)
if !ok {
return nil, fmt.Errorf("artifact %q depends on unknown configured artifact %q", selectedKey, trimmedDep)
dependencyErrors = append(dependencyErrors, fmt.Sprintf("artifact %q depends on unknown configured artifact %q", selectedKey, trimmedDep))
continue
}
entry, ok := catalog.Lookup(sourceID)
if !ok || !entry.Available {
return nil, fmt.Errorf("artifact %q depends on %q, but %q is unavailable", selectedKey, trimmedDep, sourceID)
dependencyErrors = append(dependencyErrors, fmt.Sprintf("artifact %q depends on %q, but %q is unavailable", selectedKey, trimmedDep, sourceID))
}
}
}
if len(dependencyErrors) > 0 {
return nil, errors.New(strings.Join(dependencyErrors, "; "))
}
indegree := map[string]int{}
edges := map[string][]string{}
for key := range selectedSet {
for _, key := range selected {
indegree[key] = 0
}
for key := range selectedSet {
for _, key := range selected {
cfg := artifactsCfg[key]
for _, dep := range cfg.DependsOn {
trimmedDep := strings.TrimSpace(dep)
@@ -308,16 +335,14 @@ func orderSelectedScriptoriumArtifacts(
func executeAnalyzeArtifact(
ctx context.Context,
env *Env,
m *manifest.Manifest,
paths artifacts.SessionPaths,
runLayout runStageLayout,
sessionID string,
sessionDir string,
execution analyzeExecutionContext,
plan analyzeArtifactExecutionPlan,
transcriptRefs analyzeTranscriptInputs,
runtimeCatalog *artifacts.ArtifactCatalog,
) (*analyzeArtifactExecutionResult, error) {
env := execution.Env
paths := execution.Paths
runLayout := execution.RunLayout
sessionID := execution.SessionID
transcriptRefs := execution.TranscriptRefs
artifactName := plan.Name
artifactCfg := plan.Cfg
@@ -328,24 +353,27 @@ func executeAnalyzeArtifact(
inputNames := sortedScriptoriumInputNames(artifactCfg.Inputs)
for _, inputName := range inputNames {
inputCfg := artifactCfg.Inputs[inputName]
resolvedPath, resolved, resolvedArtifact, resolveErr := resolveScriptoriumInput(inputName, inputCfg, m, paths, sessionDir, runtimeCatalog)
if resolveErr != nil {
return nil, fmt.Errorf("analyze: resolve input %q for artifact %q: %w", inputName, artifactName, resolveErr)
}
if !resolved {
resolution := resolveScriptoriumInput(inputCfg, execution)
switch resolution.State {
case analyzeInputError:
return nil, fmt.Errorf("analyze: resolve input %q for artifact %q: %w", inputName, artifactName, resolution.Err)
case analyzeInputAbsent:
if inputCfg.Required {
return nil, fmt.Errorf("analyze: required input %q for artifact %q could not be resolved", inputName, artifactName)
}
omittedOptionalInputs = append(omittedOptionalInputs, inputName)
continue
case analyzeInputPresent:
inputPaths[inputName] = resolution.Path
default:
return nil, fmt.Errorf("analyze: resolve input %q for artifact %q: invalid resolution state", inputName, artifactName)
}
inputPaths[inputName] = resolvedPath
if resolvedArtifact != nil && resolvedArtifact.Provenance == artifacts.ArtifactProvenanceDisabledFromDisk {
if resolution.Artifact != nil && resolution.Artifact.Provenance == artifacts.ArtifactProvenanceDisabledFromDisk {
reusedArtifacts = append(reusedArtifacts, map[string]any{
"name": configuredArtifactNameFromSourceID(resolvedArtifact.ID),
"source_id": resolvedArtifact.ID,
"path": resolvedArtifact.Path,
"provenance": resolvedArtifact.Provenance,
"name": configuredArtifactNameFromSourceID(resolution.Artifact.ID),
"source_id": resolution.Artifact.ID,
"path": resolution.Artifact.Path,
"provenance": resolution.Artifact.Provenance,
})
}
}
@@ -603,91 +631,96 @@ func discoverAnalyzeArtifactRef(m *manifest.Manifest, paths artifacts.SessionPat
return resolved.Path, resolved.Provenance
}
func resolveScriptoriumInput(
inputName string,
inputCfg config.ScriptoriumInputConfig,
m *manifest.Manifest,
paths artifacts.SessionPaths,
sessionDir string,
runtimeCatalog *artifacts.ArtifactCatalog,
) (string, bool, *artifacts.ResolvedSessionArtifact, error) {
func resolveScriptoriumInput(inputCfg config.ScriptoriumInputConfig, execution analyzeExecutionContext) analyzeInputResolution {
source := strings.TrimSpace(inputCfg.Source)
descriptor, describeErr := artifactpolicy.DescribeScriptoriumInputSource(source)
if describeErr != nil {
return "", false, nil, describeErr
return analyzeInputFailure(describeErr)
}
if descriptor.Source.Kind == artifactpolicy.SourceKindStableInput {
resolvedPath, ok, err := resolvePreparedStableInput(descriptor.Source.ID, paths)
resolvedPath, ok, err := resolvePreparedStableInput(descriptor.Source.ID, execution.Paths)
if err != nil {
if inputCfg.Required {
return "", false, nil, err
return analyzeInputFailure(err)
}
return "", false, nil, nil
return analyzeInputMissing()
}
return resolvedPath, ok, nil, nil
if !ok {
return analyzeInputMissing()
}
return analyzeInputFound(resolvedPath, nil)
}
if descriptor.Source.Kind == artifactpolicy.SourceKindPreviousArtifact {
resolved, err := artifacts.ResolvePreviousSessionArtifactWithCatalog(paths, m, source, runtimeCatalog)
resolved, err := artifacts.ResolvePreviousSessionArtifactWithCatalog(execution.Paths, execution.Manifest, source, execution.Catalog)
if err == nil {
copy := resolved
return resolved.Path, true, &copy, nil
return analyzeInputFound(resolved.Path, &copy)
}
if errors.Is(err, artifacts.ErrSessionArtifactNotFound) {
if inputCfg.Required {
return "", false, nil, fmt.Errorf(
"required previous-session input source %q is unavailable; run narratio run-stage --force prepare",
return analyzeInputFailure(fmt.Errorf(
"required previous-session input source %q is unavailable; run narratio run-stage prepare %s --force",
source,
)
execution.SessionID,
))
}
return "", false, nil, nil
return analyzeInputMissing()
}
return "", false, nil, err
return analyzeInputFailure(err)
}
switch source {
resolved, err := artifacts.ResolveSessionArtifactWithCatalog(execution.Paths, execution.Manifest, source, execution.Catalog)
if err == nil {
copy := resolved
return analyzeInputFound(resolved.Path, &copy)
}
if !errors.Is(err, artifacts.ErrSessionArtifactNotFound) {
return analyzeInputFailure(err)
}
if !inputCfg.Required {
return analyzeInputMissing()
}
switch descriptor.Source.Kind {
case artifactpolicy.SourceKindExtraction:
return analyzeInputFailure(fmt.Errorf(
"required extraction source %q is unavailable; enable and configure pipeline.notarius output %q, then run narratio run-stage extract %s --force",
source,
descriptor.Source.ConfiguredKey,
execution.SessionID,
))
case artifactpolicy.SourceKindConfiguredArtifact:
return analyzeInputFailure(fmt.Errorf("configured artifact source %q is unavailable", source))
default:
resolved, err := artifacts.ResolveSessionArtifactWithCatalog(paths, m, source, runtimeCatalog)
if err == nil {
copy := resolved
return resolved.Path, true, &copy, nil
}
if errors.Is(err, artifacts.ErrSessionArtifactNotFound) {
if descriptor.Source.Kind == artifactpolicy.SourceKindExtraction {
if inputCfg.Required {
return "", false, nil, fmt.Errorf(
"required extraction source %q is unavailable; enable and configure pipeline.notarius output %q, then rerun extract with --force",
source,
descriptor.Source.ConfiguredKey,
)
}
return "", false, nil, nil
}
if descriptor.Source.Kind == artifactpolicy.SourceKindConfiguredArtifact {
if inputCfg.Required {
return "", false, nil, fmt.Errorf("configured artifact source %q is unavailable", source)
}
return "", false, nil, nil
}
switch descriptor.Source.ID {
case artifacts.ArtifactTranscriptPolished:
return "", false, nil, nil
case artifacts.ArtifactTranscriptFinal:
return "", false, nil, fmt.Errorf("normalized transcript input is unavailable; run normalize stage first")
case artifacts.ArtifactTranscriptFinalTrimmed:
return "", false, nil, fmt.Errorf("trimmed transcript input is unavailable; run trim stage first")
case artifacts.ArtifactTranscriptFinalMarkdown, artifacts.ArtifactTranscriptFinalTrimmedMarkdown:
return "", false, nil, fmt.Errorf(
"rendered markdown transcript input is unavailable for source %q; run narratio run-stage render %s --force",
descriptor.Source.ID,
paths.SessionID,
)
default:
return "", false, nil, nil
}
}
return "", false, nil, err
return analyzeInputFailure(requiredBuiltInInputError(descriptor.Source.ID, execution))
}
}
func analyzeInputFound(path string, artifact *artifacts.ResolvedSessionArtifact) analyzeInputResolution {
return analyzeInputResolution{State: analyzeInputPresent, Path: path, Artifact: artifact}
}
func analyzeInputMissing() analyzeInputResolution {
return analyzeInputResolution{State: analyzeInputAbsent}
}
func analyzeInputFailure(err error) analyzeInputResolution {
return analyzeInputResolution{State: analyzeInputError, Err: err}
}
func requiredBuiltInInputError(source string, execution analyzeExecutionContext) error {
entry, ok := execution.Catalog.Lookup(source)
if !ok || strings.TrimSpace(entry.ProducerStage) == "" {
return fmt.Errorf("required built-in source %q is unavailable", source)
}
return fmt.Errorf(
"required built-in source %q is unavailable; run narratio run-stage %s %s --force",
source,
entry.ProducerStage,
execution.SessionID,
)
}
func resolvePreparedStableInput(sourceID string, paths artifacts.SessionPaths) (string, bool, error) {
filename, ok := preparedStableInputFilename(sourceID)
if !ok {
@@ -760,27 +793,6 @@ func buildAnalyzeRuntimeArtifactCatalog(
return catalog, nil
}
func resolveInputPathForRead(paths artifacts.SessionPaths, sessionDir, pathValue string) string {
trimmed := strings.TrimSpace(pathValue)
if trimmed == "" {
return ""
}
if filepath.IsAbs(trimmed) {
return filepath.Clean(trimmed)
}
candidates := []string{}
if strings.TrimSpace(sessionDir) != "" {
candidates = append(candidates, filepath.Clean(filepath.Join(sessionDir, trimmed)))
}
candidates = append(candidates, filepath.Clean(artifacts.ResolveSessionLocalPathForRead(paths, trimmed)))
for _, c := range candidates {
if info, err := os.Stat(c); err == nil && !info.IsDir() {
return c
}
}
return candidates[0]
}
func resolveScriptoriumOutputPath(paths artifacts.SessionPaths, configured string) (string, error) {
outputPath := strings.TrimSpace(configured)
if outputPath == "" {