fix(learning): stop discarding runs and undercounting CPU

docs/adaptive-scheduling.md

@@ -16,9 +16,9 @@ These identities are statistical grouping keys, not reproducible-build or whole-
 
 The resource controller is a separate process (agent-heavy scheduler) that the menu bar app starts and supervises, always restarting it when it exits. The Go queue daemon, also supervised by the app, owns jobs independently of the caller. Native Darwin counters provide host CPU intervals, available memory, pressure, swap rates and disk space. Each job's verified process group is sampled for simultaneous physical footprint and observed cumulative CPU and I/O counters. PID birth identities and repeated membership checks prevent attributing a reused process group to an older job. No new privileged service or Python runtime dependency is needed.
 
-CPU-seconds describe total observed computation, while CPU core demand and peak memory inform overlap. Summing simultaneous worker footprints avoids adding unrelated lifetime peaks. Workers born and exited entirely between samples, or peaks entirely between samples, remain invisible; observed totals are lower bounds. Completeness means the required sampled intervals were available, not kernel-enforced lifetime accounting.
+CPU-seconds describe total observed computation, while CPU core demand and peak memory inform overlap. Summing simultaneous worker footprints avoids adding unrelated lifetime peaks. Workers born and exited entirely between samples, or peaks entirely between samples, remain invisible; observed totals are lower bounds. Completeness means the required sampled intervals were available, not kernel-enforced lifetime accounting. A sample is complete when it read every process that lasted through it, so workers starting or exiting mid-sample are normal work rather than missing evidence. Native CPU counters are converted from Mach ticks to nanoseconds, because Apple Silicon reports about 41.67 ns per tick and Intel reports 1 ns.
 
-Execution start time is published with the ownership lease. A controller that first observes a job late, lacks its execution start, misses a sampling interval, or crosses sleep/clock discontinuities marks its learning history incomplete. Finalization checks the last observation against the worker's recorded execution end, including jobs that finish during an outage, and preserves actual execution duration. Healthy current intervals can recover admission readiness without turning the earlier unobserved work into complete training evidence.
+Execution start time is published with the ownership lease. A controller that first observes a job late, lacks its execution start, misses a sampling interval, or crosses sleep/clock discontinuities marks its learning history incomplete. A sample that fails for any other reason counts only if the controller samples the job again, since the final sample of nearly every job races the job's own exit. A failed sample never advances the last observation, so an earlier gap still fails finalization's check against the worker's end. Once the queue reports a job finished, samples only track what still holds its capacity and add nothing to its evidence. Finalization checks the last observation against the worker's recorded execution end, including jobs that finish during an outage, and preserves actual execution duration. Healthy current intervals can recover admission readiness without turning the earlier unobserved work into complete training evidence.
 
 Private SQLite history retains bounded numeric run records and opaque workload identities. Predictions prefer the exact variant, then related family evidence, then a conservative prior. Exact variants need five comparable successful observations within 30 days before lowering the cold-start or family envelope. Family transfers use an extra margin. Underestimates and peaks from failed or incomplete runs raise retained upper evidence without refreshing successful evidence. Old evidence gradually loses confidence. A collector restart, observation gap, cancellation or nonzero exit cannot establish that a job is cheap, and old observation timestamps are preserved across restarts.
 

docs/handoffs/learning-measurement.md

