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// Pending reports whether a child's accepted turn is waiting for scheduler
68// admission rather than running provider work.
69func (s *ResidentScheduler) Pending(childID string) bool {
70	if s == nil {
71		return false
72	}
73	s.mu.Lock()
74	defer s.mu.Unlock()
75	for _, ticket := range s.pending {
76		if ticket.ChildID == childID {
77			return true
78		}
79	}
80	return false
81}
82
83// Cancel removes an admitted-waiting ticket. Once admitted, the caller owns
84// the active reservation and must Release it instead.
85func (s *ResidentScheduler) Cancel(sequence uint64) bool {
86	s.mu.Lock()
87	defer s.mu.Unlock()
88	for i, ticket := range s.pending {
89		if ticket.Sequence != sequence {
90			continue
91		}
92		s.pending = append(s.pending[:i], s.pending[i+1:]...)
93		return true
94	}
95	return false
96}
97
98func (s *ResidentScheduler) Release(childID string) error {
99	s.mu.Lock()
100	defer s.mu.Unlock()
101	if !s.running[childID] {
102		return errors.New("resident scheduler: child has no active turn")
103	}
104	delete(s.running, childID)
105	s.active--
106	return nil
107}