Add Seriatim trim adapter support

This commit is contained in:
2026-05-08 16:58:26 +00:00
parent ea39594b33
commit 79a8bcf37b
5 changed files with 495 additions and 13 deletions

View File

@@ -29,11 +29,33 @@ func (n *NoopRunner) Run(ctx context.Context, req MergeRequest) (MergeResult, er
}, nil
}
// Trim returns the requested output path with placeholder metadata.
func (n *NoopRunner) Trim(ctx context.Context, req TrimRequest) (TrimResult, error) {
if err := ctx.Err(); err != nil {
return TrimResult{}, err
}
if err := materializeTrimPlaceholders(req); err != nil {
return TrimResult{}, err
}
return TrimResult{
OutputTrimmedPath: req.OutputTrimmedPath,
StdoutLogPath: req.StdoutLogPath,
StderrLogPath: req.StderrLogPath,
GeneratedConfigPath: req.GeneratedConfigPath,
InvokedBinary: "noop",
KeepSelector: req.KeepSelector,
Metadata: map[string]any{"placeholder": true},
}, nil
}
// FakeRunner captures merge requests and returns deterministic responses.
type FakeRunner struct {
Requests []MergeRequest
Err error
Result MergeResult
Requests []MergeRequest
Err error
Result MergeResult
TrimRequests []TrimRequest
TrimErr error
TrimResult TrimResult
}
// Run records request and returns configured response.
@@ -73,6 +95,43 @@ func (f *FakeRunner) Run(ctx context.Context, req MergeRequest) (MergeResult, er
return res, nil
}
// Trim records request and returns configured response.
func (f *FakeRunner) Trim(ctx context.Context, req TrimRequest) (TrimResult, error) {
if err := ctx.Err(); err != nil {
return TrimResult{}, err
}
f.TrimRequests = append(f.TrimRequests, req)
if f.TrimErr != nil {
return TrimResult{}, f.TrimErr
}
if err := materializeTrimPlaceholders(req); err != nil {
return TrimResult{}, err
}
res := f.TrimResult
if res.OutputTrimmedPath == "" {
res.OutputTrimmedPath = req.OutputTrimmedPath
}
if res.StdoutLogPath == "" {
res.StdoutLogPath = req.StdoutLogPath
}
if res.StderrLogPath == "" {
res.StderrLogPath = req.StderrLogPath
}
if res.GeneratedConfigPath == "" {
res.GeneratedConfigPath = req.GeneratedConfigPath
}
if res.InvokedBinary == "" {
res.InvokedBinary = "fake"
}
if res.KeepSelector == "" {
res.KeepSelector = req.KeepSelector
}
if res.Metadata == nil {
res.Metadata = map[string]any{"fake": true}
}
return res, nil
}
func materializePlaceholders(req MergeRequest) error {
if req.OutputMergedTranscriptPath != "" {
if err := subprocess.WriteFileAtomic(req.OutputMergedTranscriptPath, []byte(`{"schema":"seriatim.intermediate.v1","segments":[]}`), 0o644); err != nil {
@@ -107,3 +166,35 @@ func materializePlaceholders(req MergeRequest) error {
}
return nil
}
func materializeTrimPlaceholders(req TrimRequest) error {
if req.OutputTrimmedPath != "" {
if err := subprocess.WriteFileAtomic(req.OutputTrimmedPath, []byte(`{"schema":"seriatim.intermediate.v1","segments":[]}`), 0o644); err != nil {
return fmt.Errorf("write trimmed transcript %q: %w", req.OutputTrimmedPath, err)
}
}
if req.GeneratedConfigPath != "" {
payload := map[string]any{
"schema": "seriatim.generated.v1",
"placeholder": true,
"command": "trim",
"input_path": req.InputTranscriptPath,
"output_path": req.OutputTrimmedPath,
"keep_selector": req.KeepSelector,
}
if err := subprocess.WriteYAMLAtomic(req.GeneratedConfigPath, payload, 0o644); err != nil {
return fmt.Errorf("write generated config %q: %w", req.GeneratedConfigPath, err)
}
}
if req.StdoutLogPath != "" {
if err := subprocess.WriteFileAtomic(req.StdoutLogPath, []byte("seriatim noop/fake trim stdout placeholder\n"), 0o644); err != nil {
return fmt.Errorf("write stdout log %q: %w", req.StdoutLogPath, err)
}
}
if req.StderrLogPath != "" {
if err := subprocess.WriteFileAtomic(req.StderrLogPath, []byte("seriatim noop/fake trim stderr placeholder\n"), 0o644); err != nil {
return fmt.Errorf("write stderr log %q: %w", req.StderrLogPath, err)
}
}
return nil
}

View File

@@ -51,3 +51,51 @@ func TestFakeRunnerError(t *testing.T) {
t.Fatal("expected error, got nil")
}
}
func TestFakeRunnerTrimCapturesRequestAndReturnsPath(t *testing.T) {
fake := &FakeRunner{}
dir := t.TempDir()
req := TrimRequest{
GeneratedConfigPath: filepath.Join(dir, "config", "seriatim.trim.yml"),
InputTranscriptPath: filepath.Join(dir, "transcripts", "processed.json"),
OutputTrimmedPath: filepath.Join(dir, "transcripts", "trimmed.json"),
KeepSelector: "1-10",
StdoutLogPath: filepath.Join(dir, "logs", "seriatim.trim.stdout.log"),
StderrLogPath: filepath.Join(dir, "logs", "seriatim.trim.stderr.log"),
}
res, err := fake.Trim(context.Background(), req)
if err != nil {
t.Fatalf("Trim() error = %v", err)
}
if len(fake.TrimRequests) != 1 || fake.TrimRequests[0].GeneratedConfigPath == "" {
t.Fatalf("trim requests = %#v, want captured request", fake.TrimRequests)
}
if res.OutputTrimmedPath != req.OutputTrimmedPath {
t.Fatalf("trimmed path = %q, want %q", res.OutputTrimmedPath, req.OutputTrimmedPath)
}
if res.KeepSelector != req.KeepSelector {
t.Fatalf("keep selector = %q, want %q", res.KeepSelector, req.KeepSelector)
}
cfgData, err := os.ReadFile(req.GeneratedConfigPath)
if err != nil {
t.Fatalf("read generated config: %v", err)
}
if !strings.Contains(string(cfgData), "command: trim") {
t.Fatalf("generated config = %q, want trim command marker", string(cfgData))
}
for _, logPath := range []string{req.StdoutLogPath, req.StderrLogPath} {
if _, err := os.Stat(logPath); err != nil {
t.Fatalf("expected log file %q to exist: %v", logPath, err)
}
}
}
func TestFakeRunnerTrimError(t *testing.T) {
fake := &FakeRunner{TrimErr: errors.New("boom")}
_, err := fake.Trim(context.Background(), TrimRequest{})
if err == nil {
t.Fatal("expected error, got nil")
}
}

View File

@@ -1,4 +1,4 @@
// Package seriatim declares the adapter contract for transcript merge execution.
// Package seriatim declares the adapter contract for transcript merge/trim execution.
package seriatim
import (
@@ -8,9 +8,10 @@ import (
// TODO: implement a real Seriatim subprocess adapter.
// Runner is the adapter boundary for seriatim merge invocations.
// Runner is the adapter boundary for seriatim merge/trim invocations.
type Runner interface {
Run(ctx context.Context, req MergeRequest) (MergeResult, error)
Trim(ctx context.Context, req TrimRequest) (TrimResult, error)
}
// MergeRequest describes a seriatim merge invocation.
@@ -38,3 +39,28 @@ type MergeResult struct {
OutputSchema string
Metadata map[string]any
}
// TrimRequest describes a seriatim trim invocation.
type TrimRequest struct {
Binary string
InputTranscriptPath string
OutputTrimmedPath string
KeepSelector string
StdoutLogPath string
StderrLogPath string
GeneratedConfigPath string
Timeout time.Duration
}
// TrimResult describes a trim output.
type TrimResult struct {
OutputTrimmedPath string
StdoutLogPath string
StderrLogPath string
GeneratedConfigPath string
ExitCode int
Duration time.Duration
InvokedBinary string
KeepSelector string
Metadata map[string]any
}

View File

@@ -111,11 +111,11 @@ func (r *SubprocessRunner) Run(ctx context.Context, req MergeRequest) (MergeResu
return MergeResult{}, fmt.Errorf("seriatim report is enabled but report path is missing")
}
args := r.buildArgs(req)
args := r.buildMergeArgs(req)
env := r.buildEnvOverrides()
if req.GeneratedConfigPath != "" {
if err := r.writeInvocationConfig(req, args); err != nil {
if err := r.writeMergeInvocationConfig(req, args); err != nil {
return MergeResult{}, fmt.Errorf("write seriatim invocation config %q: %w", req.GeneratedConfigPath, err)
}
}
@@ -188,7 +188,90 @@ func (r *SubprocessRunner) Run(ctx context.Context, req MergeRequest) (MergeResu
}, nil
}
func (r *SubprocessRunner) buildArgs(req MergeRequest) []string {
// 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 {
@@ -234,10 +317,11 @@ func (r *SubprocessRunner) buildEnvOverrides() map[string]string {
return out
}
func (r *SubprocessRunner) writeInvocationConfig(req MergeRequest, args []string) error {
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,
@@ -261,6 +345,29 @@ func (r *SubprocessRunner) writeInvocationConfig(req MergeRequest, args []string
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 {
@@ -272,3 +379,24 @@ func validateJSONFile(path string) error {
}
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
}

View File

@@ -209,6 +209,172 @@ func TestSubprocessRunnerInvalidReportJSONFails(t *testing.T) {
}
}
func TestSubprocessRunnerTrimSuccessInvocationAndProvenance(t *testing.T) {
if runtime.GOOS == "windows" {
t.Skip("helper wrapper script uses /bin/sh")
}
t.Setenv("GO_WANT_SERIATIM_HELPER", "1")
t.Setenv("SERIATIM_HELPER_MODE", "trim_success")
recordPath := filepath.Join(t.TempDir(), "record.json")
t.Setenv("SERIATIM_HELPER_RECORD_PATH", recordPath)
wrapper := writeHelperWrapper(t)
runner := mustRunner(t, wrapper, false)
req := trimReqForTest(t)
res, err := runner.Trim(context.Background(), req)
if err != nil {
t.Fatalf("Trim() error = %v", err)
}
if res.OutputTrimmedPath != req.OutputTrimmedPath {
t.Fatalf("OutputTrimmedPath = %q, want %q", res.OutputTrimmedPath, req.OutputTrimmedPath)
}
if res.KeepSelector != req.KeepSelector {
t.Fatalf("KeepSelector = %q, want %q", res.KeepSelector, req.KeepSelector)
}
if res.InvokedBinary != wrapper {
t.Fatalf("InvokedBinary = %q, want %q", res.InvokedBinary, wrapper)
}
if res.ExitCode != 0 {
t.Fatalf("ExitCode = %d, want 0", res.ExitCode)
}
if res.Duration <= 0 {
t.Fatalf("Duration = %s, want >0", res.Duration)
}
if res.Metadata == nil || res.Metadata["adapter"] != "seriatim_subprocess" {
t.Fatalf("Metadata = %#v, want adapter marker", res.Metadata)
}
assertJSONFile(t, req.OutputTrimmedPath)
if _, err := os.Stat(req.StdoutLogPath); err != nil {
t.Fatalf("stdout log missing: %v", err)
}
if _, err := os.Stat(req.StderrLogPath); err != nil {
t.Fatalf("stderr log missing: %v", err)
}
if _, err := os.Stat(req.GeneratedConfigPath); err != nil {
t.Fatalf("generated config missing: %v", err)
}
stdoutData, err := os.ReadFile(req.StdoutLogPath)
if err != nil {
t.Fatalf("read stdout log: %v", err)
}
if !strings.Contains(string(stdoutData), "seriatim helper trim stdout") {
t.Fatalf("stdout log = %q, want trim helper stdout marker", string(stdoutData))
}
stderrData, err := os.ReadFile(req.StderrLogPath)
if err != nil {
t.Fatalf("read stderr log: %v", err)
}
if !strings.Contains(string(stderrData), "seriatim helper trim stderr") {
t.Fatalf("stderr log = %q, want trim helper stderr marker", string(stderrData))
}
cfgData, err := os.ReadFile(req.GeneratedConfigPath)
if err != nil {
t.Fatalf("read generated config: %v", err)
}
cfgText := string(cfgData)
if !strings.Contains(cfgText, "command: trim") {
t.Fatalf("generated config = %q, want command: trim", cfgText)
}
if !strings.Contains(cfgText, "keep_selector: 5-12") {
t.Fatalf("generated config = %q, want keep selector", cfgText)
}
rec := readHelperRecord(t, recordPath)
wantArgs := []string{
"trim",
"--input-file", req.InputTranscriptPath,
"--output-file", req.OutputTrimmedPath,
"--keep", req.KeepSelector,
}
if strings.Join(rec.Args, "\n") != strings.Join(wantArgs, "\n") {
t.Fatalf("args = %#v, want %#v", rec.Args, wantArgs)
}
}
func TestSubprocessRunnerTrimSubprocessFailure(t *testing.T) {
if runtime.GOOS == "windows" {
t.Skip("helper wrapper script uses /bin/sh")
}
t.Setenv("GO_WANT_SERIATIM_HELPER", "1")
t.Setenv("SERIATIM_HELPER_MODE", "fail")
t.Setenv("SERIATIM_HELPER_RECORD_PATH", filepath.Join(t.TempDir(), "record.json"))
runner := mustRunner(t, writeHelperWrapper(t), false)
req := trimReqForTest(t)
_, err := runner.Trim(context.Background(), req)
if err == nil {
t.Fatal("Trim() error = nil, want non-nil")
}
if !strings.Contains(err.Error(), "run seriatim trim") {
t.Fatalf("error = %q, want subprocess context", err.Error())
}
if !strings.Contains(err.Error(), "exit code") {
t.Fatalf("error = %q, want exit code context", err.Error())
}
}
func TestSubprocessRunnerTrimMissingOutputFails(t *testing.T) {
if runtime.GOOS == "windows" {
t.Skip("helper wrapper script uses /bin/sh")
}
t.Setenv("GO_WANT_SERIATIM_HELPER", "1")
t.Setenv("SERIATIM_HELPER_MODE", "missing_output")
t.Setenv("SERIATIM_HELPER_RECORD_PATH", filepath.Join(t.TempDir(), "record.json"))
runner := mustRunner(t, writeHelperWrapper(t), false)
req := trimReqForTest(t)
_, err := runner.Trim(context.Background(), req)
if err == nil {
t.Fatal("Trim() error = nil, want non-nil")
}
if !strings.Contains(err.Error(), "validate seriatim trimmed output") {
t.Fatalf("error = %q, want output validation context", err.Error())
}
}
func TestSubprocessRunnerTrimInvalidOutputJSONFails(t *testing.T) {
if runtime.GOOS == "windows" {
t.Skip("helper wrapper script uses /bin/sh")
}
t.Setenv("GO_WANT_SERIATIM_HELPER", "1")
t.Setenv("SERIATIM_HELPER_MODE", "invalid_output")
t.Setenv("SERIATIM_HELPER_RECORD_PATH", filepath.Join(t.TempDir(), "record.json"))
runner := mustRunner(t, writeHelperWrapper(t), false)
req := trimReqForTest(t)
_, err := runner.Trim(context.Background(), req)
if err == nil {
t.Fatal("Trim() error = nil, want non-nil")
}
if !strings.Contains(err.Error(), "parse json") {
t.Fatalf("error = %q, want parse json context", err.Error())
}
}
func TestSubprocessRunnerTrimOutputMissingSegmentsFails(t *testing.T) {
if runtime.GOOS == "windows" {
t.Skip("helper wrapper script uses /bin/sh")
}
t.Setenv("GO_WANT_SERIATIM_HELPER", "1")
t.Setenv("SERIATIM_HELPER_MODE", "trim_missing_segments")
t.Setenv("SERIATIM_HELPER_RECORD_PATH", filepath.Join(t.TempDir(), "record.json"))
runner := mustRunner(t, writeHelperWrapper(t), false)
req := trimReqForTest(t)
_, err := runner.Trim(context.Background(), req)
if err == nil {
t.Fatal("Trim() error = nil, want non-nil")
}
if !strings.Contains(err.Error(), "top-level segments is required") {
t.Fatalf("error = %q, want missing segments context", err.Error())
}
}
func TestSubprocessRunnerConstructorValidation(t *testing.T) {
_, err := NewSubprocessRunnerFromConfigValues("", "10m", "seriatim-intermediate", nil, true, EnvConfig{})
if err == nil {
@@ -250,14 +416,14 @@ func TestSeriatimSubprocessHelper(t *testing.T) {
_, _ = os.Stderr.WriteString("missing -- args separator\n")
os.Exit(2)
}
mergeArgs := args[start:]
cmdArgs := args[start:]
outputPath := flagValue(mergeArgs, "--output-file")
reportPath := flagValue(mergeArgs, "--report-file")
outputPath := flagValue(cmdArgs, "--output-file")
reportPath := flagValue(cmdArgs, "--report-file")
recordPath := os.Getenv("SERIATIM_HELPER_RECORD_PATH")
if strings.TrimSpace(recordPath) != "" {
rec := helperRecord{
Args: mergeArgs,
Args: cmdArgs,
Env: map[string]string{
"SERIATIM_OVERLAP_WORD_RUN_GAP": os.Getenv("SERIATIM_OVERLAP_WORD_RUN_GAP"),
"SERIATIM_OVERLAP_WORD_RUN_REORDER_WINDOW": os.Getenv("SERIATIM_OVERLAP_WORD_RUN_REORDER_WINDOW"),
@@ -280,6 +446,11 @@ func TestSeriatimSubprocessHelper(t *testing.T) {
_, _ = os.Stdout.WriteString("seriatim helper success stdout\n")
_, _ = os.Stderr.WriteString("seriatim helper success stderr\n")
os.Exit(0)
case "trim_success":
writeSeriatimHelperFile(outputPath, `{"schema":"seriatim.intermediate.v1","segments":[]}`)
_, _ = os.Stdout.WriteString("seriatim helper trim stdout\n")
_, _ = os.Stderr.WriteString("seriatim helper trim stderr\n")
os.Exit(0)
case "fail":
_, _ = os.Stderr.WriteString("seriatim helper failure\n")
os.Exit(9)
@@ -300,6 +471,9 @@ func TestSeriatimSubprocessHelper(t *testing.T) {
writeSeriatimHelperFile(reportPath, `not-json`)
}
os.Exit(0)
case "trim_missing_segments":
writeSeriatimHelperFile(outputPath, `{"schema":"seriatim.intermediate.v1"}`)
os.Exit(0)
default:
_, _ = os.Stderr.WriteString(fmt.Sprintf("unknown helper mode %q\n", mode))
os.Exit(2)
@@ -340,6 +514,21 @@ func mergeReqForTest(t *testing.T, withReport bool) MergeRequest {
return req
}
func trimReqForTest(t *testing.T) TrimRequest {
t.Helper()
dir := t.TempDir()
input := filepath.Join(dir, "processed.json")
writeSeriatimFile(t, input, `{"schema":"seriatim.intermediate.v1","segments":[]}`)
return TrimRequest{
InputTranscriptPath: input,
OutputTrimmedPath: filepath.Join(dir, "trimmed.json"),
KeepSelector: "5-12",
GeneratedConfigPath: filepath.Join(dir, "seriatim.trim.generated.yml"),
StdoutLogPath: filepath.Join(dir, "seriatim.trim.stdout.log"),
StderrLogPath: filepath.Join(dir, "seriatim.trim.stderr.log"),
}
}
func mustRunner(t *testing.T, binary string, report bool) *SubprocessRunner {
t.Helper()
coalesce := 3.0