Add subprocess runner and generated config helpers

This commit is contained in:
2026-05-02 11:32:33 -05:00
parent 1f9d8d0ef7
commit 77f421feef
7 changed files with 545 additions and 4 deletions

View File

@@ -1,6 +1,11 @@
package audita
import "context"
import (
"context"
"fmt"
"gitea.maximumdirect.net/eric/narratio/internal/adapters/subprocess"
)
// NoopRunner is a deterministic no-op audita adapter.
type NoopRunner struct{}
@@ -10,6 +15,9 @@ func (n *NoopRunner) Run(ctx context.Context, req PolishRequest) (PolishResult,
if err := ctx.Err(); err != nil {
return PolishResult{}, err
}
if err := materializePlaceholders(req); err != nil {
return PolishResult{}, err
}
return PolishResult{ProcessedTranscriptPath: req.OutputProcessedPath, Metadata: map[string]any{"placeholder": true}}, nil
}
@@ -29,6 +37,9 @@ func (f *FakeRunner) Run(ctx context.Context, req PolishRequest) (PolishResult,
if f.Err != nil {
return PolishResult{}, f.Err
}
if err := materializePlaceholders(req); err != nil {
return PolishResult{}, err
}
res := f.Result
if res.ProcessedTranscriptPath == "" {
res.ProcessedTranscriptPath = req.OutputProcessedPath
@@ -38,3 +49,28 @@ func (f *FakeRunner) Run(ctx context.Context, req PolishRequest) (PolishResult,
}
return res, nil
}
func materializePlaceholders(req PolishRequest) error {
if req.GeneratedConfigPath != "" {
payload := map[string]any{
"schema": "audita.generated.v1",
"placeholder": true,
"merged_transcript_path": req.MergedTranscriptPath,
"output_path": req.OutputProcessedPath,
}
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("audita noop/fake 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("audita noop/fake stderr placeholder\n"), 0o644); err != nil {
return fmt.Errorf("write stderr log %q: %w", req.StderrLogPath, err)
}
}
return nil
}

View File

@@ -3,12 +3,21 @@ package audita
import (
"context"
"errors"
"os"
"path/filepath"
"strings"
"testing"
)
func TestFakeRunnerCapturesRequestAndReturnsPath(t *testing.T) {
fake := &FakeRunner{}
req := PolishRequest{GeneratedConfigPath: "config/audita.yml", OutputProcessedPath: "transcripts/processed.json"}
dir := t.TempDir()
req := PolishRequest{
GeneratedConfigPath: filepath.Join(dir, "config", "audita.yml"),
OutputProcessedPath: filepath.Join(dir, "transcripts", "processed.json"),
StdoutLogPath: filepath.Join(dir, "logs", "audita.stdout.log"),
StderrLogPath: filepath.Join(dir, "logs", "audita.stderr.log"),
}
res, err := fake.Run(context.Background(), req)
if err != nil {
@@ -20,6 +29,19 @@ func TestFakeRunnerCapturesRequestAndReturnsPath(t *testing.T) {
if res.ProcessedTranscriptPath != req.OutputProcessedPath {
t.Fatalf("processed path = %q, want %q", res.ProcessedTranscriptPath, req.OutputProcessedPath)
}
cfgData, err := os.ReadFile(req.GeneratedConfigPath)
if err != nil {
t.Fatalf("read generated config: %v", err)
}
if !strings.Contains(string(cfgData), "placeholder: true") {
t.Fatalf("generated config = %q, want placeholder 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 TestFakeRunnerError(t *testing.T) {

View File

@@ -1,6 +1,11 @@
package seriatim
import "context"
import (
"context"
"fmt"
"gitea.maximumdirect.net/eric/narratio/internal/adapters/subprocess"
)
// NoopRunner is a deterministic no-op seriatim adapter.
type NoopRunner struct{}
@@ -10,6 +15,9 @@ func (n *NoopRunner) Run(ctx context.Context, req MergeRequest) (MergeResult, er
if err := ctx.Err(); err != nil {
return MergeResult{}, err
}
if err := materializePlaceholders(req); err != nil {
return MergeResult{}, err
}
return MergeResult{MergedTranscriptPath: req.OutputMergedTranscriptPath, Metadata: map[string]any{"placeholder": true}}, nil
}
@@ -29,6 +37,9 @@ func (f *FakeRunner) Run(ctx context.Context, req MergeRequest) (MergeResult, er
if f.Err != nil {
return MergeResult{}, f.Err
}
if err := materializePlaceholders(req); err != nil {
return MergeResult{}, err
}
res := f.Result
if res.MergedTranscriptPath == "" {
res.MergedTranscriptPath = req.OutputMergedTranscriptPath
@@ -38,3 +49,28 @@ func (f *FakeRunner) Run(ctx context.Context, req MergeRequest) (MergeResult, er
}
return res, nil
}
func materializePlaceholders(req MergeRequest) error {
if req.GeneratedConfigPath != "" {
payload := map[string]any{
"schema": "seriatim.generated.v1",
"placeholder": true,
"input_transcript_paths": req.InputTranscriptPaths,
"output_path": req.OutputMergedTranscriptPath,
}
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 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 stderr placeholder\n"), 0o644); err != nil {
return fmt.Errorf("write stderr log %q: %w", req.StderrLogPath, err)
}
}
return nil
}

View File

@@ -3,12 +3,21 @@ package seriatim
import (
"context"
"errors"
"os"
"path/filepath"
"strings"
"testing"
)
func TestFakeRunnerCapturesRequestAndReturnsPath(t *testing.T) {
fake := &FakeRunner{}
req := MergeRequest{GeneratedConfigPath: "config/seriatim.yml", OutputMergedTranscriptPath: "transcripts/merged.json"}
dir := t.TempDir()
req := MergeRequest{
GeneratedConfigPath: filepath.Join(dir, "config", "seriatim.yml"),
OutputMergedTranscriptPath: filepath.Join(dir, "transcripts", "merged.json"),
StdoutLogPath: filepath.Join(dir, "logs", "seriatim.stdout.log"),
StderrLogPath: filepath.Join(dir, "logs", "seriatim.stderr.log"),
}
res, err := fake.Run(context.Background(), req)
if err != nil {
@@ -20,6 +29,19 @@ func TestFakeRunnerCapturesRequestAndReturnsPath(t *testing.T) {
if res.MergedTranscriptPath != req.OutputMergedTranscriptPath {
t.Fatalf("merged path = %q, want %q", res.MergedTranscriptPath, req.OutputMergedTranscriptPath)
}
cfgData, err := os.ReadFile(req.GeneratedConfigPath)
if err != nil {
t.Fatalf("read generated config: %v", err)
}
if !strings.Contains(string(cfgData), "placeholder: true") {
t.Fatalf("generated config = %q, want placeholder 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 TestFakeRunnerError(t *testing.T) {

View File

@@ -0,0 +1,3 @@
// Package subprocess provides reusable process execution and generated-config helpers.
package subprocess

View File

@@ -0,0 +1,226 @@
package subprocess
import (
"context"
"errors"
"fmt"
"io"
"os"
"os/exec"
"path/filepath"
"sort"
"strings"
"time"
"gopkg.in/yaml.v3"
)
// RunRequest defines a subprocess invocation.
type RunRequest struct {
Executable string
Args []string
WorkingDir string
EnvOverrides map[string]string
Timeout time.Duration
StdoutLogPath string
StderrLogPath string
}
// RunResult captures subprocess execution details.
type RunResult struct {
ExitCode int
StartedAt time.Time
CompletedAt time.Time
Duration time.Duration
StdoutLogPath string
StderrLogPath string
TimedOut bool
Canceled bool
}
// Run executes a subprocess with context cancellation and optional timeout.
func Run(ctx context.Context, req RunRequest) (RunResult, error) {
if strings.TrimSpace(req.Executable) == "" {
return RunResult{}, fmt.Errorf("subprocess executable is required")
}
if ctx == nil {
ctx = context.Background()
}
runCtx := ctx
cancel := func() {}
if req.Timeout > 0 {
runCtx, cancel = context.WithTimeout(ctx, req.Timeout)
}
defer cancel()
stdoutFile, stdoutWriter, err := logWriter(req.StdoutLogPath)
if err != nil {
return RunResult{}, fmt.Errorf("open stdout log: %w", err)
}
defer closeFile(stdoutFile)
stderrFile, stderrWriter, err := logWriter(req.StderrLogPath)
if err != nil {
return RunResult{}, fmt.Errorf("open stderr log: %w", err)
}
defer closeFile(stderrFile)
cmd := exec.CommandContext(runCtx, req.Executable, req.Args...)
cmd.Dir = req.WorkingDir
cmd.Env = mergeEnv(os.Environ(), req.EnvOverrides)
cmd.Stdout = stdoutWriter
cmd.Stderr = stderrWriter
started := time.Now().UTC()
result := RunResult{
ExitCode: -1,
StartedAt: started,
StdoutLogPath: req.StdoutLogPath,
StderrLogPath: req.StderrLogPath,
}
if err := cmd.Start(); err != nil {
result.CompletedAt = time.Now().UTC()
result.Duration = result.CompletedAt.Sub(result.StartedAt)
return result, fmt.Errorf("start command %q with args %v: %w", req.Executable, req.Args, err)
}
waitErr := cmd.Wait()
result.CompletedAt = time.Now().UTC()
result.Duration = result.CompletedAt.Sub(result.StartedAt)
if cmd.ProcessState != nil {
result.ExitCode = cmd.ProcessState.ExitCode()
}
ctxErr := runCtx.Err()
if errors.Is(ctxErr, context.DeadlineExceeded) {
result.TimedOut = true
}
if errors.Is(ctxErr, context.Canceled) && !result.TimedOut {
result.Canceled = true
}
if waitErr == nil {
return result, nil
}
if result.TimedOut {
return result, fmt.Errorf("command %q timed out after %s (args=%v)", req.Executable, req.Timeout, req.Args)
}
if result.Canceled {
return result, fmt.Errorf("command %q canceled (args=%v)", req.Executable, req.Args)
}
if exitErr, ok := waitErr.(*exec.ExitError); ok {
return result, fmt.Errorf("command %q failed with exit code %d (args=%v): %w", req.Executable, exitErr.ExitCode(), req.Args, waitErr)
}
return result, fmt.Errorf("command %q failed to run (args=%v): %w", req.Executable, req.Args, waitErr)
}
// WriteYAMLAtomic marshals value as YAML and atomically writes it to path.
func WriteYAMLAtomic(path string, value any, perm os.FileMode) error {
data, err := yaml.Marshal(value)
if err != nil {
return fmt.Errorf("marshal yaml for %q: %w", path, err)
}
if err := WriteFileAtomic(path, data, perm); err != nil {
return fmt.Errorf("write yaml %q: %w", path, err)
}
return nil
}
// WriteFileAtomic writes bytes via same-directory temp file + atomic rename.
func WriteFileAtomic(path string, data []byte, perm os.FileMode) error {
if strings.TrimSpace(path) == "" {
return fmt.Errorf("write file: path is required")
}
dir := filepath.Dir(path)
if err := os.MkdirAll(dir, 0o755); err != nil {
return fmt.Errorf("create parent directory %q: %w", dir, err)
}
base := filepath.Base(path)
tmp, err := os.CreateTemp(dir, "."+base+".tmp-*")
if err != nil {
return fmt.Errorf("create temp file: %w", err)
}
tmpPath := tmp.Name()
removeTmp := true
defer func() {
if removeTmp {
_ = os.Remove(tmpPath)
}
}()
if _, err := tmp.Write(data); err != nil {
_ = tmp.Close()
return fmt.Errorf("write temp file: %w", err)
}
if err := tmp.Sync(); err != nil {
_ = tmp.Close()
return fmt.Errorf("sync temp file: %w", err)
}
if err := tmp.Close(); err != nil {
return fmt.Errorf("close temp file: %w", err)
}
if err := os.Chmod(tmpPath, perm); err != nil {
return fmt.Errorf("chmod temp file: %w", err)
}
if err := os.Rename(tmpPath, path); err != nil {
return fmt.Errorf("rename temp file: %w", err)
}
removeTmp = false
return nil
}
func logWriter(path string) (*os.File, io.Writer, error) {
if strings.TrimSpace(path) == "" {
return nil, io.Discard, nil
}
if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil {
return nil, nil, fmt.Errorf("create log directory for %q: %w", path, err)
}
f, err := os.Create(path)
if err != nil {
return nil, nil, fmt.Errorf("open log file %q: %w", path, err)
}
return f, f, nil
}
func closeFile(f *os.File) {
if f != nil {
_ = f.Close()
}
}
func mergeEnv(base []string, overrides map[string]string) []string {
if len(overrides) == 0 {
return base
}
kv := make(map[string]string, len(base)+len(overrides))
for _, item := range base {
k, v, ok := strings.Cut(item, "=")
if !ok {
continue
}
kv[k] = v
}
for k, v := range overrides {
kv[k] = v
}
keys := make([]string, 0, len(kv))
for k := range kv {
keys = append(keys, k)
}
sort.Strings(keys)
out := make([]string, 0, len(keys))
for _, k := range keys {
out = append(out, k+"="+kv[k])
}
return out
}

View File

@@ -0,0 +1,196 @@
package subprocess
import (
"context"
"os"
"path/filepath"
"strings"
"testing"
"time"
"gopkg.in/yaml.v3"
)
func TestRunSuccessCapturesStdoutStderr(t *testing.T) {
exe, err := os.Executable()
if err != nil {
t.Fatalf("os.Executable() error = %v", err)
}
dir := t.TempDir()
stdoutPath := filepath.Join(dir, "stdout.log")
stderrPath := filepath.Join(dir, "stderr.log")
req := RunRequest{
Executable: exe,
Args: []string{"-test.run=TestSubprocessHelper", "--", "success"},
EnvOverrides: map[string]string{"GO_WANT_SUBPROCESS_HELPER": "1", "SUBPROCESS_HELPER_STDOUT": "hello-out", "SUBPROCESS_HELPER_STDERR": "hello-err"},
StdoutLogPath: stdoutPath,
StderrLogPath: stderrPath,
}
res, err := Run(context.Background(), req)
if err != nil {
t.Fatalf("Run() error = %v", err)
}
if res.ExitCode != 0 {
t.Fatalf("ExitCode = %d, want 0", res.ExitCode)
}
stdoutBytes, err := os.ReadFile(stdoutPath)
if err != nil {
t.Fatalf("read stdout log: %v", err)
}
if !strings.Contains(string(stdoutBytes), "hello-out") {
t.Fatalf("stdout log = %q, want hello-out", string(stdoutBytes))
}
stderrBytes, err := os.ReadFile(stderrPath)
if err != nil {
t.Fatalf("read stderr log: %v", err)
}
if !strings.Contains(string(stderrBytes), "hello-err") {
t.Fatalf("stderr log = %q, want hello-err", string(stderrBytes))
}
}
func TestRunFailureReturnsUsefulError(t *testing.T) {
exe, err := os.Executable()
if err != nil {
t.Fatalf("os.Executable() error = %v", err)
}
req := RunRequest{
Executable: exe,
Args: []string{"-test.run=TestSubprocessHelper", "--", "fail"},
EnvOverrides: map[string]string{
"GO_WANT_SUBPROCESS_HELPER": "1",
},
}
res, err := Run(context.Background(), req)
if err == nil {
t.Fatal("Run() error = nil, want non-nil")
}
if res.ExitCode == 0 {
t.Fatalf("ExitCode = %d, want non-zero", res.ExitCode)
}
if !strings.Contains(err.Error(), "exit code") {
t.Fatalf("error = %q, want exit code context", err.Error())
}
if !strings.Contains(err.Error(), exe) {
t.Fatalf("error = %q, want executable context", err.Error())
}
}
func TestRunTimeout(t *testing.T) {
exe, err := os.Executable()
if err != nil {
t.Fatalf("os.Executable() error = %v", err)
}
req := RunRequest{
Executable: exe,
Args: []string{"-test.run=TestSubprocessHelper", "--", "sleep"},
EnvOverrides: map[string]string{
"GO_WANT_SUBPROCESS_HELPER": "1",
},
Timeout: 50 * time.Millisecond,
}
res, err := Run(context.Background(), req)
if err == nil {
t.Fatal("Run() error = nil, want timeout error")
}
if !res.TimedOut {
t.Fatalf("TimedOut = %v, want true", res.TimedOut)
}
if !strings.Contains(err.Error(), "timed out") {
t.Fatalf("error = %q, want timeout context", err.Error())
}
}
func TestWriteYAMLAtomic(t *testing.T) {
dir := t.TempDir()
path := filepath.Join(dir, "config.generated.yml")
if err := WriteYAMLAtomic(path, map[string]any{"name": "narratio", "stage": "merge"}, 0o644); err != nil {
t.Fatalf("WriteYAMLAtomic() error = %v", err)
}
data, err := os.ReadFile(path)
if err != nil {
t.Fatalf("read yaml: %v", err)
}
var got map[string]any
if err := yaml.Unmarshal(data, &got); err != nil {
t.Fatalf("yaml unmarshal: %v", err)
}
if got["name"] != "narratio" {
t.Fatalf("name = %#v, want narratio", got["name"])
}
}
func TestWriteYAMLAtomicOverwriteNoTempResidue(t *testing.T) {
dir := t.TempDir()
path := filepath.Join(dir, "config.generated.yml")
if err := WriteYAMLAtomic(path, map[string]any{"value": "one"}, 0o644); err != nil {
t.Fatalf("first write: %v", err)
}
if err := WriteYAMLAtomic(path, map[string]any{"value": "two"}, 0o644); err != nil {
t.Fatalf("second write: %v", err)
}
data, err := os.ReadFile(path)
if err != nil {
t.Fatalf("read yaml: %v", err)
}
if !strings.Contains(string(data), "two") {
t.Fatalf("yaml = %q, want overwritten value", string(data))
}
entries, err := os.ReadDir(dir)
if err != nil {
t.Fatalf("ReadDir() error = %v", err)
}
for _, entry := range entries {
name := entry.Name()
if strings.Contains(name, ".tmp-") {
t.Fatalf("temp file residue found: %q", name)
}
}
}
func TestSubprocessHelper(t *testing.T) {
if os.Getenv("GO_WANT_SUBPROCESS_HELPER") != "1" {
return
}
args := os.Args
mode := ""
for i := range args {
if args[i] == "--" && i+1 < len(args) {
mode = args[i+1]
break
}
}
if mode == "" {
os.Exit(2)
}
switch mode {
case "success":
_, _ = os.Stdout.WriteString(os.Getenv("SUBPROCESS_HELPER_STDOUT") + "\n")
_, _ = os.Stderr.WriteString(os.Getenv("SUBPROCESS_HELPER_STDERR") + "\n")
os.Exit(0)
case "fail":
_, _ = os.Stderr.WriteString("intentional failure\n")
os.Exit(3)
case "sleep":
time.Sleep(500 * time.Millisecond)
os.Exit(0)
default:
os.Exit(2)
}
}