dreampiler/kandev
1
0
Fork 0
Code Issues 43 Pull Requests Actions 6 Packages Projects Releases Wiki Activity

fix(codex): resolve deadlock in adapter by not holding mutex during RPC calls

Browse Source 
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.
This commit is contained in:
jcfs
2026-01-16 02:39:05 +00:00
parent 0599800c40
commit bb7cb66bd9
2 changed files with 28 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/backend/internal/agent/registry/defaults.go
+1 -1
View File
@@ -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
},
},
{
apps/backend/internal/agentctl/adapter/codex_adapter.go
+27 -13
View File
@@ -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,
})