narratio/internal/stage/merge.go

package stage

import (
	"context"
	"fmt"
	"os"
	"path/filepath"
	"sort"
	"strings"

	"gitea.maximumdirect.net/eric/narratio/internal/adapters/seriatim"
	"gitea.maximumdirect.net/eric/narratio/internal/artifacts"
	"gitea.maximumdirect.net/eric/narratio/internal/manifest"
)

type mergeStage struct{}

func (mergeStage) Name() string { return "merge" }

func (mergeStage) Declares() IODecl {
	return IODecl{
		Inputs: []artifacts.Ref{
			{Kind: "transcript_raw", Category: "transcripts", RelativePath: "transcripts/raw/*.json"},
			{Kind: "speakers", Category: "inputs", RelativePath: "inputs/speakers.yml"},
			{Kind: "autocorrect", Category: "inputs", RelativePath: "inputs/autocorrect.yml"},
		},
		Outputs: []artifacts.Ref{
			{Kind: "transcript_merged", Category: "transcripts", RelativePath: "transcripts/merged.json"},
			{Kind: "seriatim_report", Category: "artifacts", RelativePath: "artifacts/seriatim.report.json"},
		},
	}
}

func (mergeStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*StageResult, error) {
	if env == nil || env.Config == nil {
		return nil, fmt.Errorf("merge: stage environment config is required")
	}
	if env.ArtifactStore == nil {
		return nil, fmt.Errorf("merge: artifact store is required")
	}
	if env.Config.Pipeline == nil || env.Config.Session == nil {
		return nil, fmt.Errorf("merge: resolved config must include pipeline and session")
	}
	if env.Seriatim == nil {
		return nil, fmt.Errorf("merge: 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("merge: session id is required")
	}

	paths := env.ArtifactStore.SessionPaths(sessionID)

	inputs, err := discoverRawTranscripts(m, paths)
	if err != nil {
		return nil, fmt.Errorf("merge: resolve raw transcripts: %w", err)
	}
	if len(inputs) == 0 {
		return nil, fmt.Errorf("merge: no raw transcript inputs found")
	}
	for _, in := range inputs {
		if err := validateTranscriptJSONFile(in); err != nil {
			return nil, fmt.Errorf("merge: input transcript %q invalid: %w", in, err)
		}
	}

	speakersPath := filepath.Join(paths.InputsDir, "speakers.yml")
	if err := requireFile(speakersPath, "speakers.yml"); err != nil {
		return nil, fmt.Errorf("merge: %w", err)
	}
	autocorrectPath := filepath.Join(paths.InputsDir, "autocorrect.yml")
	if err := requireFile(autocorrectPath, "autocorrect.yml"); err != nil {
		return nil, fmt.Errorf("merge: %w", err)
	}

	mergedPath := filepath.Join(paths.TranscriptsDir, "merged.json")
	reportPath := filepath.Join(paths.ArtifactsDir, "seriatim.report.json")
	stdoutPath := filepath.Join(paths.LogsDir, "seriatim.stdout.log")
	stderrPath := filepath.Join(paths.LogsDir, "seriatim.stderr.log")
	genCfgPath := filepath.Join(paths.ConfigDir, "seriatim.generated.yml")

	reportEnabled := env.Config.Pipeline.Seriatim.Report != nil && *env.Config.Pipeline.Seriatim.Report
	req := seriatim.MergeRequest{
		GeneratedConfigPath:        genCfgPath,
		InputTranscriptPaths:       inputs,
		OutputMergedTranscriptPath: mergedPath,
		ReportPath:                 "",
		SpeakersPath:               speakersPath,
		AutocorrectPath:            autocorrectPath,
		StdoutLogPath:              stdoutPath,
		StderrLogPath:              stderrPath,
	}
	if reportEnabled {
		req.ReportPath = reportPath
	}

	res, err := env.Seriatim.Run(ctx, req)
	if err != nil {
		return nil, fmt.Errorf("merge: seriatim merge failed: %w", err)
	}

	finalMergedPath := mergedPath
	if strings.TrimSpace(res.MergedTranscriptPath) != "" {
		finalMergedPath = res.MergedTranscriptPath
	}
	if err := validateTranscriptJSONFile(finalMergedPath); err != nil {
		return nil, fmt.Errorf("merge: merged transcript %q invalid: %w", finalMergedPath, err)
	}

	finalReportPath := req.ReportPath
	if strings.TrimSpace(res.ReportPath) != "" {
		finalReportPath = res.ReportPath
	}
	if reportEnabled {
		if err := validateTranscriptJSONFile(finalReportPath); err != nil {
			return nil, fmt.Errorf("merge: report %q invalid: %w", finalReportPath, err)
		}
	}

	outputs := []artifacts.Ref{{
		Kind:         "transcript_merged",
		Category:     "transcripts",
		SessionID:    sessionID,
		AbsolutePath: finalMergedPath,
	}}
	if reportEnabled {
		outputs = append(outputs, artifacts.Ref{
			Kind:         "seriatim_report",
			Category:     "artifacts",
			SessionID:    sessionID,
			AbsolutePath: finalReportPath,
		})
	}

	coalesceGap := any(nil)
	if env.Config.Pipeline.Seriatim.CoalesceGap != nil {
		coalesceGap = *env.Config.Pipeline.Seriatim.CoalesceGap
	}

	meta := map[string]any{
		"stage":                    "merge",
		"input_transcripts_count":  len(inputs),
		"input_transcript_paths":   inputs,
		"output_schema":            env.Config.Pipeline.Seriatim.OutputSchema,
		"coalesce_gap":             coalesceGap,
		"report_enabled":           reportEnabled,
		"output_path":              finalMergedPath,
		"report_path":              finalReportPath,
		"timeout":                  env.Config.Pipeline.Seriatim.Timeout,
		"binary":                   env.Config.Pipeline.Seriatim.Binary,
		"adapter_duration_ms":      res.Duration.Milliseconds(),
		"adapter_exit_code":        res.ExitCode,
		"adapter_invoked_binary":   res.InvokedBinary,
		"adapter_output_schema":    res.OutputSchema,
		"generated_config_path":    genCfgPath,
		"stdout_log_path":          stdoutPath,
		"stderr_log_path":          stderrPath,
		"adapter_report_path":      res.ReportPath,
		"adapter_merged_out_path":  res.MergedTranscriptPath,
		"adapter_generated_config": res.GeneratedConfigPath,
	}
	if res.Metadata != nil {
		meta["adapter_metadata"] = res.Metadata
	}

	return &StageResult{
		Outputs:          outputs,
		Logs:             []string{stdoutPath, stderrPath},
		GeneratedConfigs: []string{genCfgPath},
		Metadata:         meta,
	}, nil
}

func discoverRawTranscripts(m *manifest.Manifest, paths artifacts.SessionPaths) ([]string, error) {
	fromManifest := make([]string, 0)
	if m != nil && m.Stages != nil {
		if tr := m.Stages["transcribe"]; tr != nil {
			for _, out := range tr.Outputs {
				if out.Kind != "transcript_raw" {
					continue
				}
				p := strings.TrimSpace(out.LocalPath)
				if p == "" {
					continue
				}
				if !filepath.IsAbs(p) {
					p = filepath.Join(paths.Root, p)
				}
				fromManifest = append(fromManifest, filepath.Clean(p))
			}
		}
	}

	if len(fromManifest) > 0 {
		deduped := dedupeAndSortPaths(fromManifest)
		return deduped, nil
	}

	entries, err := os.ReadDir(paths.TranscriptsRawDir)
	if err != nil {
		if os.IsNotExist(err) {
			return nil, nil
		}
		return nil, fmt.Errorf("read raw transcript directory %q: %w", paths.TranscriptsRawDir, err)
	}

	out := make([]string, 0, len(entries))
	for _, entry := range entries {
		if entry.IsDir() {
			continue
		}
		p := filepath.Join(paths.TranscriptsRawDir, entry.Name())
		if !strings.EqualFold(filepath.Ext(p), ".json") {
			continue
		}
		out = append(out, p)
	}
	return dedupeAndSortPaths(out), nil
}

func dedupeAndSortPaths(paths []string) []string {
	if len(paths) == 0 {
		return nil
	}
	set := make(map[string]struct{}, len(paths))
	out := make([]string, 0, len(paths))
	for _, p := range paths {
		clean := filepath.Clean(strings.TrimSpace(p))
		if clean == "" {
			continue
		}
		if _, exists := set[clean]; exists {
			continue
		}
		set[clean] = struct{}{}
		out = append(out, clean)
	}
	sort.Strings(out)
	return out
}