Implement render stage runtime and integrate it into pipeline execution
This commit is contained in:
288
internal/stage/render.go
Normal file
288
internal/stage/render.go
Normal file
@@ -0,0 +1,288 @@
|
||||
package stage
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
|
||||
"gitea.maximumdirect.net/eric/narratio/internal/adapters/seriatim"
|
||||
"gitea.maximumdirect.net/eric/narratio/internal/artifacts"
|
||||
"gitea.maximumdirect.net/eric/narratio/internal/config"
|
||||
"gitea.maximumdirect.net/eric/narratio/internal/manifest"
|
||||
)
|
||||
|
||||
type renderStage struct{}
|
||||
|
||||
func (renderStage) Name() string { return "render" }
|
||||
|
||||
func (renderStage) Declares() IODecl {
|
||||
return IODecl{
|
||||
Inputs: []artifacts.Ref{
|
||||
{Kind: "transcript_final", Category: "transcripts", RelativePath: "transcripts/final.json"},
|
||||
{Kind: "transcript_final_trimmed", Category: "transcripts", RelativePath: "transcripts/final.trimmed.json"},
|
||||
},
|
||||
Outputs: []artifacts.Ref{
|
||||
{Kind: artifacts.TranscriptOutputKindFinalMarkdown, Category: "transcripts", RelativePath: artifacts.TranscriptPathFinalMarkdown},
|
||||
{Kind: artifacts.TranscriptOutputKindFinalTrimmedMarkdown, Category: "transcripts", RelativePath: artifacts.TranscriptPathFinalTrimmedMarkdown},
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
func (renderStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*StageResult, error) {
|
||||
if env == nil || env.Config == nil {
|
||||
return nil, fmt.Errorf("render: stage environment config is required")
|
||||
}
|
||||
if env.ArtifactStore == nil {
|
||||
return nil, fmt.Errorf("render: artifact store is required")
|
||||
}
|
||||
if env.Config.Pipeline == nil || env.Config.Session == nil {
|
||||
return nil, fmt.Errorf("render: resolved config must include pipeline and session")
|
||||
}
|
||||
if env.Seriatim == nil {
|
||||
return nil, fmt.Errorf("render: seriatim 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("render: session id is required")
|
||||
}
|
||||
|
||||
paths := sessionPathsForEnv(env, sessionID)
|
||||
runLayout, err := resolveRunStageLayout(env, m, paths, sessionID, "render")
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("render: resolve run-stage layout: %w", err)
|
||||
}
|
||||
|
||||
renderCfg := renderConfigOrDefault(env.Config.Pipeline.Render)
|
||||
enabled := renderCfg.Enabled == nil || *renderCfg.Enabled
|
||||
format := strings.TrimSpace(renderCfg.Format)
|
||||
if format == "" {
|
||||
format = config.DefaultRenderFormat
|
||||
}
|
||||
title := resolveRenderTitle(renderCfg, env.Config.Session)
|
||||
includeTimestamps := renderCfg.IncludeTimestamps == nil || *renderCfg.IncludeTimestamps
|
||||
includeSegmentIDs := renderCfg.IncludeSegmentIDs
|
||||
includeMetadata := renderCfg.IncludeMetadata
|
||||
|
||||
meta := map[string]any{
|
||||
"stage": "render",
|
||||
"render_enabled": enabled,
|
||||
"format": format,
|
||||
"title": title,
|
||||
"include_timestamps": includeTimestamps,
|
||||
"include_segment_ids": includeSegmentIDs,
|
||||
"include_metadata": includeMetadata,
|
||||
"binary": env.Config.Pipeline.Seriatim.Binary,
|
||||
"timeout": env.Config.Pipeline.Seriatim.Timeout,
|
||||
}
|
||||
if !enabled {
|
||||
meta["skipped"] = true
|
||||
meta["reason"] = "pipeline.render.enabled is false"
|
||||
return &StageResult{Metadata: meta}, nil
|
||||
}
|
||||
|
||||
finalInput, err := artifacts.ResolveSessionArtifact(paths, m, artifacts.ArtifactTranscriptFinal)
|
||||
if err != nil {
|
||||
return nil, wrapRenderInputResolveError(err, sessionID, artifacts.ArtifactTranscriptFinal, "normalize")
|
||||
}
|
||||
finalTrimmedInput, err := artifacts.ResolveSessionArtifact(paths, m, artifacts.ArtifactTranscriptFinalTrimmed)
|
||||
if err != nil {
|
||||
return nil, wrapRenderInputResolveError(err, sessionID, artifacts.ArtifactTranscriptFinalTrimmed, "trim")
|
||||
}
|
||||
|
||||
canonicalFinalMarkdownPath, err := resolveScriptoriumOutputPath(paths, artifacts.TranscriptPathFinalMarkdown)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("render: resolve canonical final markdown output path: %w", err)
|
||||
}
|
||||
canonicalFinalTrimmedMarkdownPath, err := resolveScriptoriumOutputPath(paths, artifacts.TranscriptPathFinalTrimmedMarkdown)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("render: resolve canonical final trimmed markdown output path: %w", err)
|
||||
}
|
||||
|
||||
runFinalMarkdownPath, err := runLocalPathForCanonical(runLayout, paths, canonicalFinalMarkdownPath)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("render: resolve run-local final markdown output path: %w", err)
|
||||
}
|
||||
runFinalTrimmedMarkdownPath, err := runLocalPathForCanonical(runLayout, paths, canonicalFinalTrimmedMarkdownPath)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("render: resolve run-local final trimmed markdown output path: %w", err)
|
||||
}
|
||||
|
||||
finalStdoutLogPath := filepath.Join(paths.LogsDir, "seriatim.render.final.stdout.log")
|
||||
finalStderrLogPath := filepath.Join(paths.LogsDir, "seriatim.render.final.stderr.log")
|
||||
finalGeneratedConfigPath := filepath.Join(paths.ConfigDir, "seriatim.render.final.generated.yml")
|
||||
finalTrimmedStdoutLogPath := filepath.Join(paths.LogsDir, "seriatim.render.final_trimmed.stdout.log")
|
||||
finalTrimmedStderrLogPath := filepath.Join(paths.LogsDir, "seriatim.render.final_trimmed.stderr.log")
|
||||
finalTrimmedGeneratedConfigPath := filepath.Join(paths.ConfigDir, "seriatim.render.final_trimmed.generated.yml")
|
||||
if runLayout.Enabled {
|
||||
finalStdoutLogPath = filepath.Join(runLayout.LogsDir, "seriatim.render.final.stdout.log")
|
||||
finalStderrLogPath = filepath.Join(runLayout.LogsDir, "seriatim.render.final.stderr.log")
|
||||
finalGeneratedConfigPath = filepath.Join(runLayout.ConfigDir, "seriatim.render.final.generated.yml")
|
||||
finalTrimmedStdoutLogPath = filepath.Join(runLayout.LogsDir, "seriatim.render.final_trimmed.stdout.log")
|
||||
finalTrimmedStderrLogPath = filepath.Join(runLayout.LogsDir, "seriatim.render.final_trimmed.stderr.log")
|
||||
finalTrimmedGeneratedConfigPath = filepath.Join(runLayout.ConfigDir, "seriatim.render.final_trimmed.generated.yml")
|
||||
}
|
||||
|
||||
timeout, err := resolveTrimSeriatimTimeout(env.Config.Pipeline.Seriatim.Timeout)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("render: resolve seriatim timeout: %w", err)
|
||||
}
|
||||
|
||||
finalReq := seriatim.RenderRequest{
|
||||
Binary: env.Config.Pipeline.Seriatim.Binary,
|
||||
InputTranscriptPath: finalInput.Path,
|
||||
OutputRenderedPath: runFinalMarkdownPath,
|
||||
Format: format,
|
||||
Title: title,
|
||||
IncludeTimestamps: includeTimestamps,
|
||||
IncludeSegmentIDs: includeSegmentIDs,
|
||||
IncludeMetadata: includeMetadata,
|
||||
StdoutLogPath: finalStdoutLogPath,
|
||||
StderrLogPath: finalStderrLogPath,
|
||||
GeneratedConfigPath: finalGeneratedConfigPath,
|
||||
Timeout: timeout,
|
||||
}
|
||||
finalRes, err := env.Seriatim.Render(ctx, finalReq)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("render: seriatim render failed for %q: %w", artifacts.ArtifactTranscriptFinal, err)
|
||||
}
|
||||
finalRenderedPath := coalesceString(finalRes.OutputRenderedPath, finalReq.OutputRenderedPath)
|
||||
if err := requireNonEmptyFile(finalRenderedPath, "final transcript markdown output"); err != nil {
|
||||
return nil, fmt.Errorf("render: %w", err)
|
||||
}
|
||||
|
||||
finalTrimmedReq := seriatim.RenderRequest{
|
||||
Binary: env.Config.Pipeline.Seriatim.Binary,
|
||||
InputTranscriptPath: finalTrimmedInput.Path,
|
||||
OutputRenderedPath: runFinalTrimmedMarkdownPath,
|
||||
Format: format,
|
||||
Title: title,
|
||||
IncludeTimestamps: includeTimestamps,
|
||||
IncludeSegmentIDs: includeSegmentIDs,
|
||||
IncludeMetadata: includeMetadata,
|
||||
StdoutLogPath: finalTrimmedStdoutLogPath,
|
||||
StderrLogPath: finalTrimmedStderrLogPath,
|
||||
GeneratedConfigPath: finalTrimmedGeneratedConfigPath,
|
||||
Timeout: timeout,
|
||||
}
|
||||
finalTrimmedRes, err := env.Seriatim.Render(ctx, finalTrimmedReq)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("render: seriatim render failed for %q: %w", artifacts.ArtifactTranscriptFinalTrimmed, err)
|
||||
}
|
||||
finalTrimmedRenderedPath := coalesceString(finalTrimmedRes.OutputRenderedPath, finalTrimmedReq.OutputRenderedPath)
|
||||
if err := requireNonEmptyFile(finalTrimmedRenderedPath, "final trimmed transcript markdown output"); err != nil {
|
||||
return nil, fmt.Errorf("render: %w", err)
|
||||
}
|
||||
|
||||
materializedFinalMarkdown, err := materializeRunLocalOutput(env.ArtifactStore, finalRenderedPath, canonicalFinalMarkdownPath, artifacts.Ref{
|
||||
Kind: artifacts.TranscriptOutputKindFinalMarkdown,
|
||||
Category: "transcripts",
|
||||
SessionID: sessionID,
|
||||
})
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("render: materialize canonical final markdown output: %w", err)
|
||||
}
|
||||
materializedFinalTrimmedMarkdown, err := materializeRunLocalOutput(env.ArtifactStore, finalTrimmedRenderedPath, canonicalFinalTrimmedMarkdownPath, artifacts.Ref{
|
||||
Kind: artifacts.TranscriptOutputKindFinalTrimmedMarkdown,
|
||||
Category: "transcripts",
|
||||
SessionID: sessionID,
|
||||
})
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("render: materialize canonical final trimmed markdown output: %w", err)
|
||||
}
|
||||
|
||||
meta["final_input_path"] = finalInput.Path
|
||||
meta["final_input_provenance"] = finalInput.Provenance
|
||||
meta["final_input_run_id"] = finalInput.ProducerRunID
|
||||
meta["final_trimmed_input_path"] = finalTrimmedInput.Path
|
||||
meta["final_trimmed_input_provenance"] = finalTrimmedInput.Provenance
|
||||
meta["final_trimmed_input_run_id"] = finalTrimmedInput.ProducerRunID
|
||||
meta["run_final_markdown_path"] = finalRenderedPath
|
||||
meta["final_markdown_path"] = canonicalFinalMarkdownPath
|
||||
meta["run_final_trimmed_markdown_path"] = finalTrimmedRenderedPath
|
||||
meta["final_trimmed_markdown_path"] = canonicalFinalTrimmedMarkdownPath
|
||||
populateRenderAdapterMetadata(meta, "final_adapter_", finalRes)
|
||||
populateRenderAdapterMetadata(meta, "final_trimmed_adapter_", finalTrimmedRes)
|
||||
|
||||
return &StageResult{
|
||||
Outputs: []artifacts.Ref{
|
||||
materializedFinalMarkdown,
|
||||
materializedFinalTrimmedMarkdown,
|
||||
},
|
||||
Logs: dedupeAndSortPaths([]string{
|
||||
finalStdoutLogPath,
|
||||
finalStderrLogPath,
|
||||
finalTrimmedStdoutLogPath,
|
||||
finalTrimmedStderrLogPath,
|
||||
}),
|
||||
GeneratedConfigs: dedupeAndSortPaths([]string{
|
||||
finalGeneratedConfigPath,
|
||||
finalTrimmedGeneratedConfigPath,
|
||||
}),
|
||||
Metadata: meta,
|
||||
}, nil
|
||||
}
|
||||
|
||||
func renderConfigOrDefault(cfg *config.RenderConfig) *config.RenderConfig {
|
||||
if cfg != nil {
|
||||
return cfg
|
||||
}
|
||||
enabled := true
|
||||
includeTimestamps := true
|
||||
return &config.RenderConfig{
|
||||
Enabled: &enabled,
|
||||
Format: config.DefaultRenderFormat,
|
||||
IncludeTimestamps: &includeTimestamps,
|
||||
IncludeSegmentIDs: config.DefaultRenderSegmentIDs,
|
||||
IncludeMetadata: config.DefaultRenderMetadata,
|
||||
}
|
||||
}
|
||||
|
||||
func resolveRenderTitle(renderCfg *config.RenderConfig, sessionCfg *config.SessionConfig) string {
|
||||
if renderCfg != nil && strings.TrimSpace(renderCfg.Title) != "" {
|
||||
return strings.TrimSpace(renderCfg.Title)
|
||||
}
|
||||
if sessionCfg != nil && strings.TrimSpace(sessionCfg.Title) != "" {
|
||||
return strings.TrimSpace(sessionCfg.Title)
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
func wrapRenderInputResolveError(err error, sessionID, sourceID, guidanceStage string) error {
|
||||
var notFound *artifacts.SessionArtifactNotFoundError
|
||||
if errors.As(err, ¬Found) {
|
||||
return fmt.Errorf(
|
||||
"render: required input %q is unavailable; run narratio run-stage --force %s %s",
|
||||
sourceID,
|
||||
guidanceStage,
|
||||
sessionID,
|
||||
)
|
||||
}
|
||||
return fmt.Errorf("render: resolve %q input: %w", sourceID, err)
|
||||
}
|
||||
|
||||
func populateRenderAdapterMetadata(meta map[string]any, prefix string, result seriatim.RenderResult) {
|
||||
if meta == nil {
|
||||
return
|
||||
}
|
||||
meta[prefix+"duration_ms"] = result.Duration.Milliseconds()
|
||||
meta[prefix+"exit_code"] = result.ExitCode
|
||||
meta[prefix+"invoked_binary"] = result.InvokedBinary
|
||||
meta[prefix+"format"] = result.Format
|
||||
meta[prefix+"title"] = result.Title
|
||||
meta[prefix+"output_path"] = result.OutputRenderedPath
|
||||
meta[prefix+"generated_config"] = result.GeneratedConfigPath
|
||||
meta[prefix+"stdout_log_path"] = result.StdoutLogPath
|
||||
meta[prefix+"stderr_log_path"] = result.StderrLogPath
|
||||
if result.Metadata != nil {
|
||||
meta[prefix+"metadata"] = result.Metadata
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user