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

Consolidate analyze session execution state

Browse Source
This commit is contained in:
Eric Rakestraw
2026-09-04 15:57:54 +00:00
parent 5c520dc0d4
commit 172e2d6a4f
12 changed files with 401 additions and 268 deletions
Download Patch File Download Diff File Expand all files Collapse all files

172
internal/stage/analyze.go
View File

@@ -142,25 +142,7 @@ func (analyzeStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*S
return nil, fmt.Errorf("analyze: scriptorium adapter is required")
}
logs := []string{}
generatedConfigs := []string{}
artifactMetadata := make([]map[string]any, 0, len(workPlan.ExecutionOrder))
reusedArtifacts := []map[string]any{}
reusedSeen := map[string]struct{}{}
sessionRecords := manifest.CloneAnalyzeArtifactCollection(workPlan.ProjectedRecords)
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
}
}
currentKeys := make(map[string]struct{}, len(workPlan.ExecutionOrder)+len(workPlan.ReusedCurrent))
for _, item := range workPlan.ExecutionOrder {
currentKeys[item.Key] = struct{}{}
}
for _, item := range workPlan.ReusedCurrent {
currentKeys[item.Key] = struct{}{}
}
state := newAnalyzeExecutionState(execution, workPlan, reconciliation)
for _, item := range workPlan.ExecutionOrder {
artifactCfg := env.Config.Pipeline.Scriptorium.Artifacts[item.Key]
@@ -171,158 +153,18 @@ func (analyzeStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*S
)
if err != nil {
executionErr := fmt.Errorf("analyze: compute execution fingerprint for artifact %q: %w", item.Key, err)
return failedAnalyzeResult(
execution,
item,
artifactCfg,
"",
executionErr,
sessionRecords,
), executionErr
return state.failedResult(item, artifactCfg, "", executionErr), executionErr
}
priorRecord, hadPriorRecord := priorCurrentRecords[item.Key]
plan := analyzeArtifactExecutionPlan{Name: item.Key, Cfg: artifactCfg}
artifactResult, err := executeAnalyzeArtifact(ctx, execution, plan)
if err != nil {
return failedAnalyzeResult(
execution,
item,
artifactCfg,
fingerprint,
err,
sessionRecords,
), err
return state.failedResult(item, artifactCfg, fingerprint, err), err
}
logs = append(logs, artifactResult.Logs...)
generatedConfigs = append(generatedConfigs, artifactResult.GeneratedConfigs...)
if origin, ok := analyzeArtifactOrigin(execution, plan.Name); ok {
artifactResult.Metadata["family"] = origin.Family
artifactResult.Metadata["character_id"] = origin.CharacterID
}
artifactMetadata = append(artifactMetadata, artifactResult.Metadata)
for _, reused := range artifactResult.ReusedArtifacts {
sourceID, _ := reused["source_id"].(string)
path, _ := reused["path"].(string)
key := sourceID + "|" + path
if _, exists := reusedSeen[key]; exists {
continue
}
reusedSeen[key] = struct{}{}
reusedArtifacts = append(reusedArtifacts, reused)
}
record, err := currentAnalyzeArtifactRecord(
execution,
plan,
fingerprint,
artifactResult,
)
if err != nil {
recordErr := fmt.Errorf("analyze: record artifact %q: %w", plan.Name, err)
return failedAnalyzeResult(
execution,
item,
artifactCfg,
fingerprint,
recordErr,
sessionRecords,
), recordErr
}
sessionRecords[plan.Name] = record
sourceID, ok := runtimeCatalog.SourceIDForConfiguredKey(plan.Name)
if !ok {
catalogErr := fmt.Errorf("analyze: source id not found for artifact %q", plan.Name)
return failedAnalyzeResult(
execution, item, artifactCfg, fingerprint, catalogErr,
sessionRecords,
), catalogErr
}
if err := runtimeCatalog.MarkAvailableGeneratedEvidence(
sourceID,
artifactResult.Output.AbsolutePath,
record.ProducerRunID,
record.Output.Checksum,
record.OutputSize,
record.Output.Contract,
); err != nil {
catalogErr := fmt.Errorf("analyze: mark generated artifact %q available: %w", sourceID, err)
return failedAnalyzeResult(
execution, item, artifactCfg, fingerprint, catalogErr,
sessionRecords,
), catalogErr
}
if !hadPriorRecord || !sameAnalyzeOutputIdentity(priorRecord, record) {
staleUnscheduledAnalyzeDependents(
env.Config.Pipeline.Scriptorium.Artifacts,
plan.Name,
currentKeys,
sessionRecords,
)
if err := state.acceptArtifact(plan, fingerprint, artifactResult); err != nil {
return state.failedResult(item, artifactCfg, fingerprint, err), err
}
}
metadata := map[string]any{
"stage": "analyze",
"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),
"reused_artifact_count": len(reusedArtifacts),
}
if len(artifactMetadata) == 1 {
for k, v := range artifactMetadata[0] {
metadata[k] = v
}
}
return &StageResult{
Logs: dedupeAndSortPaths(logs),
GeneratedConfigs: dedupeAndSortPaths(generatedConfigs),
Metadata: metadata,
AnalyzeState: &AnalyzeStateProjection{
Session: sessionRecords,
},
}, nil
}
func failedAnalyzeResult(
execution analyzeExecutionContext,
item analyzePlanItem,
artifactCfg config.ScriptoriumArtifactConfig,
fingerprint string,
cause error,
sessionRecords map[string]manifest.AnalyzeArtifactRecord,
) *StageResult {
record := manifest.AnalyzeArtifactRecord{
Key: item.Key,
Status: manifest.AnalyzeArtifactFailed,
Dependencies: normalizedAnalyzeDependencyKeys(artifactCfg.DependsOn),
ProducerRunID: analyzeProducerRunID(execution),
UpdatedAt: time.Now().UTC(),
Error: NonResumable(cause.Error()).Reason,
}
if origin, ok := analyzeArtifactOrigin(execution, item.Key); ok {
record.Family = origin.Family
record.CharacterID = origin.CharacterID
}
if fingerprint != "" {
record.FingerprintVersion = manifest.AnalyzeFingerprintContractVersion
record.Fingerprint = fingerprint
}
if artifactCfg.PromptID != "" || artifactCfg.ProfileID != "" {
record.Scriptorium = &manifest.AnalyzeArtifactProvenance{
PromptID: artifactCfg.PromptID, ProfileID: artifactCfg.ProfileID,
}
}
sessionRecords[item.Key] = record
staleAnalyzeDependents(artifactCfgMap(execution), item.Key, sessionRecords)
return &StageResult{AnalyzeState: &AnalyzeStateProjection{
Session: manifest.CloneAnalyzeArtifactCollection(sessionRecords),
}}
return state.successResult(), nil
}
func analyzeArtifactOrigin(execution analyzeExecutionContext, key string) (config.ArtifactFamilyMemberOrigin, bool) {
@@ -464,7 +306,7 @@ func staleAnalyzeDependentClosure(
if eligible != nil && !eligible(key) {
continue
}
staleProjectedAnalyzeRecord(records, key)
staleAnalyzeRecord(records, key)
queue = append(queue, reverse[key]...)
}
}