Close piped processes on cancellation

docs/roadmap/implementation.md

@@ -40,7 +40,7 @@ the agent policy; an intermediate compilation dependency is recorded accurately.
 | 2 | A02, A12: common Git mechanics | Complete — focused Git, candidate, identity, and public build/publication/delivery tests passed; A02 regression evidence follows in Stage 3. |
 | 3 | A02: source identity regression evidence | Complete — commit, tree, and blob replacement regressions and public publication, build, and delivery cases passed. |
 | 4 | A03: buffered command teardown | Complete — inherited-pipe, cancellation, healthy-command, native build, and publication recovery checks passed. |
-| 5 | A03, A18: piped lifetime and queued cancellation | Pending |
+| 5 | A03, A18: piped lifetime and queued cancellation | Complete — process and Git lifecycle tests passed under the race detector; public preparation cancellation passed. |
 | 6 | A04 | Pending |
 | 7 | A05 | Pending |
 | 8 | A06, A14 | Pending |

internal/git/blob_reader.go

@@ -51,6 +51,9 @@ func (reader *BlobReader) Read(ctx context.Context, objectID string) ([]byte, er
 
 	reader.mu.Lock()
 	defer reader.mu.Unlock()
+	if err := ctx.Err(); err != nil {
+		return nil, fmt.Errorf("read blob %s: %w", objectID, err)
+	}
 	if reader.terminal != nil {
 		return nil, reader.terminal
 	}

internal/git/blob_reader_test.go

@@ -10,6 +10,7 @@ import (
 	"os/exec"
 	"path/filepath"
 	"strings"
+	"sync"
 	"testing"
 	"time"
 )
@@ -156,6 +157,46 @@ exec sleep 60`)
 	}
 }
 
+func TestBlobReaderCanceledQueuedReadKeepsSessionUsable(t *testing.T) {
+	fixture := newRemoteFixture(t)
+	object := gitText(t, fixture.work, "rev-parse", "HEAD:source.txt")
+	reader := NewBlobReader(t.Context(), fixture.work)
+	defer reader.Close()
+	if body, err := reader.Read(t.Context(), object); err != nil || string(body) != "candidate" {
+		t.Fatalf("first read = %q, %v", body, err)
+	}
+	reader.mu.Lock()
+	ctx, cancel := context.WithCancel(t.Context())
+	observed := make(chan struct{})
+	queued := &observedContext{Context: ctx, observed: observed}
+	result := make(chan error, 1)
+	go func() {
+		_, err := reader.Read(queued, object)
+		result <- err
+	}()
+	<-observed
+	cancel()
+	reader.mu.Unlock()
+	if err := <-result; !errors.Is(err, context.Canceled) {
+		t.Fatalf("queued read error = %v", err)
+	}
+	if body, err := reader.Read(t.Context(), object); err != nil || string(body) != "candidate" {
+		t.Fatalf("later read = %q, %v", body, err)
+	}
+}
+
+type observedContext struct {
+	context.Context
+	once     sync.Once
+	observed chan struct{}
+}
+
+func (ctx *observedContext) Err() error {
+	err := ctx.Context.Err()
+	ctx.once.Do(func() { close(ctx.observed) })
+	return err
+}
+
 func TestBlobReaderPreservesGitFailureAndStderr(t *testing.T) {
 	installFakeGit(t, `printf 'fixture Git failure\n' >&2
 exit 23`)

internal/process/process.go

@@ -55,6 +55,8 @@ type PipedProcess struct {
 	stdout   io.ReadCloser
 	parent   context.Context
 	cancel   context.CancelFunc
+	stopIO   func() bool
+	ioDone   <-chan struct{}
 	waitOnce sync.Once
 	waitErr  error
 }
@@ -80,6 +82,9 @@ func (child *PipedProcess) Cancel() {
 func (child *PipedProcess) Wait() error {
 	child.waitOnce.Do(func() {
 		err := child.command.Wait()
+		if !child.stopIO() {
+			<-child.ioDone
+		}
 		child.cancel()
 		if err == nil {
 			return
@@ -121,12 +126,20 @@ func StartPiped(ctx context.Context, invocation Invocation) (*PipedProcess, erro
 		cancel()
 		return nil, fmt.Errorf("start %q: %w", invocation.Args[0], err)
 	}
+	done := make(chan struct{})
+	stopIO := context.AfterFunc(childContext, func() {
+		defer close(done)
+		_ = stdin.Close()
+		_ = stdout.Close()
+	})
 	return &PipedProcess{
 		command: command,
 		stdin:   stdin,
 		stdout:  stdout,
 		parent:  ctx,
 		cancel:  cancel,
+		stopIO:  stopIO,
+		ioDone:  done,
 	}, nil
 }
 

internal/process/process_test.go

@@ -260,6 +260,52 @@ func TestStartPipedCancellationAfterReadinessReapsChild(t *testing.T) {
 	}
 }
 
+func TestStartPipedCancellationInterruptsInheritedOutputDrain(t *testing.T) {
+	ready := filepath.Join(t.TempDir(), "descendant")
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	child, err := StartPiped(ctx, Invocation{
+		Args: []string{helperExecutable(t), "-test.run=TestProcessHelper", "--", "unused", "exit-descendant"},
+		Env:  map[string]string{"RELEASEKIT_PROCESS_HELPER": "1", "RELEASEKIT_DESCENDANT_READY": ready},
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	_ = child.Stdin().Close()
+	drained := make(chan error, 1)
+	go func() {
+		_, err := io.Copy(io.Discard, child.Stdout())
+		drained <- err
+	}()
+	deadline := time.After(10 * time.Second)
+	var pid int
+	for pid == 0 {
+		if data, err := os.ReadFile(ready); err == nil {
+			pid, _ = strconv.Atoi(string(data))
+		}
+		select {
+		case <-deadline:
+			t.Fatal("descendant did not become ready")
+		default:
+			time.Sleep(10 * time.Millisecond)
+		}
+	}
+	t.Cleanup(func() {
+		if descendant, err := os.FindProcess(pid); err == nil {
+			_ = descendant.Kill()
+		}
+	})
+	cancel()
+	select {
+	case <-drained:
+	case <-time.After(10 * time.Second):
+		t.Fatal("inherited stdout did not close after cancellation")
+	}
+	if err := child.Wait(); err != nil && !errors.Is(err, context.Canceled) {
+		t.Fatalf("wait after cancellation: %v", err)
+	}
+}
+
 func TestStartPipedCancelDoesNotCancelParentContext(t *testing.T) {
 	ctx := context.Background()
 	child, err := StartPiped(ctx, Invocation{

preparation_test.go

@@ -6,12 +6,58 @@ import (
 	"os"
 	"os/exec"
 	"path/filepath"
+	"strconv"
 	"strings"
 	"testing"
+	"time"
 
 	releasekit "go.maximumdirect.net/releasekit"
 )
 
+func TestPrepareBuildDependenciesCancellationReleasesInheritedOutput(t *testing.T) {
+	directory := t.TempDir()
+	ready := filepath.Join(t.TempDir(), "descendant")
+	gitPath := filepath.Join(t.TempDir(), "git")
+	if err := os.WriteFile(gitPath, []byte("#!/bin/sh\nsleep 60 &\nprintf '%s' \"$!\" > \"$RELEASEKIT_PREPARE_DESCENDANT\"\nwait\n"), 0o700); err != nil {
+		t.Fatal(err)
+	}
+	t.Setenv("PATH", filepath.Dir(gitPath)+string(os.PathListSeparator)+os.Getenv("PATH"))
+	t.Setenv("RELEASEKIT_PREPARE_DESCENDANT", ready)
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	result := make(chan error, 1)
+	go func() {
+		result <- releasekit.PrepareBuildDependencies(ctx, releasekit.DependencyPreparationRequest{RepositoryDir: directory})
+	}()
+	deadline := time.After(10 * time.Second)
+	var pid int
+	for pid == 0 {
+		if data, err := os.ReadFile(ready); err == nil {
+			pid, _ = strconv.Atoi(string(data))
+		}
+		select {
+		case <-deadline:
+			t.Fatal("preparation descendant did not start")
+		default:
+			time.Sleep(10 * time.Millisecond)
+		}
+	}
+	t.Cleanup(func() {
+		if descendant, err := os.FindProcess(pid); err == nil {
+			_ = descendant.Kill()
+		}
+	})
+	cancel()
+	select {
+	case err := <-result:
+		if !errors.Is(err, context.Canceled) {
+			t.Fatalf("preparation error = %v, want cancellation", err)
+		}
+	case <-time.After(10 * time.Second):
+		t.Fatal("preparation did not return after cancellation")
+	}
+}
+
 func TestPrepareTaggedCheckoutFetchesExpectedCommitAndDetaches(t *testing.T) {
 	repository := newTaggedDeliveryConsumer(t)
 	tagCommit := publishPreparationTag(t, repository)