fix(orch): exempt managers from per-target worker capacity

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
+}

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,25 +188,51 @@ 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",
 		fmt.Sprintf("placed on %s", target.Name), model.JSONMap{"target_id": target.ID})
 	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
+}

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

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
 }
 

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
+}