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) }