From 36704b726f2606535018af7366a17618815a5a30 Mon Sep 17 00:00:00 2001 From: Abir Abbas Date: Thu, 29 Oct 2026 14:16:11 -0400 Subject: [PATCH 1/8] processgroup: read macOS process groups and job subtrees without cgo Members lists a group through kern.proc.pgrp with a bounded retry, and Group.Usage walks the recorded leader's descendants through the kernel's proc_info interface, converting task time with hw.tbfrequency. The new darwinproc package owns the raw trap, the SDK layouts and the argv and listening-socket readers the executor uses next. Co-Authored-By: Claude Opus 5.5 --- internal/darwinproc/doc.go | 3 + internal/darwinproc/procinfo_darwin.go | 243 +++++++++++++++++++ internal/darwinproc/procinfo_darwin_test.go | 206 ++++++++++++++++ internal/processgroup/members_darwin.go | 69 ++++++ internal/processgroup/members_darwin_test.go | 71 ++++++ internal/processgroup/members_other.go | 12 +- internal/processgroup/usage.go | 8 +- internal/processgroup/usage_darwin.go | 49 ++++ internal/processgroup/usage_darwin_test.go | 80 ++++++ internal/processgroup/usage_other.go | 10 +- internal/session/jobbound.go | 4 +- 11 files changed, 736 insertions(+), 19 deletions(-) create mode 100644 internal/darwinproc/doc.go create mode 100644 internal/darwinproc/procinfo_darwin.go create mode 100644 internal/darwinproc/procinfo_darwin_test.go create mode 100644 internal/processgroup/members_darwin.go create mode 100644 internal/processgroup/members_darwin_test.go create mode 100644 internal/processgroup/usage_darwin.go create mode 100644 internal/processgroup/usage_darwin_test.go diff --git a/internal/darwinproc/doc.go b/internal/darwinproc/doc.go new file mode 100644 index 0000000000..88c2428641 --- /dev/null +++ b/internal/darwinproc/doc.go @@ -0,0 +1,3 @@ +// Package darwinproc reads macOS process identity, descendants, CPU usage, +// arguments and listening sockets without cgo. +package darwinproc diff --git a/internal/darwinproc/procinfo_darwin.go b/internal/darwinproc/procinfo_darwin.go new file mode 100644 index 0000000000..0bc7acd9b5 --- /dev/null +++ b/internal/darwinproc/procinfo_darwin.go @@ -0,0 +1,243 @@ +//go:build darwin + +package darwinproc + +import ( + "bytes" + "encoding/binary" + "slices" + "sync" + "unsafe" + + "golang.org/x/sys/unix" +) + +// The raw proc_info trap uses Go's existing per-architecture syscall stubs, +// avoiding assembly, dynamic imports and linkname dependencies that garbling +// could break. Unlike libproc's wrappers, the trap preserves errno directly. +// Apple promises stability for neither the trap numbers nor libproc, but both +// have remained unchanged since macOS 10.5. The private XNU call selectors and +// SDK layouts below are identical on arm64 and amd64; both are little-endian. +const ( + callListPIDs = 1 + callPIDInfo = 2 + callPIDFDInfo = 3 + ppidOnly = 6 + pidListFDs = 1 + pidTaskAllInfo = 2 + pidFDSocketInfo = 3 + fdTypeSocket = 2 + socketInfoTCP = 2 + tcpStateListen = 1 + familyIPv4 = 2 + familyIPv6 = 30 + + // StatusZombie marks exited processes, which cannot contribute live work. + StatusZombie = 5 + + pidSize = 4 + taskAllInfoSize = 232 + bsdInfoOffset = 0 + bsdInfoSize = 136 + bsdStatusOffset = bsdInfoOffset + 4 + bsdPIDOffset = bsdInfoOffset + 12 + bsdPPIDOffset = bsdInfoOffset + 16 + bsdPGIDOffset = bsdInfoOffset + 100 + bsdStartSecondsOffset = bsdInfoOffset + 120 + bsdStartMicrosOffset = bsdInfoOffset + 128 + taskInfoOffset = bsdInfoSize + taskInfoSize = 96 + taskResidentOffset = taskInfoOffset + 8 + taskUserOffset = taskInfoOffset + 16 + taskSystemOffset = taskInfoOffset + 24 + fdInfoSize = 8 + fdNumberOffset = 0 + fdTypeOffset = 4 + socketFDInfoSize = 792 + socketInfoOffset = 24 + socketInfoSize = 768 + socketFamilyOffset = socketInfoOffset + 160 + socketKindOffset = socketInfoOffset + 232 + socketProtocolOffset = socketInfoOffset + 240 + socketLocalPortOffset = socketProtocolOffset + 4 + socketTCPStateOffset = socketProtocolOffset + 80 + + childrenInitialPIDs = 64 + childrenGrowth = 4 + childrenAttempts = 4 +) + +// procInfo keeps pointer conversion in the syscall expression so the buffer +// remains alive and cannot move while the kernel writes it. A nil buffer is +// used only by interfaces that return a sizing estimate. +func procInfo(call, a1, a2, a3 int, buf []byte) (int, error) { + if len(buf) > 1<<31-1 { + return 0, unix.EINVAL + } + var n uintptr + var errno unix.Errno + if len(buf) == 0 { + n, _, errno = unix.Syscall6(unix.SYS_PROC_INFO, uintptr(call), uintptr(a1), uintptr(a2), uintptr(a3), 0, 0) + } else { + n, _, errno = unix.Syscall6(unix.SYS_PROC_INFO, uintptr(call), uintptr(a1), uintptr(a2), uintptr(a3), uintptr(unsafe.Pointer(&buf[0])), uintptr(len(buf))) + } + if errno != 0 { + return 0, errno + } + if n > 1<<31-1 || (len(buf) > 0 && n > uintptr(len(buf))) { + return 0, unix.EIO + } + return int(n), nil +} + +// Children lists immediate children, bounding retries when the tree grows +// between reads. A full final buffer is unknown rather than a partial tree. +func Children(ppid int) ([]int, error) { + if ppid <= 0 { + return nil, unix.EINVAL + } + var initial [childrenInitialPIDs * pidSize]byte + buf := initial[:] + for attempt := 0; attempt < childrenAttempts; attempt++ { + n, err := procInfo(callListPIDs, ppidOnly, ppid, 0, buf) + if err != nil { + return nil, err + } + if n%pidSize != 0 { + return nil, unix.EIO + } + if n < len(buf) { + pids := make([]int, 0, n/pidSize) + for offset := 0; offset < n; offset += pidSize { + if pid := int(int32(binary.LittleEndian.Uint32(buf[offset:]))); pid > 0 { + pids = append(pids, pid) + } + } + return pids, nil + } + if attempt+1 < childrenAttempts { + buf = make([]byte, len(buf)*childrenGrowth) + } + } + return nil, unix.ENOMEM +} + +// TaskInfo combines identity and CPU from one kernel snapshot so callers can +// validate ancestry before attributing a process's accumulated CPU time. +type TaskInfo struct { + PID, PPID, PGID, Status int + CPUSeconds float64 + Resident uint64 +} + +var ( + frequencyOnce sync.Once + tickFrequency uint64 +) + +// Task reads one process. Kernel task times are absolute ticks even under +// Rosetta; hw.tbfrequency converts them correctly on both supported CPUs. +func Task(pid int) (TaskInfo, error) { + if pid <= 0 { + return TaskInfo{}, unix.EINVAL + } + frequencyOnce.Do(func() { tickFrequency, _ = unix.SysctlUint64("hw.tbfrequency") }) + if tickFrequency == 0 { + return TaskInfo{}, unix.EIO + } + var buf [taskAllInfoSize]byte + n, err := procInfo(callPIDInfo, pid, pidTaskAllInfo, 0, buf[:]) + if err != nil { + return TaskInfo{}, err + } + if n != taskAllInfoSize { + return TaskInfo{}, unix.EIO + } + return TaskInfo{ + PID: int(int32(binary.LittleEndian.Uint32(buf[bsdPIDOffset:]))), + PPID: int(int32(binary.LittleEndian.Uint32(buf[bsdPPIDOffset:]))), + PGID: int(int32(binary.LittleEndian.Uint32(buf[bsdPGIDOffset:]))), + Status: int(binary.LittleEndian.Uint32(buf[bsdStatusOffset:])), + CPUSeconds: float64(binary.LittleEndian.Uint64(buf[taskUserOffset:]))/float64(tickFrequency) + float64(binary.LittleEndian.Uint64(buf[taskSystemOffset:]))/float64(tickFrequency), + Resident: binary.LittleEndian.Uint64(buf[taskResidentOffset:]), + }, nil +} + +// ListenPorts inspects only this process's socket descriptors. Descriptor races +// are skipped individually, and a shared socket buffer avoids one allocation +// per descriptor. +func ListenPorts(pid int) []int { + if pid <= 0 { + return nil + } + size, err := procInfo(callPIDInfo, pid, pidListFDs, 0, nil) + if err != nil || size == 0 || size%fdInfoSize != 0 { + return nil + } + fds := make([]byte, size) + n, err := procInfo(callPIDInfo, pid, pidListFDs, 0, fds) + if err != nil || n%fdInfoSize != 0 { + return nil + } + var socket [socketFDInfoSize]byte + var ports []int + for offset := 0; offset < n; offset += fdInfoSize { + fd := fds[offset : offset+fdInfoSize] + if binary.LittleEndian.Uint32(fd[fdTypeOffset:]) != fdTypeSocket { + continue + } + number := int(int32(binary.LittleEndian.Uint32(fd[fdNumberOffset:]))) + count, err := procInfo(callPIDFDInfo, pid, pidFDSocketInfo, number, socket[:]) + if err != nil || count != socketFDInfoSize { + continue + } + family := binary.LittleEndian.Uint32(socket[socketFamilyOffset:]) + if (family != familyIPv4 && family != familyIPv6) || binary.LittleEndian.Uint32(socket[socketKindOffset:]) != socketInfoTCP || binary.LittleEndian.Uint32(socket[socketTCPStateOffset:]) != tcpStateListen { + continue + } + // The low sixteen bits contain network-order bytes on both macOS CPUs. + port := int(binary.BigEndian.Uint16(socket[socketLocalPortOffset:])) + ports = append(ports, port) + } + slices.Sort(ports) + return slices.Compact(ports) +} + +// Argv discards the raw sysctl response after parsing because it also contains +// environment data that must never enter a service record or diagnostic. +func Argv(pid int) []string { + if pid <= 0 { + return nil + } + raw, err := unix.SysctlRaw("kern.procargs2", pid) + if err != nil { + return nil + } + return parseProcArgs(raw) +} + +func parseProcArgs(raw []byte) []string { + if len(raw) < pidSize { + return nil + } + argc := int(int32(binary.LittleEndian.Uint32(raw))) + if argc <= 0 || argc > len(raw)-pidSize { + return nil + } + rest := raw[pidSize:] + end := bytes.IndexByte(rest, 0) + if end < 0 { + return nil + } + rest = bytes.TrimLeft(rest[end+1:], "\x00") + argv := make([]string, 0, argc) + for range argc { + end = bytes.IndexByte(rest, 0) + if end < 0 { + return nil + } + argv = append(argv, string(rest[:end])) + rest = rest[end+1:] + } + return argv +} diff --git a/internal/darwinproc/procinfo_darwin_test.go b/internal/darwinproc/procinfo_darwin_test.go new file mode 100644 index 0000000000..1da8307dc7 --- /dev/null +++ b/internal/darwinproc/procinfo_darwin_test.go @@ -0,0 +1,206 @@ +//go:build darwin + +package darwinproc + +import ( + "encoding/binary" + "math" + "net" + "os" + "os/exec" + "path/filepath" + "reflect" + "slices" + "syscall" + "testing" + "time" +) + +func TestDarwinProcArgsStopsBeforeEnvironment(t *testing.T) { + rawArgs := func(argc int32, body string) []byte { + raw := make([]byte, 4, 4+len(body)) + binary.LittleEndian.PutUint32(raw, uint32(argc)) + return append(raw, body...) + } + for _, tc := range []struct { + name string + raw []byte + want []string + }{ + {"padding and environment", rawArgs(4, "/bin/tool\x00\x00\x00tool\x00\x00two words\x00last\x00PRIVATE=value\x00"), []string{"tool", "", "two words", "last"}}, + {"zero", rawArgs(0, "/bin/tool\x00tool\x00"), nil}, + {"negative", rawArgs(-1, "/bin/tool\x00tool\x00"), nil}, + {"absurd", rawArgs(1<<30, "/bin/tool\x00tool\x00"), nil}, + {"missing argument", rawArgs(2, "/bin/tool\x00tool\x00"), nil}, + {"truncated argument", rawArgs(1, "/bin/tool\x00tool"), nil}, + {"truncated executable", rawArgs(1, "/bin/tool"), nil}, + {"short header", []byte{1, 0, 0}, nil}, + } { + t.Run(tc.name, func(t *testing.T) { + if got := parseProcArgs(tc.raw); !reflect.DeepEqual(got, tc.want) { + t.Fatalf("argv = %q, want %q", got, tc.want) + } + }) + } +} + +func sleepChild(t testing.TB) *exec.Cmd { + t.Helper() + cmd := exec.Command("sleep", "30") + cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true} + if err := cmd.Start(); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = cmd.Process.Kill(); _ = cmd.Wait() }) + return cmd +} + +func TestDarwinArgvReadsALiveChild(t *testing.T) { + cmd := sleepChild(t) + if got := Argv(cmd.Process.Pid); !reflect.DeepEqual(got, []string{"sleep", "30"}) { + t.Fatalf("argv = %q", got) + } +} + +func TestDarwinTaskMatchesSelf(t *testing.T) { + deadline := time.Now().Add(150 * time.Millisecond) + var work uint64 + for time.Now().Before(deadline) { + for range 10000 { + work = work*1664525 + 1013904223 + } + } + if work == 0 { + t.Log("bounded loop finished at zero") + } + task, err := Task(os.Getpid()) + if err != nil { + t.Fatal(err) + } + if task.PID != os.Getpid() || task.PPID != os.Getppid() { + t.Fatalf("identity = %+v", task) + } + var usage syscall.Rusage + if err := syscall.Getrusage(syscall.RUSAGE_SELF, &usage); err != nil { + t.Fatal(err) + } + seconds := float64(usage.Utime.Sec+usage.Stime.Sec) + float64(usage.Utime.Usec+usage.Stime.Usec)/1e6 + if math.Abs(task.CPUSeconds-seconds) > math.Max(seconds*.2, .03) { + t.Fatalf("CPU seconds = %f, getrusage = %f", task.CPUSeconds, seconds) + } + gone := exec.Command("true") + gone.SysProcAttr = &syscall.SysProcAttr{Setpgid: true} + if err := gone.Run(); err != nil { + t.Fatal(err) + } + if _, err := Task(gone.Process.Pid); err == nil { + t.Fatal("gone process returned a task") + } +} + +func TestDarwinChildrenListsASpawnedChild(t *testing.T) { + cmd := sleepChild(t) + pids, err := Children(os.Getpid()) + if err != nil { + t.Fatal(err) + } + if !slices.Contains(pids, cmd.Process.Pid) { + t.Fatalf("children %v omit %d", pids, cmd.Process.Pid) + } +} + +func TestDarwinListeningPorts(t *testing.T) { + listener, err := net.Listen("tcp4", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = listener.Close() }) + want := []int{listener.Addr().(*net.TCPAddr).Port} + listener6, err := net.Listen("tcp6", "[::1]:0") + if err == nil { + t.Cleanup(func() { _ = listener6.Close() }) + want = append(want, listener6.Addr().(*net.TCPAddr).Port) + } else { + t.Logf("IPv6 listener unavailable: %v", err) + } + conn, err := net.DialTimeout("tcp4", listener.Addr().String(), time.Second) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = conn.Close() }) + _ = listener.(*net.TCPListener).SetDeadline(time.Now().Add(time.Second)) + accepted, err := listener.Accept() + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = accepted.Close() }) + udp, err := net.ListenPacket("udp4", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = udp.Close() }) + // A short directory keeps the socket path within macOS's sockaddr_un limit. + socketDir, err := os.MkdirTemp("", "dp-") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = os.RemoveAll(socketDir) }) + local, err := net.Listen("unix", filepath.Join(socketDir, "s")) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = local.Close() }) + got := ListenPorts(os.Getpid()) + for _, p := range want { + if !slices.Contains(got, p) { + t.Fatalf("ports %v omit listener %d", got, p) + } + } + for _, p := range []int{conn.LocalAddr().(*net.TCPAddr).Port, udp.LocalAddr().(*net.UDPAddr).Port} { + // A UDP or ephemeral TCP port can coincide with an independently allocated listener. + if !slices.Contains(want, p) && slices.Contains(got, p) { + t.Fatalf("ports %v include non-listener %d", got, p) + } + } + if !slices.IsSorted(got) || len(slices.Compact(slices.Clone(got))) != len(got) { + t.Fatalf("ports not sorted and unique: %v", got) + } +} + +func BenchmarkTask(b *testing.B) { + b.ReportAllocs() + for b.Loop() { + if _, err := Task(os.Getpid()); err != nil { + b.Fatal(err) + } + } +} +func BenchmarkChildren(b *testing.B) { + _ = sleepChild(b) + b.ReportAllocs() + for b.Loop() { + if _, err := Children(os.Getpid()); err != nil { + b.Fatal(err) + } + } +} +func BenchmarkListenPorts(b *testing.B) { + listener, err := net.Listen("tcp4", "127.0.0.1:0") + if err != nil { + b.Fatal(err) + } + b.Cleanup(func() { _ = listener.Close() }) + b.ReportAllocs() + for b.Loop() { + _ = ListenPorts(os.Getpid()) + } +} +func BenchmarkArgv(b *testing.B) { + cmd := sleepChild(b) + b.ReportAllocs() + for b.Loop() { + if len(Argv(cmd.Process.Pid)) == 0 { + b.Fatal("argv unavailable") + } + } +} diff --git a/internal/processgroup/members_darwin.go b/internal/processgroup/members_darwin.go new file mode 100644 index 0000000000..0677879697 --- /dev/null +++ b/internal/processgroup/members_darwin.go @@ -0,0 +1,69 @@ +//go:build darwin + +package processgroup + +import ( + "encoding/binary" + "errors" + "sort" + "unsafe" + + "github.com/Agent-Field/codeaf/internal/darwinproc" + "golang.org/x/sys/unix" +) + +// Members lists live members in pid order. A growing process table gets only +// three attempts so a completion inspection cannot retry indefinitely. +func Members(pgid int) []int { + if pgid <= 0 { + return nil + } + for range 3 { + raw, err := unix.SysctlRaw("kern.proc.pgrp", pgid) + if errors.Is(err, unix.ENOMEM) { + continue + } + if err != nil || len(raw) == 0 || len(raw)%unix.SizeofKinfoProc != 0 { + return nil + } + return darwinMemberPIDs(pgid, darwinKinfoRecords(raw)) + } + return nil +} + +// Only the three fields membership reads are taken from each kernel record. +// Copying whole records would put the kernel's bytes into the x/sys type's +// pointer-typed fields, which the garbage collector then scans. +const ( + kinfoPIDOffset = unsafe.Offsetof(unix.KinfoProc{}.Proc) + unsafe.Offsetof(unix.ExternProc{}.P_pid) + kinfoStatOffset = unsafe.Offsetof(unix.KinfoProc{}.Proc) + unsafe.Offsetof(unix.ExternProc{}.P_stat) + kinfoPGIDOffset = unsafe.Offsetof(unix.KinfoProc{}.Eproc) + unsafe.Offsetof(unix.Eproc{}.Pgid) +) + +func darwinKinfoRecords(raw []byte) []unix.KinfoProc { + records := make([]unix.KinfoProc, len(raw)/unix.SizeofKinfoProc) + for i := range records { + record := raw[i*unix.SizeofKinfoProc : (i+1)*unix.SizeofKinfoProc] + records[i].Proc.P_pid = int32(binary.LittleEndian.Uint32(record[kinfoPIDOffset:])) + records[i].Proc.P_stat = int8(record[kinfoStatOffset]) + records[i].Eproc.Pgid = int32(binary.LittleEndian.Uint32(record[kinfoPGIDOffset:])) + } + return records +} + +func darwinMemberPIDs(pgid int, records []unix.KinfoProc) []int { + pids := make([]int, 0, len(records)) + for _, record := range records { + if record.Proc.P_pid > 0 && int(record.Eproc.Pgid) == pgid && record.Proc.P_stat != darwinproc.StatusZombie { + pids = append(pids, int(record.Proc.P_pid)) + } + } + sort.Ints(pids) + return pids +} + +// The shell is reaped before a sweep on macOS (internal/exec/shellwait_other.go), +// so the existing signal probe has no zombie leader to mistake for live work. +func groupHasLiveMember(pgid int) bool { + return Alive(pgid) +} diff --git a/internal/processgroup/members_darwin_test.go b/internal/processgroup/members_darwin_test.go new file mode 100644 index 0000000000..f9b1c5b1e0 --- /dev/null +++ b/internal/processgroup/members_darwin_test.go @@ -0,0 +1,71 @@ +//go:build darwin + +package processgroup + +import ( + "os/exec" + "reflect" + "strings" + "testing" + + "github.com/Agent-Field/codeaf/internal/darwinproc" + "golang.org/x/sys/unix" +) + +func TestDarwinMembersExcludeZombies(t *testing.T) { + records := make([]unix.KinfoProc, 6) + for i, pid := range []int32{30, 10, 20, 40, 0, -1} { + records[i].Proc.P_pid = pid + records[i].Proc.P_stat = 3 + records[i].Eproc.Pgid = 100 + } + records[0].Proc.P_stat = 4 + records[2].Proc.P_stat = darwinproc.StatusZombie + records[3].Eproc.Pgid = 101 + if got := darwinMemberPIDs(100, records); !reflect.DeepEqual(got, []int{10, 30}) { + t.Fatalf("Members = %v, want sleeping and stopped members [10 30]", got) + } +} + +func TestDarwinMembersReadsALiveGroup(t *testing.T) { + cmd := startDarwinGroup(t, "sleep", "30") + pid := cmd.Process.Pid + if got := Members(pid); !reflect.DeepEqual(got, []int{pid}) { + t.Fatalf("Members(%d) = %v, want [%d]", pid, got, pid) + } + if err := cmd.Process.Kill(); err != nil { + t.Fatal(err) + } + _ = cmd.Wait() + if got := Members(pid); len(got) != 0 { + t.Fatalf("Members(%d) after wait = %v, want empty", pid, got) + } +} + +func startDarwinGroup(tb testing.TB, name string, args ...string) *exec.Cmd { + tb.Helper() + cmd := exec.Command(name, args...) + Configure(cmd) + if err := cmd.Start(); err != nil { + tb.Fatal(err) + } + tb.Cleanup(func() { + if cmd.ProcessState == nil { + _ = Kill(cmd.Process.Pid) + _ = cmd.Wait() + } + }) + return cmd +} + +func BenchmarkMembers(b *testing.B) { + cmd := startDarwinGroup(b, "sh", "-c", strings.Repeat("sleep 30 & ", 9)+"wait") + waitForDarwinUsage(b, CaptureGroup(cmd.Process.Pid), 10) + b.ReportAllocs() + b.ResetTimer() + for b.Loop() { + if got := Members(cmd.Process.Pid); len(got) != 10 || got[0] != cmd.Process.Pid { + b.Fatalf("Members = %v, want the leader and nine sleeping children", got) + } + } +} diff --git a/internal/processgroup/members_other.go b/internal/processgroup/members_other.go index a0777f0b37..137975cd77 100644 --- a/internal/processgroup/members_other.go +++ b/internal/processgroup/members_other.go @@ -1,14 +1,14 @@ -//go:build !windows && !linux +//go:build !windows && !linux && !darwin package processgroup -// groupHasLiveMember is the signal probe off Linux: there the shell is reaped -// before any sweep looks at its group (internal/exec, shellwait_other.go), so -// no zombie leader is ever in the group when this is asked, and kill(-pgid, 0) -// answers correctly. +// groupHasLiveMember retains the signal probe on other Unix platforms: the +// shell is reaped before any sweep looks at its group (internal/exec, +// shellwait_other.go), so no zombie leader is in the group when this is asked, +// and kill(-pgid, 0) answers correctly. func groupHasLiveMember(pgid int) bool { return Alive(pgid) } -// Members is unknown off Linux; callers fall back to what they started. +// Members is unknown off Linux and macOS; callers fall back to what they started. func Members(pgid int) []int { return nil } diff --git a/internal/processgroup/usage.go b/internal/processgroup/usage.go index b3d418024b..1d5e1b6a85 100644 --- a/internal/processgroup/usage.go +++ b/internal/processgroup/usage.go @@ -16,11 +16,9 @@ package processgroup // cores the subtree held, which is the reading this product bounds. One // reading on its own is not a rate and must not be read as one. // -// THE READING IS /proc, WHICH IS LINUX'S. A platform without it answers "cannot -// say" rather than a guess, and a bound that cannot say never cuts — the same -// silence rule the admission governor keeps (session's task_pressure.go): a -// person on a machine this cannot measure gets exactly the scheduler they had -// before the bound existed. +// Linux reads /proc and macOS reads the kernel's process-info interface. Other +// platforms answer "cannot say" rather than guess, and a bound that cannot say +// never cuts, preserving the scheduler the person had before the bound existed. // Usage is what one job subtree is using at one instant. type Usage struct { diff --git a/internal/processgroup/usage_darwin.go b/internal/processgroup/usage_darwin.go new file mode 100644 index 0000000000..92829e5a32 --- /dev/null +++ b/internal/processgroup/usage_darwin.go @@ -0,0 +1,49 @@ +//go:build darwin + +package processgroup + +import "github.com/Agent-Field/codeaf/internal/darwinproc" + +// Usage reports the recorded leader's whole subtree, or says it cannot say. +// Identity is checked around the walk because a reading of a reused pid could +// cause the bound to cut somebody else's work. +func (g Group) Usage() (Usage, bool) { + if ok, _ := g.owned(); !ok { + return Usage{}, false + } + root, err := darwinproc.Task(g.pid) + if err != nil || root.PID != g.pid || root.Status == darwinproc.StatusZombie { + return Usage{}, false + } + usage := Usage{} + seen := map[int]bool{g.pid: true} + queue := []darwinproc.TaskInfo{root} + for next := 0; next < len(queue); next++ { + parent := queue[next] + usage.Processes++ + usage.CPUSeconds += parent.CPUSeconds + // Follow descendants even after a child changes process groups, because + // the bound covers the whole job rather than only its original group. + children, err := darwinproc.Children(parent.PID) + if err != nil { + continue + } + for _, pid := range children { + if pid <= 0 || seen[pid] { + continue + } + seen[pid] = true + child, err := darwinproc.Task(pid) + // A child can exit, be reparented or have its pid reused between reads. + // Skip that snapshot rather than charging an unrelated process to us. + if err != nil || child.PID != pid || child.PPID != parent.PID || child.Status == darwinproc.StatusZombie { + continue + } + queue = append(queue, child) + } + } + if ok, _ := g.owned(); !ok { + return Usage{}, false + } + return usage, true +} diff --git a/internal/processgroup/usage_darwin_test.go b/internal/processgroup/usage_darwin_test.go new file mode 100644 index 0000000000..ec769fff9d --- /dev/null +++ b/internal/processgroup/usage_darwin_test.go @@ -0,0 +1,80 @@ +//go:build darwin + +package processgroup + +import ( + "os" + "strings" + "testing" + "time" +) + +func TestDarwinUsageReadsDescendants(t *testing.T) { + cmd := startDarwinGroup(t, "sh", "-c", `sleep 30 & sleep 30 & sh -c 'sleep 30 & wait' & wait`) + group := CaptureGroup(cmd.Process.Pid) + usage := waitForDarwinUsage(t, group, 5) + if usage.CPUSeconds < 0 { + t.Fatalf("Usage CPUSeconds = %v, want nonnegative", usage.CPUSeconds) + } +} + +func TestDarwinUsageRefusesUnownedGroup(t *testing.T) { + if usage, ok := (Group{}).Usage(); ok || usage != (Usage{}) { + t.Fatalf("unowned Usage = %+v, %v, want zero, false", usage, ok) + } + cmd := startDarwinGroup(t, "sleep", "30") + id, ok := CaptureGroup(cmd.Process.Pid).Identity() + if !ok { + t.Fatal("could not capture sleeping leader identity") + } + id.Start++ + if usage, ok := Restore(id).Usage(); ok || usage != (Usage{}) { + t.Fatalf("wrong identity Usage = %+v, %v, want zero, false", usage, ok) + } + + t.Run("identity changes during sample", func(t *testing.T) { + group := CaptureGroup(os.Getpid()) + saved := processStart + t.Cleanup(func() { processStart = saved }) + reads := 0 + processStart = func(pid int) (uint64, bool) { + reads++ + return group.start, reads == 1 + } + if usage, ok := group.Usage(); ok || usage != (Usage{}) { + t.Fatalf("changed identity Usage = %+v, %v, want zero, false", usage, ok) + } + if reads != 2 { + t.Fatalf("identity read %d times, want before and after the sample", reads) + } + }) +} + +func waitForDarwinUsage(tb testing.TB, group Group, processes int) Usage { + tb.Helper() + deadline := time.Now().Add(5 * time.Second) + var usage Usage + var ok bool + for time.Now().Before(deadline) { + usage, ok = group.Usage() + if ok && usage.Processes >= processes { + return usage + } + time.Sleep(10 * time.Millisecond) + } + tb.Fatalf("Usage = %+v, %v, want at least %d processes", usage, ok, processes) + return Usage{} +} + +func BenchmarkUsage(b *testing.B) { + cmd := startDarwinGroup(b, "sh", "-c", strings.Repeat("sleep 30 & ", 9)+"wait") + group := CaptureGroup(cmd.Process.Pid) + waitForDarwinUsage(b, group, 10) + b.ReportAllocs() + b.ResetTimer() + for b.Loop() { + if usage, ok := group.Usage(); !ok || usage.Processes < 10 || usage.CPUSeconds < 0 { + b.Fatalf("Usage = %+v, %v, want the leader and sleeping children", usage, ok) + } + } +} diff --git a/internal/processgroup/usage_other.go b/internal/processgroup/usage_other.go index 4ecf7f23a7..84077706f8 100644 --- a/internal/processgroup/usage_other.go +++ b/internal/processgroup/usage_other.go @@ -1,10 +1,8 @@ -//go:build !linux +//go:build !linux && !darwin package processgroup -// Usage has no answer where there is no /proc to read: the subtree's CPU time -// and process count are Linux's to report here. A platform that cannot say -// answers so, and a bound that cannot say never cuts — it gets the scheduler it -// had before the bound existed rather than a guess at somebody else's process -// group. +// Usage has no answer where the subtree's CPU time and process count cannot be +// read. A bound that cannot say never cuts, preserving the existing scheduler +// rather than guessing at somebody else's process group. func (g Group) Usage() (Usage, bool) { return Usage{}, false } diff --git a/internal/session/jobbound.go b/internal/session/jobbound.go index 8248be1fc7..a782ee49f2 100644 --- a/internal/session/jobbound.go +++ b/internal/session/jobbound.go @@ -58,8 +58,8 @@ package session // // The watch is this program's own and runs everywhere. It reads the subtree // through the process-group handle #1155 gave, which works wherever a process -// group is a process group; the CPU-and-process reading itself is /proc, so off -// Linux it says it cannot say and the bound never cuts ([processgroup.Usage]). +// group is a process group. Linux and macOS read the subtree's CPU and process +// count; other platforms cannot say and the bound never cuts ([processgroup.Usage]). // The subtree's USAGE needs no cgroup — a job is a process group and not a // cgroup. Whether a quota BINDS is a different question — how many cores the // subtree could ever reach — answered from the process's own cgroup CPU quota and From d9eaf71f1d16ca34e249ef43b73c441fb1d16fcd Mon Sep 17 00:00:00 2001 From: Abir Abbas Date: Thu, 29 Oct 2026 14:16:11 -0400 Subject: [PATCH 2/8] executor: name and port left-running services on macOS cmdline reads kern.procargs2 and the ports come from each member's TCP listeners. describe reads each member's sockets once for both the naming member and the port set. The two service tests run on Linux and macOS, and the listening-service test polls the record until the backgrounded shell has exec'd its server. Co-Authored-By: Claude Opus 5.5 --- internal/executor/executor.go | 6 +- internal/executor/service.go | 48 ++++++++++------ internal/executor/service_darwin.go | 22 ++++++++ internal/executor/service_describe_test.go | 64 ++++++++++++++++++++++ internal/executor/service_linux.go | 18 ++++-- internal/executor/service_linux_test.go | 4 -- internal/executor/service_members_test.go | 7 +++ internal/executor/service_other.go | 6 +- internal/executor/service_other_test.go | 2 +- internal/executor/service_unix_test.go | 16 +++--- internal/executor/watched_unix_test.go | 4 +- 11 files changed, 155 insertions(+), 42 deletions(-) create mode 100644 internal/executor/service_darwin.go create mode 100644 internal/executor/service_describe_test.go create mode 100644 internal/executor/service_members_test.go diff --git a/internal/executor/executor.go b/internal/executor/executor.go index 8fd16ec67c..34dfb48de2 100644 --- a/internal/executor/executor.go +++ b/internal/executor/executor.go @@ -115,10 +115,10 @@ type Chunk struct { // Service is a process group still alive after its call returned. type Service struct { - Name string // basename of the first member's argv[0] - Argv []string // the first member's command line (the request's off Linux) + Name string // Basename of the naming member's argv[0]. + Argv []string // The naming member's command line on Linux and macOS; otherwise the request's. PGID int - Ports []int // listening TCP ports, when known (Linux only) + Ports []int // Listening TCP ports, when known on Linux and macOS. DataDir string // cell-relative data directory, when the service keeps one } diff --git a/internal/executor/service.go b/internal/executor/service.go index 42ac58bf38..d0e13c9871 100644 --- a/internal/executor/service.go +++ b/internal/executor/service.go @@ -4,6 +4,7 @@ import ( "context" "os/exec" "path/filepath" + "slices" "strings" "github.com/Agent-Field/codeaf/internal/processgroup" @@ -37,16 +38,42 @@ func Left(call Call) []Service { // listens names it, because it is the server; with none listening the first // member does. The ports of all members are its ports. func describe(req ExecRequest, root string, pgid int, members []int) Service { + service := describeWith(req, root, pgid, members, pidPorts, cmdline) + if len(service.Ports) == 0 { + // An empty query preserves each platform's unknown-versus-empty result + // without inspecting any member's sockets again. + service.Ports = listenPorts(nil) + } + return service +} + +// describeWith uses one socket snapshot for naming and aggregation so a member +// is not inspected twice while a service is recorded. +func describeWith(req ExecRequest, root string, pgid int, members []int, ports func(int) []int, cmdline func(int) []string) Service { + allPorts := make([]int, 0) + naming := 0 + for _, member := range members { + listening := ports(member) + if naming == 0 && len(listening) > 0 { + naming = member + } + allPorts = append(allPorts, listening...) + } + if naming == 0 && len(members) > 0 { + naming = members[0] + } + slices.Sort(allPorts) + allPorts = slices.Compact(allPorts) argv := req.Argv - if name := namingMember(members); name > 0 { - if a := cmdline(name); len(a) > 0 { + if naming > 0 { + if a := cmdline(naming); len(a) > 0 { argv = a } } name := filepath.Base(argv[0]) return Service{ Name: name, Argv: argv, PGID: pgid, - Ports: listenPorts(members), + Ports: allPorts, DataDir: dataDir(name, argv, root, req.Dir), } } @@ -100,18 +127,3 @@ func inside(root, dir, path string) string { } return filepath.ToSlash(rel) } - -// namingMember is the member that names a service: the first that listens on a -// port, else the first member, else 0. A shell that started a server lingers -// beside it, and the shell's command line says nothing of what is being served. -func namingMember(members []int) int { - for _, m := range members { - if len(listenPorts([]int{m})) > 0 { - return m - } - } - if len(members) > 0 { - return members[0] - } - return 0 -} diff --git a/internal/executor/service_darwin.go b/internal/executor/service_darwin.go new file mode 100644 index 0000000000..aa748b30b0 --- /dev/null +++ b/internal/executor/service_darwin.go @@ -0,0 +1,22 @@ +//go:build darwin + +package executor + +import ( + "slices" + + "github.com/Agent-Field/codeaf/internal/darwinproc" +) + +func cmdline(pid int) []string { return darwinproc.Argv(pid) } + +func pidPorts(pid int) []int { return darwinproc.ListenPorts(pid) } + +func listenPorts(pids []int) []int { + ports := make([]int, 0) + for _, pid := range pids { + ports = append(ports, pidPorts(pid)...) + } + slices.Sort(ports) + return slices.Compact(ports) +} diff --git a/internal/executor/service_describe_test.go b/internal/executor/service_describe_test.go new file mode 100644 index 0000000000..4ac262a0cc --- /dev/null +++ b/internal/executor/service_describe_test.go @@ -0,0 +1,64 @@ +package executor + +import ( + "reflect" + "testing" +) + +func TestDescribeInspectsEachMemberOnce(t *testing.T) { + members := []int{10, 11, 12} + wantArgv := []string{"postgres", "-D", "data"} + reads := make(map[int]int) + ports := func(pid int) []int { + reads[pid]++ + return map[int][]int{11: {8080, 5432}, 12: {8080}}[pid] + } + argv := func(pid int) []string { + if pid != 11 { + t.Fatalf("argv read for member %d, want first listener 11", pid) + } + return wantArgv + } + got := describeWith(ExecRequest{Argv: []string{"sh"}}, "/w", 10, members, ports, argv) + for _, pid := range members { + if reads[pid] != 1 { + t.Errorf("member %d inspected %d times, want once", pid, reads[pid]) + } + } + if got.Name != "postgres" || !reflect.DeepEqual(got.Argv, wantArgv) || + !reflect.DeepEqual(got.Ports, []int{5432, 8080}) || got.DataDir != "data" || got.PGID != 10 { + t.Fatalf("service = %+v", got) + } + + t.Run("no listeners", func(t *testing.T) { + for _, fallback := range []bool{false, true} { + reads := make(map[int]int) + req := ExecRequest{Argv: []string{"/bin/request", "argument"}} + wantArgv := []string{"/bin/sleep", "30"} + wantName := "sleep" + if fallback { + wantArgv, wantName = req.Argv, "request" + } + got := describeWith(req, "/w", 10, members, func(pid int) []int { + reads[pid]++ + return nil + }, func(pid int) []string { + if pid != members[0] { + t.Fatalf("argv read for member %d, want first member %d", pid, members[0]) + } + if fallback { + return nil + } + return wantArgv + }) + for _, pid := range members { + if reads[pid] != 1 { + t.Errorf("member %d inspected %d times, want once", pid, reads[pid]) + } + } + if got.Name != wantName || !reflect.DeepEqual(got.Argv, wantArgv) || len(got.Ports) != 0 { + t.Fatalf("fallback=%t: service = %+v", fallback, got) + } + } + }) +} diff --git a/internal/executor/service_linux.go b/internal/executor/service_linux.go index 559a307076..f51c85e82f 100644 --- a/internal/executor/service_linux.go +++ b/internal/executor/service_linux.go @@ -26,11 +26,8 @@ func cmdline(pid int) []string { func listenPorts(pids []int) []int { seen := map[int]bool{} for _, pid := range pids { - listening := listeningInodes(pid) - for inode := range socketInodes(pid) { - if port, ok := listening[inode]; ok { - seen[port] = true - } + for _, port := range pidPorts(pid) { + seen[port] = true } } ports := make([]int, 0, len(seen)) @@ -41,6 +38,17 @@ func listenPorts(pids []int) []int { return ports } +func pidPorts(pid int) []int { + listening := listeningInodes(pid) + ports := make([]int, 0) + for inode := range socketInodes(pid) { + if port, ok := listening[inode]; ok { + ports = append(ports, port) + } + } + return ports +} + func listeningInodes(pid int) map[uint64]int { out := map[uint64]int{} for _, table := range []string{"tcp", "tcp6"} { diff --git a/internal/executor/service_linux_test.go b/internal/executor/service_linux_test.go index 0eaf13ae71..b913825cb6 100644 --- a/internal/executor/service_linux_test.go +++ b/internal/executor/service_linux_test.go @@ -5,12 +5,8 @@ package executor import ( "reflect" "testing" - - "github.com/Agent-Field/codeaf/internal/processgroup" ) -func membersOf(pgid int) []int { return processgroup.Members(pgid) } - const tcpFixture = ` sl local_address rem_address st tx_queue rx_queue tr tm->when retrnsmt uid timeout inode 0: 0100007F:1F90 00000000:0000 0A 00000000:00000000 00:00000000 00000000 1000 0 12345 1 0000 100 0 0 10 0 1: 0100007F:0050 0100007F:9C40 01 00000000:00000000 00:00000000 00000000 1000 0 22222 1 0000 100 0 0 10 0 diff --git a/internal/executor/service_members_test.go b/internal/executor/service_members_test.go new file mode 100644 index 0000000000..526f50fbb0 --- /dev/null +++ b/internal/executor/service_members_test.go @@ -0,0 +1,7 @@ +//go:build linux || darwin + +package executor + +import "github.com/Agent-Field/codeaf/internal/processgroup" + +func membersOf(pgid int) []int { return processgroup.Members(pgid) } diff --git a/internal/executor/service_other.go b/internal/executor/service_other.go index 0a4ffe3f19..f16013f462 100644 --- a/internal/executor/service_other.go +++ b/internal/executor/service_other.go @@ -1,8 +1,10 @@ -//go:build !linux +//go:build !linux && !darwin package executor -// Off Linux the record keeps the request's argv and no ports. +// Where process inspection is unavailable, the record keeps the request's argv and no ports. func cmdline(int) []string { return nil } +func pidPorts(int) []int { return nil } + func listenPorts([]int) []int { return nil } diff --git a/internal/executor/service_other_test.go b/internal/executor/service_other_test.go index 328bd36598..1817e7ed78 100644 --- a/internal/executor/service_other_test.go +++ b/internal/executor/service_other_test.go @@ -1,4 +1,4 @@ -//go:build !linux +//go:build !linux && !darwin package executor diff --git a/internal/executor/service_unix_test.go b/internal/executor/service_unix_test.go index 80e1a8d473..f4887ff489 100644 --- a/internal/executor/service_unix_test.go +++ b/internal/executor/service_unix_test.go @@ -1,6 +1,6 @@ -//go:build !windows +//go:build linux || darwin -// K6 keeps Unix process-group service inspection on Unix. +// K6 keeps process-group service inspection on Linux and macOS. package executor import ( @@ -19,15 +19,17 @@ func TestLocalRecordsAListeningServiceFully(t *testing.T) { } s := res.Services[0] t.Cleanup(func() { _ = syscall.Kill(-s.PGID, syscall.SIGKILL) }) - if s.Name != "python3" || len(s.Argv) < 3 || s.Argv[1] != "-c" { - t.Fatalf("name/argv not recorded: %+v", s) - } - // The listener may bind just after the call returned; poll the record. + // The call can return before the backgrounded shell has exec'd python3 + // (macOS loses that race most times) and before the listener binds, so the + // record is polled until it names the server and its port. deadline := time.Now().Add(5 * time.Second) - for len(s.Ports) == 0 && time.Now().Before(deadline) { + for (s.Name != "python3" || len(s.Ports) == 0) && time.Now().Before(deadline) { time.Sleep(50 * time.Millisecond) s = describe(sh(script), root, s.PGID, membersOf(s.PGID)) } + if s.Name != "python3" || len(s.Argv) < 3 || s.Argv[1] != "-c" { + t.Fatalf("name/argv not recorded: %+v", s) + } if len(s.Ports) != 1 || s.Ports[0] == 0 { t.Fatalf("port not recorded: %+v", s) } diff --git a/internal/executor/watched_unix_test.go b/internal/executor/watched_unix_test.go index 5ce4d860d2..151d71e0a1 100644 --- a/internal/executor/watched_unix_test.go +++ b/internal/executor/watched_unix_test.go @@ -1,6 +1,6 @@ -//go:build !windows +//go:build linux || darwin -// K6 keeps Unix process, lock and mode assertions on the systems that provide them. +// K6 keeps process, lock and mode assertions on Linux and macOS. package executor import ( From 4dd98b53df26f1236d2a5e49744f175ed1e0138b Mon Sep 17 00:00:00 2001 From: Abir Abbas Date: Thu, 29 Oct 2026 14:16:11 -0400 Subject: [PATCH 3/8] docs: the job subtree bound, and that macOS now applies it Co-Authored-By: Claude Opus 5.5 --- .../0-app-darwin-process-inspection.md | 19 +++++++++++++++++++ internal/manual/chat/what-i-can-do.md | 19 +++++++++++++++++++ 2 files changed, 38 insertions(+) create mode 100644 docs/changes/unreleased/0-app-darwin-process-inspection.md diff --git a/docs/changes/unreleased/0-app-darwin-process-inspection.md b/docs/changes/unreleased/0-app-darwin-process-inspection.md new file mode 100644 index 0000000000..55a300f5b5 --- /dev/null +++ b/docs/changes/unreleased/0-app-darwin-process-inspection.md @@ -0,0 +1,19 @@ +--- +kind: fixed +title: macOS reads process groups for left-running services and the job subtree bound +pr: 0 +surface: [engine, docs] +invalidates: + - "On macOS, a server left running by a shell command was never reported, and service records named the shell without its server's argv or listening ports. These records now inspect the surviving processes." + - "The job subtree CPU and process bound never cut on macOS because subtree usage could not be read. macOS now measures the whole tree and applies the existing bound." +--- + +On macOS, a registered bash job that sustains more than three fifths of the +machine's logical cores for thirty seconds, or grows past 128 processes for +that long, is now ended like on Linux. The CPU bound applies when no CPU quota +or affinity mask holds codeaf below the machine's cores. Both limits require +15 consecutive readings two seconds apart, and the run is told why it ended. + +Process inspection works without cgo, including release and sidecar builds. +Service descriptions read each surviving member's sockets once, select the +first listening member's argv and collect the sorted, unique listening ports. diff --git a/internal/manual/chat/what-i-can-do.md b/internal/manual/chat/what-i-can-do.md index 7d40311e6f..1f897bfee6 100644 --- a/internal/manual/chat/what-i-can-do.md +++ b/internal/manual/chat/what-i-can-do.md @@ -519,6 +519,25 @@ parts it has, and however long it takes — belongs to a task instead, which giv to watch and a report you can read. If a multi-part piece of work was started as a background command, say so: it can be handed to a task instead. +## Why was my background job ended — the CPU and process bound on a job + +The bound applies to `bash` jobs: both background commands and foreground +commands kept as jobs. It measures the job's whole process tree, including its +children, rather than just the shell that started it. + +A job's tree may not hold more than **128 processes**. When no CPU quota or +affinity mask holds codeaf below the machine's logical core count, the tree +also may not sustain more than **three fifths of those logical cores**. Under +a binding quota or affinity mask, only the process-count bound applies. + +Either limit exceeded for **15 consecutive readings, 2 seconds apart** +(**thirty seconds**) ends the job. A brief spike does not end it. The run is +told **"the job subtree was not settled within its bound"**, with the job's +process count and CPU usage, so the ending is explained. + +The tree is read on **Linux and macOS**. Where the tree cannot be read, +including **Windows**, this bound never cuts a job. + ## How much disk space do finished job logs keep? Finished logs do not pile up forever either. Each session's logs/jobs From f4b6f9bfffc6321e691bbaa2cc70c10ca04bbe57 Mon Sep 17 00:00:00 2001 From: Abir Abbas Date: Thu, 29 Oct 2026 14:29:41 -0400 Subject: [PATCH 4/8] processgroup, darwinproc: bound the macOS walk and keep strangers out Review fixes: - A queued parent is re-read after its children are listed and its start time compared, so a pid handed to another process cannot bring that process's children into the job's reading. - A child listing that fills the largest buffer, and a walk that reaches 1024 processes, count as lower bounds; any unexpected listing failure makes the reading unknown instead of partial. - kern.procargs2 is parsed at the kernel's eight-byte boundary after the executable path, so an empty argv[0] can never shift an environment string into argv (checked for path lengths 1-40 on arm64 and under Rosetta). - The listening-service test's shell now waits for the server's port before it exits, so the record the call returns is asserted directly; the membersOf helpers it no longer needs are gone. Co-Authored-By: Claude Opus 5.5 --- internal/darwinproc/procinfo_darwin.go | 48 +++++-- internal/darwinproc/procinfo_darwin_test.go | 67 +++++++-- internal/executor/service_members_test.go | 7 - internal/executor/service_other_test.go | 5 - internal/executor/service_unix_test.go | 17 +-- internal/processgroup/usage_darwin.go | 81 +++++++++-- .../processgroup/usage_darwin_walk_test.go | 134 ++++++++++++++++++ 7 files changed, 304 insertions(+), 55 deletions(-) delete mode 100644 internal/executor/service_members_test.go delete mode 100644 internal/executor/service_other_test.go create mode 100644 internal/processgroup/usage_darwin_walk_test.go diff --git a/internal/darwinproc/procinfo_darwin.go b/internal/darwinproc/procinfo_darwin.go index 0bc7acd9b5..b190e614d3 100644 --- a/internal/darwinproc/procinfo_darwin.go +++ b/internal/darwinproc/procinfo_darwin.go @@ -65,6 +65,10 @@ const ( childrenInitialPIDs = 64 childrenGrowth = 4 childrenAttempts = 4 + + // procArgsAlign is the boundary the kernel pads the executable path to + // before the arguments begin: the pointer size of every supported Mac. + procArgsAlign = 8 ) // procInfo keeps pointer conversion in the syscall expression so the buffer @@ -91,7 +95,9 @@ func procInfo(call, a1, a2, a3 int, buf []byte) (int, error) { } // Children lists immediate children, bounding retries when the tree grows -// between reads. A full final buffer is unknown rather than a partial tree. +// between reads. When even the last buffer fills, the children it holds are +// returned with ENOMEM: they are a lower bound on the children, never the +// whole list, and a caller counting them must say so. func Children(ppid int) ([]int, error) { if ppid <= 0 { return nil, unix.EINVAL @@ -107,27 +113,35 @@ func Children(ppid int) ([]int, error) { return nil, unix.EIO } if n < len(buf) { - pids := make([]int, 0, n/pidSize) - for offset := 0; offset < n; offset += pidSize { - if pid := int(int32(binary.LittleEndian.Uint32(buf[offset:]))); pid > 0 { - pids = append(pids, pid) - } - } - return pids, nil + return childPIDs(buf[:n]), nil } if attempt+1 < childrenAttempts { buf = make([]byte, len(buf)*childrenGrowth) } } - return nil, unix.ENOMEM + return childPIDs(buf), unix.ENOMEM +} + +func childPIDs(buf []byte) []int { + pids := make([]int, 0, len(buf)/pidSize) + for offset := 0; offset+pidSize <= len(buf); offset += pidSize { + if pid := int(int32(binary.LittleEndian.Uint32(buf[offset:]))); pid > 0 { + pids = append(pids, pid) + } + } + return pids } // TaskInfo combines identity and CPU from one kernel snapshot so callers can // validate ancestry before attributing a process's accumulated CPU time. type TaskInfo struct { PID, PPID, PGID, Status int - CPUSeconds float64 - Resident uint64 + // Start is the process's start time in microseconds. A pid handed out + // again names a process with another start, so a reader that must not + // confuse the two compares it. + Start uint64 + CPUSeconds float64 + Resident uint64 } var ( @@ -158,6 +172,7 @@ func Task(pid int) (TaskInfo, error) { PPID: int(int32(binary.LittleEndian.Uint32(buf[bsdPPIDOffset:]))), PGID: int(int32(binary.LittleEndian.Uint32(buf[bsdPGIDOffset:]))), Status: int(binary.LittleEndian.Uint32(buf[bsdStatusOffset:])), + Start: binary.LittleEndian.Uint64(buf[bsdStartSecondsOffset:])*1_000_000 + binary.LittleEndian.Uint64(buf[bsdStartMicrosOffset:]), CPUSeconds: float64(binary.LittleEndian.Uint64(buf[taskUserOffset:]))/float64(tickFrequency) + float64(binary.LittleEndian.Uint64(buf[taskSystemOffset:]))/float64(tickFrequency), Resident: binary.LittleEndian.Uint64(buf[taskResidentOffset:]), }, nil @@ -229,7 +244,16 @@ func parseProcArgs(raw []byte) []string { if end < 0 { return nil } - rest = bytes.TrimLeft(rest[end+1:], "\x00") + // The kernel pads the executable path with NULs so the arguments start on + // a pointer-size boundary of the string area. Skipping every NUL instead + // would swallow an empty argv[0] and shift the first environment string, + // which can hold a secret, into the arguments. Padding that is not all NUL + // is a layout this parser does not know, and it reads as nothing. + start := (end + 1 + procArgsAlign - 1) &^ (procArgsAlign - 1) + if start > len(rest) || len(bytes.Trim(rest[end+1:start], "\x00")) != 0 { + return nil + } + rest = rest[start:] argv := make([]string, 0, argc) for range argc { end = bytes.IndexByte(rest, 0) diff --git a/internal/darwinproc/procinfo_darwin_test.go b/internal/darwinproc/procinfo_darwin_test.go index 1da8307dc7..f4ce0b1c5c 100644 --- a/internal/darwinproc/procinfo_darwin_test.go +++ b/internal/darwinproc/procinfo_darwin_test.go @@ -16,8 +16,26 @@ import ( "time" ) +// procArgsFixture lays out a kern.procargs2 answer the way the kernel does: +// argc, the executable path padded with NULs to the next eight-byte boundary, +// then the arguments and the environment. +func procArgsFixture(argc int32, exec string, strings ...string) []byte { + raw := make([]byte, 4) + binary.LittleEndian.PutUint32(raw, uint32(argc)) + raw = append(raw, exec...) + raw = append(raw, 0) + for (len(raw)-4)%procArgsAlign != 0 { + raw = append(raw, 0) + } + for _, s := range strings { + raw = append(raw, s...) + raw = append(raw, 0) + } + return raw +} + func TestDarwinProcArgsStopsBeforeEnvironment(t *testing.T) { - rawArgs := func(argc int32, body string) []byte { + header := func(argc int32, body string) []byte { raw := make([]byte, 4, 4+len(body)) binary.LittleEndian.PutUint32(raw, uint32(argc)) return append(raw, body...) @@ -27,13 +45,17 @@ func TestDarwinProcArgsStopsBeforeEnvironment(t *testing.T) { raw []byte want []string }{ - {"padding and environment", rawArgs(4, "/bin/tool\x00\x00\x00tool\x00\x00two words\x00last\x00PRIVATE=value\x00"), []string{"tool", "", "two words", "last"}}, - {"zero", rawArgs(0, "/bin/tool\x00tool\x00"), nil}, - {"negative", rawArgs(-1, "/bin/tool\x00tool\x00"), nil}, - {"absurd", rawArgs(1<<30, "/bin/tool\x00tool\x00"), nil}, - {"missing argument", rawArgs(2, "/bin/tool\x00tool\x00"), nil}, - {"truncated argument", rawArgs(1, "/bin/tool\x00tool"), nil}, - {"truncated executable", rawArgs(1, "/bin/tool"), nil}, + {"padding and environment", procArgsFixture(4, "/bin/tool", "tool", "", "two words", "last", "PRIVATE=value"), []string{"tool", "", "two words", "last"}}, + {"empty argv[0] keeps the environment out", procArgsFixture(2, "/bin/tool", "", "argument", "PRIVATE=value"), []string{"", "argument"}}, + {"path that fills its boundary", procArgsFixture(1, "/bin/to", "tool", "PRIVATE=value"), []string{"tool"}}, + {"padding that is not NUL", header(1, "/bin/tool\x00\x00tool\x00PRIVATE=value\x00"), nil}, + {"zero", procArgsFixture(0, "/bin/tool", "tool"), nil}, + {"negative", procArgsFixture(-1, "/bin/tool", "tool"), nil}, + {"absurd", procArgsFixture(1<<30, "/bin/tool", "tool"), nil}, + {"missing argument", procArgsFixture(2, "/bin/tool", "tool"), nil}, + {"truncated argument", header(1, "/bin/tool\x00\x00\x00\x00\x00\x00\x00tool"), nil}, + {"truncated padding", header(1, "/bin/tool\x00\x00"), nil}, + {"truncated executable", header(1, "/bin/tool"), nil}, {"short header", []byte{1, 0, 0}, nil}, } { t.Run(tc.name, func(t *testing.T) { @@ -44,6 +66,35 @@ func TestDarwinProcArgsStopsBeforeEnvironment(t *testing.T) { } } +// An empty argv[0] is read in place on a live process, with the environment +// left out, whatever the length of the executable path. +func TestDarwinArgvKeepsAnEmptyFirstArgument(t *testing.T) { + dir := t.TempDir() + for _, name := range []string{"s", "sl", "sleeper", "sleeperx"} { + link := filepath.Join(dir, name) + if err := os.Symlink("/bin/sleep", link); err != nil { + t.Fatal(err) + } + cmd := exec.Command(link, "30") + cmd.Args[0] = "" + cmd.Env = []string{"PRIVATE=value"} + cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true} + if err := cmd.Start(); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = cmd.Process.Kill(); _ = cmd.Wait() }) + var got []string + for deadline := time.Now().Add(2 * time.Second); time.Now().Before(deadline); time.Sleep(5 * time.Millisecond) { + if got = Argv(cmd.Process.Pid); reflect.DeepEqual(got, []string{"", "30"}) { + break + } + } + if !reflect.DeepEqual(got, []string{"", "30"}) { + t.Fatalf("%s: argv = %q, want [\"\" \"30\"]", name, got) + } + } +} + func sleepChild(t testing.TB) *exec.Cmd { t.Helper() cmd := exec.Command("sleep", "30") diff --git a/internal/executor/service_members_test.go b/internal/executor/service_members_test.go deleted file mode 100644 index 526f50fbb0..0000000000 --- a/internal/executor/service_members_test.go +++ /dev/null @@ -1,7 +0,0 @@ -//go:build linux || darwin - -package executor - -import "github.com/Agent-Field/codeaf/internal/processgroup" - -func membersOf(pgid int) []int { return processgroup.Members(pgid) } diff --git a/internal/executor/service_other_test.go b/internal/executor/service_other_test.go deleted file mode 100644 index 1817e7ed78..0000000000 --- a/internal/executor/service_other_test.go +++ /dev/null @@ -1,5 +0,0 @@ -//go:build !linux && !darwin - -package executor - -func membersOf(int) []int { return nil } diff --git a/internal/executor/service_unix_test.go b/internal/executor/service_unix_test.go index f4887ff489..679d03064a 100644 --- a/internal/executor/service_unix_test.go +++ b/internal/executor/service_unix_test.go @@ -7,11 +7,16 @@ import ( "context" "syscall" "testing" - "time" ) func TestLocalRecordsAListeningServiceFully(t *testing.T) { - script := "exec python3 -c 'import socket,time;s=socket.socket();s.bind((\"127.0.0.1\",0));s.listen();print(s.getsockname()[1],flush=True);time.sleep(60)' >port.txt 2>&1 port.txt 2>&1 = usageWalkLimit { + continue + } + children, err := usageReader.children(parent.PID) + switch { + case err == nil, errors.Is(err, unix.ENOMEM): + case errors.Is(err, unix.ESRCH): + continue + 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 + } continue } for _, pid := range children { + if len(seen) >= usageWalkLimit { + break + } if pid <= 0 || seen[pid] { continue } seen[pid] = true - child, err := darwinproc.Task(pid) + child, err := usageReader.task(pid) // A child can exit, be reparented or have its pid reused between reads. // Skip that snapshot rather than charging an unrelated process to us. if err != nil || child.PID != pid || child.PPID != parent.PID || child.Status == darwinproc.StatusZombie { @@ -42,8 +100,5 @@ func (g Group) Usage() (Usage, bool) { queue = append(queue, child) } } - if ok, _ := g.owned(); !ok { - return Usage{}, false - } return usage, true } diff --git a/internal/processgroup/usage_darwin_walk_test.go b/internal/processgroup/usage_darwin_walk_test.go new file mode 100644 index 0000000000..a0d5d73794 --- /dev/null +++ b/internal/processgroup/usage_darwin_walk_test.go @@ -0,0 +1,134 @@ +//go:build darwin + +package processgroup + +import ( + "testing" + + "github.com/Agent-Field/codeaf/internal/darwinproc" + "golang.org/x/sys/unix" +) + +// scriptedTree answers the walk from a fixed table, so its limits and refusals +// are proven on a tree that exists only as numbers and never as real load. +type scriptedTree struct { + tasks map[int]darwinproc.TaskInfo + children map[int][]int + listErr map[int]error + // restart gives a pid a new start time after its children are listed: the + // pid was handed to another process in between. + restart map[int]bool + listed map[int]bool + taskRead int +} + +func (s *scriptedTree) use(t *testing.T) { + t.Helper() + saved := usageReader + t.Cleanup(func() { usageReader = saved }) + s.listed = map[int]bool{} + usageReader.task = func(pid int) (darwinproc.TaskInfo, error) { + s.taskRead++ + task, ok := s.tasks[pid] + if !ok { + return darwinproc.TaskInfo{}, unix.ESRCH + } + if s.restart[pid] && s.listed[pid] { + task.Start++ + } + return task, nil + } + usageReader.children = func(ppid int) ([]int, error) { + s.listed[ppid] = true + return s.children[ppid], s.listErr[ppid] + } +} + +func task(pid, ppid, pgid int, cpu float64) darwinproc.TaskInfo { + return darwinproc.TaskInfo{PID: pid, PPID: ppid, PGID: pgid, Start: uint64(pid) * 1000, CPUSeconds: cpu, Status: 2} +} + +// The job's tree is its children's children too, including one that moved to +// a process group of its own; each live process is counted once. +func TestDarwinUsageCountsTheWholeTreeOnce(t *testing.T) { + tree := &scriptedTree{ + tasks: map[int]darwinproc.TaskInfo{ + 10: task(10, 1, 10, 1), 11: task(11, 10, 10, 2), 12: task(12, 10, 99, 3), 13: task(13, 11, 10, 4), + }, + children: map[int][]int{10: {11, 12, 11}, 11: {13}, 12: {13}}, + } + tree.use(t) + usage, ok := subtreeUsage(10) + if !ok || usage.Processes != 4 || usage.CPUSeconds != 10 { + t.Fatalf("usage = %+v, %v, want 4 processes and 10 CPU seconds", usage, ok) + } +} + +// A queued process whose pid went to another process before its children were +// listed contributes itself but none of the stranger's children. +func TestDarwinUsageKeepsAReusedParentsChildrenOut(t *testing.T) { + tree := &scriptedTree{ + tasks: map[int]darwinproc.TaskInfo{ + 10: task(10, 1, 10, 1), 11: task(11, 10, 10, 1), 30: task(30, 11, 30, 50), + }, + children: map[int][]int{10: {11}, 11: {30}}, + restart: map[int]bool{11: true}, + } + tree.use(t) + 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", usage, ok) + } + tree.restart, tree.listed = map[int]bool{10: true}, map[int]bool{} + if usage, ok := subtreeUsage(10); ok { + t.Fatalf("a root whose pid changed hands mid-walk was read: %+v", usage) + } +} + +// A listing that filled the largest buffer is a lower bound and still counts; +// an unexpected failure makes the reading unknown rather than partial. +func TestDarwinUsageCountsATruncatedListingAndRefusesAFailedOne(t *testing.T) { + tasks := map[int]darwinproc.TaskInfo{10: task(10, 1, 10, 0)} + var kids []int + for pid := 100; pid < 400; pid++ { + tasks[pid] = task(pid, 10, 10, 0) + kids = append(kids, pid) + } + tree := &scriptedTree{tasks: tasks, children: map[int][]int{10: kids}, listErr: map[int]error{10: unix.ENOMEM}} + tree.use(t) + if usage, ok := subtreeUsage(10); !ok || usage.Processes != 301 { + t.Fatalf("truncated listing: usage = %+v, %v, want at least the 301 it named", usage, ok) + } + tree.listErr = map[int]error{10: unix.EPERM} + if usage, ok := subtreeUsage(10); ok { + t.Fatalf("a listing that failed was read as a partial tree: %+v", usage) + } +} + +// A tree past the walk limit is read as far as the limit, with a bounded +// number of kernel reads, and is still far over the job bound's ceiling. +func TestDarwinUsageStopsAtTheWalkLimit(t *testing.T) { + tasks := map[int]darwinproc.TaskInfo{10: task(10, 1, 10, 0)} + children := map[int][]int{} + parent, next := 10, 11 + for i := 0; i < 5000; i++ { + tasks[next] = task(next, parent, 10, 0) + children[parent] = append(children[parent], next) + if i%3 == 2 { + parent = next + } + next++ + } + tree := &scriptedTree{tasks: tasks, children: children} + tree.use(t) + usage, ok := subtreeUsage(10) + if !ok || usage.Processes != usageWalkLimit { + t.Fatalf("usage = %+v, %v, want exactly the walk limit %d", usage, ok, usageWalkLimit) + } + if tree.taskRead > 2*usageWalkLimit+1 { + t.Fatalf("%d task reads for a walk limited to %d processes", tree.taskRead, usageWalkLimit) + } + if len(tree.listed) > usageWalkLimit { + t.Fatalf("%d child listings for a walk limited to %d processes", len(tree.listed), usageWalkLimit) + } +} From c449441bd44a9c970a8800a1bd337c3edb3d5e4e Mon Sep 17 00:00:00 2001 From: Abir Abbas Date: Thu, 29 Oct 2026 14:45:04 -0400 Subject: [PATCH 5/8] processgroup, session: a partial CPU sum yields no rate Review fix: a reading that left part of the tree out (the walk limit, a truncated child listing, a process that could not be read) still counts processes as a lower bound, but its CPU sum is marked partial. The job bound takes no rate from it and does not use it as the next baseline, so a process missing from one reading and present in the next cannot read as a burst of its whole lifetime's CPU and invent a strike. Linux readings are always complete, so nothing changes there. Co-Authored-By: Claude Opus 5.5 --- internal/processgroup/usage.go | 7 ++++ internal/processgroup/usage_darwin.go | 31 ++++++++++------ .../processgroup/usage_darwin_walk_test.go | 35 +++++++++++++++---- internal/session/jobbound.go | 14 ++++++-- internal/session/jobbound_test.go | 34 ++++++++++++++++++ 5 files changed, 102 insertions(+), 19 deletions(-) diff --git a/internal/processgroup/usage.go b/internal/processgroup/usage.go index 1d5e1b6a85..7eb8c794f0 100644 --- a/internal/processgroup/usage.go +++ b/internal/processgroup/usage.go @@ -28,4 +28,11 @@ type Usage struct { // every process in it. Subtract two readings and divide by the wall between // them to get the cores the subtree held. CPUSeconds float64 + // CPUPartial says CPUSeconds left out part of the tree: the walk stopped at + // its limit, a listing was truncated, or a process could not be read. Such + // a sum is not comparable with another reading, because a process missing + // from one and present in the next would read as a burst of its whole + // lifetime's CPU, so no rate may be taken from it. Processes is still a + // lower bound. + CPUPartial bool } diff --git a/internal/processgroup/usage_darwin.go b/internal/processgroup/usage_darwin.go index c18c419168..24bb2b770f 100644 --- a/internal/processgroup/usage_darwin.go +++ b/internal/processgroup/usage_darwin.go @@ -9,10 +9,11 @@ import ( "golang.org/x/sys/unix" ) -// usageWalkLimit is how many processes one reading visits at most. Past it the -// reading is a lower bound on both figures: still far over the 128-process -// ceiling the job bound states, so a fork storm is cut on the count, while a -// tree that keeps growing cannot make every two-second sample a longer walk. +// usageWalkLimit is how many processes one reading looks at most. Past it the +// process count is a lower bound and the CPU sum is marked partial, so a tree +// that keeps growing cannot make every two-second sample a longer walk. A storm +// that size normally still counts far over the job bound's 128-process ceiling +// and is cut on the count; a lower count can only hold a cut back. const usageWalkLimit = 1024 // usageReader is the kernel's answer for one process and for its children. It @@ -48,11 +49,13 @@ func (g Group) Usage() (Usage, bool) { // 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 listing that filled even the largest buffer, and a walk that reached -// [usageWalkLimit], are counted as far as they got: lower bounds can only hold -// a cut back, and a count that large is already over any process ceiling. A -// process that went away between two reads is skipped, as the Linux reading -// skips it; any other failure makes the whole reading unknown. +// 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 +// than having exited are counted as far as they got, with the CPU sum marked +// partial ([Usage.CPUPartial]): the count is a lower bound and no rate is taken +// from the sum. A process that exited between two reads is skipped, as the +// Linux reading skips it. A listing that fails outright makes the whole +// reading unknown. func subtreeUsage(root int) (Usage, bool) { top, err := usageReader.task(root) if err != nil || top.PID != root || top.Status == darwinproc.StatusZombie { @@ -66,11 +69,14 @@ func subtreeUsage(root int) (Usage, bool) { usage.Processes++ usage.CPUSeconds += parent.CPUSeconds if len(seen) >= usageWalkLimit { + usage.CPUPartial = true continue } children, err := usageReader.children(parent.PID) switch { - case err == nil, errors.Is(err, unix.ENOMEM): + case err == nil: + case errors.Is(err, unix.ENOMEM): + usage.CPUPartial = true case errors.Is(err, unix.ESRCH): continue default: @@ -85,6 +91,7 @@ func subtreeUsage(root int) (Usage, bool) { } for _, pid := range children { if len(seen) >= usageWalkLimit { + usage.CPUPartial = true break } if pid <= 0 || seen[pid] { @@ -92,6 +99,10 @@ func subtreeUsage(root int) (Usage, bool) { } seen[pid] = true child, err := usageReader.task(pid) + if err != nil && !errors.Is(err, unix.ESRCH) { + usage.CPUPartial = true + continue + } // A child can exit, be reparented or have its pid reused between reads. // Skip that snapshot rather than charging an unrelated process to us. if err != nil || child.PID != pid || child.PPID != parent.PID || child.Status == darwinproc.StatusZombie { diff --git a/internal/processgroup/usage_darwin_walk_test.go b/internal/processgroup/usage_darwin_walk_test.go index a0d5d73794..5d5fd46b26 100644 --- a/internal/processgroup/usage_darwin_walk_test.go +++ b/internal/processgroup/usage_darwin_walk_test.go @@ -59,8 +59,8 @@ func TestDarwinUsageCountsTheWholeTreeOnce(t *testing.T) { } tree.use(t) usage, ok := subtreeUsage(10) - if !ok || usage.Processes != 4 || usage.CPUSeconds != 10 { - t.Fatalf("usage = %+v, %v, want 4 processes and 10 CPU seconds", usage, ok) + if !ok || usage.Processes != 4 || usage.CPUSeconds != 10 || usage.CPUPartial { + t.Fatalf("usage = %+v, %v, want 4 processes and a complete 10 CPU seconds", usage, ok) } } @@ -96,8 +96,8 @@ func TestDarwinUsageCountsATruncatedListingAndRefusesAFailedOne(t *testing.T) { } tree := &scriptedTree{tasks: tasks, children: map[int][]int{10: kids}, listErr: map[int]error{10: unix.ENOMEM}} tree.use(t) - if usage, ok := subtreeUsage(10); !ok || usage.Processes != 301 { - t.Fatalf("truncated listing: usage = %+v, %v, want at least the 301 it named", usage, ok) + if usage, ok := subtreeUsage(10); !ok || usage.Processes != 301 || !usage.CPUPartial { + t.Fatalf("truncated listing: usage = %+v, %v, want the 301 it named and a partial CPU sum", usage, ok) } tree.listErr = map[int]error{10: unix.EPERM} if usage, ok := subtreeUsage(10); ok { @@ -122,8 +122,8 @@ func TestDarwinUsageStopsAtTheWalkLimit(t *testing.T) { tree := &scriptedTree{tasks: tasks, children: children} tree.use(t) usage, ok := subtreeUsage(10) - if !ok || usage.Processes != usageWalkLimit { - t.Fatalf("usage = %+v, %v, want exactly the walk limit %d", usage, ok, usageWalkLimit) + if !ok || usage.Processes != usageWalkLimit || !usage.CPUPartial { + t.Fatalf("usage = %+v, %v, want exactly the walk limit %d and a partial CPU sum", usage, ok, usageWalkLimit) } if tree.taskRead > 2*usageWalkLimit+1 { t.Fatalf("%d task reads for a walk limited to %d processes", tree.taskRead, usageWalkLimit) @@ -132,3 +132,26 @@ func TestDarwinUsageStopsAtTheWalkLimit(t *testing.T) { t.Fatalf("%d child listings for a walk limited to %d processes", len(tree.listed), usageWalkLimit) } } + +// A child that exited between two reads leaves the tree, as on Linux; one that +// could not be read for another reason makes the CPU sum partial. +func TestDarwinUsageMarksAnUnreadableChildPartial(t *testing.T) { + tree := &scriptedTree{ + tasks: map[int]darwinproc.TaskInfo{10: task(10, 1, 10, 1)}, + children: map[int][]int{10: {11}}, + } + tree.use(t) + if usage, ok := subtreeUsage(10); !ok || usage.Processes != 1 || usage.CPUPartial { + t.Fatalf("an exited child: usage = %+v, %v, want the leader alone and a complete sum", usage, ok) + } + saved := usageReader.task + usageReader.task = func(pid int) (darwinproc.TaskInfo, error) { + if pid == 11 { + return darwinproc.TaskInfo{}, unix.EPERM + } + return saved(pid) + } + if usage, ok := subtreeUsage(10); !ok || usage.Processes != 1 || !usage.CPUPartial { + t.Fatalf("an unreadable child: usage = %+v, %v, want the leader and a partial sum", usage, ok) + } +} diff --git a/internal/session/jobbound.go b/internal/session/jobbound.go index a782ee49f2..d0ef847d13 100644 --- a/internal/session/jobbound.go +++ b/internal/session/jobbound.go @@ -179,10 +179,18 @@ func (b *subtreeBound) strike(now time.Time, usage processgroup.Usage, ok bool) return false } cores := 0.0 - if b.have && now.After(b.last) { - cores = (usage.CPUSeconds - b.lastCPU) / now.Sub(b.last).Seconds() + if usage.CPUPartial { + // A sum that left part of the tree out yields no rate, and it is no + // baseline for the next reading either: only the process count is + // judged, and the next complete reading arms the CPU half again. + b.have = false + } else { + if b.have && now.After(b.last) { + cores = (usage.CPUSeconds - b.lastCPU) / now.Sub(b.last).Seconds() + } + b.last, b.lastCPU, b.have = now, usage.CPUSeconds, true } - b.last, b.lastCPU, b.have, b.lastCores = now, usage.CPUSeconds, true, cores + b.lastCores = cores if usage.Processes > jobBoundProcesses || cores > b.ceilingCores() { b.strikes++ } else { diff --git a/internal/session/jobbound_test.go b/internal/session/jobbound_test.go index cfa0898075..38933cf368 100644 --- a/internal/session/jobbound_test.go +++ b/internal/session/jobbound_test.go @@ -230,3 +230,37 @@ func TestASubtreeThatCannotBeReadIsNeverCut(t *testing.T) { t.Fatal("a subtree whose usage could not be read was cut: silence is not pressure") } } + +// TestAPartialCPUSumNeverMakesARate: a reading that left part of the tree out +// (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. +func TestAPartialCPUSumNeverMakesARate(t *testing.T) { + registry := &jobRegistry{} + one := &job{id: 1, kind: jobKindBash, state: jobRunning, done: make(chan struct{})} + var readings []processgroup.Usage + for i := 0; i < jobBoundStrikes-1; i++ { + readings = append(readings, processgroup.Usage{Processes: 1100, CPUSeconds: 1, CPUPartial: true}) + } + readings = append(readings, processgroup.Usage{Processes: 2, CPUSeconds: 1001}, processgroup.Usage{Processes: 2, CPUSeconds: 1001}) + subtreeReadings(t, readings, 2*time.Second) + if driveBound(registry, one, newTestBound(), len(readings)) { + t.Fatal("a jump between a partial and a complete CPU sum was read as a storm and cut") + } +} + +// TestAPartialReadingStillCountsTowardsTheProcessBackstop: the count from a +// partial reading is a lower bound, and a fork storm read that way is cut. +func TestAPartialReadingStillCountsTowardsTheProcessBackstop(t *testing.T) { + registry := &jobRegistry{} + one := &job{id: 1, kind: jobKindBash, state: jobRunning, done: make(chan struct{})} + readings := storm(0.5, 1024, long, time.Second) + for i := range readings { + readings[i].CPUPartial = true + } + subtreeReadings(t, readings, time.Second) + if !driveBound(registry, one, newTestBound(), long) { + t.Fatal("a fork storm read partially was never cut by the process backstop") + } +} From 37f3a47cd4a03ba3c63f992a84bce1bdce648c8a Mon Sep 17 00:00:00 2001 From: Abir Abbas Date: Thu, 29 Oct 2026 14:48:03 -0400 Subject: [PATCH 6/8] processgroup: a parent that cannot be re-read makes the sum partial Review fix: when a queued parent read once but its re-read after the child listing fails for a reason other than having exited, its children are left out of this reading only, so the CPU sum is marked partial and no rate is taken from it. Co-Authored-By: Claude Opus 5.5 --- internal/processgroup/usage_darwin.go | 6 +++++ .../processgroup/usage_darwin_walk_test.go | 27 +++++++++++++++++++ 2 files changed, 33 insertions(+) diff --git a/internal/processgroup/usage_darwin.go b/internal/processgroup/usage_darwin.go index 24bb2b770f..52aaa9c534 100644 --- a/internal/processgroup/usage_darwin.go +++ b/internal/processgroup/usage_darwin.go @@ -87,6 +87,12 @@ func subtreeUsage(root int) (Usage, bool) { 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 } for _, pid := range children { diff --git a/internal/processgroup/usage_darwin_walk_test.go b/internal/processgroup/usage_darwin_walk_test.go index 5d5fd46b26..0d7a63d28f 100644 --- a/internal/processgroup/usage_darwin_walk_test.go +++ b/internal/processgroup/usage_darwin_walk_test.go @@ -155,3 +155,30 @@ func TestDarwinUsageMarksAnUnreadableChildPartial(t *testing.T) { t.Fatalf("an unreadable child: usage = %+v, %v, want the leader and a partial sum", usage, ok) } } + +// A parent that read once and then could not be re-read leaves its children out +// of this reading only, so the sum is partial; one that exited does not. +func TestDarwinUsageMarksAnUnreadableParentPartial(t *testing.T) { + tree := &scriptedTree{ + tasks: map[int]darwinproc.TaskInfo{10: task(10, 1, 10, 1), 11: task(11, 10, 10, 1), 12: task(12, 11, 10, 9)}, + children: map[int][]int{10: {11}, 11: {12}}, + } + tree.use(t) + saved := usageReader.task + reads := map[int]int{} + failWith := unix.EPERM + usageReader.task = func(pid int) (darwinproc.TaskInfo, error) { + reads[pid]++ + if pid == 11 && reads[pid] == 2 { + return darwinproc.TaskInfo{}, failWith + } + return saved(pid) + } + if usage, ok := subtreeUsage(10); !ok || usage.Processes != 2 || !usage.CPUPartial { + t.Fatalf("a parent that could not be re-read: usage = %+v, %v, want 2 processes and a partial sum", usage, ok) + } + reads, failWith = map[int]int{}, unix.ESRCH + if usage, ok := subtreeUsage(10); !ok || usage.Processes != 2 || usage.CPUPartial { + t.Fatalf("a parent that exited: usage = %+v, %v, want 2 processes and a complete sum", usage, ok) + } +} From 6ac0fb7c7742acadadc5ac033df3bb4a6a49eac8 Mon Sep 17 00:00:00 2001 From: Abir Abbas Date: Thu, 29 Oct 2026 19:16:25 -0400 Subject: [PATCH 7/8] docs: number the change entry for #2039 Co-Authored-By: Claude Opus 5.5 --- ...-process-inspection.md => 2039-darwin-process-inspection.md} | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) rename docs/changes/unreleased/{0-app-darwin-process-inspection.md => 2039-darwin-process-inspection.md} (99%) diff --git a/docs/changes/unreleased/0-app-darwin-process-inspection.md b/docs/changes/unreleased/2039-darwin-process-inspection.md similarity index 99% rename from docs/changes/unreleased/0-app-darwin-process-inspection.md rename to docs/changes/unreleased/2039-darwin-process-inspection.md index 55a300f5b5..5a91eb265e 100644 --- a/docs/changes/unreleased/0-app-darwin-process-inspection.md +++ b/docs/changes/unreleased/2039-darwin-process-inspection.md @@ -1,7 +1,7 @@ --- kind: fixed title: macOS reads process groups for left-running services and the job subtree bound -pr: 0 +pr: 2039 surface: [engine, docs] invalidates: - "On macOS, a server left running by a shell command was never reported, and service records named the shell without its server's argv or listening ports. These records now inspect the surviving processes." From 0cf71525152ee05798bebf5bed33dcad0a00a784 Mon Sep 17 00:00:00 2001 From: Abir Abbas Date: Thu, 29 Oct 2026 19:29:07 -0400 Subject: [PATCH 8/8] 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 --- internal/processgroup/usage_darwin.go | 40 ++++++++-------- .../processgroup/usage_darwin_walk_test.go | 46 +++++++++++++++++++ internal/session/jobbound_test.go | 6 ++- 3 files changed, 72 insertions(+), 20 deletions(-) diff --git a/internal/processgroup/usage_darwin.go b/internal/processgroup/usage_darwin.go index 52aaa9c534..cd87c8d9df 100644 --- a/internal/processgroup/usage_darwin.go +++ b/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 } diff --git a/internal/processgroup/usage_darwin_walk_test.go b/internal/processgroup/usage_darwin_walk_test.go index 0d7a63d28f..cf521e767a 100644 --- a/internal/processgroup/usage_darwin_walk_test.go +++ b/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) + } +} diff --git a/internal/session/jobbound_test.go b/internal/session/jobbound_test.go index 38933cf368..2228ce7108 100644 --- a/internal/session/jobbound_test.go +++ b/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}) }