From ed62caced55fd113b130948dd40db366e3f54824 Mon Sep 17 00:00:00 2001 From: Levi Neely Date: Sun, 26 Apr 2026 19:40:06 +0200 Subject: [PATCH] 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. --- pkg/agent/core.go | 18 ++++++++++++++++++ pkg/store/store_test.go | 7 ++++++- 2 files changed, 24 insertions(+), 1 deletion(-) diff --git a/pkg/agent/core.go b/pkg/agent/core.go index ce36572..22407c2 100644 --- a/pkg/agent/core.go +++ b/pkg/agent/core.go @@ -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 } diff --git a/pkg/store/store_test.go b/pkg/store/store_test.go index 63bf30a..9444cfe 100644 --- a/pkg/store/store_test.go +++ b/pkg/store/store_test.go @@ -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) }