fix(launcher): repeat the group SIGKILL until reaped so a missed member cannot hold Done open (#1652)

Closes #1649

## Root cause
`TestRetire_LauncherKillReachesDescendants` failure is a real (rare)
production slow path, not only a tight test window. A SIGKILL to the
launcher's process group can miss a group member that is being forked at
that instant (reproduced locally under CPU load: a `sleep 30` orphan
survived the group kill, ppid 1, same pgid). That member inherits the
launcher's stdout/stderr, so `reap()` blocks in `pumpWG.Wait()` and then
in `awaitGroupEmpty` (the exited leader is deliberately kept unreaped
while a group member runs). Done stayed open until the member exited: 30
s here, forever for a daemon. The test's `treeAlive` cannot see it
(`sleep 30` has no dir in its command line), so only the Done wait
failed.

## Fix
- `handle.killed` is set when a SIGKILL is sent (Kill, Stop escalation,
retirement).
- `repeatKillUntilReaped` (started before the pump drain, stopped after
the reap) re-sends SIGKILL every 50 ms under `reapMu` only while the
leader is unreaped, so the zombie/live leader still pins the pgid (no
reuse risk). No-op on Windows (Job Object).
- Test: `TestHandle_KillMissedMember_RepeatedUntilGroupEmpty` (fails
without the fix); the core test's Done wait widened to 15 s.

## Verification
race tests ./internal/upstream/... bare and -tags server, GOOS=windows
vet, golangci-lint (only pre-existing govet inline in
client_secret_test.go), 40x TestRetire_ under 30 CPU spinners.
Cross-reviewed with opencode Sol 6.1: round 1 ISSUES (retry must start
before pump drain; fixed), round 2 CLEAN.

Author: Dumbris

internal/upstream/core/retire_kill_test.go

@@ -119,7 +119,7 @@ func TestRetire_LauncherKillReachesDescendants(t *testing.T) {
 		"Retire kills the launcher child's whole tree")
 	select {
 	case <-h.Done():
-	case <-time.After(2 * time.Second):
+	case <-time.After(15 * time.Second):
 		t.Fatal("the killed launcher child was not reaped")
 	}
 	assert.NoError(t, h.Kill(), "Kill after exit is a no-op")

internal/upstream/launcher/kill_unix_test.go

@@ -428,3 +428,37 @@ func TestHandle_ObservationFailure_KillNotBlockedAndExitStatusKept(t *testing.T)
 	<-h3.Done()
 	require.NoError(t, h3.Wait())
 }
+
+// A SIGKILL can miss a group member that was being forked while the signal
+// was delivered (#1649). The kill was requested, so the reaper repeats it
+// while it keeps the exited leader unreaped, and Done closes instead of
+// staying open for as long as the missed member runs.
+func TestHandle_KillMissedMember_RepeatedUntilGroupEmpty(t *testing.T) {
+	dir := t.TempDir()
+	pidFile := filepath.Join(dir, "descendant.pid")
+	cmd := exec.Command("/bin/sh", "-c",
+		`sleep 300 & echo $! > "`+pidFile+`"`)
+	h, err := Spawn(context.Background(), &Spec{Cmd: cmd, Name: "bg-missed"}, zap.NewNop())
+	require.NoError(t, err)
+	hh := h.(*handle)
+	pid := readPid(t, pidFile)
+	t.Cleanup(func() { _ = syscall.Kill(pid, syscall.SIGKILL) })
+	require.Eventually(t, func() bool { return !liveNonZombie(hh.cmd.Process.Pid) },
+		10*time.Second, 20*time.Millisecond, "the launcher leader did not exit")
+	select {
+	case <-h.Done():
+		t.Fatal("control: Done closed while the descendant runs")
+	case <-time.After(300 * time.Millisecond):
+	}
+
+	// The member keeps the launcher's output pipes, so the pumps cannot finish
+	// either. The kill was requested but its signal missed the descendant.
+	hh.killed.Store(true)
+	select {
+	case <-h.Done():
+	case <-time.After(10 * time.Second):
+		t.Fatal("a requested kill was not repeated to the member it missed")
+	}
+	require.Eventually(t, func() bool { return !liveNonZombie(pid) },
+		5*time.Second, 20*time.Millisecond, "the missed descendant must be killed")
+}