@@ -0,0 +1,24 @@
+# Learning measurement handoff
+
+## Result
+
+Runs now reach learning history. Before this change most runs recorded as incomplete, so no workload variant earned the evidence that adaptive overlap and exact predictions need. CPU on Apple Silicon also read about 42 times too low.
+
+## What changed
+
+- `internal/backend/monitor.go`: a sample is complete when it read every process that lasted through it.
+- `internal/backend/monitor_darwin.go`: CPU counters are converted from Mach ticks with the timebase. A process that has exited is classified apart from one that cannot be read.
+- `internal/backend/scheduler_controller.go`: a failed sample counts only if the job is sampled again, and never advances the last observation. Samples of a job the queue reports as finished stay out of its evidence, but admission still reserves what it holds. Finalization tolerates a sample landing just after the recorded end, and duration is the recorded end minus the start.
+
+## Known limits
+
+- Existing CPU history is not migrated. It stays low until each variant runs again, because envelopes keep the maximum.
+- Correct CPU figures mean Adaptive mode refuses more heavy jobs: a peak above about 64% of cores now predicts over budget.
+- Workers that start and exit between samples are invisible, and counters of a worker that exits during a sample are dropped. Observed totals are lower bounds, and such runs can now become evidence.
+- A job needs about three good samples to count, so jobs under roughly 4 to 6 seconds still record as incomplete. This predates the change.
+- A peak seen after a job finished is not kept once its footprint drops, because a finished job cannot climb back to it.
+- The `values[1] != 0` branch of the Darwin exit classification has no test that reaches it on current macOS.
+
+## Next steps
+
+The grouping work that this enables is separate: per-repository learning groups and classification, then the project and operation browse hierarchy. Re-measure group counts after it lands, on a copy of the history database.

internal/backend/monitor.go

@@ -2,6 +2,7 @@ package backend
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"io"
 	"math"
@@ -185,6 +186,10 @@ func intersect(a, b map[string]bool) bool {
 	}
 	return false
 }
