From bb7cb66bd9f612b1d68e41b28a419f2d557c0d50 Mon Sep 17 00:00:00 2001 From: jcfs Date: Fri, 16 Jan 2026 02:39:05 +0000 Subject: [PATCH] fix(codex): resolve deadlock in adapter by not holding mutex during RPC calls The LoadSession, NewSession, Cancel, and Prompt methods were holding a.mu.Lock() while waiting for client.Call() to return. The codex client's readLoop processes incoming messages synchronously, and when codex sends notifications (like mcp_startup_complete), handleNotification tries to acquire a.mu.RLock() - which blocks because the exclusive lock is held. This prevents readLoop from processing the RPC response, causing deadlock. Fix: Don't hold the mutex while waiting for RPC calls. Instead: 1. Acquire lock briefly to read client and needed fields 2. Release lock before RPC call 3. Acquire lock briefly to write results after call completes Also reverts codex-agent SessionConfig to ResumeViaACP: true since thread/resume works correctly when not deadlocked. --- .../internal/agent/registry/defaults.go | 2 +- .../agentctl/adapter/codex_adapter.go | 40 +++++++++++++------ 2 files changed, 28 insertions(+), 14 deletions(-) diff --git a/apps/backend/internal/agent/registry/defaults.go b/apps/backend/internal/agent/registry/defaults.go index 42df572305..d3c9d70a6a 100644 --- a/apps/backend/internal/agent/registry/defaults.go +++ b/apps/backend/internal/agent/registry/defaults.go @@ -86,7 +86,7 @@ func DefaultAgents() []*AgentTypeConfig { ModelFlag: "--model", Protocol: agent.ProtocolCodex, SessionConfig: SessionConfig{ - ResumeViaACP: true, // Codex uses ACP session/load for resume + ResumeViaACP: true, // Codex uses thread/resume via ACP }, }, { diff --git a/apps/backend/internal/agentctl/adapter/codex_adapter.go b/apps/backend/internal/agentctl/adapter/codex_adapter.go index ca1147472f..c3a8fbc5ac 100644 --- a/apps/backend/internal/agentctl/adapter/codex_adapter.go +++ b/apps/backend/internal/agentctl/adapter/codex_adapter.go @@ -131,16 +131,19 @@ func (a *CodexAdapter) GetAgentInfo() *AgentInfo { // NewSession creates a new Codex thread (session). func (a *CodexAdapter) NewSession(ctx context.Context) (string, error) { - a.mu.Lock() - defer a.mu.Unlock() + // Check client under lock, but don't hold lock during Call() to avoid deadlock + // with handleNotification which also needs the lock + a.mu.RLock() + client := a.client + a.mu.RUnlock() - if a.client == nil { + if client == nil { return "", fmt.Errorf("adapter not initialized") } - resp, err := a.client.Call(ctx, codex.MethodThreadStart, &codex.ThreadStartParams{ + resp, err := client.Call(ctx, codex.MethodThreadStart, &codex.ThreadStartParams{ Cwd: a.cfg.WorkDir, - ApprovalPolicy: "never", // Valid values: untrusted, on-failure, on-request, never + ApprovalPolicy: "never", // Valid values: untrusted, on-failure, on-request, never Sandbox: "danger-full-access", // Disable sandbox to avoid landlock permission issues }) if err != nil { @@ -155,7 +158,10 @@ func (a *CodexAdapter) NewSession(ctx context.Context) (string, error) { return "", fmt.Errorf("failed to parse thread start result: %w", err) } + a.mu.Lock() a.threadID = result.Thread.ID + a.mu.Unlock() + a.logger.Info("created new thread", zap.String("thread_id", a.threadID)) return a.threadID, nil @@ -163,14 +169,17 @@ func (a *CodexAdapter) NewSession(ctx context.Context) (string, error) { // LoadSession resumes an existing Codex thread. func (a *CodexAdapter) LoadSession(ctx context.Context, sessionID string) error { - a.mu.Lock() - defer a.mu.Unlock() + // Check client under lock, but don't hold lock during Call() to avoid deadlock + // with handleNotification which also needs the lock + a.mu.RLock() + client := a.client + a.mu.RUnlock() - if a.client == nil { + if client == nil { return fmt.Errorf("adapter not initialized") } - resp, err := a.client.Call(ctx, codex.MethodThreadResume, &codex.ThreadResumeParams{ + resp, err := client.Call(ctx, codex.MethodThreadResume, &codex.ThreadResumeParams{ ThreadID: sessionID, }) if err != nil { @@ -180,7 +189,10 @@ func (a *CodexAdapter) LoadSession(ctx context.Context, sessionID string) error return fmt.Errorf("thread resume error: %s", resp.Error.Message) } + a.mu.Lock() a.threadID = sessionID + a.mu.Unlock() + a.logger.Info("resumed thread", zap.String("thread_id", a.threadID)) return nil @@ -189,6 +201,7 @@ func (a *CodexAdapter) LoadSession(ctx context.Context, sessionID string) error // Prompt sends a prompt to the agent, starting a new turn. func (a *CodexAdapter) Prompt(ctx context.Context, message string) error { a.mu.Lock() + client := a.client threadID := a.threadID // Reset accumulators for new turn a.messageBuffer = "" @@ -196,7 +209,7 @@ func (a *CodexAdapter) Prompt(ctx context.Context, message string) error { a.summaryBuffer = "" a.mu.Unlock() - if a.client == nil { + if client == nil { return fmt.Errorf("adapter not initialized") } @@ -209,7 +222,7 @@ func (a *CodexAdapter) Prompt(ctx context.Context, message string) error { }, } - resp, err := a.client.Call(ctx, codex.MethodTurnStart, params) + resp, err := client.Call(ctx, codex.MethodTurnStart, params) if err != nil { return fmt.Errorf("failed to start turn: %w", err) } @@ -244,18 +257,19 @@ func (a *CodexAdapter) Prompt(ctx context.Context, message string) error { // Cancel interrupts the current turn. func (a *CodexAdapter) Cancel(ctx context.Context) error { a.mu.RLock() + client := a.client threadID := a.threadID turnID := a.turnID a.mu.RUnlock() - if a.client == nil { + if client == nil { return fmt.Errorf("adapter not initialized") } a.logger.Info("cancelling turn", zap.String("thread_id", threadID), zap.String("turn_id", turnID)) // Codex uses turn/interrupt to cancel - _, err := a.client.Call(ctx, codex.MethodTurnInterrupt, map[string]string{ + _, err := client.Call(ctx, codex.MethodTurnInterrupt, map[string]string{ "threadId": threadID, "turnId": turnID, })