fix(backend): stop workflow step reset from double-starting the agent (#2263)
This commit is contained in:
@@ -3,10 +3,12 @@ package orchestrator
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"strings"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/kandev/kandev/internal/agent/runtime/lifecycle"
|
||||
"github.com/kandev/kandev/internal/orchestrator/executor"
|
||||
"github.com/kandev/kandev/internal/orchestrator/messagequeue"
|
||||
"github.com/kandev/kandev/internal/orchestrator/watcher"
|
||||
@@ -527,3 +529,159 @@ func TestAutoStartTransientError_AutoResumesWhenAgentDead(t *testing.T) {
|
||||
t.Errorf("expected exactly 1 Merge user_message, got %d", mergeUserMsgs)
|
||||
}
|
||||
}
|
||||
|
||||
// TestAutoStartCreatedLaunch_QueuesPromptWhenAgentAlreadyRunning covers the
|
||||
// CREATED branch of autoStartStepPrompt, which short-circuits before the
|
||||
// PromptTask retry loop and therefore never reached that loop's
|
||||
// isAgentAlreadyRunningError → queue handling.
|
||||
//
|
||||
// recordAutoStartMessage has already written the prompt to chat history by the
|
||||
// time StartCreatedSession fails. Before the fix the only recovery was
|
||||
// requeueTaken(), which restores a *taken handoff* message and not the recorded
|
||||
// prompt — so the prompt existed as a chat row that no agent would ever be
|
||||
// given, and the session sat idle after recovery booted a fresh agent.
|
||||
//
|
||||
// ErrAgentAlreadyRunning is the synchronous shape of exactly that collision:
|
||||
// a concurrent path already owns a start for this session.
|
||||
func TestAutoStartCreatedLaunch_QueuesPromptWhenAgentAlreadyRunning(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
const (
|
||||
taskID = "task-created"
|
||||
sessionID = "session-created"
|
||||
profile = "profile-created"
|
||||
)
|
||||
|
||||
repo := setupTestRepo(t)
|
||||
seedTaskAndSession(t, repo, taskID, sessionID, models.TaskSessionStateCreated)
|
||||
|
||||
taskRepo := newMockTaskRepo()
|
||||
taskRepo.tasks[taskID] = &v1.Task{
|
||||
ID: taskID, Title: "Created Task", State: v1.TaskStateInProgress,
|
||||
}
|
||||
|
||||
// No in-memory execution (the shape after a backend restart), so
|
||||
// startAgentOnExistingWorkspace returns ErrStaleExecution and the full
|
||||
// LaunchAgent path runs — the one route that can still fail synchronously.
|
||||
// A concurrent path already owns the start for this session.
|
||||
agentMgr := &mockAgentManager{
|
||||
launchAgentFunc: func(context.Context, *executor.LaunchAgentRequest) (*executor.LaunchAgentResponse, error) {
|
||||
return nil, lifecycle.ErrAgentAlreadyRunning
|
||||
},
|
||||
}
|
||||
|
||||
step := &wfmodels.WorkflowStep{ID: "step-work", WorkflowID: "wf1", Name: "Work"}
|
||||
stepGetter := newMockStepGetter()
|
||||
stepGetter.steps[step.ID] = step
|
||||
|
||||
svc := createTestServiceWithScheduler(repo, stepGetter, taskRepo, agentMgr)
|
||||
msgCreator := &mockMessageCreator{}
|
||||
svc.messageCreator = msgCreator
|
||||
|
||||
session, err := repo.GetTaskSession(ctx, sessionID)
|
||||
if err != nil {
|
||||
t.Fatalf("get session: %v", err)
|
||||
}
|
||||
session.AgentProfileID = profile
|
||||
if err := repo.UpdateTaskSession(ctx, session); err != nil {
|
||||
t.Fatalf("set session profile: %v", err)
|
||||
}
|
||||
|
||||
err = svc.autoStartStepPrompt(ctx, taskID, session, step, "Do the work", false, true)
|
||||
if err == nil {
|
||||
t.Fatal("expected autoStartStepPrompt to return the launch error")
|
||||
}
|
||||
if !isAgentAlreadyRunningError(err) {
|
||||
t.Fatalf("expected an already-running launch error, got %v", err)
|
||||
}
|
||||
|
||||
// The recorded prompt must survive as a queued message the boot-ready
|
||||
// drain can deliver, not only as an undeliverable chat row.
|
||||
if got := svc.messageQueue.GetStatus(ctx, sessionID).Count; got != 1 {
|
||||
t.Fatalf("expected the prompt to be queued for redelivery, queue count = %d", got)
|
||||
}
|
||||
|
||||
// The chat row was already written, so the drain must not write a second.
|
||||
if len(msgCreator.userMessages) != 1 {
|
||||
t.Fatalf("expected exactly 1 recorded user message, got %d", len(msgCreator.userMessages))
|
||||
}
|
||||
}
|
||||
|
||||
// Regression for the same CREATED branch, with a hand-off message in play.
|
||||
//
|
||||
// autoStartStepPrompt takes any queued hand-off (e.g. from move_task_kandev)
|
||||
// and merges it into the auto-start prompt, so the queued prompt on the failure
|
||||
// path already carries the hand-off content. requeueTaken then restored the
|
||||
// *original* hand-off on top of it, leaving two entries whose combined text
|
||||
// contains the hand-off twice — so the boot-ready drain delivered it twice.
|
||||
//
|
||||
// Only a failure that did NOT preserve the merged prompt may restore the taken
|
||||
// message; the shouldQueueIfBusy branch above and the PromptTask retry loop
|
||||
// below both already follow that rule.
|
||||
func TestAutoStartCreatedLaunch_DoesNotRestoreHandoffAfterQueueingMergedPrompt(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
const (
|
||||
taskID = "task-handoff"
|
||||
sessionID = "session-handoff"
|
||||
profile = "profile-handoff"
|
||||
autoStart = "Do the work"
|
||||
handoffText = "Also update the changelog"
|
||||
)
|
||||
|
||||
repo := setupTestRepo(t)
|
||||
seedTaskAndSession(t, repo, taskID, sessionID, models.TaskSessionStateCreated)
|
||||
|
||||
taskRepo := newMockTaskRepo()
|
||||
taskRepo.tasks[taskID] = &v1.Task{
|
||||
ID: taskID, Title: "Handoff Task", State: v1.TaskStateInProgress,
|
||||
}
|
||||
|
||||
agentMgr := &mockAgentManager{
|
||||
launchAgentFunc: func(context.Context, *executor.LaunchAgentRequest) (*executor.LaunchAgentResponse, error) {
|
||||
return nil, lifecycle.ErrAgentAlreadyRunning
|
||||
},
|
||||
}
|
||||
|
||||
step := &wfmodels.WorkflowStep{ID: "step-work", WorkflowID: "wf1", Name: "Work"}
|
||||
stepGetter := newMockStepGetter()
|
||||
stepGetter.steps[step.ID] = step
|
||||
|
||||
svc := createTestServiceWithScheduler(repo, stepGetter, taskRepo, agentMgr)
|
||||
svc.messageCreator = &mockMessageCreator{}
|
||||
|
||||
session, err := repo.GetTaskSession(ctx, sessionID)
|
||||
if err != nil {
|
||||
t.Fatalf("get session: %v", err)
|
||||
}
|
||||
session.AgentProfileID = profile
|
||||
if err := repo.UpdateTaskSession(ctx, session); err != nil {
|
||||
t.Fatalf("set session profile: %v", err)
|
||||
}
|
||||
|
||||
// The hand-off the user attached to the step move, waiting in the queue.
|
||||
if _, err := svc.messageQueue.QueueMessage(
|
||||
ctx, sessionID, taskID, handoffText, "", "user-1", false, nil,
|
||||
); err != nil {
|
||||
t.Fatalf("seed handoff message: %v", err)
|
||||
}
|
||||
|
||||
err = svc.autoStartStepPrompt(ctx, taskID, session, step, autoStart, false, true)
|
||||
if err == nil {
|
||||
t.Fatal("expected autoStartStepPrompt to return the launch error")
|
||||
}
|
||||
|
||||
status := svc.messageQueue.GetStatus(ctx, sessionID)
|
||||
if status.Count != 1 {
|
||||
t.Fatalf("expected exactly 1 queued entry after the failed launch, got %d: %+v",
|
||||
status.Count, status.Entries)
|
||||
}
|
||||
// The single entry must be the merged prompt — the hand-off is preserved
|
||||
// inside it, not alongside it.
|
||||
content := status.Entries[0].Content
|
||||
if !strings.Contains(content, autoStart) || !strings.Contains(content, handoffText) {
|
||||
t.Fatalf("queued entry lost content, want both the auto-start prompt and the hand-off, got %q", content)
|
||||
}
|
||||
if strings.Count(content, handoffText) != 1 {
|
||||
t.Fatalf("hand-off text appears %d times in the queued prompt, want 1: %q",
|
||||
strings.Count(content, handoffText), content)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -223,15 +223,18 @@ type mockAgentManager struct {
|
||||
// optional rowLivenessProber so reconciliation tests can drive runtime-aware
|
||||
// liveness per row. Nil → the mock is not a prober and reconciliation treats
|
||||
// every row as Unknown.
|
||||
rowLivenessFn func(*models.ExecutorRunning) models.ProcessLiveness
|
||||
resolveProfileInfo *executor.AgentProfileInfo
|
||||
resolveProfileErr error
|
||||
restartProcessCalls []string // tracks execution IDs passed to RestartAgentProcess
|
||||
restartProcessErr error
|
||||
promptErr error
|
||||
promptResult *executor.PromptResult
|
||||
promptAgentFunc func(context.Context, string, string, []v1.MessageAttachment, bool) (*executor.PromptResult, error)
|
||||
launchAgentFunc func(context.Context, *executor.LaunchAgentRequest) (*executor.LaunchAgentResponse, error)
|
||||
rowLivenessFn func(*models.ExecutorRunning) models.ProcessLiveness
|
||||
resolveProfileInfo *executor.AgentProfileInfo
|
||||
resolveProfileErr error
|
||||
restartProcessCalls []string // tracks execution IDs passed to RestartAgentProcess
|
||||
restartProcessErr error
|
||||
promptErr error
|
||||
promptResult *executor.PromptResult
|
||||
promptAgentFunc func(context.Context, string, string, []v1.MessageAttachment, bool) (*executor.PromptResult, error)
|
||||
launchAgentFunc func(context.Context, *executor.LaunchAgentRequest) (*executor.LaunchAgentResponse, error)
|
||||
startAgentProcessCalls []string
|
||||
startAgentProcessErr error
|
||||
startAgentProcessFunc func(context.Context, string) error
|
||||
|
||||
mu sync.Mutex
|
||||
stopAgentWithReasonArgs []stopAgentCall // tracks StopAgentWithReason calls
|
||||
@@ -349,8 +352,18 @@ func (m *mockAgentManager) LaunchAgent(ctx context.Context, req *executor.Launch
|
||||
}
|
||||
return nil, nil
|
||||
}
|
||||
func (m *mockAgentManager) StartAgentProcess(_ context.Context, _ string) error { return nil }
|
||||
func (m *mockAgentManager) IsAgentCommandConfigured(_ string) bool { return true }
|
||||
func (m *mockAgentManager) StartAgentProcess(ctx context.Context, sessionID string) error {
|
||||
m.mu.Lock()
|
||||
m.startAgentProcessCalls = append(m.startAgentProcessCalls, sessionID)
|
||||
hook := m.startAgentProcessFunc
|
||||
err := m.startAgentProcessErr
|
||||
m.mu.Unlock()
|
||||
if hook != nil {
|
||||
return hook(ctx, sessionID)
|
||||
}
|
||||
return err
|
||||
}
|
||||
func (m *mockAgentManager) IsAgentCommandConfigured(_ string) bool { return true }
|
||||
func (m *mockAgentManager) StopAgent(_ context.Context, agentExecutionID string, force bool) error {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
|
||||
@@ -1151,8 +1151,8 @@ func (s *Service) processOnEnter(ctx context.Context, taskID string, session *mo
|
||||
}
|
||||
|
||||
switch {
|
||||
case hasAutoStart && isPassthrough:
|
||||
// Passthrough path: write prompt directly to PTY stdin.
|
||||
case hasAutoStart && isPassthrough && session.State != models.TaskSessionStateCreated:
|
||||
// Started passthrough path: write prompt directly to PTY stdin.
|
||||
// By the time processOnEnter runs (from an on_turn_complete transition),
|
||||
// the agent has finished its previous turn and the PTY is waiting for input.
|
||||
effectivePrompt := s.buildWorkflowPrompt(ctx, taskDescription, step, taskID, sessionID, isPassthrough)
|
||||
@@ -1585,6 +1585,42 @@ func workflowMessageMetadata(planMode bool, origin workflowMessageOrigin, refere
|
||||
return meta
|
||||
}
|
||||
|
||||
// handleCreatedAutoStartLaunchFailure preserves a workflow prompt when the
|
||||
// created-session launch loses a concurrent-start race. The prompt was already
|
||||
// recorded in chat history, so queueing it is the only way to deliver it after
|
||||
// the session becomes ready. If queueing is not applicable or fails, restore
|
||||
// the hand-off message that was taken before the prompt was merged.
|
||||
func (s *Service) handleCreatedAutoStartLaunchFailure(
|
||||
ctx context.Context,
|
||||
taskID, sessionID, stepName, prompt string,
|
||||
launchErr error,
|
||||
planMode, shouldQueueIfBusy, userMessageRecorded bool,
|
||||
attachments []v1.MessageAttachment,
|
||||
origin workflowMessageOrigin,
|
||||
references []v1.EntityReference,
|
||||
takenMsg *messagequeue.QueuedMessage,
|
||||
) {
|
||||
promptQueued := false
|
||||
if shouldQueueIfBusy && (isAgentAlreadyRunningError(launchErr) ||
|
||||
isSessionBusyError(launchErr) || isTransientPromptError(launchErr)) {
|
||||
if queueErr := s.queueAutoStartPrompt(
|
||||
ctx, taskID, sessionID, prompt, planMode,
|
||||
attachments, origin, userMessageRecorded, references,
|
||||
); queueErr != nil {
|
||||
s.logger.Warn("failed to queue auto-start prompt after launch failure",
|
||||
zap.String("task_id", taskID),
|
||||
zap.String("session_id", sessionID),
|
||||
zap.String("step_name", stepName),
|
||||
zap.Error(queueErr))
|
||||
} else {
|
||||
promptQueued = true
|
||||
}
|
||||
}
|
||||
if !promptQueued && takenMsg != nil {
|
||||
s.requeueMessage(ctx, takenMsg, takenMsg.QueuedBy)
|
||||
}
|
||||
}
|
||||
|
||||
func (s *Service) autoStartStepPrompt(
|
||||
ctx context.Context,
|
||||
taskID string, session *models.TaskSession, step *wfmodels.WorkflowStep, prompt string,
|
||||
@@ -1686,7 +1722,11 @@ func (s *Service) autoStartStepPrompt(
|
||||
recordedPrompt, true, planMode, true, attachments, references,
|
||||
)
|
||||
if err != nil {
|
||||
requeueTaken()
|
||||
s.handleCreatedAutoStartLaunchFailure(
|
||||
ctx, taskID, sessionID, stepName, prompt, err,
|
||||
planMode, shouldQueueIfBusy, userMsgRecorded,
|
||||
attachments, origin, references, takenMsg,
|
||||
)
|
||||
}
|
||||
return err
|
||||
}
|
||||
@@ -2141,6 +2181,22 @@ func (s *Service) markIdleAfterReset(
|
||||
func (s *Service) resetAgentContext(ctx context.Context, taskID string, session *models.TaskSession, stepName string) bool {
|
||||
sessionID := session.ID
|
||||
|
||||
// A CREATED session has never been prompted, so there is no agent
|
||||
// conversation to clear. Its execution may still be workspace-only
|
||||
// (prepared, never started), in which case a "restart" here would *start*
|
||||
// the subprocess — and auto_start_agent, which reads State==CREATED as
|
||||
// "the agent was never started", would then start it a second time. The
|
||||
// second start is rejected by agentctl's running-process guard, failing
|
||||
// the task and dropping the step prompt. Leave the start to
|
||||
// autoStartStepPrompt: the process it launches begins on a fresh ACP
|
||||
// session regardless, so nothing is lost by skipping.
|
||||
if session.State == models.TaskSessionStateCreated {
|
||||
s.logger.Debug("session has no agent context to reset, skipping",
|
||||
zap.String("session_id", sessionID),
|
||||
zap.String("step_name", stepName))
|
||||
return true
|
||||
}
|
||||
|
||||
executionID, err := s.agentManager.GetExecutionIDForSession(ctx, sessionID)
|
||||
if err != nil || executionID == "" {
|
||||
s.logger.Debug("no agent execution for context reset, skipping",
|
||||
|
||||
@@ -1004,6 +1004,156 @@ func TestProcessOnEnterResetAgentContext(t *testing.T) {
|
||||
}
|
||||
})
|
||||
|
||||
// Regression: a CREATED session has never been prompted, so there is no
|
||||
// agent conversation to clear — but its prepared execution row made
|
||||
// resetAgentContext restart (in practice: start) the subprocess anyway.
|
||||
// markIdleAfterReset then left the state at CREATED, so autoStartStepPrompt
|
||||
// concluded the agent was never started and started it a second time. The
|
||||
// second start hit agentctl's "cannot configure while agent is running"
|
||||
// guard, which failed the task, force-killed the agent, and dropped the
|
||||
// step prompt. Reset must own no start for a never-prompted session.
|
||||
t.Run("reset_agent_context skipped for CREATED session", func(t *testing.T) {
|
||||
repo := setupTestRepo(t)
|
||||
seedSession(t, repo, "t1", "s1", "step1")
|
||||
|
||||
session, _ := repo.GetTaskSession(ctx, "s1")
|
||||
session.State = models.TaskSessionStateCreated
|
||||
session.AgentExecutionID = "exec-created"
|
||||
seedExecutorRunning(t, repo, session.ID, session.TaskID, "exec-created")
|
||||
_ = repo.UpdateTaskSession(ctx, session)
|
||||
|
||||
agentMgr := &mockAgentManager{repoForExecutionLookup: repo}
|
||||
svc := createTestServiceWithAgent(repo, newMockStepGetter(), newMockTaskRepo(), agentMgr)
|
||||
|
||||
step := &wfmodels.WorkflowStep{
|
||||
ID: "step2", WorkflowID: "wf1", Name: "Work",
|
||||
Events: wfmodels.StepEvents{
|
||||
OnEnter: []wfmodels.OnEnterAction{
|
||||
{Type: wfmodels.OnEnterResetAgentContext},
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
session, _ = repo.GetTaskSession(ctx, "s1")
|
||||
if !svc.resetAgentContext(ctx, "t1", session, step.Name) {
|
||||
t.Fatal("expected reset to report success for a never-prompted session")
|
||||
}
|
||||
|
||||
if len(agentMgr.restartProcessCalls) != 0 {
|
||||
t.Fatalf("expected 0 RestartAgentProcess calls for a CREATED session, got %d",
|
||||
len(agentMgr.restartProcessCalls))
|
||||
}
|
||||
|
||||
// A never-prompted session has no ACP session; the skip must not have
|
||||
// been reached by way of a clear that happened to leave it empty.
|
||||
updated, _ := repo.GetTaskSession(ctx, "s1")
|
||||
if updated.Metadata != nil {
|
||||
if acp, _ := updated.Metadata["acp_session_id"].(string); acp != "" {
|
||||
t.Errorf("expected acp_session_id to remain empty, got %q", acp)
|
||||
}
|
||||
}
|
||||
})
|
||||
|
||||
// Same skip for passthrough: the CLI has no conversation to reset before
|
||||
// its first prompt either.
|
||||
t.Run("reset_agent_context skipped for CREATED passthrough session", func(t *testing.T) {
|
||||
repo := setupTestRepo(t)
|
||||
seedSession(t, repo, "t1", "s1", "step1")
|
||||
|
||||
session, _ := repo.GetTaskSession(ctx, "s1")
|
||||
session.State = models.TaskSessionStateCreated
|
||||
session.AgentExecutionID = "exec-created-pt"
|
||||
seedExecutorRunning(t, repo, session.ID, session.TaskID, "exec-created-pt")
|
||||
_ = repo.UpdateTaskSession(ctx, session)
|
||||
|
||||
agentMgr := &mockAgentManager{repoForExecutionLookup: repo, isPassthrough: true}
|
||||
svc := createTestServiceWithAgent(repo, newMockStepGetter(), newMockTaskRepo(), agentMgr)
|
||||
|
||||
session, _ = repo.GetTaskSession(ctx, "s1")
|
||||
if !svc.resetAgentContext(ctx, "t1", session, "Work") {
|
||||
t.Fatal("expected reset to report success for a never-prompted passthrough session")
|
||||
}
|
||||
if len(agentMgr.restartProcessCalls) != 0 {
|
||||
t.Fatalf("expected 0 RestartAgentProcess calls for a CREATED passthrough session, got %d",
|
||||
len(agentMgr.restartProcessCalls))
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("CREATED passthrough session starts before workflow prompt", func(t *testing.T) {
|
||||
repo := setupTestRepo(t)
|
||||
seedSession(t, repo, "t1", "s1", "step1")
|
||||
|
||||
session, _ := repo.GetTaskSession(ctx, "s1")
|
||||
session.State = models.TaskSessionStateCreated
|
||||
session.IsPassthrough = true
|
||||
session.AgentProfileID = "profile-created-pt-start"
|
||||
session.AgentExecutionID = "exec-created-pt-start"
|
||||
seedExecutorRunning(t, repo, session.ID, session.TaskID, session.AgentExecutionID)
|
||||
if err := repo.UpdateTaskSession(ctx, session); err != nil {
|
||||
t.Fatalf("seed passthrough session: %v", err)
|
||||
}
|
||||
|
||||
taskRepo := newMockTaskRepo()
|
||||
taskRepo.tasks["t1"] = &v1.Task{
|
||||
ID: "t1", Title: "Created passthrough task", Description: "Run the work",
|
||||
State: v1.TaskStateInProgress,
|
||||
}
|
||||
agentMgr := &mockAgentManager{
|
||||
isPassthrough: true,
|
||||
isAgentRunning: true,
|
||||
launchAgentFunc: func(context.Context, *executor.LaunchAgentRequest) (*executor.LaunchAgentResponse, error) {
|
||||
return &executor.LaunchAgentResponse{
|
||||
AgentExecutionID: "exec-created-pt-start",
|
||||
Status: v1.AgentStatusStarting,
|
||||
WorkspacePath: t.TempDir(),
|
||||
}, nil
|
||||
},
|
||||
}
|
||||
startAgentProcessCalled := make(chan struct{}, 1)
|
||||
agentMgr.startAgentProcessFunc = func(_ context.Context, _ string) error {
|
||||
select {
|
||||
case startAgentProcessCalled <- struct{}{}:
|
||||
default:
|
||||
}
|
||||
started, err := repo.GetTaskSession(ctx, session.ID)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
started.State = models.TaskSessionStateWaitingForInput
|
||||
return repo.UpdateTaskSession(ctx, started)
|
||||
}
|
||||
stepGetter := newMockStepGetter()
|
||||
stepGetter.steps["step1"] = &wfmodels.WorkflowStep{ID: "step1", WorkflowID: "wf1", Name: "Current"}
|
||||
svc := createTestServiceWithScheduler(repo, stepGetter, taskRepo, agentMgr)
|
||||
|
||||
step := &wfmodels.WorkflowStep{
|
||||
ID: "step2", WorkflowID: "wf1", Name: "Work", Prompt: "Do the work",
|
||||
Events: wfmodels.StepEvents{OnEnter: []wfmodels.OnEnterAction{
|
||||
{Type: wfmodels.OnEnterResetAgentContext},
|
||||
{Type: wfmodels.OnEnterAutoStartAgent},
|
||||
}},
|
||||
}
|
||||
|
||||
session, _ = repo.GetTaskSession(ctx, "s1")
|
||||
svc.processOnEnter(ctx, "t1", session, step, "Run the work")
|
||||
select {
|
||||
case <-startAgentProcessCalled:
|
||||
case <-time.After(time.Second):
|
||||
t.Fatal("expected StartAgentProcess to be called")
|
||||
}
|
||||
|
||||
agentMgr.mu.Lock()
|
||||
startCalls := append([]string(nil), agentMgr.startAgentProcessCalls...)
|
||||
stdinCalls := append([]passthroughStdinCall(nil), agentMgr.passthroughStdinCalls...)
|
||||
agentMgr.mu.Unlock()
|
||||
if len(startCalls) != 1 || startCalls[0] != session.AgentExecutionID {
|
||||
t.Fatalf("expected one agent start for CREATED passthrough session, got %v", startCalls)
|
||||
}
|
||||
if len(stdinCalls) != 0 {
|
||||
t.Fatalf("expected no PTY write before passthrough process start, got %v", stdinCalls)
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("reset_agent_context works for passthrough sessions", func(t *testing.T) {
|
||||
repo := setupTestRepo(t)
|
||||
seedSession(t, repo, "t1", "s1", "step1")
|
||||
|
||||
@@ -0,0 +1,137 @@
|
||||
---
|
||||
spec: docs/specs/workflow-step-agent-start-ownership/spec.md
|
||||
created: 2026-08-05
|
||||
status: complete
|
||||
---
|
||||
|
||||
# Implementation Plan: Workflow Step Agent Start Ownership
|
||||
|
||||
## Overview
|
||||
|
||||
Two small, ordered changes in `internal/orchestrator/event_handlers_workflow.go`.
|
||||
Task 01 removes the double start at its source by making context reset a no-op
|
||||
for sessions that have never been prompted. Task 02 closes the prompt-loss hole
|
||||
the same failure exposed, by queueing the recorded auto-start prompt when the
|
||||
`CREATED` launch fails — reusing the queue-and-drain machinery that already
|
||||
exists for the sibling `PromptTask` branch.
|
||||
|
||||
Both tasks edit the same file, so they are sequential. No schema, contract, API,
|
||||
or persistence change.
|
||||
|
||||
## Confirmed root cause
|
||||
|
||||
On entering a step whose `on_enter` has both `reset_agent_context` and
|
||||
`auto_start_agent`:
|
||||
|
||||
1. `resetAgentContext` (`event_handlers_workflow.go:2137`) early-returns only
|
||||
when there is no agent execution ID. A `CREATED` session with a
|
||||
workspace-only execution therefore reaches `ResetAgentContext`, whose
|
||||
reset-unsupported fallback **starts** the subprocess and mints a fresh ACP
|
||||
session.
|
||||
2. `markIdleAfterReset` (`:2117`) flips only `RUNNING`/`STARTING`, so the
|
||||
session stays `CREATED`. Its comment asserts the opposite invariant.
|
||||
3. `autoStartStepPrompt` (`:1680`) reads `CREATED`, records the prompt, and
|
||||
calls `StartCreatedSession` → `StartAgentProcess` on the running execution.
|
||||
4. `Manager.Configure`
|
||||
(`internal/agentctl/server/process/manager.go:1388`) rejects with
|
||||
`cannot configure while agent is running`.
|
||||
5. `handleAgentProcessStartFailure`
|
||||
(`internal/orchestrator/executor/executor_execute.go`) marks the session
|
||||
`FAILED` and force-stops the agent. The `CREATED` branch's only recovery is
|
||||
`requeueTaken()`, which restores a taken handoff message and not the prompt
|
||||
recorded by `recordAutoStartMessage`, so the prompt is dropped.
|
||||
6. Recovery's `session/load` for that ACP session fails `-32002` (it died with
|
||||
the killed process); the fallback `session/new` boots an idle, unprompted
|
||||
agent and the session parks in `WAITING_FOR_INPUT`.
|
||||
|
||||
Evidence: backend logs from a local run, spanning ~13 seconds from the step
|
||||
move to the idle boot, plus the orphaned prompt row left in
|
||||
`task_session_messages` with no matching delivery.
|
||||
|
||||
Confirming detail: the execution was created workspace-only hours earlier and
|
||||
has no `StartAgentProcess` entry before the step move — the reset path is what
|
||||
started it.
|
||||
|
||||
## Backend
|
||||
|
||||
### Task 01 — reset is a no-op for never-prompted sessions
|
||||
|
||||
`resetAgentContext` gains a `session.State == CREATED` early-return, before the
|
||||
execution lookup, returning `true` (success, nothing to do). This restores the
|
||||
invariant `markIdleAfterReset` already documents and leaves `auto_start_agent`
|
||||
as the single starter.
|
||||
|
||||
Safety: a `CREATED` session has no agent conversation to clear, and the process
|
||||
auto-start launches begins on a fresh ACP session regardless. Nothing is lost by
|
||||
skipping.
|
||||
|
||||
Existing coverage is unaffected: `seedSession`
|
||||
(`event_handlers_test.go:637`) seeds sessions as `RUNNING`, so
|
||||
`TestProcessOnEnterResetAgentContext`'s restart assertions keep exercising the
|
||||
unchanged path. The regression test sets `CREATED` explicitly.
|
||||
|
||||
### Task 02 — a recorded auto-start prompt is never dropped
|
||||
|
||||
**Scope corrected during implementation.** The plan assumed
|
||||
`StartCreatedSession` surfaces the incident's failure synchronously. It does
|
||||
not: `startAgentOnExistingWorkspace` ends with `startAgentProcessAsync` and
|
||||
returns `nil`, and its log line "agent starting on existing workspace" is the
|
||||
one immediately preceding the async failure in the production trace. So the
|
||||
`CREATED` branch's error clause was never reached for the reported bug.
|
||||
|
||||
The synchronous branch is still reachable — and still drops the prompt — when
|
||||
there is no in-memory execution (the post-restart shape), where
|
||||
`startAgentOnExistingWorkspace` returns `ErrStaleExecution` and the full
|
||||
`LaunchAgent` path runs. Task 02 closes that case: a busy or already-running
|
||||
error queues the prompt via `queueAutoStartPrompt(..., userMsgRecorded, ...)`,
|
||||
mirroring the selectivity of the `PromptTask` retry loop below it. Permanent
|
||||
rejections are not queued. The queued prompt is drained by the existing
|
||||
`handleAgentBootReady` drain that
|
||||
`TestAutoStartTransientError_BootReadyDrainsOrphanedQueue` already pins, and
|
||||
the error is still returned so `FAILED` state and surfacing are unchanged.
|
||||
|
||||
The async path remains uncovered — see the spec's **Known gap**. Task 01
|
||||
removes the only known trigger for it on a first-turn launch, so the incident
|
||||
cannot recur through it.
|
||||
|
||||
## Frontend
|
||||
|
||||
No frontend changes. The symptom is entirely backend prompt dispatch; the UI
|
||||
renders whatever session state the backend reports.
|
||||
|
||||
## Waves
|
||||
|
||||
| Wave | Task | Parallel-safe |
|
||||
|---|---|---|
|
||||
| 1 | 01 — reset skips CREATED sessions | no (shared file) |
|
||||
| 2 | 02 — queue prompt on failed CREATED launch | no (shared file) |
|
||||
|
||||
`parallelism: sequential` on both. Task 02 depends on Task 01 only for edit
|
||||
ordering in the same file, not for behavior.
|
||||
|
||||
## Validation
|
||||
|
||||
```bash
|
||||
(cd apps/backend && go test ./internal/orchestrator/... -race)
|
||||
```
|
||||
|
||||
```bash
|
||||
(cd apps/backend && golangci-lint run ./... --new-from-rev=origin/main --timeout=5m)
|
||||
```
|
||||
|
||||
## Risks
|
||||
|
||||
- `resetAgentContext` is also reachable from the session-scoped
|
||||
`Service.ResetAgentContext` entry point (`session_scope_matrix_test.go:61`)
|
||||
covering a different, user-invoked reset. Task 01 changes only the
|
||||
workflow-step helper, not that public method — verify the distinction holds
|
||||
before editing.
|
||||
- Skipping the reset means a `CREATED` session no longer has `acp_session_id`
|
||||
cleared on step entry. That value is already empty for a never-prompted
|
||||
session; the regression test asserts it stays empty rather than assuming it.
|
||||
|
||||
## Out of scope
|
||||
|
||||
Carried from the spec: the agentctl `Configure` guard stays as-is, native ACP
|
||||
`session/reset` is unchanged, the `-32002` resume failure is a consequence not a
|
||||
defect, and task `FAILED` / recovery semantics are untouched.
|
||||
@@ -0,0 +1,72 @@
|
||||
---
|
||||
id: "01-reset-skips-created"
|
||||
title: "Context reset skips never-prompted sessions"
|
||||
status: done
|
||||
wave: 1
|
||||
parallelism: sequential
|
||||
depends_on: []
|
||||
plan: "plan.md"
|
||||
spec: "../../specs/workflow-step-agent-start-ownership/spec.md"
|
||||
---
|
||||
|
||||
# Task 01: Context reset skips never-prompted sessions
|
||||
|
||||
## Acceptance
|
||||
|
||||
- `resetAgentContext` returns success without restarting the subprocess when
|
||||
`session.State` is `CREATED`, so `auto_start_agent` remains the only path that
|
||||
starts the agent on that step entry.
|
||||
- The skip happens before the agent-execution lookup, so a `CREATED` session
|
||||
with a prepared workspace-only execution never reaches `ResetAgentContext`.
|
||||
- Sessions in every other state keep today's behavior: subprocess restarted and
|
||||
`acp_session_id` cleared.
|
||||
- Passthrough `CREATED` sessions are skipped on the same rule.
|
||||
- The session-scoped public `Service.ResetAgentContext` entry point is not
|
||||
changed by this task.
|
||||
|
||||
## Regression test
|
||||
|
||||
Add to `TestProcessOnEnterResetAgentContext` in
|
||||
`apps/backend/internal/orchestrator/event_handlers_workflow_triggers_test.go`:
|
||||
|
||||
- **Red first.** A subtest seeding a session with `State = CREATED` and a
|
||||
prepared execution, entering a step whose `on_enter` is
|
||||
`reset_agent_context` + `auto_start_agent`, asserting
|
||||
`len(agentMgr.restartProcessCalls) == 0`. Before the fix this reads 1.
|
||||
- Assert `acp_session_id` is still empty afterward (it is empty for a
|
||||
never-prompted session; the test must not assume the clear ran).
|
||||
- Keep the existing `RUNNING` subtests untouched — `seedSession`
|
||||
(`event_handlers_test.go:637`) seeds `RUNNING`, so they continue to pin the
|
||||
restart path.
|
||||
|
||||
## Verification
|
||||
|
||||
```bash
|
||||
(cd apps/backend && go test ./internal/orchestrator/ -race -run 'TestProcessOnEnterResetAgentContext|TestAutoStart')
|
||||
```
|
||||
|
||||
```bash
|
||||
(cd apps/backend && go test ./internal/orchestrator/... -race)
|
||||
```
|
||||
|
||||
## Files Likely Touched
|
||||
|
||||
- `apps/backend/internal/orchestrator/event_handlers_workflow.go`
|
||||
- `apps/backend/internal/orchestrator/event_handlers_workflow_triggers_test.go`
|
||||
|
||||
## Dependencies
|
||||
|
||||
None.
|
||||
|
||||
## Inputs
|
||||
|
||||
- Spec scenarios 1–4.
|
||||
- `resetAgentContext` at `event_handlers_workflow.go:2137` and its call site at
|
||||
`:1123`.
|
||||
- `markIdleAfterReset` at `:2117`, whose comment already documents the intended
|
||||
invariant.
|
||||
|
||||
## Output Contract
|
||||
|
||||
`resetAgentContext` is a no-op returning `true` for `CREATED` sessions; all
|
||||
other states unchanged. No signature change.
|
||||
+88
@@ -0,0 +1,88 @@
|
||||
---
|
||||
id: "02-queue-prompt-on-failed-launch"
|
||||
title: "Queue the auto-start prompt when a CREATED launch fails"
|
||||
status: done
|
||||
wave: 2
|
||||
parallelism: sequential
|
||||
depends_on: ["01-reset-skips-created"]
|
||||
plan: "plan.md"
|
||||
spec: "../../specs/workflow-step-agent-start-ownership/spec.md"
|
||||
---
|
||||
|
||||
# Task 02: Queue the auto-start prompt when a CREATED launch fails
|
||||
|
||||
> **Scope corrected during implementation.** The incident's failure is
|
||||
> *asynchronous* — `startAgentOnExistingWorkspace` calls
|
||||
> `startAgentProcessAsync` and returns `nil`, so `StartCreatedSession` returned
|
||||
> no error and this branch was never reached in production. The synchronous
|
||||
> branch is still reachable and still drops prompts, but only when there is no
|
||||
> in-memory execution (the post-restart shape) and the full `LaunchAgent` path
|
||||
> runs. This task closes that case; the async gap is recorded under the spec's
|
||||
> **Known gap** and needs its own cycle.
|
||||
|
||||
## Acceptance
|
||||
|
||||
- When `autoStartStepPrompt`'s `CREATED` branch gets a busy or already-running
|
||||
error from `StartCreatedSession`, the already-recorded prompt is queued for
|
||||
the session via `queueAutoStartPrompt`, passing `userMsgRecorded` so the
|
||||
drain does not duplicate the chat row.
|
||||
- Permanent rejections (Office scheduler guard, missing agent profile) are not
|
||||
queued — nothing would drain them.
|
||||
- The existing `requeueTaken()` restoration of a taken handoff message is
|
||||
preserved.
|
||||
- The original error is still returned, so session `FAILED` transition, error
|
||||
surfacing, and the caller's behavior are unchanged.
|
||||
- A failure to queue is logged and does not mask the original launch error.
|
||||
- The queued prompt is recoverable by the existing `handleAgentBootReady` drain
|
||||
— no new drain path is introduced.
|
||||
|
||||
## Regression test
|
||||
|
||||
Add to `apps/backend/internal/orchestrator/event_handlers_duplicate_autostart_test.go`,
|
||||
alongside `TestAutoStartTransientError_BootReadyDrainsOrphanedQueue`:
|
||||
|
||||
- **Red first.** A `CREATED` session whose `StartCreatedSession` fails; assert a
|
||||
queued message exists for the session afterward. Before the fix the queue is
|
||||
empty and the prompt is only a chat row.
|
||||
- Assert the returned error is the launch error, unchanged.
|
||||
- Extend through the existing boot-ready drain to show the prompt is delivered
|
||||
once the session becomes promptable, mirroring the sibling test's end-to-end
|
||||
shape.
|
||||
|
||||
## Verification
|
||||
|
||||
```bash
|
||||
(cd apps/backend && go test ./internal/orchestrator/ -race -run 'TestAutoStart|TestProcessOnEnterResetAgentContext')
|
||||
```
|
||||
|
||||
```bash
|
||||
(cd apps/backend && go test ./internal/orchestrator/... -race)
|
||||
```
|
||||
|
||||
```bash
|
||||
(cd apps/backend && golangci-lint run ./... --new-from-rev=origin/main --timeout=5m)
|
||||
```
|
||||
|
||||
## Files Likely Touched
|
||||
|
||||
- `apps/backend/internal/orchestrator/event_handlers_workflow.go`
|
||||
- `apps/backend/internal/orchestrator/event_handlers_duplicate_autostart_test.go`
|
||||
|
||||
## Dependencies
|
||||
|
||||
Task 01 (same file; edit ordering only).
|
||||
|
||||
## Inputs
|
||||
|
||||
- Spec scenario 5 and the matching failure mode.
|
||||
- The `CREATED` branch at `event_handlers_workflow.go:1680`.
|
||||
- The precedent in the same function's `PromptTask` retry loop: the
|
||||
`isAgentAlreadyRunningError` branch calls `queueAutoStartPrompt` with a
|
||||
comment describing this exact situation.
|
||||
- `handleAgentBootReady`'s drain, pinned by
|
||||
`TestAutoStartTransientError_BootReadyDrainsOrphanedQueue`.
|
||||
|
||||
## Output Contract
|
||||
|
||||
The `CREATED` branch queues before returning its error. No signature change, no
|
||||
new queue mechanism.
|
||||
@@ -0,0 +1,124 @@
|
||||
---
|
||||
status: draft
|
||||
created: 2026-08-05
|
||||
owner: Kandev
|
||||
---
|
||||
|
||||
# Workflow Step Agent Start Ownership
|
||||
|
||||
## Why
|
||||
|
||||
A workflow step whose `on_enter` combines `reset_agent_context` and
|
||||
`auto_start_agent` can have two independent paths start the same agent
|
||||
execution. The second start is rejected, the task is marked `FAILED`, the agent
|
||||
is force-killed, and the step prompt that was already written to chat history is
|
||||
never delivered. Recovery boots a fresh agent, so the user is left looking at a
|
||||
healthy "Resumed agent" banner above a session that will never do anything.
|
||||
|
||||
Observed in a local run, where the whole sequence — step move, double start,
|
||||
force-kill, idle reboot — completed in about 13 seconds.
|
||||
|
||||
## Broken behavior
|
||||
|
||||
`resetAgentContext` decides whether to restart purely on the presence of an
|
||||
agent execution ID. A `CREATED` session whose execution is workspace-only —
|
||||
prepared but never started — therefore has its subprocess *started* by the reset
|
||||
path, which reaches agentctl as a restart.
|
||||
|
||||
`markIdleAfterReset` only flips sessions in `RUNNING`/`STARTING`, so the session
|
||||
stays `CREATED`. Its own comment documents the opposite invariant: that
|
||||
`resetAgentContext` early-returns for `CREATED` "without restarting".
|
||||
|
||||
`autoStartStepPrompt` then reads `CREATED`, concludes the agent was never
|
||||
started, and calls `StartCreatedSession` → `StartAgentProcess` against the
|
||||
now-running execution. agentctl's `Manager.Configure` rejects it with
|
||||
`cannot configure while agent is running`; the executor marks the session
|
||||
`FAILED` and force-stops the agent.
|
||||
|
||||
The prompt recorded by `recordAutoStartMessage` is not requeued on that failure
|
||||
— only a taken handoff message is — so it is lost. Recovery's `session/load`
|
||||
then fails with `-32002 Resource not found` because the ACP session died with
|
||||
the killed process, and the fallback `session/new` boots an agent that is idle
|
||||
and unprompted.
|
||||
|
||||
## What
|
||||
|
||||
- On a step entry, exactly one path starts the agent subprocess for a session.
|
||||
- Context reset performs no start or restart for a session that has never been
|
||||
prompted (`CREATED`). Such a session has no agent conversation to clear, and
|
||||
the process the auto-start path launches already begins on a new ACP session.
|
||||
- Context reset for a session that has been prompted continues to restart the
|
||||
subprocess and clear `acp_session_id` exactly as it does today.
|
||||
- When a `CREATED` launch fails *synchronously* with a busy or
|
||||
already-running condition, the recorded prompt is queued for the session so
|
||||
the existing boot-ready drain delivers it once the session is promptable
|
||||
again. Permanent rejections (Office scheduler guard, missing agent profile)
|
||||
are not queued — nothing would ever drain them.
|
||||
- A failed start still marks the session `FAILED` and surfaces the error.
|
||||
|
||||
## Failure modes
|
||||
|
||||
- Reset is skipped for a `CREATED` session and auto-start then fails for an
|
||||
unrelated reason: the session is `FAILED` with the error surfaced, and the
|
||||
prompt is queued rather than orphaned.
|
||||
- The step has `reset_agent_context` but no `auto_start_agent` and the session
|
||||
is `CREATED`: no restart occurs and the session remains promptable. No agent
|
||||
context is lost, because none exists.
|
||||
- The session is passthrough and `CREATED`: the same skip applies; the CLI has
|
||||
no conversation to reset before its first prompt.
|
||||
- Execution lookup fails or returns an empty ID: unchanged — reset is skipped,
|
||||
as today.
|
||||
- Queueing the prompt after a failed start itself fails: the failure is logged
|
||||
and the session's `FAILED` state is unchanged.
|
||||
|
||||
## Scenarios
|
||||
|
||||
- **GIVEN** a session in `CREATED` with a prepared, not-yet-started agent
|
||||
execution, **WHEN** the task enters a step whose `on_enter` has both
|
||||
`reset_agent_context` and `auto_start_agent`, **THEN** the reset performs no
|
||||
subprocess restart, auto-start performs the single start, and the step prompt
|
||||
reaches the agent.
|
||||
- **GIVEN** a session in `CREATED` with a prepared execution, **WHEN** the task
|
||||
enters a step with `reset_agent_context` and no `auto_start_agent`, **THEN**
|
||||
no restart occurs and the session remains promptable.
|
||||
- **GIVEN** a session in `RUNNING` with a started agent execution, **WHEN** the
|
||||
task enters a step with `reset_agent_context`, **THEN** the subprocess is
|
||||
restarted and `acp_session_id` is cleared, unchanged from today.
|
||||
- **GIVEN** a passthrough session in `CREATED`, **WHEN** the task enters a step
|
||||
with `reset_agent_context`, **THEN** no restart occurs.
|
||||
- **GIVEN** a `CREATED` session with no in-memory execution (the shape after a
|
||||
backend restart), **WHEN** its auto-start launch fails synchronously because
|
||||
a concurrent path already owns the start, **THEN** the recorded prompt is
|
||||
queued for the session rather than dropped, and the session's later
|
||||
promptable transition delivers it.
|
||||
- **GIVEN** a `CREATED` launch rejected permanently (Office scheduler guard),
|
||||
**WHEN** the failure is handled, **THEN** the error is returned and no
|
||||
message is queued.
|
||||
|
||||
## Known gap
|
||||
|
||||
Prompt preservation across an **asynchronous** start failure is not covered
|
||||
here. `startAgentOnExistingWorkspace` ends with `startAgentProcessAsync` and
|
||||
returns `nil`, so a session with a prepared execution — the incident's own shape
|
||||
— never surfaces a synchronous error for the clause above to catch. That failure
|
||||
is handled entirely by `handleAgentProcessStartFailure` →
|
||||
`Service.handleAgentStartFailed`, which has no access to the prompt the launch
|
||||
was carrying and therefore cannot re-queue it.
|
||||
|
||||
Task 01 removes the only known trigger for that path on a first-turn launch, so
|
||||
the incident cannot recur through it. The residual gap — any *other* async start
|
||||
failure on a first-turn launch drops the step prompt — is a distinct pre-existing
|
||||
defect. Closing it needs a pending-prompt handle the async failure path can
|
||||
read, which is a design change beyond this repair.
|
||||
|
||||
## Out of scope
|
||||
|
||||
- Relaxing agentctl's `Manager.Configure` running-process guard. Reconfiguring
|
||||
command or environment under a live subprocess is unsafe; the guard is
|
||||
correct and stays as-is.
|
||||
- Native ACP `session/reset` support in the adapter. The
|
||||
reset-unsupported-fallback-to-restart path is unchanged.
|
||||
- The `-32002 Resource not found` resume failure, which is a consequence of the
|
||||
force-kill rather than an independent defect.
|
||||
- Changing task `FAILED` semantics, the reconciliation path, or the
|
||||
lazy-recovery boot that follows a backend restart.
|
||||
Reference in New Issue
Block a user