agent: serialize Submit with submitMu to prevent command/turn races
Commands like /compact arriving via ctl and prompts arriving via prompt both call Core.Submit in separate goroutines. handleCommand for mutating commands (compact, clear, agent, backend) modifies session state without setting currentAction, so IsRunning() returns false and a concurrent executeTurn can race. Add submitMu held during the serialized path of Submit. The fast path (inject/FIFO push while running) remains lock-free. Also fix pre-existing race in TestSessionFileStoreWritePrompt: replace time.Sleep synchronization with a channel signal.
This commit is contained in:
parent
c32ea01f60
commit
ed62caced5
|
|
@ -180,6 +180,7 @@ type agent struct {
|
|||
env map[string]string // session-scoped env vars
|
||||
changeMu sync.Mutex
|
||||
changeCond *sync.Cond
|
||||
submitMu sync.Mutex // serializes Submit calls (commands + turns)
|
||||
}
|
||||
|
||||
// SetEnv stores a session-scoped variable and propagates it to the execute server.
|
||||
|
|
@ -673,6 +674,23 @@ func (s *agent) Submit(ctx context.Context, input string, handler EventHandler)
|
|||
if input == "" {
|
||||
return
|
||||
}
|
||||
|
||||
// Fast path: inject and FIFO push use atomics and are safe without
|
||||
// the submit lock. Handle them before acquiring submitMu so they
|
||||
// don't block behind a long-running turn or command.
|
||||
if s.IsRunning() {
|
||||
if s.handleCommand(ctx, input, handler) {
|
||||
return
|
||||
}
|
||||
s.fifo.Push(input)
|
||||
return
|
||||
}
|
||||
|
||||
// Serialize commands and turns so that e.g. a /compact arriving via
|
||||
// ctl cannot race with an executeTurn arriving via prompt.
|
||||
s.submitMu.Lock()
|
||||
defer s.submitMu.Unlock()
|
||||
|
||||
if s.handleCommand(ctx, input, handler) {
|
||||
return
|
||||
}
|
||||
|
|
|
|||
|
|
@ -41,6 +41,7 @@ type stubCore struct {
|
|||
queued []string
|
||||
interrupted bool
|
||||
setSessionIDErr error
|
||||
submitCh chan struct{} // closed after each Submit call
|
||||
}
|
||||
|
||||
func (c *stubCore) Submit(_ context.Context, input string, handler agent.EventHandler) {
|
||||
|
|
@ -48,6 +49,9 @@ func (c *stubCore) Submit(_ context.Context, input string, handler agent.EventHa
|
|||
if handler != nil && c.reply != "" {
|
||||
handler(agent.Event{Role: "assistant", Content: c.reply})
|
||||
}
|
||||
if c.submitCh != nil {
|
||||
close(c.submitCh)
|
||||
}
|
||||
}
|
||||
func (c *stubCore) Interrupt(error) bool { c.interrupted = true; return c.running }
|
||||
func (c *stubCore) Inject(string) {}
|
||||
|
|
@ -815,8 +819,9 @@ func TestSessionFileStoreWritePrompt(t *testing.T) {
|
|||
defer sess.Cancel()
|
||||
sf, core := newSessionFileStore(t, sess)
|
||||
|
||||
core.submitCh = make(chan struct{})
|
||||
storeWrite(t, sf, "prompt", []byte("hello agent"))
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
<-core.submitCh
|
||||
if len(core.submitted) != 1 || core.submitted[0] != "hello agent" {
|
||||
t.Errorf("submitted = %v; want [hello agent]", core.submitted)
|
||||
}
|
||||
|
|
|
|||
Reference in New Issue