diff --git a/apps/backend/internal/agentctl/server/adapter/codex_adapter.go b/apps/backend/internal/agentctl/server/adapter/codex_adapter.go index 02914acfed..221858b4aa 100644 --- a/apps/backend/internal/agentctl/server/adapter/codex_adapter.go +++ b/apps/backend/internal/agentctl/server/adapter/codex_adapter.go @@ -49,11 +49,20 @@ type CodexAdapter struct { reasoningBuffer string summaryBuffer string + // Turn completion signaling + turnCompleteCh chan turnCompleteResult + // Synchronization mu sync.RWMutex closed bool } +// turnCompleteResult holds the result of a completed turn +type turnCompleteResult struct { + success bool + err string +} + // NewCodexAdapter creates a new Codex protocol adapter. // stdin and stdout are the subprocess's stdin/stdout pipes (managed by process.Manager). func NewCodexAdapter(stdin io.Writer, stdout io.Reader, cfg *Config, log *logger.Logger) *CodexAdapter { @@ -236,6 +245,7 @@ func (a *CodexAdapter) LoadSession(ctx context.Context, sessionID string) error } // Prompt sends a prompt to the agent, starting a new turn. +// This method blocks until the turn completes (turn/completed notification received). func (a *CodexAdapter) Prompt(ctx context.Context, message string) error { a.mu.Lock() client := a.client @@ -244,6 +254,8 @@ func (a *CodexAdapter) Prompt(ctx context.Context, message string) error { a.messageBuffer = "" a.reasoningBuffer = "" a.summaryBuffer = "" + // Create channel to wait for turn completion + a.turnCompleteCh = make(chan turnCompleteResult, 1) a.mu.Unlock() if client == nil { @@ -261,10 +273,16 @@ func (a *CodexAdapter) Prompt(ctx context.Context, message string) error { resp, err := client.Call(ctx, codex.MethodTurnStart, params) if err != nil { + a.mu.Lock() + a.turnCompleteCh = nil + a.mu.Unlock() return fmt.Errorf("failed to start turn: %w", err) } if resp.Error != nil { + a.mu.Lock() + a.turnCompleteCh = nil + a.mu.Unlock() return fmt.Errorf("turn start error: %s", resp.Error.Message) } @@ -280,15 +298,32 @@ func (a *CodexAdapter) Prompt(ctx context.Context, message string) error { a.mu.Lock() a.turnID = turnID + completeCh := a.turnCompleteCh a.mu.Unlock() if result.Turn != nil { - a.logger.Info("started turn", zap.String("turn_id", turnID), zap.String("status", result.Turn.Status)) + a.logger.Info("started turn, waiting for completion", zap.String("turn_id", turnID), zap.String("status", result.Turn.Status)) } else { - a.logger.Info("started turn", zap.String("turn_id", turnID)) + a.logger.Info("started turn, waiting for completion", zap.String("turn_id", turnID)) } - return nil + // Wait for turn completion or context cancellation + select { + case <-ctx.Done(): + a.mu.Lock() + a.turnCompleteCh = nil + a.mu.Unlock() + return ctx.Err() + case completeResult := <-completeCh: + a.mu.Lock() + a.turnCompleteCh = nil + a.mu.Unlock() + if !completeResult.success && completeResult.err != "" { + return fmt.Errorf("turn failed: %s", completeResult.err) + } + a.logger.Info("turn completed", zap.String("turn_id", turnID), zap.Bool("success", completeResult.success)) + return nil + } } // Cancel interrupts the current turn. @@ -439,16 +474,30 @@ func (a *CodexAdapter) handleNotification(method string, params json.RawMessage) a.logger.Warn("failed to parse turn completed", zap.Error(err)) return } - update := AgentEvent{ - Type: EventTypeComplete, - SessionID: threadID, - OperationID: p.TurnID, + + // Signal turn completion to the waiting Prompt() call + a.mu.RLock() + completeCh := a.turnCompleteCh + a.mu.RUnlock() + + if completeCh != nil { + select { + case completeCh <- turnCompleteResult{success: p.Success, err: p.Error}: + a.logger.Debug("signaled turn completion", zap.String("turn_id", p.TurnID), zap.Bool("success", p.Success)) + default: + a.logger.Warn("turn complete channel full, dropping signal") + } } + + // Send error event if the turn failed (for UI notification) if !p.Success && p.Error != "" { - update.Type = EventTypeError - update.Error = p.Error + a.sendUpdate(AgentEvent{ + Type: EventTypeError, + SessionID: threadID, + OperationID: p.TurnID, + Error: p.Error, + }) } - a.sendUpdate(update) case codex.NotifyTurnDiffUpdated: var p codex.TurnDiffUpdatedParams diff --git a/apps/backend/internal/orchestrator/service.go b/apps/backend/internal/orchestrator/service.go index 2c2b2322d1..4ab1829c89 100644 --- a/apps/backend/internal/orchestrator/service.go +++ b/apps/backend/internal/orchestrator/service.go @@ -90,14 +90,6 @@ type Service struct { mu sync.RWMutex running bool startedAt time.Time - - readyMu sync.Mutex - readyStates map[string]*readyState -} - -type readyState struct { - readySeen bool - promptCompleteSeen bool } // Status contains orchestrator status information @@ -149,7 +141,6 @@ func NewService( executor: exec, scheduler: sched, streamHandlers: make([]func(payload *lifecycle.AgentStreamEventPayload), 0), - readyStates: make(map[string]*readyState), } // Create the watcher with event handlers that wire everything together @@ -723,17 +714,13 @@ func (s *Service) handleTaskStateChanged(ctx context.Context, data watcher.TaskE } // handleAgentReady handles agent ready events (prompt completed, ready for follow-up) -// This is the definitive signal that the agent has finished processing a prompt. -// AgentReady fires ONCE after each prompt completes (unlike PromptComplete which -// fires for intermediate messages before tool calls). +// Now that both ACP and Codex Prompt() calls are synchronous, this event fires after +// the "complete" stream event. The state transition is already handled by handleAgentStreamEvent +// on the "complete" event, so this handler just logs the event for debugging. func (s *Service) handleAgentReady(ctx context.Context, data watcher.AgentEventData) { - s.logger.Info("handling agent ready", + s.logger.Debug("agent ready event received (state transition handled by complete event)", zap.String("task_id", data.TaskID), zap.String("agent_execution_id", data.AgentExecutionID)) - - if s.markAgentReady(data.TaskID) { - s.finalizeAgentReady(data.TaskID, "") - } } func (s *Service) handleACPSessionCreated(ctx context.Context, data watcher.ACPSessionEventData) { @@ -881,10 +868,10 @@ func (s *Service) handleAgentStreamEvent(ctx context.Context, payload *lifecycle case "complete": // Save any accumulated text as an agent message s.saveAgentTextIfPresent(ctx, payload) - // Mark prompt as complete and finalize agent ready state - if s.markPromptComplete(taskID) { - s.finalizeAgentReady(taskID, sessionID) - } + // Now that both ACP and Codex Prompt() calls are synchronous (return when turn is done), + // we can directly finalize the agent ready state here. + // The "complete" event is now the single source of truth for turn completion. + s.finalizeAgentReady(taskID, sessionID) case "error": // Handle error events @@ -1039,44 +1026,6 @@ func (s *Service) updateTaskSessionState(ctx context.Context, taskID, sessionID } } -func (s *Service) markAgentReady(taskID string) bool { - s.readyMu.Lock() - defer s.readyMu.Unlock() - - state, ok := s.readyStates[taskID] - if !ok { - state = &readyState{} - s.readyStates[taskID] = state - } - - state.readySeen = true - readyToFinalize := state.promptCompleteSeen - if readyToFinalize { - delete(s.readyStates, taskID) - } - - return readyToFinalize -} - -func (s *Service) markPromptComplete(taskID string) bool { - s.readyMu.Lock() - defer s.readyMu.Unlock() - - state, ok := s.readyStates[taskID] - if !ok { - state = &readyState{} - s.readyStates[taskID] = state - } - - state.promptCompleteSeen = true - readyToFinalize := state.readySeen - if readyToFinalize { - delete(s.readyStates, taskID) - } - - return readyToFinalize -} - func (s *Service) finalizeAgentReady(taskID, sessionID string) { ctx := context.Background() if sessionID == "" {