package stage import ( "context" "encoding/json" "errors" "fmt" "os" "path/filepath" "sort" "strings" "time" "gitea.maximumdirect.net/eric/narratio/internal/adapters/scriptorium" "gitea.maximumdirect.net/eric/narratio/internal/artifacts" "gitea.maximumdirect.net/eric/narratio/internal/config" "gitea.maximumdirect.net/eric/narratio/internal/manifest" ) type analyzeStage struct{} func (analyzeStage) Name() string { return "analyze" } func (analyzeStage) Declares() IODecl { return IODecl{ Inputs: []artifacts.Ref{ {Kind: "transcript_processed", Category: "transcripts", RelativePath: "transcripts/processed.json"}, {Kind: "transcript_normalized", Category: "transcripts", RelativePath: "transcripts/normalized.json"}, {Kind: "transcript_trimmed", Category: "transcripts", RelativePath: "transcripts/trimmed.json"}, }, Outputs: []artifacts.Ref{ {Kind: "session_recap", Category: "artifacts", RelativePath: "artifacts/session_recap.md"}, }, } } 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") } if env.ArtifactStore == nil { return nil, fmt.Errorf("analyze: artifact store is required") } 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) } if sessionID == "" { sessionID = strings.TrimSpace(env.Config.Session.SessionID) } if sessionID == "" { return nil, fmt.Errorf("analyze: session id is required") } paths := sessionPathsForEnv(env, sessionID) runLayout, err := resolveRunStageLayout(env, m, paths, sessionID, "analyze") if err != nil { return nil, fmt.Errorf("analyze: resolve run-stage layout: %w", err) } if env.Config.Pipeline.Scriptorium == nil { return &StageResult{ Metadata: map[string]any{ "stage": "analyze", "skipped": true, "reason": "pipeline.scriptorium is not configured", }, }, nil } artifactName, artifactCfg, skipReason, err := selectAnalyzeArtifact(env.Config.Pipeline.Scriptorium) if err != nil { return nil, fmt.Errorf("analyze: %w", err) } if skipReason != "" { return &StageResult{ Metadata: map[string]any{ "stage": "analyze", "skipped": true, "reason": skipReason, }, }, nil } runtimeCatalog, err := buildAnalyzeRuntimeArtifactCatalog(paths, env.Config.Pipeline.Scriptorium, env.SelectedAnalyzeArtifacts) if err != nil { return nil, fmt.Errorf("analyze: build runtime artifact catalog: %w", err) } transcriptRefs := discoverAnalyzeTranscriptRefs(m, paths) inputPaths := map[string]string{} omittedOptionalInputs := []string{} sessionDir := filepath.Dir(strings.TrimSpace(env.Config.SessionPath)) inputNames := sortedScriptoriumInputNames(artifactCfg.Inputs) for _, inputName := range inputNames { inputCfg := artifactCfg.Inputs[inputName] resolvedPath, resolved, resolveErr := resolveScriptoriumInput(inputName, inputCfg, m, paths, sessionDir, runtimeCatalog) if resolveErr != nil { return nil, fmt.Errorf("analyze: resolve input %q: %w", inputName, resolveErr) } if !resolved { if inputCfg.Required { return nil, fmt.Errorf("analyze: required input %q could not be resolved", inputName) } omittedOptionalInputs = append(omittedOptionalInputs, inputName) continue } inputPaths[inputName] = resolvedPath } vars, err := buildScriptoriumVars(artifactCfg.Vars, env.Config.Session) if err != nil { return nil, fmt.Errorf("analyze: resolve vars: %w", err) } canonicalOutputPath, err := resolveScriptoriumOutputPath(paths, artifactCfg.OutputPath) if err != nil { return nil, fmt.Errorf("analyze: resolve output path: %w", err) } outputPath, err := runLocalPathForCanonical(runLayout, paths, canonicalOutputPath) if err != nil { return nil, fmt.Errorf("analyze: resolve run-local output path: %w", err) } stdoutLogPath := filepath.Join(paths.LogsDir, "scriptorium."+artifactName+".stdout.log") stderrLogPath := filepath.Join(paths.LogsDir, "scriptorium."+artifactName+".stderr.log") generatedConfigPath := filepath.Join(paths.ConfigDir, "scriptorium."+artifactName+".generated.yml") if runLayout.Enabled { stdoutLogPath = filepath.Join(runLayout.LogsDir, "scriptorium."+artifactName+".stdout.log") stderrLogPath = filepath.Join(runLayout.LogsDir, "scriptorium."+artifactName+".stderr.log") generatedConfigPath = filepath.Join(runLayout.ConfigDir, "scriptorium."+artifactName+".generated.yml") } timeout, err := resolveScriptoriumTimeout(env.Config.Pipeline.Scriptorium.Timeout, artifactCfg.Timeout) if err != nil { return nil, fmt.Errorf("analyze: resolve timeout: %w", err) } logPaths := []string{} generatedConfigs := []string{} meta := map[string]any{ "stage": "analyze", "artifact_name": artifactName, "prompt_id": artifactCfg.PromptID, "profile_id": artifactCfg.ProfileID, "binary": env.Config.Pipeline.Scriptorium.Binary, "config_path": env.Config.Pipeline.Scriptorium.ConfigPath, "input_paths": inputPaths, "input_names": sortedMapKeys(inputPaths), "omitted_optional_inputs": omittedOptionalInputs, "vars": vars, "timeout": timeout.String(), "processed_transcript_path": transcriptRefs.ProcessedPath, "processed_transcript_source": transcriptRefs.ProcessedSource, "normalized_transcript_path": transcriptRefs.NormalizedPath, "normalized_transcript_source": transcriptRefs.NormalizedSource, "trimmed_transcript_path": transcriptRefs.TrimmedPath, "trimmed_transcript_source": transcriptRefs.TrimmedSource, "render_debug_enabled": resolveRenderDebugEnabled(env.Config.Pipeline.Scriptorium.RenderDebug, artifactCfg.RenderDebug), } if meta["render_debug_enabled"] == true { renderOutputPath := filepath.Join(paths.ArtifactsDir, artifactName+".render.json") if runLayout.Enabled { renderOutputPath = filepath.Join(runLayout.ReportsDir, artifactName+".render.json") } renderStdoutPath := filepath.Join(paths.LogsDir, "scriptorium."+artifactName+".render.stdout.log") renderStderrPath := filepath.Join(paths.LogsDir, "scriptorium."+artifactName+".render.stderr.log") renderGeneratedConfigPath := filepath.Join(paths.ConfigDir, "scriptorium."+artifactName+".render.generated.yml") if runLayout.Enabled { renderStdoutPath = filepath.Join(runLayout.LogsDir, "scriptorium."+artifactName+".render.stdout.log") renderStderrPath = filepath.Join(runLayout.LogsDir, "scriptorium."+artifactName+".render.stderr.log") renderGeneratedConfigPath = filepath.Join(runLayout.ConfigDir, "scriptorium."+artifactName+".render.generated.yml") } renderReq := scriptorium.RenderArtifactRequest{ Binary: env.Config.Pipeline.Scriptorium.Binary, ConfigPath: env.Config.Pipeline.Scriptorium.ConfigPath, PromptID: artifactCfg.PromptID, ProfileID: artifactCfg.ProfileID, InputPaths: inputPaths, Vars: vars, OutputPath: renderOutputPath, StdoutLogPath: renderStdoutPath, StderrLogPath: renderStderrPath, GeneratedConfigPath: renderGeneratedConfigPath, Timeout: timeout, } renderRes, renderErr := env.Scriptorium.RenderArtifact(ctx, renderReq) if renderErr != nil { return nil, fmt.Errorf("analyze: scriptorium render failed: %w", renderErr) } if renderRes.ValidationFailed { return nil, fmt.Errorf("analyze: scriptorium render returned validation_failed=true") } finalRenderOutputPath := coalesceString(renderRes.OutputPath, renderReq.OutputPath) if err := requireNonEmptyFile(finalRenderOutputPath, artifactName+" render output"); err != nil { return nil, fmt.Errorf("analyze: %w", err) } if err := validateJSONFile(finalRenderOutputPath); err != nil { return nil, fmt.Errorf("analyze: render diagnostics %q invalid json: %w", finalRenderOutputPath, err) } logPaths = append(logPaths, renderStdoutPath, renderStderrPath) generatedConfigs = append(generatedConfigs, renderGeneratedConfigPath) meta["render_output_path"] = finalRenderOutputPath meta["render_stdout_log_path"] = renderStdoutPath meta["render_stderr_log_path"] = renderStderrPath meta["render_generated_config_path"] = renderGeneratedConfigPath meta["render_adapter_exit_code"] = renderRes.ExitCode meta["render_adapter_duration_ms"] = renderRes.Duration.Milliseconds() meta["render_adapter_command_mode"] = renderRes.CommandMode meta["render_adapter_prompt_id"] = renderRes.PromptID meta["render_adapter_profile_id"] = renderRes.ProfileID meta["render_adapter_output_path"] = renderRes.OutputPath meta["render_adapter_generated_config"] = renderRes.GeneratedConfigPath meta["render_adapter_stdout_log_path"] = renderRes.StdoutLogPath meta["render_adapter_stderr_log_path"] = renderRes.StderrLogPath if renderRes.Metadata != nil { meta["render_adapter_metadata"] = renderRes.Metadata } } req := scriptorium.RunArtifactRequest{ Binary: env.Config.Pipeline.Scriptorium.Binary, ConfigPath: env.Config.Pipeline.Scriptorium.ConfigPath, PromptID: artifactCfg.PromptID, ProfileID: artifactCfg.ProfileID, InputPaths: inputPaths, Vars: vars, OutputPath: outputPath, StdoutLogPath: stdoutLogPath, StderrLogPath: stderrLogPath, GeneratedConfigPath: generatedConfigPath, Timeout: timeout, } res, runErr := env.Scriptorium.RunArtifact(ctx, req) if runErr != nil { if res.ValidationFailed { return nil, fmt.Errorf( "analyze: scriptorium validation failed (prompt_id=%q, output_path=%q, exit_code=%d, stdout_log=%q, stderr_log=%q): %w", req.PromptID, coalesceString(res.OutputPath, req.OutputPath), res.ExitCode, coalesceString(res.StdoutLogPath, req.StdoutLogPath), coalesceString(res.StderrLogPath, req.StderrLogPath), runErr, ) } return nil, fmt.Errorf("analyze: scriptorium run failed: %w", runErr) } if res.ValidationFailed { return nil, fmt.Errorf("analyze: scriptorium run returned validation_failed=true") } finalOutputPath := coalesceString(res.OutputPath, req.OutputPath) if err := requireNonEmptyFile(finalOutputPath, artifactName+" output"); err != nil { return nil, fmt.Errorf("analyze: %w", err) } promotedArtifact, err := promoteRunLocalOutput(env.ArtifactStore, finalOutputPath, canonicalOutputPath, artifacts.Ref{ Kind: artifactName, Category: "artifacts", SessionID: sessionID, }) if err != nil { return nil, fmt.Errorf("analyze: promote artifact output: %w", err) } logPaths = append(logPaths, stdoutLogPath, stderrLogPath) generatedConfigs = append(generatedConfigs, generatedConfigPath) meta["run_output_path"] = finalOutputPath meta["output_path"] = canonicalOutputPath meta["generated_config_path"] = generatedConfigPath meta["stdout_log_path"] = stdoutLogPath meta["stderr_log_path"] = stderrLogPath meta["adapter_exit_code"] = res.ExitCode meta["adapter_duration_ms"] = res.Duration.Milliseconds() meta["adapter_command_mode"] = res.CommandMode meta["adapter_prompt_id"] = res.PromptID meta["adapter_profile_id"] = res.ProfileID meta["adapter_validation_failed"] = res.ValidationFailed meta["adapter_output_path"] = res.OutputPath meta["adapter_generated_config"] = res.GeneratedConfigPath meta["adapter_stdout_log_path"] = res.StdoutLogPath meta["adapter_stderr_log_path"] = res.StderrLogPath if res.Metadata != nil { meta["adapter_metadata"] = res.Metadata } return &StageResult{ Outputs: []artifacts.Ref{promotedArtifact}, Logs: logPaths, GeneratedConfigs: generatedConfigs, Metadata: meta, }, nil } func selectAnalyzeArtifact(cfg *config.ScriptoriumConfig) (string, config.ScriptoriumArtifactConfig, string, error) { if cfg == nil { return "", config.ScriptoriumArtifactConfig{}, "pipeline.scriptorium is not configured", nil } enabled := []string{} for name, artifact := range cfg.Artifacts { if artifact.Enabled { enabled = append(enabled, name) } } sort.Strings(enabled) if len(enabled) == 0 { return "", config.ScriptoriumArtifactConfig{}, "no enabled scriptorium artifacts configured", nil } sessionRecapCfg, ok := cfg.Artifacts["session_recap"] if !ok || !sessionRecapCfg.Enabled { return "", config.ScriptoriumArtifactConfig{}, "", fmt.Errorf("only artifacts.session_recap is supported in this analyze implementation; enabled=%s", strings.Join(enabled, ",")) } return "session_recap", sessionRecapCfg, "", nil } func discoverProcessedTranscript(m *manifest.Manifest, paths artifacts.SessionPaths) (string, string, error) { candidates := []string{} if m != nil && m.Stages != nil { if sr := m.Stages["polish"]; sr != nil { for _, out := range sr.Outputs { if out.Kind != "transcript_processed" { continue } p := strings.TrimSpace(out.LocalPath) if p == "" { continue } resolved := artifacts.ResolveSessionLocalPathForRead(paths, p) candidates = append(candidates, filepath.Clean(resolved)) } } } deduped := dedupeAndSortPaths(candidates) for _, p := range deduped { if info, err := os.Stat(p); err == nil && !info.IsDir() { return p, "manifest.polish.outputs", nil } } fallback := filepath.Join(paths.TranscriptsDir, "processed.json") if info, err := os.Stat(fallback); err == nil && !info.IsDir() { return filepath.Clean(fallback), "fallback.transcripts_dir", nil } if len(deduped) > 0 { return deduped[0], "manifest.polish.outputs", nil } return "", "", nil } type analyzeTranscriptInputs struct { ProcessedPath string ProcessedSource string NormalizedPath string NormalizedSource string TrimmedPath string TrimmedSource string } func discoverAnalyzeTranscriptRefs(m *manifest.Manifest, paths artifacts.SessionPaths) analyzeTranscriptInputs { processedPath, processedSource := discoverAnalyzeArtifactRef(m, paths, artifacts.ArtifactTranscriptPolished) normalizedPath, normalizedSource := discoverAnalyzeArtifactRef(m, paths, artifacts.ArtifactTranscriptFull) trimmedPath, trimmedSource := discoverAnalyzeArtifactRef(m, paths, artifacts.ArtifactTranscriptTrimmed) return analyzeTranscriptInputs{ ProcessedPath: processedPath, ProcessedSource: processedSource, NormalizedPath: normalizedPath, NormalizedSource: normalizedSource, TrimmedPath: trimmedPath, TrimmedSource: trimmedSource, } } func discoverAnalyzeArtifactRef(m *manifest.Manifest, paths artifacts.SessionPaths, source string) (string, string) { resolved, err := artifacts.ResolveSessionArtifact(paths, m, source) if err != nil { return "", "" } 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, error) { switch strings.TrimSpace(inputCfg.Source) { case "previous_session_artifact": if strings.TrimSpace(inputCfg.Path) == "" { return "", false, nil } resolved := resolveInputPathForRead(paths, sessionDir, inputCfg.Path) if err := requireFile(resolved, "scriptorium input "+inputName); err != nil { return "", false, nil } return resolved, true, nil default: resolved, err := artifacts.ResolveSessionArtifactWithCatalog(paths, m, inputCfg.Source, runtimeCatalog) if err == nil { return resolved.Path, true, nil } if errors.Is(err, artifacts.ErrSessionArtifactNotFound) { if artifacts.IsConfiguredArtifactSource(inputCfg.Source) { if inputCfg.Required { return "", false, fmt.Errorf("configured artifact source %q is unavailable", inputCfg.Source) } return "", false, nil } normalized, normalizeErr := artifacts.NormalizeSessionArtifactSource(inputCfg.Source) if normalizeErr != nil { return "", false, normalizeErr } switch normalized { case artifacts.ArtifactTranscriptPolished: return "", false, nil case artifacts.ArtifactTranscriptFull: return "", false, fmt.Errorf("normalized transcript input is unavailable; run normalize stage first") case artifacts.ArtifactTranscriptTrimmed: return "", false, fmt.Errorf("trimmed transcript input is unavailable; run trim stage first") default: return "", false, nil } } return "", false, err } } func buildAnalyzeRuntimeArtifactCatalog( paths artifacts.SessionPaths, scriptoriumCfg *config.ScriptoriumConfig, selectedArtifacts []string, ) (*artifacts.ArtifactCatalog, error) { catalog := artifacts.NewArtifactCatalog() if err := catalog.RegisterBuiltIns(); err != nil { return nil, err } if scriptoriumCfg == nil { return catalog, nil } configured := map[string]artifacts.ConfiguredArtifactDefinition{} for key, artifactCfg := range scriptoriumCfg.Artifacts { configured[key] = artifacts.ConfiguredArtifactDefinition{ Enabled: artifactCfg.Enabled, OutputPath: artifactCfg.OutputPath, } } if err := catalog.RegisterConfiguredArtifacts(configured, selectedArtifacts); err != nil { return nil, err } for _, entry := range catalog.ListConfigured() { if entry.Executable { continue } if strings.TrimSpace(entry.CanonicalRelPath) == "" { continue } resolvedPath, err := resolveScriptoriumOutputPath(paths, entry.CanonicalRelPath) if err != nil { continue } if err := requireNonEmptyFile(resolvedPath, "configured artifact "+entry.SourceID); err != nil { continue } if err := catalog.MarkAvailableFromDisk(entry.SourceID, resolvedPath); err != nil { return nil, err } } 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 == "" { return "", fmt.Errorf("scriptorium artifact output path is required") } if filepath.IsAbs(outputPath) { return filepath.Clean(outputPath), nil } rel := filepath.Clean(outputPath) if rel == "." || rel == "" { return "", fmt.Errorf("relative output path is required") } if rel == ".." || strings.HasPrefix(rel, ".."+string(filepath.Separator)) { return "", fmt.Errorf("relative output path escapes session root: %q", outputPath) } return filepath.Join(paths.Root, rel), nil } func resolveScriptoriumTimeout(topLevel, artifact string) (time.Duration, error) { raw := strings.TrimSpace(artifact) if raw == "" { raw = strings.TrimSpace(topLevel) } if raw == "" { raw = "10m" } d, err := time.ParseDuration(raw) if err != nil { return 0, fmt.Errorf("parse duration %q: %w", raw, err) } if d <= 0 { return 0, fmt.Errorf("duration must be > 0") } return d, nil } func buildScriptoriumVars(varsCfg map[string]any, session *config.SessionConfig) (map[string]string, error) { if len(varsCfg) == 0 { return nil, nil } vars := map[string]string{} keys := make([]string, 0, len(varsCfg)) for key := range varsCfg { keys = append(keys, key) } sort.Strings(keys) for _, key := range keys { value := varsCfg[key] switch typed := value.(type) { case bool: if !typed { continue } derived, ok, err := deriveSessionVarValue(key, session) if err != nil { return nil, err } if ok { vars[key] = derived } case string: vars[key] = typed default: return nil, fmt.Errorf("var %q has unsupported type %T", key, value) } } if len(vars) == 0 { return nil, nil } return vars, nil } func deriveSessionVarValue(name string, session *config.SessionConfig) (string, bool, error) { switch name { case "session_id": if session == nil || strings.TrimSpace(session.SessionID) == "" { return "", false, nil } return strings.TrimSpace(session.SessionID), true, nil case "session_date": if session == nil || strings.TrimSpace(session.Date) == "" { return "", false, nil } return strings.TrimSpace(session.Date), true, nil case "campaign_name": if session == nil || strings.TrimSpace(session.Campaign) == "" { return "", false, nil } return strings.TrimSpace(session.Campaign), true, nil case "previous_session_id": return "", false, nil default: return "", false, fmt.Errorf("unsupported boolean var %q", name) } } func requireNonEmptyFile(path string, label string) error { if err := requireFile(path, label); err != nil { return err } info, err := os.Stat(path) if err != nil { return fmt.Errorf("%s %q stat failed: %w", label, path, err) } if info.Size() <= 0 { return fmt.Errorf("%s %q is empty", label, path) } return nil } func sortedScriptoriumInputNames(inputs map[string]config.ScriptoriumInputConfig) []string { if len(inputs) == 0 { return nil } names := make([]string, 0, len(inputs)) for name := range inputs { names = append(names, name) } sort.Strings(names) return names } func sortedMapKeys(values map[string]string) []string { if len(values) == 0 { return nil } keys := make([]string, 0, len(values)) for key := range values { keys = append(keys, key) } sort.Strings(keys) return keys } func coalesceString(primary, fallback string) string { if strings.TrimSpace(primary) != "" { return primary } return fallback } func resolveRenderDebugEnabled(global bool, perArtifact *bool) bool { if perArtifact == nil { return global } return *perArtifact } func validateJSONFile(path string) error { data, err := os.ReadFile(path) if err != nil { return fmt.Errorf("read file: %w", err) } var payload any if err := json.Unmarshal(data, &payload); err != nil { return fmt.Errorf("decode json: %w", err) } return nil }