From 991fcde65b2f0c58c56c017a8c6086bb2b906556 Mon Sep 17 00:00:00 2001 From: Ludovico Russo Date: Thu, 30 Apr 2026 14:25:00 +0200 Subject: [PATCH 1/4] Rename CLAUDE.md to AGENTS.md and create symlink --- AGENTS.md | 1 + CLAUDE.md | 2 +- 2 files changed, 2 insertions(+), 1 deletion(-) create mode 100644 AGENTS.md mode change 100644 => 120000 CLAUDE.md diff --git a/AGENTS.md b/AGENTS.md new file mode 100644 index 0000000..7874d79 --- /dev/null +++ b/AGENTS.md @@ -0,0 +1 @@ +Use subtesting instead of pattern like TextXXX_YyY diff --git a/CLAUDE.md b/CLAUDE.md deleted file mode 100644 index 7874d79..0000000 --- a/CLAUDE.md +++ /dev/null @@ -1 +0,0 @@ -Use subtesting instead of pattern like TextXXX_YyY diff --git a/CLAUDE.md b/CLAUDE.md new file mode 120000 index 0000000..47dc3e3 --- /dev/null +++ b/CLAUDE.md @@ -0,0 +1 @@ +AGENTS.md \ No newline at end of file From 9fd014a2343e859ea3b78dffcf9f03749a971e17 Mon Sep 17 00:00:00 2001 From: Ludovico Russo Date: Thu, 30 Apr 2026 14:31:40 +0200 Subject: [PATCH 2/4] refactor(daemon): privatize ManagedSession messages/todos behind locked accessors (#85) Encapsulates the conversation history and todo list behind Messages() and Todos() so external readers (status, todos commands) take a consistent snapshot under sess.mu instead of racing with an in-flight turn. - messages and todos fields are now unexported. - Messages() returns a shallow copy via slices.Clone under the lock; the append-only invariant on the slice is documented at the field. - Todos() returns the (internally thread-safe) list pointer under the lock. - statusCmd and todosCmd in daemon.go switch to the accessors. - runTurnInternal and Close() use the unexported fields; Close() semantics unchanged in this slice (handled in slice 2, #86). --- pkg/daemon/daemon.go | 7 +++-- pkg/daemon/sessions.go | 55 +++++++++++++++++++++++++------------ pkg/daemon/sessions_test.go | 13 +++++---- 3 files changed, 48 insertions(+), 27 deletions(-) diff --git a/pkg/daemon/daemon.go b/pkg/daemon/daemon.go index 1b7b141..9ac94e5 100644 --- a/pkg/daemon/daemon.go +++ b/pkg/daemon/daemon.go @@ -103,9 +103,10 @@ func Run(ctx context.Context, cfg Config) error { if sess == nil { return command.StatusInfo{}, fmt.Errorf("session not found: %s", id) } + messages := sess.Messages() var toolCalls int uniqueSkills := make(map[string]struct{}) - for _, msg := range sess.Messages { + for _, msg := range messages { toolCalls += len(msg.ToolCalls) for _, tc := range msg.ToolCalls { if tc.Name == "read_skill" { @@ -118,7 +119,7 @@ func Run(ctx context.Context, cfg Config) error { providerName, modelName, _ := strings.Cut(cfg.DefaultModel, "/") return command.StatusInfo{ SessionID: sess.ID, - MessageCount: len(sess.Messages), + MessageCount: len(messages), ToolCalls: toolCalls, SkillsLoaded: len(uniqueSkills), Provider: providerName, @@ -132,7 +133,7 @@ func Run(ctx context.Context, cfg Config) error { if sess == nil { return nil, fmt.Errorf("session not found: %s", id) } - return sess.Todos, nil + return sess.Todos(), nil }) cmdRegistry.Register(todosCmd) srv.SetCommands(cmdRegistry) diff --git a/pkg/daemon/sessions.go b/pkg/daemon/sessions.go index 9d570cf..aa92ee5 100644 --- a/pkg/daemon/sessions.go +++ b/pkg/daemon/sessions.go @@ -3,6 +3,7 @@ package daemon import ( "context" "fmt" + "slices" "sync" "time" @@ -25,11 +26,15 @@ type SessionManager struct { mu sync.RWMutex } -// ManagedSession holds the state for a single chat session. +// ManagedSession holds the state for a single chat session. Read messages and +// todos via Messages() and Todos(); the unexported fields require mu. type ManagedSession struct { - ID string - Messages []provider.Message - Todos *todo.List + ID string + // messages is append-only: a turn replaces the slice header but never + // mutates existing elements, so Messages() can hand out a shallow copy + // that is safe to read without further locking. + messages []provider.Message + todos *todo.List cfg *session.Config // per-session config (nil = use shared chatCfg) welcomeText string // per-session welcome text mu sync.Mutex @@ -38,6 +43,20 @@ type ManagedSession struct { cancelMu sync.Mutex } +// Messages returns a shallow copy of the session's conversation history. +func (s *ManagedSession) Messages() []provider.Message { + s.mu.Lock() + defer s.mu.Unlock() + return slices.Clone(s.messages) +} + +// Todos returns the session's todo list. The list is internally thread-safe. +func (s *ManagedSession) Todos() *todo.List { + s.mu.Lock() + defer s.mu.Unlock() + return s.todos +} + // NewSessionManager calls agent.Prepare to build the shared session.Config. func NewSessionManager(ctx context.Context, agentCfg agent.Config, ctr *container.Container) (*SessionManager, error) { chatCfg, dbg, err := agent.Prepare(ctx, agentCfg) @@ -63,8 +82,8 @@ func (sm *SessionManager) Create() *ManagedSession { sess := &ManagedSession{ ID: uuid.New().String(), - Messages: append([]provider.Message{}, sm.chatCfg.InitialMessages...), - Todos: todo.New(), + messages: append([]provider.Message{}, sm.chatCfg.InitialMessages...), + todos: todo.New(), createdAt: time.Now(), } sm.sessions[sess.ID] = sess @@ -87,11 +106,11 @@ func (sm *SessionManager) CreateCode(workDir string) (*ManagedSession, error) { defer sm.mu.Unlock() sess := &ManagedSession{ ID: uuid.New().String(), - Messages: append([]provider.Message{}, codeCfg.InitialMessages...), + messages: append([]provider.Message{}, codeCfg.InitialMessages...), cfg: codeCfg, welcomeText: codeCfg.WelcomeText, createdAt: time.Now(), - Todos: todo.New(), + todos: todo.New(), } sm.sessions[sess.ID] = sess return sess, nil @@ -113,11 +132,11 @@ func (sm *SessionManager) CreateWithModel(model string) (*ManagedSession, error) defer sm.mu.Unlock() sess := &ManagedSession{ ID: uuid.New().String(), - Messages: append([]provider.Message{}, chatCfg.InitialMessages...), + messages: append([]provider.Message{}, chatCfg.InitialMessages...), cfg: chatCfg, welcomeText: chatCfg.WelcomeText, createdAt: time.Now(), - Todos: todo.New(), + todos: todo.New(), } sm.sessions[sess.ID] = sess return sess, nil @@ -139,11 +158,11 @@ func (sm *SessionManager) CreateCodeWithModel(workDir, model string) (*ManagedSe defer sm.mu.Unlock() sess := &ManagedSession{ ID: uuid.New().String(), - Messages: append([]provider.Message{}, codeCfg.InitialMessages...), + messages: append([]provider.Message{}, codeCfg.InitialMessages...), cfg: codeCfg, welcomeText: codeCfg.WelcomeText, createdAt: time.Now(), - Todos: todo.New(), + todos: todo.New(), } sm.sessions[sess.ID] = sess return sess, nil @@ -165,12 +184,12 @@ func (sm *SessionManager) Close(ctx context.Context, id string) { } sm.mu.Unlock() - if !ok || len(sess.Messages) == 0 { + if !ok || len(sess.messages) == 0 { return } // Best-effort finalize (update memory). - _ = agent.Finalize(ctx, sm.agentCfg, sess.Messages) + _ = agent.Finalize(ctx, sm.agentCfg, sess.messages) } @@ -274,8 +293,8 @@ func (sm *SessionManager) runTurnInternal(ctx context.Context, id, input, extraS if onToolCall != nil { cfg.OnToolCall = provider.ToolCallCallback(onToolCall) } - if sess.Todos != nil { - todos := sess.Todos + if sess.todos != nil { + todos := sess.todos cfg.RequestReminder = func() string { return todos.RenderSystemReminder() } ctx = todo.WithList(ctx, todos) if onTodoSnapshot != nil && cfg.Executor != nil { @@ -302,12 +321,12 @@ func (sm *SessionManager) runTurnInternal(ctx context.Context, id, input, extraS } }) - updated, resp, err := session.RunTurnStream(ctx, &cfg, sess.Messages, input, chunkCb) + updated, resp, err := session.RunTurnStream(ctx, &cfg, sess.messages, input, chunkCb) if err != nil { return "", err } - sess.Messages = updated + sess.messages = updated return resp.Message.Content, nil } diff --git a/pkg/daemon/sessions_test.go b/pkg/daemon/sessions_test.go index 6ba6020..a5218e4 100644 --- a/pkg/daemon/sessions_test.go +++ b/pkg/daemon/sessions_test.go @@ -65,7 +65,7 @@ func TestReset(t *testing.T) { oldID := old.ID // Add a message so the old session is non-empty. - old.Messages = append(old.Messages, provider.Message{Role: "user", Content: "hi"}) + old.messages = append(old.messages, provider.Message{Role: "user", Content: "hi"}) newSess, err := sm.Reset(context.Background(), oldID) if err != nil { @@ -110,7 +110,7 @@ func TestReset(t *testing.T) { } old := sm.Create() // Simulate conversation history. - old.Messages = append(old.Messages, + old.messages = append(old.messages, provider.Message{Role: "user", Content: "hello"}, provider.Message{Role: "assistant", Content: "hi there"}, ) @@ -120,11 +120,12 @@ func TestReset(t *testing.T) { t.Fatalf("Reset() error: %v", err) } // New session should only have the initial messages, not the old conversation. - if len(newSess.Messages) != 1 { - t.Errorf("expected 1 initial message, got %d", len(newSess.Messages)) + messages := newSess.Messages() + if len(messages) != 1 { + t.Errorf("expected 1 initial message, got %d", len(messages)) } - if newSess.Messages[0].Content != "You are helpful." { - t.Errorf("expected initial system message, got %q", newSess.Messages[0].Content) + if messages[0].Content != "You are helpful." { + t.Errorf("expected initial system message, got %q", messages[0].Content) } }) From 09b87aba7e3eaab83a483a4cf0a1c0f3b6037b2d Mon Sep 17 00:00:00 2001 From: Ludovico Russo Date: Thu, 30 Apr 2026 14:37:28 +0200 Subject: [PATCH 3/4] refactor(daemon): cancel-then-finalize SessionManager.Close() (#86) Close() previously deleted the session from the manager map first and then read sess.messages without synchronization, racing with an in-flight turn's append. It also blocked on whatever the running turn was doing because there was no cancellation step. The new flow is: 1. Look up the session under sm.mu. 2. Cancel any in-flight turn so it releases sess.mu promptly. 3. Take sess.mu and run agent.Finalize on the now-stable message history (the lock is held across Finalize so the snapshot is consistent for the entire call). 4. Remove the session from the manager map. The duplicated cancel-under-mutex pattern shared by Close() and Interrupt() is factored into a new (*ManagedSession).cancelInflight() helper. The redundant len(messages)==0 guard is dropped: Finalize already short-circuits on empty messages and the prior read was racy. Reset() inherits the new behavior unchanged. --- pkg/daemon/sessions.go | 40 +++++++++++++++++++++++++--------------- 1 file changed, 25 insertions(+), 15 deletions(-) diff --git a/pkg/daemon/sessions.go b/pkg/daemon/sessions.go index aa92ee5..230edc3 100644 --- a/pkg/daemon/sessions.go +++ b/pkg/daemon/sessions.go @@ -57,6 +57,15 @@ func (s *ManagedSession) Todos() *todo.List { return s.todos } +// cancelInflight cancels the current turn if one is running. +func (s *ManagedSession) cancelInflight() { + s.cancelMu.Lock() + defer s.cancelMu.Unlock() + if s.cancel != nil { + s.cancel() + } +} + // NewSessionManager calls agent.Prepare to build the shared session.Config. func NewSessionManager(ctx context.Context, agentCfg agent.Config, ctr *container.Container) (*SessionManager, error) { chatCfg, dbg, err := agent.Prepare(ctx, agentCfg) @@ -175,23 +184,28 @@ func (sm *SessionManager) Get(id string) *ManagedSession { return sm.sessions[id] } -// Close finalizes the session (updates memory) and removes it. +// Close cancels any in-flight turn, finalizes the session's memory on a +// stable view of the message history, and removes it from the manager map. func (sm *SessionManager) Close(ctx context.Context, id string) { - sm.mu.Lock() + sm.mu.RLock() sess, ok := sm.sessions[id] - if ok { - delete(sm.sessions, id) - } - sm.mu.Unlock() - - if !ok || len(sess.messages) == 0 { + sm.mu.RUnlock() + if !ok { return } - // Best-effort finalize (update memory). + // Cancel first so a running turn releases sess.mu. + sess.cancelInflight() + + // Hold sess.mu across Finalize so the message slice can't change mid-call. + sess.mu.Lock() _ = agent.Finalize(ctx, sm.agentCfg, sess.messages) -} + sess.mu.Unlock() + sm.mu.Lock() + delete(sm.sessions, id) + sm.mu.Unlock() +} // Reset closes the given session (with finalize) and creates a fresh one. func (sm *SessionManager) Reset(ctx context.Context, id string) (*ManagedSession, error) { @@ -218,11 +232,7 @@ func (sm *SessionManager) Interrupt(id string) { if sess == nil { return } - sess.cancelMu.Lock() - defer sess.cancelMu.Unlock() - if sess.cancel != nil { - sess.cancel() - } + sess.cancelInflight() } // WelcomeText returns the configured welcome text. From baca0c1565aac651c884d2040fd861b8db70ebc5 Mon Sep 17 00:00:00 2001 From: Ludovico Russo Date: Thu, 30 Apr 2026 14:44:49 +0200 Subject: [PATCH 4/4] test(daemon): concurrency suite for SessionManager (#87) Adds TestSessionConcurrency in pkg/daemon/sessions_concurrency_test.go to lock in the behavior introduced by slices 1 and 2 (#85, #86) and provide regression protection under -race. Subtests: - messages snapshot serializes with in-flight turn: Messages() blocks while a turn holds sess.mu and returns a coherent slice once the turn completes. - close cancels in-flight turn and removes session: Close() unblocks the turn, finalizes, and deletes the session from the manager map. - interrupt unblocks in-flight turn: Interrupt() cancels the turn and subsequent readers proceed. - stress: hammered readers and cancellers do not race or panic: 16 hammers x 3 op kinds x 10 iters racing against an active turn plus Close(); under -race the assertion is "no race report, no panic". Helpers: - gatedProvider exposes started/release channels so subtests can deterministically synchronize on an in-flight turn without time.Sleep. - slowProvider models a turn that finishes naturally if no one cancels, used by the stress subtest so both natural-completion and cancellation paths are exercised. - newConcurrencyTestSM uses fakeProvider for agentCfg.Provider so Close()'s Finalize step never blocks regardless of the chat provider. --- pkg/daemon/sessions_concurrency_test.go | 254 ++++++++++++++++++++++++ 1 file changed, 254 insertions(+) create mode 100644 pkg/daemon/sessions_concurrency_test.go diff --git a/pkg/daemon/sessions_concurrency_test.go b/pkg/daemon/sessions_concurrency_test.go new file mode 100644 index 0000000..706da14 --- /dev/null +++ b/pkg/daemon/sessions_concurrency_test.go @@ -0,0 +1,254 @@ +package daemon + +import ( + "context" + "sync" + "testing" + "time" + + "github.com/ludusrusso/wildgecu/pkg/agent" + "github.com/ludusrusso/wildgecu/pkg/home" + "github.com/ludusrusso/wildgecu/pkg/provider" + "github.com/ludusrusso/wildgecu/pkg/session" +) + +const ( + concurrencyUserMsg = "user input" + concurrencyAssistantMsg = "ok" +) + +// gatedProvider blocks Generate until release is closed (or ctx is canceled), +// closing started when it first enters Generate so callers can deterministically +// wait for the turn to be in flight. +type gatedProvider struct { + started chan struct{} + release chan struct{} + once sync.Once +} + +func newGatedProvider() *gatedProvider { + return &gatedProvider{ + started: make(chan struct{}), + release: make(chan struct{}), + } +} + +func (p *gatedProvider) Generate(ctx context.Context, _ *provider.GenerateParams) (*provider.Response, error) { + p.once.Do(func() { close(p.started) }) + select { + case <-p.release: + return &provider.Response{Message: provider.Message{Role: provider.RoleModel, Content: concurrencyAssistantMsg}}, nil + case <-ctx.Done(): + return nil, ctx.Err() + } +} + +// slowProvider returns a canned response after sleeping for delay or until ctx +// is canceled. Used by the stress subtest to model an in-flight turn that +// finishes naturally if no one cancels it. +type slowProvider struct { + delay time.Duration +} + +func (p slowProvider) Generate(ctx context.Context, _ *provider.GenerateParams) (*provider.Response, error) { + select { + case <-time.After(p.delay): + return &provider.Response{Message: provider.Message{Role: provider.RoleModel, Content: concurrencyAssistantMsg}}, nil + case <-ctx.Done(): + return nil, ctx.Err() + } +} + +func newConcurrencyTestSM(t *testing.T, p provider.Provider) *SessionManager { + t.Helper() + h, err := home.New(t.TempDir()) + if err != nil { + t.Fatalf("home.New: %v", err) + } + return &SessionManager{ + agentCfg: agent.Config{ + Provider: fakeProvider{}, // used by Finalize; must not block + Home: h, + }, + chatCfg: &session.Config{ + Provider: p, + WelcomeText: "hello", + }, + sessions: make(map[string]*ManagedSession), + } +} + +// startGatedTurn launches RunTurnStream in a goroutine and returns a channel +// receiving the turn's terminal error. Blocks until the gatedProvider has +// entered Generate so the caller can synchronize on an in-flight turn. +func startGatedTurn(sm *SessionManager, p *gatedProvider, id, input string) <-chan error { + done := make(chan error, 1) + go func() { + _, err := sm.RunTurnStream(context.Background(), id, input, nil, nil, nil, nil) + done <- err + }() + <-p.started + return done +} + +func TestSessionConcurrency(t *testing.T) { + t.Run("messages snapshot serializes with in-flight turn", func(t *testing.T) { + p := newGatedProvider() + sm := newConcurrencyTestSM(t, p) + sess := sm.Create() + + turnDone := startGatedTurn(sm, p, sess.ID, concurrencyUserMsg) + + msgsCh := make(chan []provider.Message, 1) + go func() { msgsCh <- sess.Messages() }() + + select { + case <-msgsCh: + t.Fatal("Messages() returned while turn was in flight") + case <-time.After(20 * time.Millisecond): + } + + close(p.release) + + select { + case err := <-turnDone: + if err != nil { + t.Fatalf("turn returned error: %v", err) + } + case <-time.After(time.Second): + t.Fatal("turn did not return after release") + } + + var msgs []provider.Message + select { + case msgs = <-msgsCh: + case <-time.After(time.Second): + t.Fatal("Messages() did not return after turn finished") + } + + var sawUser, sawAssistant bool + for _, m := range msgs { + if m.Role == provider.RoleUser && m.Content == concurrencyUserMsg { + sawUser = true + } + if m.Role == provider.RoleModel && m.Content == concurrencyAssistantMsg { + sawAssistant = true + } + } + if !sawUser { + t.Errorf("snapshot missing user message; got %+v", msgs) + } + if !sawAssistant { + t.Errorf("snapshot missing assistant message; got %+v", msgs) + } + }) + + t.Run("close cancels in-flight turn and removes session", func(t *testing.T) { + p := newGatedProvider() + sm := newConcurrencyTestSM(t, p) + sess := sm.Create() + + turnDone := startGatedTurn(sm, p, sess.ID, "hi") + + closeReturned := make(chan struct{}) + go func() { + sm.Close(context.Background(), sess.ID) + close(closeReturned) + }() + + select { + case err := <-turnDone: + if err == nil { + t.Error("expected turn to return error after Close") + } + case <-time.After(time.Second): + t.Fatal("turn did not return after Close") + } + + select { + case <-closeReturned: + case <-time.After(time.Second): + t.Fatal("Close did not return") + } + + if sm.Get(sess.ID) != nil { + t.Error("expected session to be removed after Close") + } + }) + + t.Run("interrupt unblocks in-flight turn", func(t *testing.T) { + p := newGatedProvider() + sm := newConcurrencyTestSM(t, p) + sess := sm.Create() + + turnDone := startGatedTurn(sm, p, sess.ID, "hi") + + sm.Interrupt(sess.ID) + + select { + case err := <-turnDone: + if err == nil { + t.Error("expected turn to return error after Interrupt") + } + case <-time.After(time.Second): + t.Fatal("turn did not return after Interrupt") + } + + readDone := make(chan struct{}) + go func() { + _ = sess.Messages() + _ = sess.Todos() + close(readDone) + }() + select { + case <-readDone: + case <-time.After(time.Second): + t.Fatal("post-interrupt readers did not unblock") + } + }) + + t.Run("stress: hammered readers and cancellers do not race or panic", func(t *testing.T) { + sm := newConcurrencyTestSM(t, slowProvider{delay: 50 * time.Millisecond}) + sess := sm.Create() + + var wg sync.WaitGroup + + wg.Add(1) + go func() { + defer wg.Done() + _, _ = sm.RunTurnStream(context.Background(), sess.ID, "hi", nil, nil, nil, nil) + }() + + const hammers = 16 + const iters = 10 + for range hammers { + wg.Add(3) + go func() { + defer wg.Done() + for range iters { + _ = sess.Messages() + } + }() + go func() { + defer wg.Done() + for range iters { + _ = sess.Todos() + } + }() + go func() { + defer wg.Done() + for range iters { + sm.Interrupt(sess.ID) + } + }() + } + + wg.Add(1) + go func() { + defer wg.Done() + sm.Close(context.Background(), sess.ID) + }() + + wg.Wait() + }) +}