463 lines
14 KiB
Go
463 lines
14 KiB
Go
package stage
|
|
|
|
import (
|
|
"context"
|
|
"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: "artifact", Category: "artifacts", RelativePath: "artifacts/session_recap.md"},
|
|
},
|
|
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 := env.ArtifactStore.SessionPaths(sessionID)
|
|
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
|
|
}
|
|
|
|
processedTranscriptPath, processedSource, err := discoverProcessedTranscript(m, paths)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("analyze: resolve processed transcript: %w", err)
|
|
}
|
|
if processedTranscriptPath == "" {
|
|
return nil, fmt.Errorf("analyze: processed transcript input is required")
|
|
}
|
|
if err := validateProcessedTranscriptOutput(processedTranscriptPath); err != nil {
|
|
return nil, fmt.Errorf("analyze: processed transcript %q invalid: %w", processedTranscriptPath, err)
|
|
}
|
|
|
|
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, processedTranscriptPath, paths, sessionDir)
|
|
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)
|
|
}
|
|
|
|
outputPath, err := resolveScriptoriumOutputPath(paths, artifactCfg.OutputPath)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("analyze: resolve 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")
|
|
|
|
timeout, err := resolveScriptoriumTimeout(env.Config.Pipeline.Scriptorium.Timeout, artifactCfg.Timeout)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("analyze: resolve timeout: %w", err)
|
|
}
|
|
|
|
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, "session recap output"); err != nil {
|
|
return nil, fmt.Errorf("analyze: %w", err)
|
|
}
|
|
|
|
artifactRef := artifacts.Ref{
|
|
Kind: artifactName,
|
|
Category: "artifacts",
|
|
SessionID: sessionID,
|
|
AbsolutePath: finalOutputPath,
|
|
}
|
|
|
|
meta := map[string]any{
|
|
"stage": "analyze",
|
|
"artifact_name": artifactName,
|
|
"prompt_id": artifactCfg.PromptID,
|
|
"profile_id": artifactCfg.ProfileID,
|
|
"output_path": finalOutputPath,
|
|
"generated_config_path": generatedConfigPath,
|
|
"stdout_log_path": stdoutLogPath,
|
|
"stderr_log_path": stderrLogPath,
|
|
"input_paths": inputPaths,
|
|
"input_names": sortedMapKeys(inputPaths),
|
|
"omitted_optional_inputs": omittedOptionalInputs,
|
|
"vars": vars,
|
|
"timeout": timeout.String(),
|
|
"binary": env.Config.Pipeline.Scriptorium.Binary,
|
|
"config_path": env.Config.Pipeline.Scriptorium.ConfigPath,
|
|
"processed_transcript_path": processedTranscriptPath,
|
|
"processed_transcript_source": processedSource,
|
|
"adapter_exit_code": res.ExitCode,
|
|
"adapter_duration_ms": res.Duration.Milliseconds(),
|
|
"adapter_command_mode": res.CommandMode,
|
|
"adapter_prompt_id": res.PromptID,
|
|
"adapter_profile_id": res.ProfileID,
|
|
"adapter_validation_failed": res.ValidationFailed,
|
|
"adapter_output_path": res.OutputPath,
|
|
"adapter_generated_config": res.GeneratedConfigPath,
|
|
"adapter_stdout_log_path": res.StdoutLogPath,
|
|
"adapter_stderr_log_path": res.StderrLogPath,
|
|
}
|
|
if res.Metadata != nil {
|
|
meta["adapter_metadata"] = res.Metadata
|
|
}
|
|
|
|
return &StageResult{
|
|
Outputs: []artifacts.Ref{artifactRef},
|
|
Logs: []string{stdoutLogPath, stderrLogPath},
|
|
GeneratedConfigs: []string{generatedConfigPath},
|
|
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
|
|
}
|
|
|
|
func resolveScriptoriumInput(
|
|
inputName string,
|
|
inputCfg config.ScriptoriumInputConfig,
|
|
processedTranscriptPath string,
|
|
paths artifacts.SessionPaths,
|
|
sessionDir string,
|
|
) (string, bool, error) {
|
|
switch strings.TrimSpace(inputCfg.Source) {
|
|
case "processed_transcript":
|
|
return processedTranscriptPath, true, nil
|
|
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:
|
|
return "", false, fmt.Errorf("unsupported source %q", inputCfg.Source)
|
|
}
|
|
}
|
|
|
|
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 == "" {
|
|
outputPath = "artifacts/session_recap.md"
|
|
}
|
|
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
|
|
}
|