fix(tools): wait for pipe readers before reaping detached bash shells (#716)

## Summary

Detaching a streaming bash command to the background — chat `ctrl+b`, or
directexec `!cmd &` — could truncate its trailing output. The
pipe-reader goroutines drain the command's `stdout`/`stderr` into the
shared shell buffer, but the live `*exec.Cmd` was handed to the
`jobs.Supervisor`, whose `shellJob.Run` called `cmd.Wait()` in a
separate goroutine with no coordination. Per `os/exec`, `Wait` closes
the pipes on process exit, so output still buffered in the pipe at exit
was lost. The `time.Sleep(20ms)` band-aid ran at detach time, unrelated
to exit time, so it never addressed the race.

## Fix

`shellJob.Run` now waits for the pipe readers to reach EOF before
`cmd.Wait()`, mirroring the already-correct foreground path (`wg.Wait()`
then `cmd.Wait()`). The wait `select`s on `ctx.Done()` so
`WindStop`/`Stop()` can't wedge if a grandchild holds the pipe open (a
detached shell has no timeout). Both detach callers thread a
readers-done channel through `DetachToBackground`.

- `internal/domain/shell.go` — `ReadersDone` on `BackgroundShell`;
widened `DetachToBackground`
- `internal/services/background_shell_service.go` — store `readersDone`
on the shell
- `internal/services/jobs/shell_job.go` — cancellable readers-done wait
before `cmd.Wait()`
- `internal/agent/tools/bash.go`, `internal/services/directexec/bash.go`
— pass the signal; drop the 20ms sleep
- regenerated `fake_background_shell_service.go`

## Testing

- Two regression tests in `shell_job_test.go`: ordering (Run parks until
the readers drain, even after the process has exited) and the
cancellable escape (`WindStop` still reaps when the readers never
signal). Confirmed the latter deadlocks without the `ctx.Done()` case.
- `task build`, `go test -race` on the affected packages, full `task
test`, and `task lint` (0 issues) — all clean.

Closes #696

Author: edenreich

internal/agent/tools/bash.go

