narratio/internal/stage/polish.go

package stage

import (
	"context"
	"encoding/json"
	"fmt"
	"path/filepath"
	"strings"

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

type polishStage struct{}

func (polishStage) Name() string { return "polish" }

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

	paths := sessionPathsForEnv(env, sessionID)
	runLayout, err := resolveRunStageLayout(env, m, paths, sessionID, "polish")
	if err != nil {
		return nil, fmt.Errorf("polish: resolve run-stage layout: %w", err)
	}
	mergedPath, source, err := discoverMergedTranscript(m, paths)
	if err != nil {
		return nil, fmt.Errorf("polish: resolve merged transcript: %w", err)
	}
	if mergedPath == "" {
		return nil, fmt.Errorf("polish: merged transcript input is required")
	}
	if err := validateTranscriptJSONFile(mergedPath); err != nil {
		return nil, fmt.Errorf("polish: merged transcript %q invalid: %w", mergedPath, err)
	}

	glossaryPath := filepath.Join(paths.InputsDir, "glossary.yml")
	if err := requireFile(glossaryPath, "glossary.yml"); err != nil {
		return nil, fmt.Errorf("polish: %w", err)
	}

	canonicalProcessedPath := filepath.Join(paths.TranscriptsDir, "polished.json")
	processedPath, err := runLocalPathForCanonical(runLayout, paths, canonicalProcessedPath)
	if err != nil {
		return nil, fmt.Errorf("polish: resolve run-local processed transcript path: %w", err)
	}
	stdoutPath := filepath.Join(paths.LogsDir, "audita.stdout.log")
	stderrPath := filepath.Join(paths.LogsDir, "audita.stderr.log")
	generatedConfigPath := filepath.Join(paths.ConfigDir, "audita.generated.yml")
	if runLayout.Enabled {
		stdoutPath = filepath.Join(runLayout.LogsDir, "audita.stdout.log")
		stderrPath = filepath.Join(runLayout.LogsDir, "audita.stderr.log")
		generatedConfigPath = filepath.Join(runLayout.ConfigDir, "audita.generated.yml")
	}

	req := audita.PolishRequest{
		GeneratedConfigPath:  generatedConfigPath,
		MergedTranscriptPath: mergedPath,
		OutputProcessedPath:  processedPath,
		GlossaryPath:         glossaryPath,
		StdoutLogPath:        stdoutPath,
		StderrLogPath:        stderrPath,
	}

	res, err := env.Audita.Run(ctx, req)
	if err != nil {
		return nil, fmt.Errorf("polish: audita polish failed (output_dir=%q, stdout_log=%q, stderr_log=%q): %w",
			filepath.Dir(processedPath), stdoutPath, stderrPath, err)
	}

	finalProcessedPath, err := authoritativeOutputPath(processedPath, res.ProcessedTranscriptPath)
	if err != nil {
		return nil, fmt.Errorf("polish: %w", err)
	}
	if err := validateProcessedTranscriptOutput(finalProcessedPath); err != nil {
		return nil, fmt.Errorf("polish: processed transcript %q invalid: %w", finalProcessedPath, err)
	}
	reportPaths := make([]string, 0, 3)
	reportMetadata := map[string]string{}
	for _, report := range []struct{ name, returned string }{
		{"errors.yml", res.Reports.ErrorsPath},
		{"recoveries.yml", res.Reports.RecoveriesPath},
		{"report.md", res.Reports.OverviewPath},
	} {
		if report.returned == "" {
			continue
		}
		expected := filepath.Join(filepath.Dir(processedPath), report.name)
		path, err := authoritativeOutputPath(expected, report.returned)
		if err != nil {
			return nil, fmt.Errorf("polish: report %s: %w", report.name, err)
		}
		if err := requireRegularReadableFile(path, "audita report"); err != nil {
			return nil, fmt.Errorf("polish: %w", err)
		}
		reportPaths = append(reportPaths, path)
		reportMetadata[report.name] = path
	}

	materializedProcessed, err := materializeRunLocalOutput(env.ArtifactStore, finalProcessedPath, canonicalProcessedPath, artifacts.Ref{
		Kind:      "transcript_polished",
		Category:  "transcripts",
		SessionID: sessionID,
	})
	if err != nil {
		return nil, fmt.Errorf("polish: materialize canonical polished transcript: %w", err)
	}
	outputs := []artifacts.Ref{materializedProcessed}
	meta := map[string]any{
		"stage":                         "polish",
		"merged_transcript_path":        mergedPath,
		"merged_transcript_source":      source,
		"glossary_path":                 glossaryPath,
		"run_output_path":               finalProcessedPath,
		"output_path":                   canonicalProcessedPath,
		"modules":                       append(env.Config.Pipeline.Audita.Modules[:0:0], env.Config.Pipeline.Audita.Modules...),
		"overlay":                       env.Config.Pipeline.Audita.Overlay,
		"proposal_profile":              env.Config.Pipeline.Audita.ProposalProfile,
		"validation_profile":            env.Config.Pipeline.Audita.ValidationProfile,
		"transcript_description":        env.Config.Pipeline.Audita.TranscriptDescription,
		"config_path":                   env.Config.Pipeline.Audita.ConfigPath,
		"output_schema":                 env.Config.Pipeline.Audita.OutputSchema,
		"timeout":                       env.Config.Pipeline.Audita.Timeout,
		"binary":                        env.Config.Pipeline.Audita.Binary,
		"generated_config_path":         generatedConfigPath,
		"stdout_log_path":               stdoutPath,
		"stderr_log_path":               stderrPath,
		"adapter_duration_ms":           res.Duration.Milliseconds(),
		"adapter_exit_code":             res.ExitCode,
		"adapter_invoked_binary":        res.InvokedBinary,
		"adapter_processed_output_path": res.ProcessedTranscriptPath,
		"adapter_generated_config_path": res.GeneratedConfigPath,
		"adapter_stdout_log_path":       res.StdoutLogPath,
		"adapter_stderr_log_path":       res.StderrLogPath,
		"report_paths":                  reportMetadata,
	}
	if res.Metadata != nil {
		meta["adapter_metadata"] = res.Metadata
	}

	return &StageResult{
		Outputs:          outputs,
		Logs:             append([]string{stdoutPath, stderrPath}, reportPaths...),
		GeneratedConfigs: []string{generatedConfigPath},
		Metadata:         meta,
	}, nil
}

func discoverMergedTranscript(m *manifest.Manifest, paths artifacts.SessionPaths) (string, string, error) {
	return resolveSingletonTranscript(m, paths, artifacts.ArtifactTranscriptBase)
}

func validateProcessedTranscriptOutput(path string) error {
	if err := requireFile(path, "processed transcript"); err != nil {
		return err
	}
	data, err := readExternalResult(path, "audita processed transcript result")
	if err != nil {
		return err
	}

	var payload map[string]any
	if err := json.Unmarshal(data, &payload); err != nil {
		return fmt.Errorf("decode json: %w", err)
	}
	segments, ok := payload["segments"]
	if !ok {
		return fmt.Errorf("top-level segments is required")
	}
	if _, ok := segments.([]any); !ok {
		return fmt.Errorf("top-level segments must be an array")
	}
	return nil
}