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

fix: make Codex Prompt() synchronous to fix premature task state transition (#35)

Browse Source
This commit is contained in:
João Salavisa
2026-01-18 10:43:13 +00:00
committed by GitHub
parent ee90f79102
commit 5488f0bb5d
2 changed files with 67 additions and 69 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/backend/internal/agentctl/server/adapter/codex_adapter.go
+59 -10
View File
@@ -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
apps/backend/internal/orchestrator/service.go
+8 -59
View File
@@ -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 == "" {