session: support multiple concurrent bypass requests
Previously a session tracked only one pending bypass request. If agent A1 had a pending bypass, agent A2's bypass request would block waiting to be read from toolsrv's channel, effectively blocking all agents in the session. Now bypassPending is a map keyed by request ID: - SetBypassPending adds to the map instead of overwriting - BypassPending returns all pending requests (slice) - BypassPendingByID returns a specific request - ResolveBypass removes from the map by ID - 9P bypass file returns JSON array of all pending The bypass loop reads requests continuously without waiting for resolution, so multiple agents can have concurrent pending requests.
This commit is contained in:
parent
cc56fda862
commit
412b8513e6
|
|
@ -474,13 +474,13 @@ func buildSessionChildren(
|
|||
}, s.GoalSignal),
|
||||
),
|
||||
virtfs.FileNode("bypass", 0666,
|
||||
virtfs.Doc("Bypass request handling. Read: pending request JSON or empty. Write: 'id approve' or 'id deny'."),
|
||||
virtfs.Doc("Bypass request handling. Read: pending requests JSON array. Write: 'id approve' or 'id deny'."),
|
||||
virtfs.Read(func() ([]byte, error) {
|
||||
req := s.BypassPending()
|
||||
if req == nil {
|
||||
return nil, nil
|
||||
reqs := s.BypassPending()
|
||||
if len(reqs) == 0 {
|
||||
return []byte("[]\n"), nil
|
||||
}
|
||||
data, err := json.Marshal(req)
|
||||
data, err := json.Marshal(reqs)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
|
|
|||
|
|
@ -63,9 +63,9 @@ type Session struct {
|
|||
modelsCache string
|
||||
modelsCacheAt time.Time
|
||||
|
||||
// Bypass request tracking
|
||||
// Bypass request tracking (multiple concurrent requests supported)
|
||||
bypassMu sync.RWMutex
|
||||
bypassPending *protocol.BypassRequest
|
||||
bypassPending map[string]*protocol.BypassRequest // keyed by request ID
|
||||
bypassDialFn func() *toolclient.ToolsrvConn
|
||||
bypassSignalCh chan struct{}
|
||||
}
|
||||
|
|
@ -252,7 +252,10 @@ func (s *Session) GoalSignal() <-chan struct{} {
|
|||
// SetBypassPending stores a pending bypass request and publishes an event.
|
||||
func (s *Session) SetBypassPending(req *protocol.BypassRequest, dialFn func() *toolclient.ToolsrvConn) {
|
||||
s.bypassMu.Lock()
|
||||
s.bypassPending = req
|
||||
if s.bypassPending == nil {
|
||||
s.bypassPending = make(map[string]*protocol.BypassRequest)
|
||||
}
|
||||
s.bypassPending[req.ID] = req
|
||||
s.bypassDialFn = dialFn
|
||||
if s.bypassSignalCh != nil {
|
||||
close(s.bypassSignalCh)
|
||||
|
|
@ -270,24 +273,38 @@ func (s *Session) SetBypassPending(req *protocol.BypassRequest, dialFn func() *t
|
|||
PublishEvent("session."+s.ID+".agent."+agentID+".bypass.request", req.ID+"\t"+cmd+"\t"+req.Cwd)
|
||||
}
|
||||
|
||||
// BypassPending returns the current pending bypass request, or nil if none.
|
||||
func (s *Session) BypassPending() *protocol.BypassRequest {
|
||||
// BypassPending returns all pending bypass requests.
|
||||
func (s *Session) BypassPending() []*protocol.BypassRequest {
|
||||
s.bypassMu.RLock()
|
||||
defer s.bypassMu.RUnlock()
|
||||
return s.bypassPending
|
||||
out := make([]*protocol.BypassRequest, 0, len(s.bypassPending))
|
||||
for _, req := range s.bypassPending {
|
||||
out = append(out, req)
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// BypassPendingByID returns a specific pending request by ID.
|
||||
func (s *Session) BypassPendingByID(id string) *protocol.BypassRequest {
|
||||
s.bypassMu.RLock()
|
||||
defer s.bypassMu.RUnlock()
|
||||
return s.bypassPending[id]
|
||||
}
|
||||
|
||||
// ResolveBypass resolves a pending bypass request and clears it.
|
||||
func (s *Session) ResolveBypass(id string, approved bool) error {
|
||||
s.bypassMu.Lock()
|
||||
req := s.bypassPending
|
||||
req, ok := s.bypassPending[id]
|
||||
dialFn := s.bypassDialFn
|
||||
if req == nil || req.ID != id {
|
||||
if !ok {
|
||||
s.bypassMu.Unlock()
|
||||
return fmt.Errorf("no pending bypass with id %s", id)
|
||||
}
|
||||
s.bypassPending = nil
|
||||
s.bypassDialFn = nil
|
||||
delete(s.bypassPending, id)
|
||||
// Only clear dialFn if no more pending requests
|
||||
if len(s.bypassPending) == 0 {
|
||||
s.bypassDialFn = nil
|
||||
}
|
||||
if s.bypassSignalCh != nil {
|
||||
close(s.bypassSignalCh)
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in New Issue