From 04283e99d409d958533583eb19a9e167cb7caba0 Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Mon, 31 Aug 2026 23:48:49 +0000 Subject: [PATCH] Separate subprocess execution lifecycle from policy --- docs/internal/adapters.md | 4 +- docs/roadmap/implementation.md | 2 +- internal/adapters/subprocess/diagnostics.go | 6 +- internal/adapters/subprocess/doc.go | 4 +- internal/adapters/subprocess/execution.go | 90 +++++++++++++++++++++ internal/adapters/subprocess/run.go | 71 ++++------------ internal/adapters/subprocess/run_test.go | 6 +- 7 files changed, 120 insertions(+), 63 deletions(-) create mode 100644 internal/adapters/subprocess/execution.go diff --git a/docs/internal/adapters.md b/docs/internal/adapters.md index f8aabc1..d2fd007 100644 --- a/docs/internal/adapters.md +++ b/docs/internal/adapters.md @@ -71,7 +71,9 @@ configured filesystem secrets before adapter initialization. - Constructor errors fail stage execution setup early. - Runtime adapter errors propagate to stage code and then manifest failure handling. - Subprocess adapters persist stage logs/generated configs through stage-managed paths. -- Shared subprocess execution starts an owned process group on Linux/macOS or a +- Shared subprocess execution keeps private lifecycle mechanics separate from + Narratio's environment, diagnostics, redaction, and error policy. It starts + an owned process group on Linux/macOS or a kill-on-close job object on Windows. Every terminal path disposes of that owned tree before returning. After a natural leader exit, Unix checks for remaining group members and uses bounded graceful then forceful termination; diff --git a/docs/roadmap/implementation.md b/docs/roadmap/implementation.md index dccca35..a5a2a18 100644 --- a/docs/roadmap/implementation.md +++ b/docs/roadmap/implementation.md @@ -190,7 +190,7 @@ Go 1.26-only process APIs. ## Stage 2 — Separate Subprocess Lifecycle Mechanics From Policy -**Status: Pending** +**Status: Completed** ### Goal diff --git a/internal/adapters/subprocess/diagnostics.go b/internal/adapters/subprocess/diagnostics.go index f1e47b2..a56eccc 100644 --- a/internal/adapters/subprocess/diagnostics.go +++ b/internal/adapters/subprocess/diagnostics.go @@ -72,7 +72,7 @@ type logWriters struct { Stdout io.Writer Stderr io.Writer - limits chan *captureLimitError + limits chan error mu sync.Mutex limit *captureLimitError stdout *diagnosticWriter @@ -92,7 +92,7 @@ type diagnosticWriter struct { } func openLogWriters(stdoutPath, stderrPath, owner string, sensitiveValues []string) (*logWriters, error) { - logs := &logWriters{limits: make(chan *captureLimitError, 1)} + logs := &logWriters{limits: make(chan error, 1)} cleanStdout := cleanLogPath(stdoutPath) cleanStderr := cleanLogPath(stderrPath) @@ -221,7 +221,7 @@ func (w *diagnosticWriter) reachLimit() error { return limit } -func (l *logWriters) Limits() <-chan *captureLimitError { return l.limits } +func (l *logWriters) Limits() <-chan error { return l.limits } func (l *logWriters) Limit() *captureLimitError { l.mu.Lock() diff --git a/internal/adapters/subprocess/doc.go b/internal/adapters/subprocess/doc.go index ded1ac8..f07f270 100644 --- a/internal/adapters/subprocess/doc.go +++ b/internal/adapters/subprocess/doc.go @@ -1,2 +1,4 @@ -// Package subprocess provides reusable process execution and generated-config helpers. +// Package subprocess provides internal process execution and Narratio-specific +// subprocess policy. Its private execution mechanism owns process lifecycle; +// the Run boundary owns environments, diagnostics, redaction, and errors. package subprocess diff --git a/internal/adapters/subprocess/execution.go b/internal/adapters/subprocess/execution.go new file mode 100644 index 0000000..d526f4d --- /dev/null +++ b/internal/adapters/subprocess/execution.go @@ -0,0 +1,90 @@ +package subprocess + +import ( + "context" + "os/exec" + "time" +) + +type executionReason uint8 + +const ( + executionNotStarted executionReason = iota + executionNaturalExit + executionCanceled + executionDeadline + executionAborted +) + +type executionOutcome struct { + reason executionReason + startedAt time.Time + completedAt time.Time + exitCode int + startErr error + waitErr error + contextErr error + abortErr error + cleanupErr error +} + +func executeOwnedCommand(ctx context.Context, cmd *exec.Cmd, abort <-chan error) (outcome executionOutcome) { + outcome = executionOutcome{startedAt: time.Now().UTC(), exitCode: -1} + defer func() { outcome.completedAt = time.Now().UTC() }() + if outcome.contextErr = ctx.Err(); outcome.contextErr != nil { + outcome.reason = executionNotStarted + return outcome + } + tree, err := newOwnedProcessTree() + if err != nil { + outcome.reason, outcome.startErr = executionNotStarted, err + return outcome + } + defer func() { outcome.cleanupErr = joinErrors(outcome.cleanupErr, tree.Dispose()) }() + cmd.WaitDelay = forcefulTerminationWait + if outcome.startErr = tree.Start(cmd); outcome.startErr != nil { + outcome.reason = executionNotStarted + return outcome + } + waitCh := make(chan error, 1) + go func() { waitCh <- cmd.Wait() }() + select { + case outcome.waitErr = <-waitCh: + outcome.reason = executionNaturalExit + case <-ctx.Done(): + outcome.contextErr = ctx.Err() + if outcome.contextErr == context.DeadlineExceeded { + outcome.reason = executionDeadline + } else { + outcome.reason = executionCanceled + } + outcome.cleanupErr = terminateAndReap(tree, waitCh) + case outcome.abortErr = <-abort: + outcome.reason = executionAborted + outcome.cleanupErr = terminateAndReap(tree, waitCh) + } + if cmd.ProcessState != nil { + outcome.exitCode = cmd.ProcessState.ExitCode() + } + return outcome +} + +func terminateAndReap(tree ownedProcessTree, waitCh <-chan error) error { + cleanupErr := tree.TerminateGracefully() + timer := time.NewTimer(gracefulTerminationWait) + defer timer.Stop() + select { + case <-waitCh: + return joinErrors(cleanupErr, tree.TerminateForcefully()) + case <-timer.C: + } + cleanupErr = joinErrors(cleanupErr, tree.TerminateForcefully()) + forcefulTimer := time.NewTimer(forcefulTerminationWait) + defer forcefulTimer.Stop() + select { + case <-waitCh: + return cleanupErr + case <-forcefulTimer.C: + return joinErrors(cleanupErr, context.DeadlineExceeded) + } +} diff --git a/internal/adapters/subprocess/run.go b/internal/adapters/subprocess/run.go index 4709684..3dddf0a 100644 --- a/internal/adapters/subprocess/run.go +++ b/internal/adapters/subprocess/run.go @@ -60,63 +60,23 @@ func Run(ctx context.Context, req RunRequest) (RunResult, error) { } defer logs.Close() - tree, err := newOwnedProcessTree() - if err != nil { - return RunResult{}, fmt.Errorf("prepare owned subprocess tree: %w", err) - } - cmd := exec.Command(req.Executable, req.Args...) cmd.Dir = req.WorkingDir cmd.Env = childEnv cmd.Stdout = logs.Stdout cmd.Stderr = logs.Stderr - // Streaming capture uses pipes. Bound their lifetime when a leader exits - // while a descendant still holds a stream descriptor. - cmd.WaitDelay = forcefulTerminationWait - - started := time.Now().UTC() - result := RunResult{ - ExitCode: -1, - StartedAt: started, - StdoutLogPath: req.StdoutLogPath, - StderrLogPath: req.StderrLogPath, - } - - if err := runCtx.Err(); err != nil { - result.CompletedAt = time.Now().UTC() - result.Duration = result.CompletedAt.Sub(result.StartedAt) - return result, fmt.Errorf("command was not started: %w", err) - } - - if err := tree.Start(cmd); 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) - } - - waitCh := make(chan error, 1) - go func() { waitCh <- cmd.Wait() }() - - waitErr, ctxErr, captureLimit, cleanupErr := waitForOwnedCommand(runCtx, tree, waitCh, logs.Limits()) - cleanupErr = joinErrors(cleanupErr, tree.Dispose()) - cleanupErr = joinErrors(cleanupErr, logs.Flush()) + outcome := executeOwnedCommand(runCtx, cmd, logs.Limits()) + cleanupErr := joinErrors(outcome.cleanupErr, logs.Flush()) + captureLimit, _ := outcome.abortErr.(*captureLimitError) if captureLimit == nil { captureLimit = logs.Limit() } - result.CompletedAt = time.Now().UTC() + result := RunResult{ExitCode: outcome.exitCode, StartedAt: outcome.startedAt, CompletedAt: outcome.completedAt, StdoutLogPath: req.StdoutLogPath, StderrLogPath: req.StderrLogPath} result.Duration = result.CompletedAt.Sub(result.StartedAt) - if cmd.ProcessState != nil { - result.ExitCode = cmd.ProcessState.ExitCode() - } + result.TimedOut = outcome.reason == executionDeadline + result.Canceled = outcome.reason == executionCanceled - if ctxErr == context.DeadlineExceeded { - result.TimedOut = true - } - if ctxErr == context.Canceled && !result.TimedOut { - result.Canceled = true - } - - if waitErr == nil && ctxErr == nil && cleanupErr == nil { + if outcome.reason == executionNaturalExit && outcome.waitErr == nil && cleanupErr == nil { return result, nil } @@ -124,25 +84,28 @@ func Run(ctx context.Context, req RunRequest) (RunResult, error) { diagnostics := buildDiagnostics(req, result, stderrTail) if captureLimit != nil { - if cause := joinErrors(waitErr, cleanupErr); cause != nil { + if cause := joinErrors(outcome.waitErr, cleanupErr); cause != nil { return result, fmt.Errorf("%w (%s): %w", captureLimit, diagnostics, cause) } return result, fmt.Errorf("%w (%s)", captureLimit, diagnostics) } if result.TimedOut { - return result, fmt.Errorf("command timed out after %s (%s): %w", req.Timeout, diagnostics, joinErrors(ctxErr, waitErr, cleanupErr)) + return result, fmt.Errorf("command timed out after %s (%s): %w", req.Timeout, diagnostics, joinErrors(outcome.contextErr, outcome.waitErr, cleanupErr)) } if result.Canceled { - return result, fmt.Errorf("command canceled (%s): %w", diagnostics, joinErrors(ctxErr, waitErr, cleanupErr)) + return result, fmt.Errorf("command canceled (%s): %w", diagnostics, joinErrors(outcome.contextErr, outcome.waitErr, cleanupErr)) } - if exitErr, ok := waitErr.(*exec.ExitError); ok { - return result, fmt.Errorf("command failed with exit code %d (%s): %w", exitErr.ExitCode(), diagnostics, joinErrors(waitErr, cleanupErr)) + if exitErr, ok := outcome.waitErr.(*exec.ExitError); ok { + return result, fmt.Errorf("command failed with exit code %d (%s): %w", exitErr.ExitCode(), diagnostics, joinErrors(outcome.waitErr, cleanupErr)) } if cleanupErr != nil { - return result, fmt.Errorf("command cleanup failed (%s): %w", diagnostics, joinErrors(waitErr, cleanupErr)) + return result, fmt.Errorf("command cleanup failed (%s): %w", diagnostics, joinErrors(outcome.waitErr, cleanupErr)) } - return result, fmt.Errorf("command failed to run (%s): %w", diagnostics, joinErrors(waitErr, cleanupErr)) + if outcome.startErr != nil { + return result, fmt.Errorf("start command %q with args %v: %w", req.Executable, req.Args, outcome.startErr) + } + return result, fmt.Errorf("command failed to run (%s): %w", diagnostics, joinErrors(outcome.waitErr, outcome.contextErr, outcome.abortErr, cleanupErr)) } // WriteYAMLAtomic marshals value as YAML and atomically writes it to path. diff --git a/internal/adapters/subprocess/run_test.go b/internal/adapters/subprocess/run_test.go index 5512c59..1ab8f33 100644 --- a/internal/adapters/subprocess/run_test.go +++ b/internal/adapters/subprocess/run_test.go @@ -382,7 +382,7 @@ func TestStreamRedactorHandlesSplitAndOverlappingSecrets(t *testing.T) { } func TestDiagnosticWriterHonorsExactLimitAndCapPlusOne(t *testing.T) { - exactLogs := &logWriters{limits: make(chan *captureLimitError, 1)} + exactLogs := &logWriters{limits: make(chan error, 1)} var exactOutput bytes.Buffer exact := newDiagnosticWriter(exactLogs, "stdout", "test", &exactOutput, 5, nil) if _, err := exact.Write([]byte("abcde")); err != nil { @@ -398,7 +398,7 @@ func TestDiagnosticWriterHonorsExactLimitAndCapPlusOne(t *testing.T) { t.Fatal("exact write recorded a capture limit") } - cappedLogs := &logWriters{limits: make(chan *captureLimitError, 1)} + cappedLogs := &logWriters{limits: make(chan error, 1)} var cappedOutput bytes.Buffer capped := newDiagnosticWriter(cappedLogs, "stderr", "test", &cappedOutput, 5, nil) if _, err := capped.Write([]byte("abcdef")); err == nil { @@ -416,7 +416,7 @@ func TestDiagnosticWriterHonorsExactLimitAndCapPlusOne(t *testing.T) { } func TestDiagnosticWriterRetainsBoundedRedactedTail(t *testing.T) { - logs := &logWriters{limits: make(chan *captureLimitError, 1)} + logs := &logWriters{limits: make(chan error, 1)} var output bytes.Buffer secret := "credential-value" writer := newDiagnosticWriter(logs, "stderr", "test", &output, 16*1024, []string{secret})