@@ -292,9 +292,7 @@ func (t *BashTool) executeBashWithStreaming(ctx context.Context, cmd *exec.Cmd,
 			detached = true
 			detachedMux.Unlock()
 
-			time.Sleep(20 * time.Millisecond)
-
-			shellID, err := t.backgroundShellService.DetachToBackground(ctx, cmd, result.Command, outputBuffer)
+			shellID, err := t.backgroundShellService.DetachToBackground(ctx, cmd, result.Command, outputBuffer, done)
 			if err != nil {
 				logger.Debug("bash: DetachToBackground failed", "error", err)
 				result.ExitCode = -1

internal/domain/shell.go

@@ -38,6 +38,8 @@ type BackgroundShell struct {
 	OutputBuffer OutputRingBuffer
 	CancelFunc   context.CancelFunc
 	ReadOffset   int64
+
+	ReadersDone <-chan struct{}
 }
 
 // OutputRingBuffer defines the interface for the circular output buffer.
@@ -110,8 +112,10 @@ func NewShellInfo(shell *BackgroundShell) *ShellInfo {
 
 // BackgroundShellService defines the interface for managing background shells
 type BackgroundShellService interface {
-	// DetachToBackground moves a running command to background
-	DetachToBackground(ctx context.Context, cmd *exec.Cmd, command string, outputBuffer OutputRingBuffer) (string, error)
+	// DetachToBackground moves a running command to background. readersDone, when
+	// non-nil, is closed once the caller's pipe readers reach EOF; the supervisor
+	// waits on it before reaping so trailing output is not truncated.
+	DetachToBackground(ctx context.Context, cmd *exec.Cmd, command string, outputBuffer OutputRingBuffer, readersDone <-chan struct{}) (string, error)
 
 	// GetShellOutput retrieves output from a shell
 	GetShellOutput(shellID string, fromOffset int64) (string, int64, ShellState, error)

internal/services/background_shell_service.go

@@ -51,6 +51,7 @@ func (s *BackgroundShellService) DetachToBackground(
 	cmd *exec.Cmd,
 	command string,
 	outputBuffer domain.OutputRingBuffer,
+	readersDone <-chan struct{},
 ) (string, error) {
 	if !s.config.Tools.Bash.BackgroundShells.Enabled {
 		return "", fmt.Errorf("background shells are disabled in configuration")
@@ -66,6 +67,7 @@ func (s *BackgroundShellService) DetachToBackground(
 		State:        domain.ShellStateRunning,
 		OutputBuffer: outputBuffer,
 		ReadOffset:   0,
+		ReadersDone:  readersDone,
 	}
 
 	if err := s.shellTracker.Add(shell); err != nil {

internal/services/directexec/bash.go

@@ -367,18 +367,25 @@ func (s *Service) executeBashCommandInBackground(commandText, command string) te
 			}
 		}()
 
+		done := make(chan struct{})
+		go func() {
+			wg.Wait()
+			close(done)
+		}()
+
 		shellID, err := s.backgroundShellService.DetachToBackground(
 			ctx,
 			cmd,
 			command,
 			outputBuffer,
+			done,
 		)
 
 		if err != nil {
 			return
 		}
 
-		wg.Wait()
+		<-done
 
 		assistantEntry := domain.ConversationEntry{
 			Message: sdk.Message{

internal/services/jobs/shell_job.go

@@ -40,7 +40,17 @@ func (j *shellJob) Meta() domain.JobMeta {
 }
 
 // Run waits for the shell process to exit and records the outcome on the shell.
+// It first waits for the pipe readers to drain (ReadersDone) so Cmd.Wait doesn't
+// close the pipes mid-read and lose output; the wait honours ctx so a kill or
+// shutdown can't wedge when a grandchild is holding the pipe open.
 func (j *shellJob) Run(ctx context.Context, _ func(domain.JobSignal)) domain.ToolExecutionResult {
+	if j.shell.ReadersDone != nil {
+		select {
+		case <-j.shell.ReadersDone:
+		case <-ctx.Done():
+		}
+	}
+
 	waitErr := j.shell.Cmd.Wait()
 
 	now := time.Now()
@@ -49,7 +59,6 @@ func (j *shellJob) Run(ctx context.Context, _ func(domain.JobSignal)) domain.Too
 
 	switch {
 	case ctx.Err() != nil:
-		// Terminated by Wind(WindStop) / supervisor shutdown rather than finishing.
 		code := -1
 		j.shell.ExitCode = &code
 		j.shell.State = domain.ShellStateCancelled

internal/services/jobs/shell_job_test.go

@@ -45,6 +45,71 @@ func waitTerminal(t *testing.T, sup *Supervisor, id string) domain.JobStatus {
 	return ""
 }
 
+// isTerminal reports whether the job is currently in a terminal state without
+// blocking, for asserting that a job has NOT yet finished.
+func isTerminal(sup *Supervisor, id string) bool {
+	for _, j := range sup.Snapshot() {
+		if j.Meta.ID == id {
+			return j.Status.IsTerminal()
+		}
+	}
+	return false
+}
+
+// TestShellJob_WaitsForReadersDoneBeforeReaping asserts Run does not call
+// Cmd.Wait until the pipe readers have drained (ReadersDone closed), even after
+// the process has already exited — otherwise Wait closes the pipes mid-drain and
+// truncates trailing output.
+func TestShellJob_WaitsForReadersDoneBeforeReaping(t *testing.T) {
+	tracker := utils.NewShellTracker(10)
+	sup := NewSupervisor(&domainmocks.FakeMessageQueue{}, &domainmocks.FakeConversationRepository{}, nil)
+	defer sup.Stop()
+
+	// `true` exits immediately; the job must still park on ReadersDone.
+	shell := startShell(t, "s-drain", "true")
+	readersDone := make(chan struct{})
+	shell.ReadersDone = readersDone
+	_ = tracker.Add(shell)
+	sup.Submit(NewShellJob(shell, tracker))
+
+	time.Sleep(50 * time.Millisecond)
+	if isTerminal(sup, "s-drain") {
+		t.Fatal("shell reaped before the pipe readers signalled done")
+	}
+
+	close(readersDone)
+	if got := waitTerminal(t, sup, "s-drain"); got != domain.JobCompleted {
+		t.Fatalf("status = %s, want completed", got)
+	}
+	if shell.State != domain.ShellStateCompleted {
+		t.Fatalf("shell state = %s, want completed", shell.State)
+	}
+}
+
+// TestShellJob_WindStopUnblocksReadersWait guards the cancellable-wait escape:
+// when ReadersDone never closes (e.g. a grandchild holds the pipe open),
+// Wind(WindStop) must still reap the job via ctx cancellation rather than
+// wedging Run — which would also hang Supervisor.Stop().
+func TestShellJob_WindStopUnblocksReadersWait(t *testing.T) {
+	tracker := utils.NewShellTracker(10)
+	sup := NewSupervisor(&domainmocks.FakeMessageQueue{}, &domainmocks.FakeConversationRepository{}, nil)
+	defer sup.Stop()
+
+	shell := startShell(t, "s-stuck", "sleep", "60")
+	shell.ReadersDone = make(chan struct{}) // never closed
+	_ = tracker.Add(shell)
+	sup.Submit(NewShellJob(shell, tracker))
+
+	time.Sleep(20 * time.Millisecond)
+	if err := sup.Wind("s-stuck", domain.WindStop); err != nil {
+		t.Fatalf("Wind: %v", err)
+	}
+	waitTerminal(t, sup, "s-stuck")
+	if shell.State != domain.ShellStateCancelled {
+		t.Fatalf("shell state = %s, want cancelled", shell.State)
+	}
+}
+
 func TestShellJob_CompletesAndNotifies(t *testing.T) {
 	tracker := utils.NewShellTracker(10)
 	queue := &domainmocks.FakeMessageQueue{}