Execute incremental analysis artifact plans

2026-08-29 19:43:28 +00:00
parent 6abdd67bb5
commit 99f4f9a0db
13 changed files with 613 additions and 211 deletions
--- a/internal/stage/analyze.go
+++ b/internal/stage/analyze.go
@@ -2,8 +2,9 @@ package stage

 import (
 	"context"
+	"crypto/sha256"
+	"encoding/hex"
 	"encoding/json"
-	"errors"
 	"fmt"
 	"path/filepath"
 	"sort"
@@ -11,6 +12,7 @@ import (
 	"time"

 	"gitea.maximumdirect.net/eric/narratio/internal/adapters/scriptorium"
+	"gitea.maximumdirect.net/eric/narratio/internal/artifactmodel"
 	"gitea.maximumdirect.net/eric/narratio/internal/artifactpolicy"
 	"gitea.maximumdirect.net/eric/narratio/internal/artifacts"
 	"gitea.maximumdirect.net/eric/narratio/internal/config"
@@ -28,6 +30,8 @@ type analyzeArtifactExecutionPlan struct {

 type analyzeArtifactExecutionResult struct {
 	Output           artifacts.Ref
+	OutputSize       int64
+	Scriptorium      manifest.AnalyzeArtifactProvenance
 	Logs             []string
 	GeneratedConfigs []string
 	Metadata         map[string]any
@@ -54,10 +58,6 @@ func (analyzeStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*S
 	if env.Config.Pipeline == nil || env.Config.Session == nil {
 		return nil, fmt.Errorf("analyze: resolved config must include pipeline and session")
 	}
-	if env.Scriptorium == nil {
-		return nil, fmt.Errorf("analyze: scriptorium adapter is required")
-	}
-
 	var sessionID string
 	if m != nil {
 		sessionID = strings.TrimSpace(m.SessionID)
@@ -81,6 +81,13 @@ func (analyzeStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*S
 			"reason":  "pipeline.scriptorium is not configured",
 		}}, nil
 	}
+	if len(env.Config.Pipeline.Scriptorium.Artifacts) == 0 {
+		return &StageResult{Metadata: map[string]any{
+			"stage":   "analyze",
+			"skipped": true,
+			"reason":  "no scriptorium artifacts configured",
+		}}, nil
+	}

 	effective := env.EffectiveArtifacts
 	if !effective.Resolved() {
@@ -104,15 +111,11 @@ func (analyzeStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*S
 		return nil, fmt.Errorf("analyze: build runtime artifact catalog: %w", err)
 	}

-	plans, skipReason, err := buildAnalyzeExecutionPlans(env.Config.Pipeline.Scriptorium, effective, runtimeCatalog)
-	if err != nil {
-		return nil, fmt.Errorf("analyze: %w", err)
-	}
-	if skipReason != "" {
+	if len(effective.Keys()) == 0 {
 		return &StageResult{Metadata: map[string]any{
 			"stage":   "analyze",
 			"skipped": true,
-			"reason":  skipReason,
+			"reason":  "no selected scriptorium artifacts to execute",
 		}}, nil
 	}

@@ -125,21 +128,64 @@ func (analyzeStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*S
 		TranscriptRefs: discoverAnalyzeTranscriptRefs(m, paths),
 		Catalog:        runtimeCatalog,
 	}
+	reconciliation, err := reconcileAnalyzeArtifacts(env.Config.Pipeline.Scriptorium, execution)
+	if err != nil {
+		return nil, fmt.Errorf("analyze: reconcile configured artifacts: %w", err)
+	}
+	workPlan, err := planAnalyzeWork(
+		env.Config.Pipeline.Scriptorium,
+		env.SelectedArtifactKeys,
+		env.Force,
+		reconciliation,
+	)
+	if err != nil {
+		return nil, fmt.Errorf("analyze: plan configured artifacts: %w", err)
+	}
+	if len(workPlan.ExecutionOrder) > 0 && env.Scriptorium == nil {
+		return nil, fmt.Errorf("analyze: scriptorium adapter is required")
+	}

-	outputs := make([]artifacts.Ref, 0, len(plans))
 	logs := []string{}
 	generatedConfigs := []string{}
-	artifactMetadata := make([]map[string]any, 0, len(plans))
+	artifactMetadata := make([]map[string]any, 0, len(workPlan.ExecutionOrder))
 	reusedArtifacts := []map[string]any{}
 	reusedSeen := map[string]struct{}{}
+	sessionRecords := manifest.CloneAnalyzeArtifactCollection(workPlan.ProjectedRecords)
+	invocationRecords := make(map[string]manifest.AnalyzeArtifactRecord)
+	priorCurrentRecords := make(map[string]manifest.AnalyzeArtifactRecord)
+	for _, item := range reconciliation.Ordered {
+		if item.Stored != nil && item.Stored.Status == manifest.AnalyzeArtifactCurrent {
+			priorCurrentRecords[item.Key] = *item.Stored
+		}
+	}
+	invocationKeys := make(map[string]struct{}, len(workPlan.ExecutionOrder)+len(workPlan.ReusedCurrent))
+	for _, item := range workPlan.ExecutionOrder {
+		invocationKeys[item.Key] = struct{}{}
+	}
+	for _, item := range workPlan.ReusedCurrent {
+		invocationKeys[item.Key] = struct{}{}
+		if record, ok := sessionRecords[item.Key]; ok {
+			invocationRecords[item.Key] = record
+		}
+	}

-	for _, plan := range plans {
+	for _, item := range workPlan.ExecutionOrder {
+		artifactCfg := env.Config.Pipeline.Scriptorium.Artifacts[item.Key]
+		fingerprint, _, _, err := computeAnalyzeArtifactFingerprint(
+			item.Key,
+			env.Config.Pipeline.Scriptorium,
+			execution,
+		)
+		if err != nil {
+			return nil, fmt.Errorf("analyze: compute execution fingerprint for artifact %q: %w", item.Key, err)
+		}
+		priorRecord, hadPriorRecord := priorCurrentRecords[item.Key]
+		plan := analyzeArtifactExecutionPlan{Name: item.Key, Cfg: artifactCfg}
 		artifactResult, err := executeAnalyzeArtifact(ctx, execution, plan)
 		if err != nil {
 			return nil, err
 		}

-		outputs = append(outputs, artifactResult.Output)
 		logs = append(logs, artifactResult.Logs...)
 		generatedConfigs = append(generatedConfigs, artifactResult.GeneratedConfigs...)
 		artifactMetadata = append(artifactMetadata, artifactResult.Metadata)
@@ -154,18 +200,47 @@ func (analyzeStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*S
 			reusedArtifacts = append(reusedArtifacts, reused)
 		}

+		record, err := currentAnalyzeArtifactRecord(
+			execution,
+			plan,
+			fingerprint,
+			artifactResult,
+		)
+		if err != nil {
+			return nil, fmt.Errorf("analyze: record artifact %q: %w", plan.Name, err)
+		}
+		sessionRecords[plan.Name] = record
+		invocationRecords[plan.Name] = record
+
 		sourceID, ok := runtimeCatalog.SourceIDForConfiguredKey(plan.Name)
 		if !ok {
 			return nil, fmt.Errorf("analyze: source id not found for artifact %q", plan.Name)
 		}
-		if err := runtimeCatalog.MarkAvailableGenerated(sourceID, artifactResult.Output.AbsolutePath); err != nil {
+		if err := runtimeCatalog.MarkAvailableGeneratedEvidence(
+			sourceID,
+			artifactResult.Output.AbsolutePath,
+			record.ProducerRunID,
+			record.Output.Checksum,
+			record.OutputSize,
+			record.Output.Contract,
+		); err != nil {
 			return nil, fmt.Errorf("analyze: mark generated artifact %q available: %w", sourceID, err)
 		}
+		if !hadPriorRecord || !sameAnalyzeOutputIdentity(priorRecord, record) {
+			staleUnscheduledAnalyzeDependents(
+				env.Config.Pipeline.Scriptorium.Artifacts,
+				plan.Name,
+				invocationKeys,
+				sessionRecords,
+			)
+		}
 	}

 	metadata := map[string]any{
 		"stage":                 "analyze",
-		"selected_artifacts":    extractPlanNames(plans),
+		"selected_artifacts":    append([]string(nil), workPlan.ExplicitTargets...),
+		"executed_artifacts":    analyzePlanKeysForMetadata(workPlan.ExecutionOrder),
+		"reused_current":        analyzePlanKeysForMetadata(workPlan.ReusedCurrent),
 		"generated_artifacts":   artifactMetadata,
 		"reused_artifacts":      reusedArtifacts,
 		"artifact_count":        len(artifactMetadata),
@@ -178,133 +253,134 @@ func (analyzeStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*S
 	}

 	return &StageResult{
-		Outputs:          outputs,
 		Logs:             dedupeAndSortPaths(logs),
 		GeneratedConfigs: dedupeAndSortPaths(generatedConfigs),
 		Metadata:         metadata,
+		AnalyzeState: &AnalyzeStateProjection{
+			Session:    sessionRecords,
+			Invocation: manifest.CloneAnalyzeArtifactCollection(invocationRecords),
+		},
 	}, nil
 }

-func buildAnalyzeExecutionPlans(
-	scriptoriumCfg *config.ScriptoriumConfig,
-	effective artifacts.EffectiveArtifactSet,
-	catalog *artifacts.ArtifactCatalog,
-) ([]analyzeArtifactExecutionPlan, string, error) {
-	if scriptoriumCfg == nil {
-		return nil, "pipeline.scriptorium is not configured", nil
+func currentAnalyzeArtifactRecord(
+	execution analyzeExecutionContext,
+	plan analyzeArtifactExecutionPlan,
+	fingerprint string,
+	result *analyzeArtifactExecutionResult,
+) (manifest.AnalyzeArtifactRecord, error) {
+	if result == nil {
+		return manifest.AnalyzeArtifactRecord{}, fmt.Errorf("execution result is required")
 	}
-	if len(scriptoriumCfg.Artifacts) == 0 {
-		return nil, "no scriptorium artifacts configured", nil
+	producerRunID := ""
+	if execution.Manifest != nil {
+		producerRunID = strings.TrimSpace(execution.Manifest.RunID)
 	}
-
-	if len(effective.Keys()) == 0 {
-		return nil, "no selected scriptorium artifacts to execute", nil
+	if producerRunID == "" {
+		// Direct stage callers predate invocation manifests. Application-owned
+		// execution always supplies the actual run identity.
+		producerRunID = "direct-analyze"
 	}
-
-	ordered, err := orderSelectedScriptoriumArtifacts(scriptoriumCfg.Artifacts, effective, catalog)
+	relativePath, err := normalizedAnalyzeOutputIdentity(plan.Cfg.OutputPath)
 	if err != nil {
-		return nil, "", err
+		return manifest.AnalyzeArtifactRecord{}, err
 	}
-
-	plans := make([]analyzeArtifactExecutionPlan, 0, len(ordered))
-	for _, name := range ordered {
-		artifactCfg, ok := scriptoriumCfg.Artifacts[name]
-		if !ok {
-			return nil, "", fmt.Errorf("selected artifact %q is not configured", name)
-		}
-		plans = append(plans, analyzeArtifactExecutionPlan{Name: name, Cfg: artifactCfg})
+	if relativePath == "" {
+		return manifest.AnalyzeArtifactRecord{}, fmt.Errorf("configured output path is required")
 	}
-	return plans, "", nil
+	contract := result.Output.Contract
+	if contract == nil {
+		return manifest.AnalyzeArtifactRecord{}, fmt.Errorf("validated output contract is required")
+	}
+	record := manifest.AnalyzeArtifactRecord{
+		Key:                plan.Name,
+		Status:             manifest.AnalyzeArtifactCurrent,
+		FingerprintVersion: manifest.AnalyzeFingerprintContractVersion,
+		Fingerprint:        fingerprint,
+		Dependencies:       normalizedAnalyzeDependencyKeys(plan.Cfg.DependsOn),
+		Output: &manifest.ArtifactRecord{
+			Kind:          "scriptorium_artifact",
+			SourceID:      artifacts.ConfiguredArtifactSourceID(plan.Name),
+			LocalPath:     relativePath,
+			Contract:      cloneAnalyzeOutputContract(contract),
+			ProducerRunID: producerRunID,
+			Checksum:      result.Output.Checksum,
+		},
+		OutputSize:       result.OutputSize,
+		ProducerRunID:    producerRunID,
+		UpdatedAt:        time.Now().UTC(),
+		Scriptorium:      &result.Scriptorium,
+		Logs:             dedupeAndSortPaths(result.Logs),
+		GeneratedConfigs: dedupeAndSortPaths(result.GeneratedConfigs),
+	}
+	if err := manifest.ValidateAnalyzeArtifactCollection(
+		manifest.AnalyzeStateContractVersion,
+		map[string]manifest.AnalyzeArtifactRecord{plan.Name: record},
+	); err != nil {
+		return manifest.AnalyzeArtifactRecord{}, err
+	}
+	return record, nil
 }

-func orderSelectedScriptoriumArtifacts(
-	artifactsCfg map[string]config.ScriptoriumArtifactConfig,
-	effective artifacts.EffectiveArtifactSet,
-	catalog *artifacts.ArtifactCatalog,
-) ([]string, error) {
-	selectedSet := map[string]struct{}{}
-	selected := effective.Keys()
-	for _, key := range selected {
-		selectedSet[key] = struct{}{}
+func sameAnalyzeOutputIdentity(left, right manifest.AnalyzeArtifactRecord) bool {
+	if left.Status != manifest.AnalyzeArtifactCurrent || right.Status != manifest.AnalyzeArtifactCurrent ||
+		left.Output == nil || right.Output == nil || left.Output.Contract == nil || right.Output.Contract == nil {
+		return false
 	}
+	return left.OutputSize == right.OutputSize &&
+		left.Output.Checksum == right.Output.Checksum &&
+		*left.Output.Contract == *right.Output.Contract
+}

-	dependencyErrors := []string{}
-	for _, selectedKey := range selected {
-		cfg, ok := artifactsCfg[selectedKey]
-		if !ok {
-			dependencyErrors = append(dependencyErrors, fmt.Sprintf("selected artifact %q is not configured", selectedKey))
+func staleUnscheduledAnalyzeDependents(
+	configured map[string]config.ScriptoriumArtifactConfig,
+	changed string,
+	invocationKeys map[string]struct{},
+	records map[string]manifest.AnalyzeArtifactRecord,
+) {
+	reverse := make(map[string][]string, len(configured))
+	for key, artifactCfg := range configured {
+		for _, dependency := range normalizedAnalyzeDependencyKeys(artifactCfg.DependsOn) {
+			reverse[dependency] = append(reverse[dependency], key)
+		}
+	}
+	for key := range reverse {
+		sort.Strings(reverse[key])
+	}
+	queue := append([]string(nil), reverse[changed]...)
+	seen := make(map[string]struct{}, len(queue))
+	for len(queue) > 0 {
+		key := queue[0]
+		queue = queue[1:]
+		if _, visited := seen[key]; visited {
 			continue
 		}
-		for _, dep := range cfg.DependsOn {
-			trimmedDep := strings.TrimSpace(dep)
-			if trimmedDep == "" {
-				continue
-			}
-			if _, ok := selectedSet[trimmedDep]; ok {
-				continue
-			}
-			sourceID, ok := catalog.SourceIDForConfiguredKey(trimmedDep)
-			if !ok {
-				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 {
-				dependencyErrors = append(dependencyErrors, fmt.Sprintf("artifact %q depends on %q, but %q is unavailable", selectedKey, trimmedDep, sourceID))
-			}
+		seen[key] = struct{}{}
+		if _, evaluated := invocationKeys[key]; evaluated {
+			continue
 		}
+		staleProjectedAnalyzeRecord(records, key)
+		queue = append(queue, reverse[key]...)
 	}
-	if len(dependencyErrors) > 0 {
-		return nil, errors.New(strings.Join(dependencyErrors, "; "))
-	}
+}

-	indegree := map[string]int{}
-	edges := map[string][]string{}
-	for _, key := range selected {
-		indegree[key] = 0
+func analyzePlanKeysForMetadata(items []analyzePlanItem) []string {
+	if len(items) == 0 {
+		return nil
 	}
-	for _, key := range selected {
-		cfg := artifactsCfg[key]
-		for _, dep := range cfg.DependsOn {
-			trimmedDep := strings.TrimSpace(dep)
-			if _, ok := selectedSet[trimmedDep]; !ok {
-				continue
-			}
-			edges[trimmedDep] = append(edges[trimmedDep], key)
-			indegree[key]++
-		}
+	keys := make([]string, 0, len(items))
+	for _, item := range items {
+		keys = append(keys, item.Key)
 	}
+	return keys
+}

-	for key := range edges {
-		sort.Strings(edges[key])
+func cloneAnalyzeOutputContract(value *artifactmodel.ContractMetadata) *artifactmodel.ContractMetadata {
+	if value == nil {
+		return nil
 	}
-
-	ready := make([]string, 0, len(indegree))
-	for key, degree := range indegree {
-		if degree == 0 {
-			ready = append(ready, key)
-		}
-	}
-	sort.Strings(ready)
-
-	order := make([]string, 0, len(selectedSet))
-	for len(ready) > 0 {
-		node := ready[0]
-		ready = ready[1:]
-		order = append(order, node)
-		for _, dep := range edges[node] {
-			indegree[dep]--
-			if indegree[dep] == 0 {
-				ready = append(ready, dep)
-				sort.Strings(ready)
-			}
-		}
-	}
-
-	if len(order) != len(selectedSet) {
-		return nil, fmt.Errorf("selected scriptorium artifacts contain a dependency cycle")
-	}
-	return order, nil
+	cloned := *value
+	return &cloned
 }

 func executeAnalyzeArtifact(
@@ -504,14 +580,32 @@ func executeAnalyzeArtifact(
 	if err := requireNonEmptyFile(finalOutputPath, artifactName+" output"); err != nil {
 		return nil, fmt.Errorf("analyze: %w", err)
 	}
+	validatedOutput, err := readExternalResult(finalOutputPath, artifactName+" output")
+	if err != nil {
+		return nil, fmt.Errorf("analyze: %w", err)
+	}
+	contract := &artifactmodel.ContractMetadata{
+		MediaType: "text/markdown", SchemaID: "narratio." + artifactName, SchemaVersion: "1",
+	}
+	relativeOutputPath, err := normalizedAnalyzeOutputIdentity(artifactCfg.OutputPath)
+	if err != nil {
+		return nil, fmt.Errorf("analyze: normalize output identity for artifact %q: %w", artifactName, err)
+	}
 	materializedArtifact, err := materializeRunLocalOutput(env.ArtifactStore, finalOutputPath, canonicalOutputPath, artifacts.Ref{
-		Kind:      artifactName,
-		Category:  "artifacts",
-		SessionID: sessionID,
+		Kind:         artifactName,
+		SourceID:     artifacts.ConfiguredArtifactSourceID(artifactName),
+		Category:     "artifacts",
+		SessionID:    sessionID,
+		RelativePath: relativeOutputPath,
+		Contract:     contract,
 	})
 	if err != nil {
 		return nil, fmt.Errorf("analyze: materialize artifact output for %q: %w", artifactName, err)
 	}
+	expectedDigest := sha256.Sum256(validatedOutput)
+	if materializedArtifact.Checksum != hex.EncodeToString(expectedDigest[:]) {
+		return nil, fmt.Errorf("analyze: materialized artifact output for %q differs from validated run-local bytes", artifactName)
+	}

 	logPaths = append(logPaths, stdoutLogPath, stderrLogPath)
 	generatedConfigs = append(generatedConfigs, generatedConfigPath)
@@ -537,7 +631,11 @@ func executeAnalyzeArtifact(
 	}

 	return &analyzeArtifactExecutionResult{
-		Output:           materializedArtifact,
+		Output:     materializedArtifact,
+		OutputSize: int64(len(validatedOutput)),
+		Scriptorium: manifest.AnalyzeArtifactProvenance{
+			PromptID: artifactCfg.PromptID, ProfileID: artifactCfg.ProfileID, CommandMode: res.CommandMode,
+		},
 		Logs:             logPaths,
 		GeneratedConfigs: generatedConfigs,
 		Metadata:         meta,
@@ -545,17 +643,6 @@ func executeAnalyzeArtifact(
 	}, nil
 }

-func extractPlanNames(plans []analyzeArtifactExecutionPlan) []string {
-	if len(plans) == 0 {
-		return nil
-	}
-	out := make([]string, 0, len(plans))
-	for _, plan := range plans {
-		out = append(out, plan.Name)
-	}
-	return out
-}
-
 func configuredArtifactNameFromSourceID(sourceID string) string {
 	name, _ := artifactpolicy.ParseConfiguredSource(sourceID)
 	return name