403 lines
12 KiB
Go
403 lines
12 KiB
Go
package seriatim
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"os"
|
|
"strconv"
|
|
"strings"
|
|
"time"
|
|
|
|
"gitea.maximumdirect.net/eric/narratio/internal/adapters/subprocess"
|
|
)
|
|
|
|
// EnvConfig defines optional Seriatim environment tuning values.
|
|
type EnvConfig struct {
|
|
OverlapWordRunGap *float64
|
|
OverlapWordRunReorderWindow *float64
|
|
BackchannelMaxDuration *float64
|
|
FillerMaxDuration *float64
|
|
}
|
|
|
|
// SubprocessRunnerConfig defines deterministic settings for Seriatim CLI execution.
|
|
type SubprocessRunnerConfig struct {
|
|
Binary string
|
|
Timeout time.Duration
|
|
OutputSchema string
|
|
CoalesceGap *float64
|
|
Report bool
|
|
Env EnvConfig
|
|
}
|
|
|
|
// SubprocessRunner invokes Seriatim via subprocess.
|
|
type SubprocessRunner struct {
|
|
binary string
|
|
timeout time.Duration
|
|
outputSchema string
|
|
coalesceGap *float64
|
|
report bool
|
|
env EnvConfig
|
|
}
|
|
|
|
// NewSubprocessRunnerFromConfigValues parses config-derived values once.
|
|
func NewSubprocessRunnerFromConfigValues(
|
|
binary string,
|
|
timeout string,
|
|
outputSchema string,
|
|
coalesceGap *float64,
|
|
report bool,
|
|
env EnvConfig,
|
|
) (*SubprocessRunner, error) {
|
|
if strings.TrimSpace(timeout) == "" {
|
|
return nil, fmt.Errorf("seriatim timeout is required")
|
|
}
|
|
parsedTimeout, err := time.ParseDuration(timeout)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("parse seriatim timeout %q: %w", timeout, err)
|
|
}
|
|
return NewSubprocessRunner(SubprocessRunnerConfig{
|
|
Binary: binary,
|
|
Timeout: parsedTimeout,
|
|
OutputSchema: outputSchema,
|
|
CoalesceGap: coalesceGap,
|
|
Report: report,
|
|
Env: env,
|
|
})
|
|
}
|
|
|
|
// NewSubprocessRunner constructs a validated Seriatim subprocess runner.
|
|
func NewSubprocessRunner(cfg SubprocessRunnerConfig) (*SubprocessRunner, error) {
|
|
if strings.TrimSpace(cfg.Binary) == "" {
|
|
return nil, fmt.Errorf("seriatim binary is required")
|
|
}
|
|
if cfg.Timeout <= 0 {
|
|
return nil, fmt.Errorf("seriatim timeout must be > 0")
|
|
}
|
|
if strings.TrimSpace(cfg.OutputSchema) == "" {
|
|
return nil, fmt.Errorf("seriatim output schema is required")
|
|
}
|
|
switch cfg.OutputSchema {
|
|
case "seriatim-minimal", "seriatim-intermediate", "seriatim-full":
|
|
default:
|
|
return nil, fmt.Errorf("seriatim output schema %q is unsupported", cfg.OutputSchema)
|
|
}
|
|
if cfg.CoalesceGap != nil && *cfg.CoalesceGap < 0 {
|
|
return nil, fmt.Errorf("seriatim coalesce gap must be >= 0")
|
|
}
|
|
|
|
return &SubprocessRunner{
|
|
binary: cfg.Binary,
|
|
timeout: cfg.Timeout,
|
|
outputSchema: cfg.OutputSchema,
|
|
coalesceGap: cfg.CoalesceGap,
|
|
report: cfg.Report,
|
|
env: cfg.Env,
|
|
}, nil
|
|
}
|
|
|
|
// Run executes Seriatim merge with deterministic flags and validates output artifacts.
|
|
func (r *SubprocessRunner) Run(ctx context.Context, req MergeRequest) (MergeResult, error) {
|
|
if r == nil {
|
|
return MergeResult{}, fmt.Errorf("seriatim subprocess runner is nil")
|
|
}
|
|
if strings.TrimSpace(req.OutputMergedTranscriptPath) == "" {
|
|
return MergeResult{}, fmt.Errorf("seriatim merge output path is required")
|
|
}
|
|
if len(req.InputTranscriptPaths) == 0 {
|
|
return MergeResult{}, fmt.Errorf("seriatim merge requires at least one input transcript")
|
|
}
|
|
if r.report && strings.TrimSpace(req.ReportPath) == "" {
|
|
return MergeResult{}, fmt.Errorf("seriatim report is enabled but report path is missing")
|
|
}
|
|
|
|
args := r.buildMergeArgs(req)
|
|
env := r.buildEnvOverrides()
|
|
|
|
if req.GeneratedConfigPath != "" {
|
|
if err := r.writeMergeInvocationConfig(req, args); err != nil {
|
|
return MergeResult{}, fmt.Errorf("write seriatim invocation config %q: %w", req.GeneratedConfigPath, err)
|
|
}
|
|
}
|
|
|
|
runRes, err := subprocess.Run(ctx, subprocess.RunRequest{
|
|
Executable: r.binary,
|
|
Args: args,
|
|
Timeout: r.timeout,
|
|
EnvOverrides: env,
|
|
StdoutLogPath: req.StdoutLogPath,
|
|
StderrLogPath: req.StderrLogPath,
|
|
})
|
|
if err != nil {
|
|
return MergeResult{
|
|
MergedTranscriptPath: req.OutputMergedTranscriptPath,
|
|
ReportPath: req.ReportPath,
|
|
StdoutLogPath: req.StdoutLogPath,
|
|
StderrLogPath: req.StderrLogPath,
|
|
GeneratedConfigPath: req.GeneratedConfigPath,
|
|
ExitCode: runRes.ExitCode,
|
|
Duration: runRes.Duration,
|
|
InvokedBinary: r.binary,
|
|
OutputSchema: r.outputSchema,
|
|
}, fmt.Errorf("run seriatim merge (binary=%q): %w", r.binary, err)
|
|
}
|
|
|
|
if err := validateJSONFile(req.OutputMergedTranscriptPath); err != nil {
|
|
return MergeResult{
|
|
MergedTranscriptPath: req.OutputMergedTranscriptPath,
|
|
ReportPath: req.ReportPath,
|
|
StdoutLogPath: req.StdoutLogPath,
|
|
StderrLogPath: req.StderrLogPath,
|
|
GeneratedConfigPath: req.GeneratedConfigPath,
|
|
ExitCode: runRes.ExitCode,
|
|
Duration: runRes.Duration,
|
|
InvokedBinary: r.binary,
|
|
OutputSchema: r.outputSchema,
|
|
}, fmt.Errorf("validate seriatim merged output %q: %w", req.OutputMergedTranscriptPath, err)
|
|
}
|
|
|
|
if r.report {
|
|
if err := validateJSONFile(req.ReportPath); err != nil {
|
|
return MergeResult{
|
|
MergedTranscriptPath: req.OutputMergedTranscriptPath,
|
|
ReportPath: req.ReportPath,
|
|
StdoutLogPath: req.StdoutLogPath,
|
|
StderrLogPath: req.StderrLogPath,
|
|
GeneratedConfigPath: req.GeneratedConfigPath,
|
|
ExitCode: runRes.ExitCode,
|
|
Duration: runRes.Duration,
|
|
InvokedBinary: r.binary,
|
|
OutputSchema: r.outputSchema,
|
|
}, fmt.Errorf("validate seriatim report output %q: %w", req.ReportPath, err)
|
|
}
|
|
}
|
|
|
|
return MergeResult{
|
|
MergedTranscriptPath: req.OutputMergedTranscriptPath,
|
|
ReportPath: req.ReportPath,
|
|
StdoutLogPath: req.StdoutLogPath,
|
|
StderrLogPath: req.StderrLogPath,
|
|
GeneratedConfigPath: req.GeneratedConfigPath,
|
|
ExitCode: runRes.ExitCode,
|
|
Duration: runRes.Duration,
|
|
InvokedBinary: r.binary,
|
|
OutputSchema: r.outputSchema,
|
|
Metadata: map[string]any{
|
|
"adapter": "seriatim_subprocess",
|
|
},
|
|
}, nil
|
|
}
|
|
|
|
// Trim executes Seriatim trim with deterministic flags and validates output artifacts.
|
|
func (r *SubprocessRunner) Trim(ctx context.Context, req TrimRequest) (TrimResult, error) {
|
|
if r == nil {
|
|
return TrimResult{}, fmt.Errorf("seriatim subprocess runner is nil")
|
|
}
|
|
if strings.TrimSpace(req.InputTranscriptPath) == "" {
|
|
return TrimResult{}, fmt.Errorf("seriatim trim input path is required")
|
|
}
|
|
if strings.TrimSpace(req.OutputTrimmedPath) == "" {
|
|
return TrimResult{}, fmt.Errorf("seriatim trim output path is required")
|
|
}
|
|
if strings.TrimSpace(req.KeepSelector) == "" {
|
|
return TrimResult{}, fmt.Errorf("seriatim trim keep selector is required")
|
|
}
|
|
|
|
binary := r.binary
|
|
if strings.TrimSpace(req.Binary) != "" {
|
|
binary = strings.TrimSpace(req.Binary)
|
|
}
|
|
|
|
timeout := r.timeout
|
|
if req.Timeout < 0 {
|
|
return TrimResult{}, fmt.Errorf("seriatim trim timeout must be >= 0")
|
|
}
|
|
if req.Timeout > 0 {
|
|
timeout = req.Timeout
|
|
}
|
|
|
|
args := buildTrimArgs(req)
|
|
if req.GeneratedConfigPath != "" {
|
|
if err := writeTrimInvocationConfig(req, args, binary, timeout); err != nil {
|
|
return TrimResult{}, fmt.Errorf("write seriatim trim invocation config %q: %w", req.GeneratedConfigPath, err)
|
|
}
|
|
}
|
|
|
|
runRes, err := subprocess.Run(ctx, subprocess.RunRequest{
|
|
Executable: binary,
|
|
Args: args,
|
|
Timeout: timeout,
|
|
StdoutLogPath: req.StdoutLogPath,
|
|
StderrLogPath: req.StderrLogPath,
|
|
})
|
|
if err != nil {
|
|
return TrimResult{
|
|
OutputTrimmedPath: req.OutputTrimmedPath,
|
|
StdoutLogPath: req.StdoutLogPath,
|
|
StderrLogPath: req.StderrLogPath,
|
|
GeneratedConfigPath: req.GeneratedConfigPath,
|
|
ExitCode: runRes.ExitCode,
|
|
Duration: runRes.Duration,
|
|
InvokedBinary: binary,
|
|
KeepSelector: req.KeepSelector,
|
|
}, fmt.Errorf("run seriatim trim (binary=%q): %w", binary, err)
|
|
}
|
|
|
|
if err := validateJSONFileWithSegments(req.OutputTrimmedPath); err != nil {
|
|
return TrimResult{
|
|
OutputTrimmedPath: req.OutputTrimmedPath,
|
|
StdoutLogPath: req.StdoutLogPath,
|
|
StderrLogPath: req.StderrLogPath,
|
|
GeneratedConfigPath: req.GeneratedConfigPath,
|
|
ExitCode: runRes.ExitCode,
|
|
Duration: runRes.Duration,
|
|
InvokedBinary: binary,
|
|
KeepSelector: req.KeepSelector,
|
|
}, fmt.Errorf("validate seriatim trimmed output %q: %w", req.OutputTrimmedPath, err)
|
|
}
|
|
|
|
return TrimResult{
|
|
OutputTrimmedPath: req.OutputTrimmedPath,
|
|
StdoutLogPath: req.StdoutLogPath,
|
|
StderrLogPath: req.StderrLogPath,
|
|
GeneratedConfigPath: req.GeneratedConfigPath,
|
|
ExitCode: runRes.ExitCode,
|
|
Duration: runRes.Duration,
|
|
InvokedBinary: binary,
|
|
KeepSelector: req.KeepSelector,
|
|
Metadata: map[string]any{
|
|
"adapter": "seriatim_subprocess",
|
|
},
|
|
}, nil
|
|
}
|
|
|
|
func (r *SubprocessRunner) buildMergeArgs(req MergeRequest) []string {
|
|
args := []string{"merge"}
|
|
|
|
for _, path := range req.InputTranscriptPaths {
|
|
args = append(args, "--input-file", path)
|
|
}
|
|
args = append(args, "--output-file", req.OutputMergedTranscriptPath)
|
|
|
|
if r.report {
|
|
args = append(args, "--report-file", req.ReportPath)
|
|
}
|
|
if strings.TrimSpace(req.SpeakersPath) != "" {
|
|
args = append(args, "--speakers", req.SpeakersPath)
|
|
}
|
|
if strings.TrimSpace(req.AutocorrectPath) != "" {
|
|
args = append(args, "--autocorrect", req.AutocorrectPath)
|
|
}
|
|
|
|
args = append(args, "--output-schema", r.outputSchema)
|
|
if r.coalesceGap != nil {
|
|
args = append(args, "--coalesce-gap", strconv.FormatFloat(*r.coalesceGap, 'f', -1, 64))
|
|
}
|
|
|
|
return args
|
|
}
|
|
|
|
func (r *SubprocessRunner) buildEnvOverrides() map[string]string {
|
|
out := map[string]string{}
|
|
if r.env.OverlapWordRunGap != nil {
|
|
out["SERIATIM_OVERLAP_WORD_RUN_GAP"] = strconv.FormatFloat(*r.env.OverlapWordRunGap, 'f', -1, 64)
|
|
}
|
|
if r.env.OverlapWordRunReorderWindow != nil {
|
|
out["SERIATIM_OVERLAP_WORD_RUN_REORDER_WINDOW"] = strconv.FormatFloat(*r.env.OverlapWordRunReorderWindow, 'f', -1, 64)
|
|
}
|
|
if r.env.BackchannelMaxDuration != nil {
|
|
out["SERIATIM_BACKCHANNEL_MAX_DURATION"] = strconv.FormatFloat(*r.env.BackchannelMaxDuration, 'f', -1, 64)
|
|
}
|
|
if r.env.FillerMaxDuration != nil {
|
|
out["SERIATIM_FILLER_MAX_DURATION"] = strconv.FormatFloat(*r.env.FillerMaxDuration, 'f', -1, 64)
|
|
}
|
|
if len(out) == 0 {
|
|
return nil
|
|
}
|
|
return out
|
|
}
|
|
|
|
func (r *SubprocessRunner) writeMergeInvocationConfig(req MergeRequest, args []string) error {
|
|
payload := map[string]any{
|
|
"schema": "seriatim.generated.v1",
|
|
"binary": r.binary,
|
|
"command": "merge",
|
|
"args": args,
|
|
"timeout": r.timeout.String(),
|
|
"output_schema": r.outputSchema,
|
|
"report_enabled": r.report,
|
|
"input_transcript_paths": req.InputTranscriptPaths,
|
|
"output_path": req.OutputMergedTranscriptPath,
|
|
}
|
|
if req.ReportPath != "" {
|
|
payload["report_path"] = req.ReportPath
|
|
}
|
|
if req.SpeakersPath != "" {
|
|
payload["speakers_path"] = req.SpeakersPath
|
|
}
|
|
if req.AutocorrectPath != "" {
|
|
payload["autocorrect_path"] = req.AutocorrectPath
|
|
}
|
|
if r.coalesceGap != nil {
|
|
payload["coalesce_gap"] = *r.coalesceGap
|
|
}
|
|
|
|
return subprocess.WriteYAMLAtomic(req.GeneratedConfigPath, payload, 0o644)
|
|
}
|
|
|
|
func buildTrimArgs(req TrimRequest) []string {
|
|
return []string{
|
|
"trim",
|
|
"--input-file", req.InputTranscriptPath,
|
|
"--output-file", req.OutputTrimmedPath,
|
|
"--keep", req.KeepSelector,
|
|
}
|
|
}
|
|
|
|
func writeTrimInvocationConfig(req TrimRequest, args []string, binary string, timeout time.Duration) error {
|
|
payload := map[string]any{
|
|
"schema": "seriatim.generated.v1",
|
|
"command": "trim",
|
|
"binary": binary,
|
|
"args": args,
|
|
"timeout": timeout.String(),
|
|
"input_path": req.InputTranscriptPath,
|
|
"output_path": req.OutputTrimmedPath,
|
|
"keep_selector": req.KeepSelector,
|
|
}
|
|
return subprocess.WriteYAMLAtomic(req.GeneratedConfigPath, payload, 0o644)
|
|
}
|
|
|
|
func validateJSONFile(path string) error {
|
|
data, err := os.ReadFile(path)
|
|
if err != nil {
|
|
return fmt.Errorf("read file: %w", err)
|
|
}
|
|
var v any
|
|
if err := json.Unmarshal(data, &v); err != nil {
|
|
return fmt.Errorf("parse json: %w", err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func validateJSONFileWithSegments(path string) error {
|
|
data, err := os.ReadFile(path)
|
|
if err != nil {
|
|
return fmt.Errorf("read file: %w", err)
|
|
}
|
|
|
|
var payload map[string]any
|
|
if err := json.Unmarshal(data, &payload); err != nil {
|
|
return fmt.Errorf("parse 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
|
|
}
|