fix(backend): allow IDLE office sessions to accept follow-up prompts (#1119)
* fix(backend): allow IDLE office sessions to accept follow-up prompts The IDLE session state (introduced with the office feature) was missing from checkSessionPromptable's accept bucket, so any prompt to an IDLE office session fell into the default branch and returned ErrAgentPromptInProgress with the misleading message "agent is currently processing a prompt, session is in IDLE state" — even though IDLE means the agent process is in fact gone (torn down between turns) and the conversation is parked waiting to resume. Net effect before this fix: once an office session went IDLE it could not be re-engaged from either the UI composer or MCP message_task_kandev. The five IDLE sessions sitting in the local DB exhibit this exactly. Fix: - Add TaskSessionStateIdle to the accept bucket in checkSessionPromptable. ensureSessionRunning already does the right thing for IDLE: it detects the missing in-memory execution, finds the preserved executors_running row, and routes through ResumeSession (which uses the stored ACP resume token). - Introduce ErrSessionNotPromptable for STARTING / CREATED / FAILED / CANCELLED rejections. The previous code wrapped those in ErrAgentPromptInProgress, which is semantically wrong and breaks any caller doing errors.Is(err, ErrAgentPromptInProgress) checks (the MCP promptWithAutoResume retry path keys on a different sentinel, so this doesn't regress, but the categorisation is now honest). Tests: - TestPromptTask_IdleSession_IsPromptable — reproduces the bug end-to-end. - TestCheckSessionPromptable_StateMatrix — pins down every state's classification, including the IDLE→accept and the STARTING/CREATED/FAILED/CANCELLED → ErrSessionNotPromptable (NOT ErrAgentPromptInProgress) split. - TestEnsureSessionRunning_IdleSessionTriggersResume — confirms IDLE sessions reach ResumeSession (the second half of the recovery path). Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com> * fix: preserve requeue for STARTING/CREATED + drop time.Sleep in test Address PR #1119 review feedback from Greptile, cubic, and Claude: P1: ErrSessionNotPromptable broke the requeue guard in executeQueuedMessage and the auto-start retry loop. Pre-PR, STARTING/CREATED bucketed into ErrAgentPromptInProgress and matched isAgentPromptInProgressError, so queued messages hitting a booting session were retried; post-PR they would have been silently dropped (no DLQ — see TODO at event_handlers_agent.go). Introduce isSessionBusyError covering both sentinels and swap it into the two requeue call sites. Pinned by TestIsSessionBusyError. P2: Replace the 50ms time.Sleep in TestEnsureSessionRunning_IdleSessionTriggersResume with a deterministic state-watching poll (waits for STARTING from persistResumeState before flipping to WAITING_FOR_INPUT). No CI flake exposure, no timing assumption. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com> * fix: replace test poll loop with ticker+select per AGENTS.md Address CodeRabbit + Claude review on PR #1119: the 5ms time.Sleep poll loop in the launchAgentFunc watcher goroutine flunked the backend convention of "channel-based synchronization over sleep-based waits." testing/synctest doesn't help here (SQLite repo runs real I/O goroutines that synctest can't observe as idle), so use a ticker + select instead, keeping the 5s timeout as the safety bound. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com> --------- Co-authored-by: Kandev Agent <agent@kandev.dev> Co-authored-by: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
João Salavisa
@@ -369,7 +369,7 @@ func (s *Service) executeQueuedMessage(callerSessionID string, queuedMsg *messag
|
||||
zap.String("queue_id", queuedMsg.ID),
|
||||
zap.Error(err))
|
||||
|
||||
if isAgentPromptInProgressError(err) || isTransientPromptError(err) || isSessionResetInProgressError(err) {
|
||||
if isSessionBusyError(err) || isTransientPromptError(err) || isSessionResetInProgressError(err) {
|
||||
s.logger.Warn("queued message execution failed transiently; requeueing",
|
||||
zap.String("session_id", callerSessionID),
|
||||
zap.String("task_id", queuedMsg.TaskID),
|
||||
|
||||
@@ -1259,7 +1259,7 @@ func (s *Service) autoStartStepPrompt(
|
||||
return nil
|
||||
}
|
||||
|
||||
if !isAgentPromptInProgressError(err) && !isTransientPromptError(err) && !isSessionResetInProgressError(err) {
|
||||
if !isSessionBusyError(err) && !isTransientPromptError(err) && !isSessionResetInProgressError(err) {
|
||||
requeueTaken()
|
||||
return err
|
||||
}
|
||||
|
||||
@@ -0,0 +1,232 @@
|
||||
package orchestrator
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/kandev/kandev/internal/orchestrator/executor"
|
||||
"github.com/kandev/kandev/internal/task/models"
|
||||
v1 "github.com/kandev/kandev/pkg/api/v1"
|
||||
)
|
||||
|
||||
// TestPromptTask_IdleSession_IsPromptable reproduces the office-mode IDLE
|
||||
// re-engagement bug. An IDLE office session has its conversation parked
|
||||
// (ACP session preserved) but no live agent process. Sending a prompt MUST
|
||||
// wake the session up rather than be bounced by the "agent is currently
|
||||
// processing a prompt" guard.
|
||||
//
|
||||
// Before the fix, checkSessionPromptable's default branch returned
|
||||
// ErrAgentPromptInProgress for IDLE — the misleading error the user reported.
|
||||
func TestPromptTask_IdleSession_IsPromptable(t *testing.T) {
|
||||
repo := setupTestRepo(t)
|
||||
taskRepo := newMockTaskRepo()
|
||||
|
||||
// Model the post-resume state: agent is running, executors_running row
|
||||
// exists. This keeps the test focused on the checkSessionPromptable bug;
|
||||
// the IDLE→resume path itself is covered by
|
||||
// TestEnsureSessionRunning_IdleSessionTriggersResume.
|
||||
agentMgr := &mockAgentManager{
|
||||
isAgentRunning: true,
|
||||
repoForExecutionLookup: repo,
|
||||
}
|
||||
svc := createTestServiceWithAgent(repo, newMockStepGetter(), taskRepo, agentMgr)
|
||||
svc.executor = executor.NewExecutor(agentMgr, repo, testLogger(), executor.ExecutorConfig{})
|
||||
|
||||
seedTaskAndSession(t, repo, "task1", "session1", models.TaskSessionStateIdle)
|
||||
session, err := repo.GetTaskSession(context.Background(), "session1")
|
||||
if err != nil {
|
||||
t.Fatalf("failed to load session: %v", err)
|
||||
}
|
||||
session.AgentExecutionID = "exec-idle-1"
|
||||
seedExecutorRunning(t, repo, session.ID, session.TaskID, "exec-idle-1")
|
||||
if err := repo.UpdateTaskSession(context.Background(), session); err != nil {
|
||||
t.Fatalf("failed to update session: %v", err)
|
||||
}
|
||||
|
||||
_, err = svc.PromptTask(context.Background(), "task1", "session1", "follow-up", "", false, nil, false)
|
||||
if err != nil {
|
||||
if errors.Is(err, ErrAgentPromptInProgress) {
|
||||
t.Fatalf("IDLE session must not surface ErrAgentPromptInProgress: %v", err)
|
||||
}
|
||||
if strings.Contains(err.Error(), "agent is currently processing a prompt") {
|
||||
t.Fatalf("must not return 'agent is currently processing a prompt' for IDLE: %v", err)
|
||||
}
|
||||
t.Fatalf("expected IDLE session to accept prompt, got: %v", err)
|
||||
}
|
||||
|
||||
// The prompt was forwarded to the (post-resume) live agent.
|
||||
if len(agentMgr.capturedPrompts) != 1 {
|
||||
t.Fatalf("expected one captured prompt, got %d", len(agentMgr.capturedPrompts))
|
||||
}
|
||||
}
|
||||
|
||||
// TestIsSessionBusyError pins down the requeue umbrella: STARTING / CREATED /
|
||||
// RUNNING all need to come back true so executeQueuedMessage and the auto-start
|
||||
// loop in event_handlers_workflow.go requeue rather than drop. Without this
|
||||
// helper, splitting ErrAgentPromptInProgress and ErrSessionNotPromptable
|
||||
// would silently lose queued messages that hit a session mid-startup.
|
||||
func TestIsSessionBusyError(t *testing.T) {
|
||||
cases := []struct {
|
||||
name string
|
||||
err error
|
||||
want bool
|
||||
}{
|
||||
{"nil", nil, false},
|
||||
{"unrelated", errors.New("something else"), false},
|
||||
{"in-progress sentinel", ErrAgentPromptInProgress, true},
|
||||
{"wrapped in-progress", fmt.Errorf("outer: %w", ErrAgentPromptInProgress), true},
|
||||
{"not-promptable sentinel", ErrSessionNotPromptable, true},
|
||||
{"wrapped not-promptable", fmt.Errorf("outer: %w", ErrSessionNotPromptable), true},
|
||||
{"reset is NOT busy", ErrSessionResetInProgress, false},
|
||||
}
|
||||
for _, tc := range cases {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
if got := isSessionBusyError(tc.err); got != tc.want {
|
||||
t.Errorf("isSessionBusyError(%v) = %v, want %v", tc.err, got, tc.want)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// TestCheckSessionPromptable_StateMatrix exercises every state to lock in the
|
||||
// IDLE acceptance and document which states are rejected and with which error.
|
||||
// RUNNING → ErrAgentPromptInProgress (someone else is mid-turn).
|
||||
// IDLE / COMPLETED / WAITING_FOR_INPUT → accepted.
|
||||
// Everything else → ErrSessionNotPromptable (NOT ErrAgentPromptInProgress).
|
||||
func TestCheckSessionPromptable_StateMatrix(t *testing.T) {
|
||||
repo := setupTestRepo(t)
|
||||
svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo())
|
||||
|
||||
cases := []struct {
|
||||
state models.TaskSessionState
|
||||
wantErr bool
|
||||
wantIs error
|
||||
}{
|
||||
{models.TaskSessionStateWaitingForInput, false, nil},
|
||||
{models.TaskSessionStateCompleted, false, nil},
|
||||
{models.TaskSessionStateIdle, false, nil},
|
||||
{models.TaskSessionStateRunning, true, ErrAgentPromptInProgress},
|
||||
{models.TaskSessionStateStarting, true, ErrSessionNotPromptable},
|
||||
{models.TaskSessionStateFailed, true, ErrSessionNotPromptable},
|
||||
{models.TaskSessionStateCancelled, true, ErrSessionNotPromptable},
|
||||
{models.TaskSessionStateCreated, true, ErrSessionNotPromptable},
|
||||
}
|
||||
for _, tc := range cases {
|
||||
t.Run(string(tc.state), func(t *testing.T) {
|
||||
err := svc.checkSessionPromptable("t", "s", tc.state)
|
||||
if tc.wantErr {
|
||||
if err == nil {
|
||||
t.Fatalf("expected error for state %q", tc.state)
|
||||
}
|
||||
if tc.wantIs != nil && !errors.Is(err, tc.wantIs) {
|
||||
t.Fatalf("state %q: expected errors.Is(%v); got %v", tc.state, tc.wantIs, err)
|
||||
}
|
||||
// Non-RUNNING states must NOT be classified as
|
||||
// "agent in progress" — that misleads the UI and any caller
|
||||
// doing errors.Is(err, ErrAgentPromptInProgress) checks.
|
||||
if tc.state != models.TaskSessionStateRunning && errors.Is(err, ErrAgentPromptInProgress) {
|
||||
t.Fatalf("state %q must not wrap ErrAgentPromptInProgress: %v", tc.state, err)
|
||||
}
|
||||
} else if err != nil {
|
||||
t.Fatalf("expected nil for state %q, got: %v", tc.state, err)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// TestEnsureSessionRunning_IdleSessionTriggersResume confirms that an IDLE
|
||||
// session with a preserved executors_running row (the normal office-mode IDLE
|
||||
// shape, since handleOfficeTurnComplete tears down the agent process but leaves
|
||||
// the row intact) routes through ResumeSession when the in-memory execution
|
||||
// store has no live entry. This pins down the second half of the bug: even
|
||||
// with checkSessionPromptable fixed, ensureSessionRunning must not bounce IDLE
|
||||
// — it must resume.
|
||||
func TestEnsureSessionRunning_IdleSessionTriggersResume(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
repo := setupTestRepo(t)
|
||||
|
||||
seedTaskAndSession(t, repo, "task1", "session1", models.TaskSessionStateIdle)
|
||||
session, err := repo.GetTaskSession(ctx, "session1")
|
||||
if err != nil {
|
||||
t.Fatalf("failed to load session: %v", err)
|
||||
}
|
||||
session.AgentExecutionID = "exec-idle-1"
|
||||
session.AgentProfileID = "profile1"
|
||||
if err := repo.UpdateTaskSession(ctx, session); err != nil {
|
||||
t.Fatalf("failed to update session: %v", err)
|
||||
}
|
||||
seedExecutorRunning(t, repo, session.ID, session.TaskID, "exec-idle-1")
|
||||
|
||||
// Mock launch: report success and capture the call so the test can confirm
|
||||
// the resume path triggered. isAgentRunning starts false (true IDLE shape).
|
||||
// After LaunchAgent returns, ResumeSession's persistResumeState forces the
|
||||
// session into STARTING; we then need to flip to WAITING_FOR_INPUT so the
|
||||
// downstream waitForSessionReady poll exits. We watch for STARTING to
|
||||
// appear (deterministic — no time.Sleep race against persistResumeState)
|
||||
// then flip, mirroring the real AgentBootReady → handleAgentBootReady flow.
|
||||
launchCalled := make(chan struct{}, 1)
|
||||
agentMgr := &mockAgentManager{
|
||||
isAgentRunning: false,
|
||||
repoForExecutionLookup: repo,
|
||||
launchAgentFunc: func(_ context.Context, req *executor.LaunchAgentRequest) (*executor.LaunchAgentResponse, error) {
|
||||
select {
|
||||
case launchCalled <- struct{}{}:
|
||||
default:
|
||||
}
|
||||
go func(sessID string) {
|
||||
// Watch for STARTING (committed by persistResumeState) then flip
|
||||
// — guarantees ordering without time.Sleep. Ticker + select per
|
||||
// AGENTS.md "channel-based synchronization over sleep-based
|
||||
// waits" guidance (testing/synctest doesn't help here because
|
||||
// the SQLite repo runs real I/O goroutines that synctest can't
|
||||
// fully observe as idle).
|
||||
tick := time.NewTicker(5 * time.Millisecond)
|
||||
defer tick.Stop()
|
||||
timeout := time.After(5 * time.Second)
|
||||
for {
|
||||
select {
|
||||
case <-tick.C:
|
||||
sess, err := repo.GetTaskSession(context.Background(), sessID)
|
||||
if err == nil && sess != nil && sess.State == models.TaskSessionStateStarting {
|
||||
sess.State = models.TaskSessionStateWaitingForInput
|
||||
sess.UpdatedAt = time.Now().UTC()
|
||||
_ = repo.UpdateTaskSession(context.Background(), sess)
|
||||
return
|
||||
}
|
||||
case <-timeout:
|
||||
return
|
||||
}
|
||||
}
|
||||
}(req.SessionID)
|
||||
return &executor.LaunchAgentResponse{AgentExecutionID: "exec-resumed-1"}, nil
|
||||
},
|
||||
}
|
||||
|
||||
taskRepo := newMockTaskRepo()
|
||||
taskRepo.tasks["task1"] = &v1.Task{
|
||||
ID: "task1",
|
||||
Title: "Test Task",
|
||||
State: v1.TaskStateInProgress,
|
||||
}
|
||||
svc := createTestServiceWithAgent(repo, newMockStepGetter(), taskRepo, agentMgr)
|
||||
svc.executor = executor.NewExecutor(agentMgr, repo, testLogger(), executor.ExecutorConfig{})
|
||||
|
||||
// Re-load
|
||||
session, err = repo.GetTaskSession(ctx, "session1")
|
||||
if err != nil {
|
||||
t.Fatalf("reload: %v", err)
|
||||
}
|
||||
|
||||
if err := svc.ensureSessionRunning(ctx, "session1", session); err != nil {
|
||||
t.Fatalf("ensureSessionRunning failed for IDLE session: %v", err)
|
||||
}
|
||||
select {
|
||||
case <-launchCalled:
|
||||
default:
|
||||
t.Fatal("expected ResumeSession to call LaunchAgent on IDLE session, but it never fired")
|
||||
}
|
||||
}
|
||||
@@ -40,10 +40,32 @@ const resumeReasonFailedSessionResumable = "failed_session_resumable"
|
||||
var ErrAgentPromptInProgress = errors.New("agent is currently processing a prompt")
|
||||
var ErrSessionResetInProgress = errors.New("session reset in progress")
|
||||
|
||||
// ErrSessionNotPromptable is returned when a session cannot accept a prompt
|
||||
// because of its lifecycle state (STARTING, CREATED, FAILED, CANCELLED).
|
||||
// Distinct from ErrAgentPromptInProgress, which is RUNNING-only — confusing
|
||||
// the two misleads the UI and any caller doing errors.Is checks.
|
||||
var ErrSessionNotPromptable = errors.New("session not promptable")
|
||||
|
||||
func isAgentPromptInProgressError(err error) bool {
|
||||
return err != nil && errors.Is(err, ErrAgentPromptInProgress)
|
||||
}
|
||||
|
||||
// isSessionBusyError reports whether the session is in a state where a queued
|
||||
// or auto-started prompt should be retried later rather than dropped. Covers
|
||||
// both "the agent is mid-turn" (ErrAgentPromptInProgress, RUNNING) and "the
|
||||
// session isn't yet ready to accept input" (ErrSessionNotPromptable —
|
||||
// STARTING, CREATED, FAILED, CANCELLED). The pre-PR code path collapsed both
|
||||
// into ErrAgentPromptInProgress; this helper preserves the requeue behaviour
|
||||
// after the error split so queued messages targeting a session that is
|
||||
// briefly STARTING/CREATED don't get silently dropped (see TODO about the
|
||||
// missing dead-letter queue in executeQueuedMessage).
|
||||
func isSessionBusyError(err error) bool {
|
||||
if err == nil {
|
||||
return false
|
||||
}
|
||||
return errors.Is(err, ErrAgentPromptInProgress) || errors.Is(err, ErrSessionNotPromptable)
|
||||
}
|
||||
|
||||
func isSessionResetInProgressError(err error) bool {
|
||||
return err != nil && errors.Is(err, ErrSessionResetInProgress)
|
||||
}
|
||||
@@ -1933,10 +1955,18 @@ func (s *Service) PromptTask(ctx context.Context, taskID, sessionID string, prom
|
||||
}
|
||||
|
||||
// checkSessionPromptable returns nil when the session's state accepts a new
|
||||
// prompt, or an ErrAgentPromptInProgress-wrapped error otherwise.
|
||||
// prompt. RUNNING is rejected with ErrAgentPromptInProgress; any other
|
||||
// non-acceptable state (STARTING / CREATED / FAILED / CANCELLED) is rejected
|
||||
// with ErrSessionNotPromptable so callers can distinguish "wait for the
|
||||
// current turn" from "this session is not in a state where it can take a
|
||||
// prompt". IDLE is acceptable: office sessions intentionally park in IDLE
|
||||
// between turns (agent torn down, ACP session preserved) and the next prompt
|
||||
// resumes them — see ensureSessionRunning.
|
||||
func (s *Service) checkSessionPromptable(taskID, sessionID string, state models.TaskSessionState) error {
|
||||
switch state {
|
||||
case models.TaskSessionStateWaitingForInput, models.TaskSessionStateCompleted:
|
||||
case models.TaskSessionStateWaitingForInput,
|
||||
models.TaskSessionStateCompleted,
|
||||
models.TaskSessionStateIdle:
|
||||
return nil
|
||||
case models.TaskSessionStateRunning:
|
||||
s.logger.Warn("rejected prompt while agent is already running",
|
||||
@@ -1949,7 +1979,7 @@ func (s *Service) checkSessionPromptable(taskID, sessionID string, state models.
|
||||
zap.String("task_id", taskID),
|
||||
zap.String("session_id", sessionID),
|
||||
zap.String("session_state", string(state)))
|
||||
return fmt.Errorf("%w, session is in %s state", ErrAgentPromptInProgress, state)
|
||||
return fmt.Errorf("%w: session is in %s state", ErrSessionNotPromptable, state)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user