From 9209127bc0e84d80bea4ec1ab9592933ebe370b3 Mon Sep 17 00:00:00 2001 From: Nathan Whitaker Date: Wed, 24 Jun 2026 18:14:24 -0700 Subject: [PATCH] fix(orch): exempt managers from per-target worker capacity MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Interactive managers are long-lived per-objective supervisors that sit idle most of their life, yet each held a per-target session slot for its whole life. With sum(capacity) slots across the fleet (e.g. 4+16=20), an objective's PR feedback could not be placed once idle managers filled a target and the only other target was load-gated — the wedge that left a PR comment unanswered: SelectTarget -> ErrNoTarget, swallowed silently in spawnPRFollowup and retried invisibly every monitor tick until a slot happened to free. The codebase already exempts managers from the GLOBAL worker-concurrency cap (CountActiveWorkerSessions / Scheduler.Tick) for exactly this reason, but left them charged against PER-TARGET capacity. Extend the same exemption: managers place without claiming a worker slot (keyed on the durable session role so claim and release stay in lockstep), and SelectTarget ignores worker capacity for a manager request. Managers stay bounded by the load gate and the per-objective respawn limit. available_sessions is an incrementally-maintained counter, so make it self-healing: ReconcileTargetSlots rebuilds it from live non-terminal, non-manager occupancy at startup (before the scheduler's first tick), healing both crash drift and the slots currently-running managers claimed under the old accounting. --- internal/orch/capacity_test.go | 132 +++++++++++++++++++++++++++++++++ internal/orch/scheduler.go | 50 +++++++++++-- internal/orch/sessions.go | 6 ++ internal/orch/targets.go | 4 + internal/store/targets.go | 40 ++++++++++ 5 files changed, 225 insertions(+), 7 deletions(-) create mode 100644 internal/orch/capacity_test.go diff --git a/internal/orch/capacity_test.go b/internal/orch/capacity_test.go new file mode 100644 index 0000000..9c2f8a3 --- /dev/null +++ b/internal/orch/capacity_test.go @@ -0,0 +1,132 @@ +package orch + +import ( + "errors" + "testing" + + "github.com/nathanwhit/orcha/internal/model" + "github.com/nathanwhit/orcha/internal/store" +) + +// A manager is exempt from per-target worker capacity: it places on a target +// whose worker slots are full, while a worker requesting the same full target is +// refused. This is what stops a pool of idle managers from filling a box and +// starving real work. +func TestSelectTarget_ManagerIgnoresFullCapacity(t *testing.T) { + o, st := newTestOrch(t) + tgt := addTarget(t, st, "only", model.TargetLocal, 1) + // Consume the single worker slot. + if err := st.ClaimTargetSlot(tgt.ID); err != nil { + t.Fatalf("claim: %v", err) + } + + // A worker can't place — the box is full. + if _, err := o.SelectTarget(TargetRequest{}); !errors.Is(err, ErrNoTarget) { + t.Fatalf("worker on full target: err=%v, want ErrNoTarget", err) + } + // A manager places anyway. + got, err := o.SelectTarget(TargetRequest{IgnoreWorkerCapacity: true}) + if err != nil { + t.Fatalf("manager on full target: %v", err) + } + if got.ID != tgt.ID { + t.Fatalf("manager placed on %s, want %s", got.Name, tgt.Name) + } +} + +// Placing a manager must not debit a target's worker capacity, and releasing it +// must not credit one back — otherwise capacity drifts and the box oversubscribes +// real workers. A worker placed on the same target debits and credits normally. +func TestPlaceSession_ManagerDoesNotConsumeWorkerSlot(t *testing.T) { + o, st := newTestOrch(t) + tgt := addTarget(t, st, "box", model.TargetLocal, 2) + + mgr := &model.Session{Role: model.RoleManager, Agent: model.AgentClaude, Status: model.SessionQueued} + if err := st.CreateSession(mgr); err != nil { + t.Fatalf("create mgr: %v", err) + } + if _, err := o.PlaceSession(mgr.ID, o.targetRequestFor(mgr)); err != nil { + t.Fatalf("place mgr: %v", err) + } + if avail := mustAvail(t, st, tgt.ID); avail != 2 { + t.Fatalf("after manager placement available=%d, want 2 (unchanged)", avail) + } + + worker := &model.Session{Role: model.RoleImplementer, Agent: model.AgentClaude, Status: model.SessionQueued} + if err := st.CreateSession(worker); err != nil { + t.Fatalf("create worker: %v", err) + } + if _, err := o.PlaceSession(worker.ID, o.targetRequestFor(worker)); err != nil { + t.Fatalf("place worker: %v", err) + } + if avail := mustAvail(t, st, tgt.ID); avail != 1 { + t.Fatalf("after worker placement available=%d, want 1", avail) + } + + // Release both: the manager is a no-op, the worker credits its slot back. + mgr, _ = st.GetSession(mgr.ID) + o.releaseTargetSlot(mgr) + if avail := mustAvail(t, st, tgt.ID); avail != 1 { + t.Fatalf("after manager release available=%d, want 1 (no credit)", avail) + } + worker, _ = st.GetSession(worker.ID) + o.releaseTargetSlot(worker) + if avail := mustAvail(t, st, tgt.ID); avail != 2 { + t.Fatalf("after worker release available=%d, want 2", avail) + } +} + +// ReconcileTargetSlots rebuilds available_sessions from live occupancy: it counts +// non-terminal workers, ignores managers (capacity-exempt) and terminal rows, and +// heals a drifted counter — the migration that frees the slots managers held +// under the old accounting. +func TestReconcileTargetSlots(t *testing.T) { + _, st := newTestOrch(t) + tgt := addTarget(t, st, "box", model.TargetLocal, 4) + + bind := func(role model.SessionRole, status model.SessionStatus) { + s := &model.Session{Role: role, Agent: model.AgentClaude, Status: status, TargetID: tgt.ID} + if err := st.CreateSession(s); err != nil { + t.Fatalf("create session: %v", err) + } + } + bind(model.RoleImplementer, model.SessionRunning) // counts + bind(model.RoleReviewer, model.SessionStarting) // counts + bind(model.RoleManager, model.SessionRunning) // exempt — does not count + bind(model.RoleImplementer, model.SessionSucceeded) // terminal — does not count + + // Drift the counter to a wrong value (e.g. managers had claimed under the old + // accounting): available 4 -> 0. + for i := 0; i < 4; i++ { + _ = st.ClaimTargetSlot(tgt.ID) + } + if avail := mustAvail(t, st, tgt.ID); avail != 0 { + t.Fatalf("precondition: available=%d, want 0", avail) + } + + n, err := st.ReconcileTargetSlots() + if err != nil { + t.Fatalf("reconcile: %v", err) + } + if n != 1 { + t.Fatalf("corrected %d targets, want 1", n) + } + // 4 capacity - 2 live workers = 2; manager + terminal worker excluded. + if avail := mustAvail(t, st, tgt.ID); avail != 2 { + t.Fatalf("after reconcile available=%d, want 2", avail) + } + + // Idempotent: a second pass corrects nothing. + if n, err := st.ReconcileTargetSlots(); err != nil || n != 0 { + t.Fatalf("second reconcile n=%d err=%v, want 0 nil", n, err) + } +} + +func mustAvail(t *testing.T, st *store.Store, targetID string) int { + t.Helper() + tgt, err := st.GetTarget(targetID) + if err != nil { + t.Fatalf("get target: %v", err) + } + return tgt.AvailableSessions +} diff --git a/internal/orch/scheduler.go b/internal/orch/scheduler.go index 57d2bc6..e2bd0fa 100644 --- a/internal/orch/scheduler.go +++ b/internal/orch/scheduler.go @@ -12,6 +12,16 @@ type TargetRequest struct { RequiredLabels []string ProjectPath string // for build/cache locality preference PinnedTargetID string // explicit user pinning + // IgnoreWorkerCapacity places this session without consuming, or being gated + // by, a target's per-session worker capacity. Set for interactive managers: + // they are long-lived per-objective supervisors that sit idle most of their + // life, so charging them the same slots workers need lets accumulated idle + // managers fill a target and starve real work — a fleet with N total slots + // could otherwise hold at most N managers and zero workers. This mirrors the + // existing exemption from the global worker-concurrency cap + // (CountActiveWorkerSessions / Scheduler.Tick). Managers remain bounded by + // the load gate and the per-objective respawn limit. + IgnoreWorkerCapacity bool } // SelectTarget picks a schedulable target satisfying the request. It considers: @@ -31,7 +41,7 @@ func (o *Orchestrator) SelectTarget(req TargetRequest) (*model.Target, error) { if !t.Status.CanSchedule() { return nil, fmt.Errorf("%w: pinned target %s is %s", ErrNoTarget, t.Name, t.Status) } - if t.AvailableSessions <= 0 { + if !req.IgnoreWorkerCapacity && t.AvailableSessions <= 0 { return nil, fmt.Errorf("%w: pinned target %s is at capacity", ErrNoTarget, t.Name) } if !hasLabels(t, req.RequiredLabels) { @@ -49,7 +59,7 @@ func (o *Orchestrator) SelectTarget(req TargetRequest) (*model.Target, error) { if !t.Status.CanSchedule() { // draining/offline/disabled excluded continue } - if t.AvailableSessions <= 0 { + if !req.IgnoreWorkerCapacity && t.AvailableSessions <= 0 { continue } if !hasLabels(t, req.RequiredLabels) { @@ -178,15 +188,26 @@ func (o *Orchestrator) PlaceSession(sessionID string, req TargetRequest) (*model if err != nil { return nil, err } - // Atomic claim — enforces capacity and draining at the store layer. - if err := o.st.ClaimTargetSlot(target.ID); err != nil { - return nil, err + // Managers don't consume a worker slot, so they bind to the target without + // claiming one. SelectTarget already excluded draining/offline targets, so + // the atomic claim — whose job is to stop concurrent schedulers + // oversubscribing capacity — has nothing to enforce for a manager. Keyed on + // the durable session role (not req) so it can never disagree with + // releaseTargetSlot, which must skip the matching release. + claimsSlot := !sessionExemptFromCapacity(sess) + if claimsSlot { + // Atomic claim — enforces capacity and draining at the store layer. + if err := o.st.ClaimTargetSlot(target.ID); err != nil { + return nil, err + } } if _, err := o.st.UpdateSessionRuntime(sessionID, func(s *model.Session) { s.TargetID = target.ID }); err != nil { // Roll back the claim so we never leak capacity. - _ = o.st.ReleaseTargetSlot(target.ID) + if claimsSlot { + _ = o.st.ReleaseTargetSlot(target.ID) + } return nil, err } o.audit(sess.ObjectiveID, sessionID, "session_placed", @@ -194,9 +215,24 @@ func (o *Orchestrator) PlaceSession(sessionID string, req TargetRequest) (*model return target, nil } -// releaseTargetSlot frees the capacity a session held, if any. +// releaseTargetSlot frees the capacity a session held, if any. A session exempt +// from worker capacity (a manager) never claimed a slot, so it must not release +// one either — crediting capacity that was never debited would let the target +// oversubscribe real workers. func (o *Orchestrator) releaseTargetSlot(sess *model.Session) { + if sessionExemptFromCapacity(sess) { + return + } if sess.TargetID != "" { _ = o.st.ReleaseTargetSlot(sess.TargetID) } } + +// sessionExemptFromCapacity reports whether a session is placed without +// consuming a target's per-session worker capacity. Interactive managers are +// long-lived supervisors that sit idle most of their life; see +// TargetRequest.IgnoreWorkerCapacity for the full rationale. Claim, release, and +// the slot reconcile all key off this single predicate so they stay in lockstep. +func sessionExemptFromCapacity(sess *model.Session) bool { + return sess.Role == model.RoleManager +} diff --git a/internal/orch/sessions.go b/internal/orch/sessions.go index fe60b7c..d7efe18 100644 --- a/internal/orch/sessions.go +++ b/internal/orch/sessions.go @@ -706,6 +706,12 @@ func (o *Orchestrator) RecoverInterrupted() int { if n, err := o.st.DeduplicatePRs(); err == nil && n > 0 { o.audit("", "", "prs_deduped", fmt.Sprintf("removed %d duplicate PR row(s)", n), nil) } + // Heal target capacity counters drifted by a crash mid-claim or by managers + // becoming capacity-exempt, so a stale available_sessions can't permanently + // starve a target. Runs before the scheduler's first tick. + if n, err := o.st.ReconcileTargetSlots(); err == nil && n > 0 { + o.audit("", "", "target_slots_reconciled", fmt.Sprintf("corrected %d target capacity counter(s)", n), nil) + } sessions, err := o.st.RequeueInterruptedSessions() if err != nil || len(sessions) == 0 { return 0 diff --git a/internal/orch/targets.go b/internal/orch/targets.go index a097162..8b19aaa 100644 --- a/internal/orch/targets.go +++ b/internal/orch/targets.go @@ -105,6 +105,10 @@ func (o *Orchestrator) targetRequestFor(sess *model.Session) TargetRequest { req.PinnedTargetID = ws.TargetID } } + // Managers don't consume per-target worker capacity (see + // TargetRequest.IgnoreWorkerCapacity), so an idle pool of them can't fill a + // target and starve workers. + req.IgnoreWorkerCapacity = sessionExemptFromCapacity(sess) return req } diff --git a/internal/store/targets.go b/internal/store/targets.go index e4684bd..78b9f43 100644 --- a/internal/store/targets.go +++ b/internal/store/targets.go @@ -237,3 +237,43 @@ func (s *Store) ReleaseTargetSlot(targetID string) error { WHERE id = ?`, targetID) return err } + +// ReconcileTargetSlots recomputes every target's available_sessions from actual +// occupancy: capacity minus the non-terminal, capacity-consuming (non-manager) +// sessions bound to it. available_sessions is otherwise maintained incrementally +// — debited on claim, credited at terminal — so a crash between claim and the +// runtime update, or a change in WHAT counts against capacity (managers became +// exempt), can drift it and permanently shrink a target's usable slots. Run once +// at startup, before the scheduler, to heal that drift. It is idempotent. +// Returns the number of targets whose counter it corrected. +func (s *Store) ReconcileTargetSlots() (int, error) { + targets, err := s.ListTargets() + if err != nil { + return 0, err + } + corrected := 0 + for _, t := range targets { + var used int + if err := s.db.QueryRow( + `SELECT COUNT(*) FROM sessions + WHERE target_id = ? AND role <> ? AND status NOT IN (?,?,?)`, + t.ID, string(model.RoleManager), + string(model.SessionSucceeded), string(model.SessionFailed), string(model.SessionCanceled), + ).Scan(&used); err != nil { + return corrected, err + } + avail := t.CapacitySessions - used + if avail < 0 { + avail = 0 + } + if avail == t.AvailableSessions { + continue + } + if _, err := s.db.Exec( + `UPDATE targets SET available_sessions = ? WHERE id = ?`, avail, t.ID); err != nil { + return corrected, err + } + corrected++ + } + return corrected, nil +}