resident_scheduler.go RawHistoryFolder

1package subagents
2
3import (
4	"errors"
5	"sync"
6)
7
8const DefaultResidentConcurrency = 6
9
10// ResidentTicket is an accepted prompt's stable global admission position.
11type ResidentTicket struct {
12	Sequence uint64
13	ChildID  string
14	Prompt   string
15	ready    chan struct{}
16}
17
18// ResidentScheduler provides fair global admission while enforcing one active
19// turn per child. It intentionally does not perform provider work.
20type ResidentScheduler struct {
21	mu      sync.Mutex
22	limit   int
23	next    uint64
24	active  int
25	running map[string]bool
26	pending []ResidentTicket
27}
28
29func NewResidentScheduler(limit int) *ResidentScheduler {
30	if limit <= 0 {
31		limit = DefaultResidentConcurrency
32	}
33	return &ResidentScheduler{limit: limit, running: make(map[string]bool)}
34}
35
36func (s *ResidentScheduler) Enqueue(childID, prompt string) ResidentTicket {
37	s.mu.Lock()
38	defer s.mu.Unlock()
39	s.next++
40	ticket := ResidentTicket{Sequence: s.next, ChildID: childID, Prompt: prompt, ready: make(chan struct{})}
41	s.pending = append(s.pending, ticket)
42	return ticket
43}
44
45// Admit returns the oldest ticket whose child is not already active. An older
46// ineligible ticket therefore cannot prevent a later independent child from
47// using an available global slot.
48func (s *ResidentScheduler) Admit() (ResidentTicket, bool) {
49	s.mu.Lock()
50	defer s.mu.Unlock()
51	if s.active >= s.limit {
52		return ResidentTicket{}, false
53	}
54	for i, ticket := range s.pending {
55		if s.running[ticket.ChildID] {
56			continue
57		}
58		s.pending = append(s.pending[:i], s.pending[i+1:]...)
59		s.running[ticket.ChildID] = true
60		s.active++
61		close(ticket.ready)
62		return ticket, true
63	}
64	return ResidentTicket{}, false
65}
66
67// Cancel removes an admitted-waiting ticket. Once admitted, the caller owns
68// the active reservation and must Release it instead.
69func (s *ResidentScheduler) Cancel(sequence uint64) bool {
70	s.mu.Lock()
71	defer s.mu.Unlock()
72	for i, ticket := range s.pending {
73		if ticket.Sequence != sequence {
74			continue
75		}
76		s.pending = append(s.pending[:i], s.pending[i+1:]...)
77		return true
78	}
79	return false
80}
81
82func (s *ResidentScheduler) Release(childID string) error {
83	s.mu.Lock()
84	defer s.mu.Unlock()
85	if !s.running[childID] {
86		return errors.New("resident scheduler: child has no active turn")
87	}
88	delete(s.running, childID)
89	s.active--
90	return nil
91}