+
+// errResourceProcessExited reports a process that ended before its counters were read.
+var errResourceProcessExited = errors.New("A process exited during resource sampling")
+
 func (s *ResourceSampler) Group(pgid int, expected []Object) (Object, error) {
 	if pgid <= 1 || pgid > math.MaxInt32 {
 		return nil, fmt.Errorf("Invalid resource process group")
@@ -226,11 +231,15 @@ func (s *ResourceSampler) Group(pgid int, expected []Object) (Object, error) {
 	complete := true
 	type reading struct{ member, usage Object }
 	sampled := []reading{}
+	exited := map[string]bool{}
 	for _, member := range members {
 		pid, _ := integer(member["pid"])
 		usage, e := reader.Process(pid, pgid)
+		if errors.Is(e, errResourceProcessExited) {
+			exited[memberKey(member)] = true
+			continue
+		}
 		if e != nil {
-			complete = false
 			continue
 		}
 		valid := true
@@ -249,23 +258,23 @@ func (s *ResourceSampler) Group(pgid int, expected []Object) (Object, error) {
 		return nil, e
 	}
 	after := memberSet(membersAfter)
-	if len(after) != len(identities) {
-		complete = false
+	if ownership != nil && !intersect(after, ownership) {
+		return nil, fmt.Errorf("Resource group ownership changed during sampling")
 	}
+	// Workers come and go during normal work. A sample is complete when it read
+	// every process that lasted through it: an exited worker has released its
+	// footprint, and a new one is read in full by the next sample.
+	lasting := 0
 	for key := range identities {
-		if !after[key] {
-			complete = false
+		if after[key] && !exited[key] {
+			lasting++
 		}
 	}
-	if ownership != nil && !intersect(after, ownership) {
-		return nil, fmt.Errorf("Resource group ownership changed during sampling")
-	}
 	beforeCPU := state.cpu
 	memory := uint64(0)
 	count := 0
 	for _, sample := range sampled {
 		if !after[memberKey(sample.member)] {
-			complete = false
 			continue
 		}
 		identity := fmt.Sprintf("%s\x00%v", memberKey(sample.member), sample.usage["start"])
@@ -308,7 +317,7 @@ func (s *ResourceSampler) Group(pgid int, expected []Object) (Object, error) {
 	if intervalValid {
 		cores = float64(state.cpu-beforeCPU) / 1e9 / elapsed
 	}
-	return Object{"memoryBytes": memory, "cpuSeconds": float64(state.cpu) / 1e9, "cpuCores": cores, "processCount": count, "complete": complete && count == len(members) && count > 0 && intervalValid, "baseline": baseline && complete && count == len(members) && count > 0 && collectionValid, "readBytes": state.read, "writeBytes": state.write}, nil
+	return Object{"memoryBytes": memory, "cpuSeconds": float64(state.cpu) / 1e9, "cpuCores": cores, "processCount": count, "complete": complete && count == lasting && count > 0 && intervalValid, "intervalValid": intervalValid, "baseline": baseline && complete && count == lasting && count > 0 && collectionValid, "readBytes": state.read, "writeBytes": state.write}, nil
 }
 
 // resourceProcessListing deliberately reads no arguments, environment or paths.

internal/backend/monitor_darwin.go

@@ -12,6 +12,7 @@ package backend
 #include <libproc.h>
 #include <mach/mach.h>
 #include <mach/mach_host.h>
+#include <mach/mach_time.h>
 #include <mach/host_info.h>
 #include <mach/vm_statistics.h>
 
@@ -32,10 +33,13 @@ static int queue_host_memory(uint64_t *values) {
  return 0;
 }
 static int queue_process_usage(int pid, uint64_t *values) {
+ // rusage CPU times are Mach ticks: 1 ns on Intel, about 41.67 ns on Apple Silicon.
+ static mach_timebase_info_data_t timebase;
+ if (timebase.denom==0 && (mach_timebase_info(&timebase)!=KERN_SUCCESS || timebase.denom==0)) return -1;
  struct rusage_info_v2 usage;
  if (proc_pid_rusage(pid,RUSAGE_INFO_V2,(rusage_info_t *)&usage)!=0) return -1;
  values[0]=usage.ri_proc_start_abstime;values[1]=usage.ri_proc_exit_abstime;values[2]=usage.ri_phys_footprint;
- values[3]=usage.ri_user_time+usage.ri_system_time;values[4]=usage.ri_diskio_bytesread;values[5]=usage.ri_diskio_byteswritten;
+ values[3]=(uint64_t)(((unsigned __int128)usage.ri_user_time+usage.ri_system_time)*timebase.numer/timebase.denom);values[4]=usage.ri_diskio_bytesread;values[5]=usage.ri_diskio_byteswritten;
  return 0;
 }
 static int queue_swap_used(uint64_t *used) {
@@ -54,6 +58,7 @@ import (
 	"crypto/sha256"
 	"encoding/binary"
 	"encoding/hex"
+	"errors"
 	"fmt"
 	"strconv"
 	"syscall"
@@ -150,15 +155,25 @@ func (b *darwinResourceBackend) Members(pgid int) ([]Object, error) {
 }
 func (b *darwinResourceBackend) Process(pid, pgid int) (Object, error) {
 	group, e := syscall.Getpgid(pid)
+	if errors.Is(e, syscall.ESRCH) {
+		return nil, errResourceProcessExited
+	}
 	if e != nil || group != pgid {
 		return nil, fmt.Errorf("A process resource identity is unavailable")
 	}
 	var values [6]C.uint64_t
 	if C.queue_process_usage(C.int(pid), &values[0]) != 0 {
+		// The counter read also fails for a process that exited just before it.
+		if _, e = syscall.Getpgid(pid); errors.Is(e, syscall.ESRCH) {
+			return nil, errResourceProcessExited
+		}
 		return nil, fmt.Errorf("A process resource counter is unavailable")
 	}
 	group, e = syscall.Getpgid(pid)
-	if e != nil || group != pgid || values[0] == 0 || values[1] != 0 {
+	if errors.Is(e, syscall.ESRCH) || values[1] != 0 {
+		return nil, errResourceProcessExited
+	}
+	if e != nil || group != pgid || values[0] == 0 {
 		return nil, fmt.Errorf("A process resource identity changed")
 	}
 	return Object{"start": uint64(values[0]), "memoryBytes": uint64(values[2]), "cpuNanoseconds": uint64(values[3]), "readBytes": uint64(values[4]), "writeBytes": uint64(values[5])}, nil

internal/backend/monitor_darwin_test.go

@@ -3,11 +3,88 @@
 package backend
 
 import (
+	"errors"
 	"os"
+	"os/exec"
 	"syscall"
 	"testing"
+	"time"
 )
 
+func TestResourceDarwinCPUTimeIsNanoseconds(t *testing.T) {
+	backend, e := newPlatformResourceBackend(t.TempDir())
+	if e != nil {
+		t.Fatal(e)
+	}
+	read := func() (reported, kernel float64) {
+		t.Helper()
+		process, e := backend.Process(os.Getpid(), syscall.Getpgrp())
+		if e != nil {
+			t.Fatal(e)
+		}
+		var usage syscall.Rusage
+		if e = syscall.Getrusage(syscall.RUSAGE_SELF, &usage); e != nil {
+			t.Fatal(e)
+		}
+		return metric(process["cpuNanoseconds"]) / 1e9, float64(usage.Utime.Nano()+usage.Stime.Nano()) / 1e9
+	}
+	reportedBefore, kernelBefore := read()
+	// Burn CPU until the kernel's own accounting has moved measurably.
+	spin := uint64(1)
+	for _, kernel := read(); kernel-kernelBefore < .25; _, kernel = read() {
+		for i := range 2000000 {
+			spin = spin*31 + uint64(i)
+		}
+	}
+	reportedAfter, kernelAfter := read()
+	if ratio := (reportedAfter - reportedBefore) / (kernelAfter - kernelBefore); ratio < .8 || ratio > 1.25 {
+		t.Fatalf("CPU time is in the wrong unit: reported %.3fs, kernel %.3fs (%d)", reportedAfter-reportedBefore, kernelAfter-kernelBefore, spin)
+	}
+}
+func TestResourceDarwinProcessExitClassification(t *testing.T) {
+	backend, e := newPlatformResourceBackend(t.TempDir())
+	if e != nil {
+		t.Fatal(e)
+	}
+	start := func(arguments ...string) *exec.Cmd {
+		t.Helper()
+		command := exec.Command(arguments[0], arguments[1:]...)
+		command.SysProcAttr = &syscall.SysProcAttr{Setpgid: true}
+		if e := command.Start(); e != nil {
+			t.Fatal(e)
+		}
+		return command
+	}
+	live := start("/bin/sleep", "30")
+	defer func() { live.Process.Kill(); live.Wait() }()
+	if _, e = backend.Process(live.Process.Pid, live.Process.Pid); e != nil {
+		t.Fatalf("Live process was unreadable: %v", e)
+	}
+	if _, e = backend.Process(live.Process.Pid, live.Process.Pid+1); e == nil || errors.Is(e, errResourceProcessExited) {
+		t.Fatalf("A process in another group was classified as exited: %v", e)
+	}
+	if os.Geteuid() != 0 {
+		if _, e = backend.Process(1, 1); e == nil || errors.Is(e, errResourceProcessExited) {
+			t.Fatalf("An unreadable live process was classified as exited: %v", e)
+		}
+	}
+	unreaped := start("/usr/bin/true")
+	deadline := time.Now().Add(10 * time.Second)
+	for {
+		_, e = backend.Process(unreaped.Process.Pid, unreaped.Process.Pid)
+		if errors.Is(e, errResourceProcessExited) {
+			break
+		}
+		if time.Now().After(deadline) {
+			t.Fatalf("Unreaped process was not classified as exited: %v", e)
+		}
+		time.Sleep(10 * time.Millisecond)
+	}
+	unreaped.Wait()
+	if _, e = backend.Process(unreaped.Process.Pid, unreaped.Process.Pid); !errors.Is(e, errResourceProcessExited) {
+		t.Fatalf("Reaped process was not classified as exited: %v", e)
+	}
+}
 func TestResourceDarwinNativeMetrics(t *testing.T) {
 	backend, e := newPlatformResourceBackend(t.TempDir())
 	if e != nil {

internal/backend/monitor_test.go

@@ -3,6 +3,7 @@ package backend
 import (
 	"fmt"
 	"math"
+	"strings"
 	"testing"
 
 	"github.com/Dunky13/agent-job-queue/internal/census"
@@ -13,6 +14,7 @@ type fakeResourceBackend struct {
 	members   []Object
 	usages    map[int]Object
 	fail      map[int]bool
+	exited    map[int]bool
 	after     []Object
 	calls     int
 	onHost    func()
@@ -52,6 +54,9 @@ func (b *fakeResourceBackend) Process(pid, pgid int) (Object, error) {
 	if b.onProcess != nil {
 		b.onProcess()
 	}
+	if b.exited[pid] {
+		return nil, errResourceProcessExited
+	}
 	if b.fail[pid] {
 		return nil, fmt.Errorf("Counter unavailable")
 	}
@@ -192,6 +197,76 @@ func TestResourceGroupWorkersExitsPIDReuseAndResets(t *testing.T) {
 		t.Fatal("Forgotten process group inherited totals")
 	}
 }
+func TestResourceGroupRejectsOwnershipLostDuringSample(t *testing.T) {
+	s, b, _, _ := resourceFixture()
+	owner := b.members
+	b.after = []Object{{"pid": 900, "birth": "stranger"}}
+	if _, e := s.Group(400, owner); e == nil || !strings.Contains(e.Error(), "changed during sampling") {
+		t.Fatalf("Ownership loss during a sample was accepted: %v", e)
+	}
+}
+func TestResourceGroupSampleToleratesWorkerChurn(t *testing.T) {
+	stable, leaving, joining := Object{"pid": 101, "birth": "stable birth"}, Object{"pid": 102, "birth": "leaving"}, Object{"pid": 103, "birth": "joining"}
+	sample := func(t *testing.T, prepare func(*fakeResourceBackend)) Object {
+		t.Helper()
+		s, b, now, _ := resourceFixture()
+		mustGroup(t, s)
+		*now += 2
+		b.calls = 0
+		prepare(b)
+		return mustGroup(t, s)
+	}
+	t.Run("worker exits during the sample", func(t *testing.T) {
+		v := sample(t, func(b *fakeResourceBackend) {
+			b.members = []Object{stable, leaving}
+			b.usages[102] = fixtureUsage(2, 500, 1)
+			b.after = []Object{stable}
+		})
+		if !boolean(v["complete"]) || metric(v["memoryBytes"]) != 100 {
+			t.Fatalf("Exited worker failed the sample or kept its footprint: %v", v)
+		}
+	})
+	t.Run("worker exits before its counters are read", func(t *testing.T) {
+		v := sample(t, func(b *fakeResourceBackend) {
+			b.members = []Object{stable, leaving}
+			b.fail[102] = true
+			b.after = []Object{stable}
+		})
+		if !boolean(v["complete"]) {
+			t.Fatalf("Exited worker failed the sample: %v", v)
+		}
+	})
+	t.Run("worker starts during the sample", func(t *testing.T) {
+		v := sample(t, func(b *fakeResourceBackend) { b.after = []Object{stable, joining} })
+		if !boolean(v["complete"]) {
+			t.Fatalf("New worker failed the sample: %v", v)
+		}
+	})
+	t.Run("unreaped worker has exited", func(t *testing.T) {
+		v := sample(t, func(b *fakeResourceBackend) {
+			b.members = []Object{stable, leaving}
+			b.exited = map[int]bool{102: true}
+		})
+		if !boolean(v["complete"]) {
+			t.Fatalf("Unreaped worker failed the sample: %v", v)
+		}
+	})
+	t.Run("lasting worker could not be read", func(t *testing.T) {
+		v := sample(t, func(b *fakeResourceBackend) {
+			b.members = []Object{stable, leaving}
+			b.fail[102] = true
+		})
+		if boolean(v["complete"]) {
+			t.Fatalf("Unmeasured worker was certified: %v", v)
+		}
+	})
+	t.Run("every worker has gone", func(t *testing.T) {
+		v := sample(t, func(b *fakeResourceBackend) { b.exited = map[int]bool{101: true} })
+		if boolean(v["complete"]) {
+			t.Fatalf("Empty sample was certified: %v", v)
+		}
+	})
+}
 func TestResourceGroupGapsAndCollectionSuspend(t *testing.T) {
 	for _, gap := range [][2]float64{{600, 600}, {2, 602}, {2, -298}, {600, 2}, {0, 600}, {-1, 2}, {2, 0}, {2, 5}} {
 		t.Run(fmt.Sprint(gap), func(t *testing.T) {