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

fix(backend): send auto-start prompt after on_turn_complete context reset (#669)

Browse Source
This commit is contained in:
Carlos Florêncio
2026-04-21 19:56:53 +01:00
committed by GitHub
parent e8781a6443
commit 222bba8d08
3 changed files with 150 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/backend/internal/orchestrator/event_handlers_workflow.go
+38
View File
@@ -594,6 +594,7 @@ func (s *Service) processOnEnter(ctx context.Context, taskID string, session *mo
s.publishSessionWaitingEvent(ctx, taskID, sessionID, step.ID, session)
return
}
s.markIdleAfterReset(ctx, taskID, sessionID, session, step, isPassthrough)
}
hasAutoStart := false
@@ -836,6 +837,43 @@ func (s *Service) queueAutoStartPrompt(
return nil
}
// markIdleAfterReset flips a freshly-reset session to WAITING_FOR_INPUT so a
// following auto_start_agent sends the prompt directly instead of queueing
// against a stale RUNNING state. processOnEnter runs from handleAgentReady,
// which loads the session before the turn finishes — the in-memory pointer
// still reads RUNNING even though the agent is now idle. Without this flip,
// queueAutoStartPromptIfRunning queues the message and PromptTask later
// rejects the drained queued send because the DB row also still reads RUNNING.
//
// Skip the flip when:
// - state was not RUNNING/STARTING (e.g. CREATED, where resetAgentContext
// early-returns true without restarting and autoStartStepPrompt routes
// the prompt through StartCreatedSession);
// - the session is passthrough AND auto_start_agent will write to PTY stdin
// next (the agent is actively processing, not idle).
//
// Uses updateTaskSessionState directly rather than setSessionWaitingForInput
// because the helper would also flip the task to TaskStateReview, which would
// be wrong here — auto_start_agent runs next and should leave the task as
// IN_PROGRESS.
func (s *Service) markIdleAfterReset(
ctx context.Context,
taskID, sessionID string,
session *models.TaskSession,
step *wfmodels.WorkflowStep,
isPassthrough bool,
) {
if session.State != models.TaskSessionStateRunning &&
session.State != models.TaskSessionStateStarting {
return
}
if isPassthrough && step.HasOnEnterAction(wfmodels.OnEnterAutoStartAgent) {
return
}
s.updateTaskSessionState(ctx, taskID, sessionID, models.TaskSessionStateWaitingForInput, "", false, session)
session.State = models.TaskSessionStateWaitingForInput
}
// resetAgentContext restarts the agent subprocess with a fresh ACP session, clearing
// the agent's conversation context. The workspace environment is preserved.
func (s *Service) resetAgentContext(ctx context.Context, taskID string, session *models.TaskSession, stepName string) bool {
apps/backend/internal/orchestrator/event_handlers_workflow_triggers_test.go
+97
View File
@@ -729,4 +729,101 @@ func TestProcessOnEnterResetAgentContext(t *testing.T) {
t.Fatalf("expected session state %q, got %q", models.TaskSessionStateWaitingForInput, updated.State)
}
})
// Regression: on_turn_complete auto-transition enters processOnEnter with an
// in-memory session still marked RUNNING (loaded at the start of
// handleAgentReady before the turn finished). After reset_agent_context we
// must flip the session to WAITING_FOR_INPUT — otherwise a following
// auto_start_agent hits queueAutoStartPromptIfRunning (sees State=RUNNING)
// and queues the prompt instead of sending it, and PromptTask rejects the
// drained queued message for the same reason. The agent then sits idle
// until the user cancels. Both the in-memory pointer and the DB must be
// updated.
t.Run("reset_agent_context flips session to WAITING_FOR_INPUT for a running session", func(t *testing.T) {
repo := setupTestRepo(t)
seedSession(t, repo, "t1", "s1", "step1") // seeds session.State = RUNNING
session, _ := repo.GetTaskSession(ctx, "s1")
session.AgentExecutionID = "exec-abc"
session.Metadata = map[string]interface{}{"acp_session_id": "old-acp"}
_ = repo.UpdateTaskSession(ctx, session)
agentMgr := &mockAgentManager{}
svc := createTestServiceWithAgent(repo, newMockStepGetter(), newMockTaskRepo(), agentMgr)
step := &wfmodels.WorkflowStep{
ID: "step2", WorkflowID: "wf1", Name: "Review Step",
Events: wfmodels.StepEvents{
OnEnter: []wfmodels.OnEnterAction{
{Type: wfmodels.OnEnterResetAgentContext},
},
},
}
session, _ = repo.GetTaskSession(ctx, "s1")
if session.State != models.TaskSessionStateRunning {
t.Fatalf("precondition: seed should start session as RUNNING, got %q", session.State)
}
svc.processOnEnter(ctx, "t1", session, step, "review task")
// In-memory session must also be updated — queueAutoStartPromptIfRunning
// (further down in processOnEnter when auto_start_agent is present)
// checks the pointer, not the DB.
if session.State != models.TaskSessionStateWaitingForInput {
t.Errorf("in-memory session.State: want %q, got %q",
models.TaskSessionStateWaitingForInput, session.State)
}
updated, _ := repo.GetTaskSession(ctx, "s1")
if updated.State != models.TaskSessionStateWaitingForInput {
t.Errorf("DB session.State: want %q, got %q",
models.TaskSessionStateWaitingForInput, updated.State)
}
if len(agentMgr.restartProcessCalls) != 1 {
t.Errorf("expected 1 RestartAgentProcess call, got %d", len(agentMgr.restartProcessCalls))
}
})
// Regression: passthrough sessions handle auto_start_agent by writing to
// PTY stdin via autoStartPassthroughPrompt — the agent is actively
// processing immediately after, not idle. We must NOT flip the session to
// WAITING_FOR_INPUT, otherwise WS subscribers see "ready for input" while
// the agent is mid-prompt.
t.Run("reset_agent_context preserves running state for passthrough+auto_start", func(t *testing.T) {
repo := setupTestRepo(t)
seedSession(t, repo, "t1", "s1", "step1") // RUNNING
session, _ := repo.GetTaskSession(ctx, "s1")
session.AgentExecutionID = "exec-pt"
_ = repo.UpdateTaskSession(ctx, session)
agentMgr := &mockAgentManager{isPassthrough: true}
svc := createTestServiceWithAgent(repo, newMockStepGetter(), newMockTaskRepo(), agentMgr)
step := &wfmodels.WorkflowStep{
ID: "step2", WorkflowID: "wf1", Name: "Review Step",
Events: wfmodels.StepEvents{
OnEnter: []wfmodels.OnEnterAction{
{Type: wfmodels.OnEnterResetAgentContext},
{Type: wfmodels.OnEnterAutoStartAgent},
},
},
}
session, _ = repo.GetTaskSession(ctx, "s1")
svc.processOnEnter(ctx, "t1", session, step, "review task")
// Positive assertion: pin the expected state to RUNNING so any other
// unintended mutation (COMPLETED, FAILED, etc.) also fails the test,
// not just an erroneous flip to WAITING_FOR_INPUT.
if session.State != models.TaskSessionStateRunning {
t.Errorf("in-memory session.State should remain RUNNING for passthrough+auto_start, got %q", session.State)
}
updated, _ := repo.GetTaskSession(ctx, "s1")
if updated.State != models.TaskSessionStateRunning {
t.Errorf("DB session.State should remain RUNNING for passthrough+auto_start, got %q", updated.State)
}
})
}
apps/backend/internal/orchestrator/workflow_e2e_test.go
+15 -5
View File
@@ -7,8 +7,10 @@ import (
"fmt"
"testing"
"github.com/kandev/kandev/internal/orchestrator/executor"
"github.com/kandev/kandev/internal/orchestrator/messagequeue"
"github.com/kandev/kandev/internal/task/models"
sqliterepo "github.com/kandev/kandev/internal/task/repository/sqlite"
"github.com/kandev/kandev/internal/workflow/engine"
wfmodels "github.com/kandev/kandev/internal/workflow/models"
)
@@ -112,15 +114,22 @@ var workflowTestCases = []workflowTestCase{
// Agent finishes at Backlog → In Progress (auto_start queues)
{Trigger: engine.TriggerOnTurnComplete, ExpectStep: "In Progress",
ExpectTransitioned: true, ExpectQueued: true, ExpectResets: 0},
// Agent finishes at In Progress → New Context (reset + auto_start)
// Agent finishes at In Progress → New Context (reset + auto_start).
// After reset_agent_context, processOnEnter flips session state to
// WAITING_FOR_INPUT so auto_start sends directly instead of queueing
// against a stale RUNNING state. PromptTask then attempts the send —
// in this mock environment the executor has no running record for
// the session, so the send fails and autoStartStepPrompt's error
// path leaves the session at WAITING_FOR_INPUT. Crucially, the
// message is NOT sitting in the queue.
{Trigger: engine.TriggerOnTurnComplete, SetRunning: true, ExpectStep: "New Context",
ExpectTransitioned: true, ExpectQueued: true, ExpectResets: 1},
ExpectTransitioned: true, ExpectQueued: false, ExpectResets: 1},
// Agent starts at New Context → on_turn_start → back to In Progress
{Trigger: engine.TriggerOnTurnStart, SetRunning: true, ExpectStep: "In Progress",
ExpectTransitioned: true, ExpectQueued: false, ExpectResets: 1},
// Agent finishes at In Progress → New Context again
// Agent finishes at In Progress → New Context again (same reset + auto_start path)
{Trigger: engine.TriggerOnTurnComplete, SetRunning: true, ExpectStep: "New Context",
ExpectTransitioned: true, ExpectQueued: true, ExpectResets: 2},
ExpectTransitioned: true, ExpectQueued: false, ExpectResets: 2},
// Agent finishes at New Context → New Step (reset, no auto_start)
{Trigger: engine.TriggerOnTurnComplete, SetRunning: true, ExpectStep: "New Step",
ExpectTransitioned: true, ExpectQueued: false, ExpectResets: 3},
@@ -283,7 +292,7 @@ func buildWorkflowFromJSON(t *testing.T, jsonStr string) (*mockStepGetter, map[s
}
// createEngineService creates a Service with the workflow engine initialized.
func createEngineService(t *testing.T, repo sessionExecutorStore, sg *mockStepGetter, agentMgr *mockAgentManager) *Service {
func createEngineService(t *testing.T, repo *sqliterepo.Repository, sg *mockStepGetter, agentMgr *mockAgentManager) *Service {
t.Helper()
log := testLogger()
svc := &Service{
@@ -292,6 +301,7 @@ func createEngineService(t *testing.T, repo sessionExecutorStore, sg *mockStepGe
taskRepo: newMockTaskRepo(),
agentManager: agentMgr,
messageQueue: messagequeue.NewService(log),
executor: executor.NewExecutor(agentMgr, repo, log, executor.ExecutorConfig{}),
}
svc.SetWorkflowStepGetter(sg)
return svc