processgroup: a child counts only while its parent is still the same

Review fix: the macOS walk now reads a parent's listed children first and
re-reads the parent after them. A pid is never handed on while its process
lives, so a parent with the same start time after its children were read was
their parent while they were read; a parent and child whose pids both changed
hands just after the parent's own re-read can no longer bring a stranger's busy
child into the job's reading.

The partial-sum regression now starts from one complete, quiet reading, so it
also fails if a partial reading ever keeps that earlier baseline.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Author: AbirAbbas

internal/processgroup/usage_darwin.go

@@ -44,10 +44,12 @@ func (g Group) Usage() (Usage, bool) {
 // subtreeUsage walks the tree below root through each process's children, so
 // a child that changed process groups is still the job's.
 //
-// A PARENT IS RE-READ AFTER ITS CHILDREN ARE LISTED. A queued process can exit
-// and its pid be handed to another process before its children are asked for;
-// the children of that stranger would pass the parent-pid check. Comparing the
-// parent's start time before and after the listing keeps them out.
+// A PARENT IS RE-READ AFTER ITS CHILDREN ARE READ. A queued process can exit
+// and its pid be handed to another process while its children are listed or
+// read; the children of that stranger would pass the parent-pid check. A pid is
+// never handed on while its process lives, so a parent that reads with the same
+// start time after its children were read was their parent while they were
+// read, and only then are they admitted.
 //
 // A listing that filled even the largest buffer, a walk that reached
 // [usageWalkLimit], and a process that could not be read for any reason other
@@ -82,19 +84,7 @@ func subtreeUsage(root int) (Usage, bool) {
 		default:
 			return Usage{}, false
 		}
-		again, err := usageReader.task(parent.PID)
-		if err != nil || again.Start != parent.Start || again.PID != parent.PID {
-			if next == 0 {
-				return Usage{}, false
-			}
-			// A parent that exited or whose pid changed hands left the tree with
-			// its children; one that merely could not be read leaves them out of
-			// this sum only, so the sum is partial.
-			if err != nil && !errors.Is(err, unix.ESRCH) {
-				usage.CPUPartial = true
-			}
-			continue
-		}
+		var read []darwinproc.TaskInfo
 		for _, pid := range children {
 			if len(seen) >= usageWalkLimit {
 				usage.CPUPartial = true
@@ -114,8 +104,22 @@ func subtreeUsage(root int) (Usage, bool) {
 			if err != nil || child.PID != pid || child.PPID != parent.PID || child.Status == darwinproc.StatusZombie {
 				continue
 			}
-			queue = append(queue, child)
+			read = append(read, child)
+		}
+		again, err := usageReader.task(parent.PID)
+		if err != nil || again.Start != parent.Start || again.PID != parent.PID {
+			if next == 0 {
+				return Usage{}, false
+			}
+			// A parent that exited or whose pid changed hands left the tree with
+			// its children; one that merely could not be read leaves them out of
+			// this sum only, so the sum is partial.
+			if err != nil && !errors.Is(err, unix.ESRCH) {
+				usage.CPUPartial = true
+			}
+			continue
 		}
+		queue = append(queue, read...)
 	}
 	return usage, true
 }

internal/processgroup/usage_darwin_walk_test.go

@@ -182,3 +182,49 @@ func TestDarwinUsageMarksAnUnreadableParentPartial(t *testing.T) {
 		t.Fatalf("a parent that exited: usage = %+v, %v, want 2 processes and a complete sum", usage, ok)
 	}
 }
+
+// A parent's pid and its listed child's pid can both change hands right after
+// the parent read the same again. The child read after that swap is a
+// stranger's busy child and must not be charged to the job: a child is
+// admitted only when its parent still reads as the same process after it.
+func TestDarwinUsageAdmitsAChildOnlyWhileItsParentIsTheSame(t *testing.T) {
+	saved := usageReader
+	t.Cleanup(func() { usageReader = saved })
+	leader, own := task(10, 1, 10, 1), task(11, 10, 10, 1)
+	strangerLeader, strangerChild := task(10, 1, 50, 0), task(11, 10, 50, 500)
+	strangerLeader.Start, strangerChild.Start = 90000, 91000
+	strangers := map[int]darwinproc.TaskInfo{12: task(12, 11, 50, 300), 13: task(13, 11, 50, 300)}
+	leaderReads, swapped := 0, false
+	usageReader.task = func(pid int) (darwinproc.TaskInfo, error) {
+		switch {
+		case pid == 10 && swapped:
+			return strangerLeader, nil
+		case pid == 10:
+			// The swap lands just after the leader's second read.
+			if leaderReads++; leaderReads == 2 {
+				swapped = true
+			}
+			return leader, nil
+		case pid == 11 && swapped:
+			return strangerChild, nil
+		case pid == 11:
+			return own, nil
+		case swapped && strangers[pid].PID == pid:
+			return strangers[pid], nil
+		}
+		return darwinproc.TaskInfo{}, unix.ESRCH
+	}
+	usageReader.children = func(ppid int) ([]int, error) {
+		switch {
+		case ppid == 10:
+			return []int{11}, nil
+		case ppid == 11 && swapped:
+			return []int{12, 13}, nil
+		}
+		return nil, nil
+	}
+	usage, ok := subtreeUsage(10)
+	if !ok || usage.Processes != 2 || usage.CPUSeconds != 2 {
+		t.Fatalf("usage = %+v, %v, want the leader and its own child only (2 processes, 2 CPU seconds)", usage, ok)
+	}
+}

internal/session/jobbound_test.go

@@ -235,11 +235,13 @@ func TestASubtreeThatCannotBeReadIsNeverCut(t *testing.T) {
 // (macOS's walk limit, a truncated listing) still counts processes, but its CPU
 // sum is not comparable with the next one. A storm of sleepers read partially
 // for fourteen readings that then shrinks to a leader and one old child — whose
-// lifetime CPU the partial sums never held — must not be cut on that jump.
+// lifetime CPU the partial sums never held — must not be cut on that jump. The
+// storm starts after one complete, quiet reading, so the partial readings must
+// also drop that earlier baseline rather than leave it waiting for the jump.
 func TestAPartialCPUSumNeverMakesARate(t *testing.T) {
 	registry := &jobRegistry{}
 	one := &job{id: 1, kind: jobKindBash, state: jobRunning, done: make(chan struct{})}
-	var readings []processgroup.Usage
+	readings := []processgroup.Usage{{Processes: 2, CPUSeconds: 1}}
 	for i := 0; i < jobBoundStrikes-1; i++ {
 		readings = append(readings, processgroup.Usage{Processes: 1100, CPUSeconds: 1, CPUPartial: true})
 	}