internal/upstream/launcher/launcher.go

@@ -246,6 +246,12 @@ type handle struct {
 	reapMu sync.Mutex
 	reaped bool
 
+	// killed is set once a SIGKILL was sent to the group (Unix). A member
+	// forked while that signal was delivered can be missed by it, so the
+	// reaper, while it keeps the exited leader unreaped, repeats the kill
+	// until the leader is reaped (repeatKillUntilReaped).
+	killed atomic.Bool
+
 	stopOnce sync.Once
 	stopErr  error
 
@@ -356,6 +362,7 @@ func (h *handle) reap() {
 	// exec.Cmd doc: "it is thus incorrect to call Wait before all reads from the
 	// pipe have completed." Pumps naturally finish when the child exits and the
 	// kernel propagates EOF — we don't need Wait() to reach that state.
+	stopRepeat := h.repeatKillUntilReaped()
 	h.pumpWG.Wait()
 
 	// Reap the leader under reapMu (review round 4). cmd.Wait releases the
@@ -392,6 +399,7 @@ func (h *handle) reap() {
 	} else {
 		exitCode, err = h.reapUnobserved()
 	}
+	stopRepeat()
 	if hook := afterReapHook.Load(); hook != nil {
 		(*hook)(h)
 	}

internal/upstream/launcher/launcher_unix.go

@@ -76,6 +76,9 @@ func (h *handle) signalUnreaped(sig syscall.Signal) error {
 	if h.reaped {
 		return nil
 	}
+	if sig == syscall.SIGKILL {
+		h.killed.Store(true)
+	}
 	return h.signalLocked(sig)
 }
 
@@ -164,6 +167,49 @@ func (h *handle) awaitGroupEmpty() {
 	}
 }
 
+// repeatKillInterval is how often repeatKillUntilReaped re-sends SIGKILL.
+const repeatKillInterval = 50 * time.Millisecond
+
+// repeatKillUntilReaped runs for the whole of reap and returns a stop
+// function. Once a SIGKILL was requested (Kill, Stop's escalation or a
+// retirement) it repeats the signal to the group until the leader is reaped.
+// A member forked while the first signal was delivered is not in the group
+// yet when the kernel walks it, so it survives that kill; without a repeat
+// the exited leader stays unreaped, and Done stays open, for as long as that
+// member runs (forever for a daemon). It must run beside the reaper's pump
+// drain too: the missed member holds the output pipes, so the pumps (and
+// with them awaitGroupEmpty) cannot finish until it dies. Every signal is
+// sent under reapMu with the leader not reaped, so the leader, alive or a
+// zombie, still pins the group ID.
+func (h *handle) repeatKillUntilReaped() (stop func()) {
+	quit := make(chan struct{})
+	finished := make(chan struct{})
+	go func() {
+		defer close(finished)
+		t := time.NewTicker(repeatKillInterval)
+		defer t.Stop()
+		for {
+			select {
+			case <-quit:
+				return
+			case <-t.C:
+			}
+			if !h.killed.Load() {
+				continue
+			}
+			h.reapMu.Lock()
+			if !h.reaped {
+				_ = h.signalLocked(syscall.SIGKILL)
+			}
+			h.reapMu.Unlock()
+		}
+	}()
+	return func() {
+		close(quit)
+		<-finished
+	}
+}
+
 // killGroupBeforeReap runs in the reaper under reapMu, after awaitGroupEmpty
 // and before cmd.Wait reaps the leader. The zombie leader still holds the
 // group ID, so the SIGKILL reaches only what is left of our own group:

internal/upstream/launcher/launcher_windows.go

@@ -133,3 +133,7 @@ func (h *handle) killGroupBeforeReap() {}
 // awaitGroupEmpty is a no-op on Windows (no process group; see
 // killGroupBeforeReap).
 func (h *handle) awaitGroupEmpty() {}
+
+// repeatKillUntilReaped is a no-op on Windows: the Job Object kill reaches
+// every member, including ones created while it was delivered.
+func (h *handle) repeatKillUntilReaped() (stop func()) { return func() {} }