fix(orchestrator): repair worktree state and prompt persistence on resume (#1121)

This commit is contained in:
Carlos Florêncio
2026-05-28 21:48:12 +01:00
committed by GitHub
parent 616bafa848
commit 21db57bdfd
11 changed files with 690 additions and 85 deletions
@@ -2049,12 +2049,9 @@ func (a *Adapter) handlePermissionRequest(ctx context.Context, req *PermissionRe
}
// Only emit a synthetic tool_call event if no ToolCall notification preceded this.
// When a ToolCall notification exists (tracked in activeToolCalls), the message
// was already created by convertToolCallUpdate → handleToolCallEvent.
// Emitting a second tool_call for the same ID creates duplicate messages in the UI.
a.mu.RLock()
_, alreadyTracked := a.activeToolCalls[req.ToolCallID]
a.mu.RUnlock()
// waitForActiveToolCall bounds the race window between a SessionUpdate.ToolCall
// notification and a same-id request_permission dispatched on separate goroutines.
alreadyTracked := a.waitForActiveToolCall(ctx, req.ToolCallID, syntheticToolCallRaceWindow)
if !alreadyTracked {
toolCallEvent := AgentEvent{
@@ -0,0 +1,41 @@
package acp
import (
"context"
"time"
)
// syntheticToolCallRaceWindow bounds how long handlePermissionRequest waits
// for a concurrently-dispatched ToolCall notification to populate
// activeToolCalls before falling back to emitting a synthetic tool_call. Sized
// to cover the goroutine-scheduling gap between the SDK delivering a
// SessionUpdate.ToolCall and a request_permission for the same toolCallID,
// without adding noticeable latency to permission prompts.
const syntheticToolCallRaceWindow = 100 * time.Millisecond
// waitForActiveToolCall polls activeToolCalls for the given id, sleeping in
// small increments, and returns true if the entry appears within timeout.
// Returns false when the wait expires without the entry materializing — the
// expected outcome when the agent really did skip the tool_call notification.
// Honors ctx so a cancelled session aborts the poll loop immediately rather
// than blocking the handler for the full timeout window.
func (a *Adapter) waitForActiveToolCall(ctx context.Context, toolCallID string, timeout time.Duration) bool {
const pollInterval = 10 * time.Millisecond
deadline := time.Now().Add(timeout)
for {
a.mu.RLock()
_, tracked := a.activeToolCalls[toolCallID]
a.mu.RUnlock()
if tracked {
return true
}
if time.Now().After(deadline) {
return false
}
select {
case <-ctx.Done():
return false
case <-time.After(pollInterval):
}
}
}
@@ -0,0 +1,124 @@
package acp
import (
"context"
"testing"
"time"
"github.com/kandev/kandev/internal/agentctl/types/streams"
)
// TestWaitForActiveToolCall_ReturnsOnContextCancel ensures a cancelled session
// breaks out of the poll loop immediately instead of blocking the handler
// until the timeout window expires.
func TestWaitForActiveToolCall_ReturnsOnContextCancel(t *testing.T) {
a := newTestAdapter()
ctx, cancel := context.WithCancel(t.Context())
go func() {
time.Sleep(5 * time.Millisecond)
cancel()
}()
start := time.Now()
tracked := a.waitForActiveToolCall(ctx, "tool-1", time.Second)
elapsed := time.Since(start)
if tracked {
t.Fatalf("expected waitForActiveToolCall to return false on cancellation, got true")
}
if elapsed >= 200*time.Millisecond {
t.Errorf("waitForActiveToolCall did not honor cancel: elapsed=%v", elapsed)
}
}
// TestHandlePermissionRequest_NoDuplicateWhenToolCallNotificationLanded verifies
// the activeToolCalls check suppresses the synthetic tool_call emit when a
// SessionUpdate.ToolCall has already populated the map. Pre-existing guard;
// covered here so the race fix below has a baseline counterpart.
func TestHandlePermissionRequest_NoDuplicateWhenToolCallNotificationLanded(t *testing.T) {
a := newTestAdapter()
a.activeToolCalls["tool-1"] = &streams.NormalizedPayload{}
_, _ = a.handlePermissionRequest(t.Context(), &PermissionRequest{
SessionID: "sess-1",
ToolCallID: "tool-1",
Title: "Kandev: List Workspaces",
ActionType: "other",
})
for _, ev := range drainEvents(a) {
if ev.Type == streams.EventTypeToolCall {
t.Fatalf("did not expect synthetic tool_call when notification already tracked, got %#v", ev)
}
}
}
// TestHandlePermissionRequest_WaitsForRacingToolCallNotification covers the
// race window: the SDK delivers a ToolCall SessionUpdate and a
// request_permission for the same toolCallID on separate goroutines, and the
// permission handler may run before the notification handler has populated
// activeToolCalls. Without the bounded wait, the handler would emit a
// synthetic tool_call that becomes a duplicate message in the chat.
func TestHandlePermissionRequest_WaitsForRacingToolCallNotification(t *testing.T) {
a := newTestAdapter()
go func() {
time.Sleep(20 * time.Millisecond)
a.mu.Lock()
a.activeToolCalls["tool-1"] = &streams.NormalizedPayload{}
a.mu.Unlock()
}()
_, _ = a.handlePermissionRequest(t.Context(), &PermissionRequest{
SessionID: "sess-1",
ToolCallID: "tool-1",
Title: "Kandev: List Workspaces",
ActionType: "other",
})
for _, ev := range drainEvents(a) {
if ev.Type == streams.EventTypeToolCall {
t.Fatalf("waited for racing tool_call notification but still emitted synthetic %#v", ev)
}
}
}
// TestHandlePermissionRequest_EmitsSyntheticAfterTimeout exercises the
// fallback path: when no ToolCall notification ever arrives, the bounded wait
// must give up and emit the synthetic tool_call so the UI still gets a row to
// attach the approval to. Agents that skip the ToolCall notification and go
// straight to request_permission rely on this path.
func TestHandlePermissionRequest_EmitsSyntheticAfterTimeout(t *testing.T) {
a := newTestAdapter()
start := time.Now()
_, _ = a.handlePermissionRequest(t.Context(), &PermissionRequest{
SessionID: "sess-1",
ToolCallID: "tool-1",
Title: "Kandev: List Workspaces",
ActionType: "other",
})
elapsed := time.Since(start)
if elapsed < syntheticToolCallRaceWindow {
t.Errorf("handler returned before the race window expired: elapsed=%v window=%v", elapsed, syntheticToolCallRaceWindow)
}
var synthetic *AgentEvent
for _, ev := range drainEvents(a) {
if ev.Type == streams.EventTypeToolCall && ev.ToolCallID == "tool-1" {
ev := ev
synthetic = &ev
}
}
if synthetic == nil {
t.Fatalf("expected synthetic tool_call after timeout, got none")
}
if synthetic.ToolStatus != "pending_permission" {
t.Errorf("synthetic ToolStatus = %q, want %q", synthetic.ToolStatus, "pending_permission")
}
if synthetic.ToolTitle != "Kandev: List Workspaces" {
t.Errorf("synthetic ToolTitle = %q, want %q", synthetic.ToolTitle, "Kandev: List Workspaces")
}
}
@@ -269,7 +269,7 @@ func TestBuildResumeRequest_ReusesTaskEnvironmentRuntimeMetadata(t *testing.T) {
},
}
req, _, _, running, err := exec.buildResumeRequest(context.Background(), task, session, true)
req, _, _, _, running, err := exec.buildResumeRequest(context.Background(), task, session, true)
if err != nil {
t.Fatalf("buildResumeRequest returned error: %v", err)
}
@@ -1071,6 +1071,12 @@ func (e *Executor) persistTaskEnvironment(
if sandboxID := extractSandboxID(resp.Metadata); sandboxID != "" {
existingEnv.SandboxID = sandboxID
}
// Refresh TaskDirName when the request carries a new value — covers
// resume-after-failure where the original env row was stamped with an
// empty task_dir_name and the resume regenerates it.
if req.TaskDirName != "" {
existingEnv.TaskDirName = req.TaskDirName
}
if err := e.repo.UpdateTaskEnvironment(ctx, existingEnv); err != nil {
e.logger.Warn("failed to update task environment",
zap.String("task_id", taskID),
@@ -9,6 +9,7 @@ import (
"github.com/kandev/kandev/internal/agent/runtime/lifecycle"
"github.com/kandev/kandev/internal/common/gitref"
"github.com/kandev/kandev/internal/task/models"
"github.com/kandev/kandev/internal/worktree"
v1 "github.com/kandev/kandev/pkg/api/v1"
"go.uber.org/zap"
)
@@ -325,7 +326,7 @@ func (e *Executor) ResumeSession(ctx context.Context, session *models.TaskSessio
}
defer unlock()
req, repositoryID, _, _, err := e.buildResumeRequest(ctx, task, session, startAgent)
req, repositoryID, execCfg, existingEnv, _, err := e.buildResumeRequest(ctx, task, session, startAgent)
if err != nil {
return nil, err
}
@@ -387,6 +388,16 @@ func (e *Executor) ResumeSession(ctx context.Context, session *models.TaskSessio
e.persistResumeState(ctx, task.ID, session, startAgent)
e.persistWorktreeAssociation(ctx, task.ID, session, repositoryID, resp)
// Refresh task_environments after a successful resume so the row reflects
// the new worktree, container, and ready status. Without this, sessions
// that failed on initial launch (empty task_dir_name / worktree_path,
// status=stopped) stay stuck on those stale values even though the resume
// just prepared a fresh worktree — the frontend keeps showing
// "Executor environment is unavailable (stopped)" and disables chat input.
// existingEnv is captured from buildResumeRequest above (resolveResumeTaskEnvironment
// already aborts on real DB errors); when nil, persistTaskEnvironment will
// re-fetch under the per-task lock and create a fresh row if needed.
e.persistTaskEnvironment(ctx, task.ID, session, existingEnv, req, resp, execCfg)
worktreePath := resp.WorktreePath
worktreeBranch := resp.WorktreeBranch
@@ -493,7 +504,7 @@ func (e *Executor) validateAndLockResume(ctx context.Context, session *models.Ta
// buildResumeRequest constructs the LaunchAgentRequest for a session resume, resolving executor config,
// repository details, worktree settings, and ACP resume token.
// Returns the request, repository ID, executor config, existing ExecutorRunning record (may be nil), and error.
func (e *Executor) buildResumeRequest(ctx context.Context, task *v1.Task, session *models.TaskSession, startAgent bool) (*LaunchAgentRequest, string, executorConfig, *models.ExecutorRunning, error) {
func (e *Executor) buildResumeRequest(ctx context.Context, task *v1.Task, session *models.TaskSession, startAgent bool) (*LaunchAgentRequest, string, executorConfig, *models.TaskEnvironment, *models.ExecutorRunning, error) {
req := &LaunchAgentRequest{
TaskID: task.ID,
SessionID: session.ID,
@@ -523,15 +534,15 @@ func (e *Executor) buildResumeRequest(ctx context.Context, task *v1.Task, sessio
existingEnv, err := e.resolveResumeTaskEnvironment(ctx, task.ID, session)
if err != nil {
return nil, "", execConfig, nil, err
return nil, "", execConfig, nil, nil, err
}
if session.TaskEnvironmentID != "" {
req.TaskEnvironmentID = session.TaskEnvironmentID
}
repositoryID, err := e.applyResumeRepoConfig(ctx, task, session, req)
repositoryID, err := e.applyResumeRepoConfig(ctx, task, session, req, existingEnv)
if err != nil {
return nil, "", execConfig, nil, err
return nil, "", execConfig, nil, nil, err
}
e.reuseExistingEnvironment(ctx, req, existingEnv)
@@ -543,7 +554,7 @@ func (e *Executor) buildResumeRequest(ctx context.Context, task *v1.Task, sessio
existingRunning := e.applyRunningRecordToResumeRequest(ctx, req, task, session, startAgent)
return req, repositoryID, execConfig, existingRunning, nil
return req, repositoryID, execConfig, existingEnv, existingRunning, nil
}
func (e *Executor) resolveResumeTaskEnvironment(ctx context.Context, taskID string, session *models.TaskSession) (*models.TaskEnvironment, error) {
@@ -642,21 +653,16 @@ func (e *Executor) applyRunningRecordToResumeRequest(ctx context.Context, req *L
}
// applyResumeRepoConfig resolves repository details and applies them to req.
// Returns the resolved repositoryID.
func (e *Executor) applyResumeRepoConfig(ctx context.Context, task *v1.Task, session *models.TaskSession, req *LaunchAgentRequest) (string, error) {
repositoryID := session.RepositoryID
if repositoryID == "" && len(task.Repositories) > 0 {
repositoryID = task.Repositories[0].RepositoryID
}
baseBranch := session.BaseBranch
if baseBranch == "" && len(task.Repositories) > 0 && task.Repositories[0].BaseBranch != "" {
baseBranch = task.Repositories[0].BaseBranch
}
// Returns the resolved repositoryID. existingEnv is the task_environments row
// resolved upstream in buildResumeRequest; it is passed in so we can derive
// TaskDirName without an extra DB round-trip and so both the single-repo and
// multi-repo branches agree on the same fallback name (the fallback uses a
// random suffix, so two independent calls would produce different names).
func (e *Executor) applyResumeRepoConfig(ctx context.Context, task *v1.Task, session *models.TaskSession, req *LaunchAgentRequest, existingEnv *models.TaskEnvironment) (string, error) {
repositoryID, baseBranch := resolveResumeRepoIDAndBranch(task, session)
if baseBranch != "" {
req.Branch = baseBranch
}
if repositoryID == "" {
return "", nil
}
@@ -671,6 +677,43 @@ func (e *Executor) applyResumeRepoConfig(ctx context.Context, task *v1.Task, ses
}
repositoryPath := repository.LocalPath
applyResumeRepoBasics(req, repository, repositoryPath)
if err := e.applyResumeCloneURL(req, repository, baseBranch); err != nil {
return "", err
}
if shouldUseWorktree(req.ExecutorType) && repositoryPath != "" {
e.applyResumeWorktreeConfig(ctx, task, req, repository, repositoryID, repositoryPath, baseBranch, existingEnv)
}
if err := e.applyResumeMultiRepoConfig(ctx, task, req, existingEnv); err != nil {
return repositoryID, err
}
return repositoryID, nil
}
// resolveResumeRepoIDAndBranch picks the primary repositoryID and baseBranch
// for a resume, preferring the session's persisted values and falling back to
// the task's primary repository when the session row was created before those
// fields existed.
func resolveResumeRepoIDAndBranch(task *v1.Task, session *models.TaskSession) (string, string) {
repositoryID := session.RepositoryID
if repositoryID == "" && len(task.Repositories) > 0 {
repositoryID = task.Repositories[0].RepositoryID
}
baseBranch := session.BaseBranch
if baseBranch == "" && len(task.Repositories) > 0 && task.Repositories[0].BaseBranch != "" {
baseBranch = task.Repositories[0].BaseBranch
}
return repositoryID, baseBranch
}
// applyResumeRepoBasics copies the repository's local path and setup script
// onto the request. Pulled out of applyResumeRepoConfig so the parent's
// cyclomatic complexity stays inside the lint budget.
func applyResumeRepoBasics(req *LaunchAgentRequest, repository *models.Repository, repositoryPath string) {
if repositoryPath != "" {
req.RepositoryURL = repositoryPath
}
@@ -680,68 +723,100 @@ func (e *Executor) applyResumeRepoConfig(ctx context.Context, task *v1.Task, ses
}
req.Metadata[lifecycle.MetadataKeyRepoSetupScript] = repository.SetupScript
}
}
if e.capabilities != nil && e.capabilities.RequiresCloneURL(req.ExecutorType) {
cloneURL := repositoryCloneURL(repository)
if cloneURL == "" {
return "", ErrNoCloneURL
}
req.RepositoryURL = cloneURL
if req.Metadata == nil {
req.Metadata = make(map[string]interface{})
}
req.Metadata["repository_clone_url"] = cloneURL
// Clone-based remote executors (Sprites, Docker, …) need BaseBranch
// in launch metadata so the prepare script's `git clone --branch <X>`
// resolves. The local executor must NOT receive BaseBranch on resume:
// LocalPreparer would force a `git fetch && git checkout` against
// the user's actual workspace and clobber the "use current state" UX.
// Worktree executors set their own BaseBranch in the block below.
if baseBranch != "" {
req.BaseBranch = baseBranch
}
// applyResumeCloneURL handles clone-based remote executors: it stamps the
// clone URL on the request and propagates BaseBranch into launch metadata so
// the prepare script's `git clone --branch <X>` resolves. Local executors skip
// this path so LocalPreparer doesn't clobber the "use current state" UX.
func (e *Executor) applyResumeCloneURL(req *LaunchAgentRequest, repository *models.Repository, baseBranch string) error {
if e.capabilities == nil || !e.capabilities.RequiresCloneURL(req.ExecutorType) {
return nil
}
if shouldUseWorktree(req.ExecutorType) && repositoryPath != "" {
req.UseWorktree = true
req.RepositoryPath = repositoryPath
req.RepositoryID = repositoryID
if baseBranch != "" {
req.BaseBranch = baseBranch
} else {
req.BaseBranch = defaultBaseBranch
}
// Carry forward CheckoutBranch from task repository (e.g. PR head branch)
primaryTaskRepo, _ := e.repo.GetPrimaryTaskRepository(ctx, task.ID)
if primaryTaskRepo != nil && primaryTaskRepo.CheckoutBranch != "" {
req.CheckoutBranch = primaryTaskRepo.CheckoutBranch
req.PRNumber = prNumberFromMetadata(primaryTaskRepo.Metadata)
}
req.WorktreeBranchPrefix = repository.WorktreeBranchPrefix
req.PullBeforeWorktree = repository.PullBeforeWorktree
cloneURL := repositoryCloneURL(repository)
if cloneURL == "" {
return ErrNoCloneURL
}
// Multi-repo: when the task has more than one repository, populate
// req.Repositories so the lifecycle preparer can resume/recreate each
// repo's worktree. The legacy top-level fields above stay populated
// from the primary for backwards compat.
if len(task.Repositories) > 1 {
allRepos, allErr := e.resolveAllRepoInfo(ctx, task.ID)
if allErr != nil {
return repositoryID, allErr
}
if len(allRepos) > 1 {
req.Repositories = buildRepoSpecs(allRepos)
// Stamp the per-repo TaskDirName so the preparer reuses the same
// task root that the original launch created.
if env, _ := e.repo.GetTaskEnvironmentByTaskID(ctx, task.ID); env != nil && env.TaskDirName != "" {
req.TaskDirName = env.TaskDirName
}
}
req.RepositoryURL = cloneURL
if req.Metadata == nil {
req.Metadata = make(map[string]interface{})
}
req.Metadata["repository_clone_url"] = cloneURL
if baseBranch != "" {
req.BaseBranch = baseBranch
}
return nil
}
return repositoryID, nil
// applyResumeMultiRepoConfig populates req.Repositories for multi-repo tasks
// so the lifecycle preparer can resume/recreate each repo's worktree. The
// legacy top-level fields stay populated from the primary for backwards
// compat.
func (e *Executor) applyResumeMultiRepoConfig(ctx context.Context, task *v1.Task, req *LaunchAgentRequest, existingEnv *models.TaskEnvironment) error {
if len(task.Repositories) <= 1 {
return nil
}
allRepos, err := e.resolveAllRepoInfo(ctx, task.ID)
if err != nil {
return err
}
if len(allRepos) > 1 {
req.Repositories = buildRepoSpecs(allRepos)
req.TaskDirName = resolveResumeTaskDirName(existingEnv, task)
}
return nil
}
// applyResumeWorktreeConfig stamps the worktree-related fields on req for a
// single-repo worktree resume. Extracted from applyResumeRepoConfig to keep
// cognitive complexity within the lint budget.
func (e *Executor) applyResumeWorktreeConfig(
ctx context.Context,
task *v1.Task,
req *LaunchAgentRequest,
repository *models.Repository,
repositoryID, repositoryPath, baseBranch string,
existingEnv *models.TaskEnvironment,
) {
req.UseWorktree = true
req.RepositoryPath = repositoryPath
req.RepositoryID = repositoryID
if baseBranch != "" {
req.BaseBranch = baseBranch
} else {
req.BaseBranch = defaultBaseBranch
}
// Carry forward CheckoutBranch from task repository (e.g. PR head branch)
primaryTaskRepo, _ := e.repo.GetPrimaryTaskRepository(ctx, task.ID)
if primaryTaskRepo != nil && primaryTaskRepo.CheckoutBranch != "" {
req.CheckoutBranch = primaryTaskRepo.CheckoutBranch
req.PRNumber = prNumberFromMetadata(primaryTaskRepo.Metadata)
}
req.WorktreeBranchPrefix = repository.WorktreeBranchPrefix
req.PullBeforeWorktree = repository.PullBeforeWorktree
// Worktree manager requires TaskDirName and RepoName. Mirror the
// initial-launch path (applyRepositoryConfig) so resumes of single-repo
// tasks don't fail with ErrTaskDirRequired. Prefer a persisted
// TaskDirName so we reuse the same on-disk task root; fall back to a
// freshly generated one when the original launch failed before the
// environment was stamped.
if repository.Name != "" {
req.RepoName = repository.Name
}
req.TaskDirName = resolveResumeTaskDirName(existingEnv, task)
}
// resolveResumeTaskDirName returns the per-task directory name to use when
// resuming. It prefers the value persisted on task_environments (so we reuse
// the same ~/.kandev/tasks/{name}/ root the initial launch created) and falls
// back to a fresh semantic name when the original launch failed before any
// environment was stamped. That fallback is what lets a previously failed
// session recover instead of looping on ErrTaskDirRequired.
func resolveResumeTaskDirName(existingEnv *models.TaskEnvironment, task *v1.Task) string {
if existingEnv != nil && existingEnv.TaskDirName != "" {
return existingEnv.TaskDirName
}
return worktree.SemanticWorktreeName(task.Title, worktree.SmallSuffix(3))
}
// persistResumeState updates the session row after a successful resume launch.
@@ -4,6 +4,7 @@ import (
"context"
"errors"
"fmt"
"strings"
"testing"
"time"
@@ -496,7 +497,7 @@ func TestApplyResumeRepoConfig_BaseBranchByExecutorType(t *testing.T) {
ExecutorType: tc.executorType,
}
if _, err := exec.applyResumeRepoConfig(context.Background(), task, session, req); err != nil {
if _, err := exec.applyResumeRepoConfig(context.Background(), task, session, req, nil); err != nil {
t.Fatalf("applyResumeRepoConfig: %v", err)
}
@@ -510,6 +511,200 @@ func TestApplyResumeRepoConfig_BaseBranchByExecutorType(t *testing.T) {
}
}
// TestApplyResumeRepoConfig_WorktreeStampsTaskDir locks in the fix for
// resumes of single-repo worktree tasks: the lifecycle preparer hands the
// request to worktree.Manager.Create, which rejects requests missing
// TaskDirName or RepoName with ErrTaskDirRequired. The initial-launch path
// (applyRepositoryConfig) sets both; the resume path must do the same, and
// must also be able to regenerate TaskDirName when the original launch
// failed before any task_environments row was stamped.
func TestApplyResumeRepoConfig_WorktreeStampsTaskDir(t *testing.T) {
t.Run("reuses persisted TaskDirName when present", func(t *testing.T) {
repo := newMockRepository()
repo.repositories["repo-1"] = &models.Repository{
ID: "repo-1",
Name: "my-repo",
LocalPath: "/tmp/repo",
}
existingEnv := &models.TaskEnvironment{
ID: "env-1",
TaskID: "task-1",
TaskDirName: "previously-stamped_abc",
}
repo.taskEnvironments["env-1"] = existingEnv
repo.tasks["task-1"] = &models.Task{ID: "task-1"}
task := &v1.Task{ID: "task-1", Title: "Fix login bug"}
session := &models.TaskSession{
ID: "sess-1",
TaskID: "task-1",
RepositoryID: "repo-1",
BaseBranch: "main",
}
exec := newTestExecutor(t, &mockAgentManager{}, repo)
req := &LaunchAgentRequest{TaskID: "task-1", SessionID: "sess-1", ExecutorType: "worktree"}
if _, err := exec.applyResumeRepoConfig(context.Background(), task, session, req, existingEnv); err != nil {
t.Fatalf("applyResumeRepoConfig: %v", err)
}
if req.TaskDirName != "previously-stamped_abc" {
t.Errorf("TaskDirName = %q, want %q", req.TaskDirName, "previously-stamped_abc")
}
if req.RepoName != "my-repo" {
t.Errorf("RepoName = %q, want %q", req.RepoName, "my-repo")
}
})
t.Run("regenerates TaskDirName when persisted value is empty", func(t *testing.T) {
// This is the failure mode from the bug report: the initial launch
// failed before stamping task_dir_name, so the env row exists with
// an empty TaskDirName. The resume must still be able to proceed.
repo := newMockRepository()
repo.repositories["repo-1"] = &models.Repository{
ID: "repo-1",
Name: "my-repo",
LocalPath: "/tmp/repo",
}
repo.tasks["task-1"] = &models.Task{ID: "task-1"}
task := &v1.Task{ID: "task-1", Title: "Fix login bug"}
session := &models.TaskSession{
ID: "sess-1",
TaskID: "task-1",
RepositoryID: "repo-1",
BaseBranch: "main",
}
exec := newTestExecutor(t, &mockAgentManager{}, repo)
req := &LaunchAgentRequest{TaskID: "task-1", SessionID: "sess-1", ExecutorType: "worktree"}
existingEnv := &models.TaskEnvironment{
ID: "env-1",
TaskID: "task-1",
TaskDirName: "",
}
if _, err := exec.applyResumeRepoConfig(context.Background(), task, session, req, existingEnv); err != nil {
t.Fatalf("applyResumeRepoConfig: %v", err)
}
if req.TaskDirName == "" {
t.Error("TaskDirName must not be empty; worktree.Manager.Create would reject the request")
}
// Semantic name format: {sanitized-title}_{suffix}, suffix is 3 chars.
if !strings.HasPrefix(req.TaskDirName, "fix-login-bug_") {
t.Errorf("TaskDirName = %q, want prefix %q", req.TaskDirName, "fix-login-bug_")
}
if req.RepoName != "my-repo" {
t.Errorf("RepoName = %q, want %q", req.RepoName, "my-repo")
}
})
t.Run("regenerates TaskDirName when no env row exists", func(t *testing.T) {
repo := newMockRepository()
repo.repositories["repo-1"] = &models.Repository{
ID: "repo-1",
Name: "my-repo",
LocalPath: "/tmp/repo",
}
repo.tasks["task-1"] = &models.Task{ID: "task-1"}
task := &v1.Task{ID: "task-1", Title: "Fix login bug"}
session := &models.TaskSession{
ID: "sess-1",
TaskID: "task-1",
RepositoryID: "repo-1",
BaseBranch: "main",
}
exec := newTestExecutor(t, &mockAgentManager{}, repo)
req := &LaunchAgentRequest{TaskID: "task-1", SessionID: "sess-1", ExecutorType: "worktree"}
if _, err := exec.applyResumeRepoConfig(context.Background(), task, session, req, nil); err != nil {
t.Fatalf("applyResumeRepoConfig: %v", err)
}
if req.TaskDirName == "" {
t.Error("TaskDirName must not be empty even without an env row")
}
if req.RepoName != "my-repo" {
t.Errorf("RepoName = %q, want %q", req.RepoName, "my-repo")
}
})
}
// TestResumeSession_RefreshesStaleEnvironmentRow locks in the fix for the
// post-restart bug where a resume of a session that had previously failed mid-
// launch left task_environments stuck at status=stopped with empty
// task_dir_name / worktree_path. The frontend polls that row to decide
// whether the chat input is enabled, so the stale state stranded the UI on
// "Executor environment is unavailable (stopped)" even after the resume
// successfully prepared a fresh worktree.
func TestResumeSession_RefreshesStaleEnvironmentRow(t *testing.T) {
repo := newMockRepository()
taskID := "task-resume-env"
sessionID := "sess-resume-env"
repo.repositories["repo-1"] = &models.Repository{
ID: "repo-1", Name: "my-repo", LocalPath: "/repos/my-repo",
}
// Worktree executor: required for the resume to hit the worktree branch in
// applyResumeRepoConfig that stamps TaskDirName onto req.
repo.executors["exec-worktree"] = &models.Executor{
ID: "exec-worktree", Type: models.ExecutorTypeWorktree,
}
repo.tasks[taskID] = &models.Task{ID: taskID, Title: "Resume after failure"}
repo.sessions[sessionID] = &models.TaskSession{
ID: sessionID,
TaskID: taskID,
AgentProfileID: "profile-1",
State: models.TaskSessionStateFailed,
ExecutorID: "exec-worktree",
RepositoryID: "repo-1",
BaseBranch: "main",
TaskEnvironmentID: "env-stale",
}
// Stale env row from the previous failed launch: status=stopped, no
// worktree_path, no task_dir_name. This is the exact shape produced when
// LaunchAgent fails before persistTaskEnvironment writes the worktree
// fields back.
repo.taskEnvironments["env-stale"] = &models.TaskEnvironment{
ID: "env-stale",
TaskID: taskID,
Status: models.TaskEnvironmentStatusStopped,
WorktreePath: "",
TaskDirName: "",
}
const newWorktreePath = "/home/u/.kandev/tasks/resume-after-failure_abc/my-repo"
agentMgr := &mockAgentManager{
launchAgentFunc: func(_ context.Context, _ *LaunchAgentRequest) (*LaunchAgentResponse, error) {
return &LaunchAgentResponse{
AgentExecutionID: "exec-new",
WorktreeID: "wt-new",
WorktreePath: newWorktreePath,
WorktreeBranch: "kandev/resume-after-failure",
Status: v1.AgentStatusStarting,
}, nil
},
}
exec := newTestExecutor(t, agentMgr, repo)
if _, err := exec.ResumeSession(context.Background(), repo.sessions[sessionID], true); err != nil {
t.Fatalf("ResumeSession: %v", err)
}
env := repo.taskEnvironments["env-stale"]
if env.Status != models.TaskEnvironmentStatusReady {
t.Errorf("env.Status = %q, want %q — without the refresh the frontend keeps showing the executor as unavailable",
env.Status, models.TaskEnvironmentStatusReady)
}
if env.WorktreePath != newWorktreePath {
t.Errorf("env.WorktreePath = %q, want %q", env.WorktreePath, newWorktreePath)
}
if env.WorktreeID != "wt-new" {
t.Errorf("env.WorktreeID = %q, want %q", env.WorktreeID, "wt-new")
}
if env.TaskDirName == "" {
t.Error("env.TaskDirName must not be empty after resume; the worktree manager needs it for the on-disk task root")
}
}
func TestIsTerminalSessionState(t *testing.T) {
cases := []struct {
state models.TaskSessionState
@@ -181,6 +181,9 @@ type sessionExecutorStore interface {
ListTaskSessions(ctx context.Context, taskID string) ([]*models.TaskSession, error)
ListNonTerminalSessionsByAgentInstance(ctx context.Context, agentInstanceID string) ([]*models.TaskSession, error)
DeleteTaskSession(ctx context.Context, id string) error
// Messages — used by resume to backfill the initial user prompt when a
// prior launch failed before recordInitialMessage ran.
ListMessages(ctx context.Context, sessionID string) ([]*models.Message, error)
// Workspace
GetWorkspace(ctx context.Context, id string) (*models.Workspace, error)
// Task environment
@@ -819,6 +819,34 @@ func (s *Service) applyWorkflowAndPlanMode(ctx context.Context, prompt string, t
return effectivePrompt, planMode || stepHasPlanMode
}
// backfillInitialUserMessageIfMissing records the task's description as the
// first user message when the session has no messages at all. This covers the
// edge case where the initial launch failed before recordInitialMessage was
// called (postLaunchStart only runs after a successful LaunchAgent), leaving
// the chat empty even though the task carries the prompt the user originally
// typed.
//
// The check requires *zero* messages, not just zero user messages: if any
// agent output already exists from a partial prior run, the backfilled
// message would land at the bottom of the chat (CreateMessage stamps
// CreatedAt=now), which is worse than leaving the chat alone.
func (s *Service) backfillInitialUserMessageIfMissing(ctx context.Context, taskID, sessionID, prompt string) {
if prompt == "" || s.messageCreator == nil {
return
}
msgs, err := s.repo.ListMessages(ctx, sessionID)
if err != nil {
s.logger.Warn("backfill initial message: list messages failed",
zap.String("session_id", sessionID),
zap.Error(err))
return
}
if len(msgs) > 0 {
return
}
s.recordInitialMessage(ctx, taskID, sessionID, prompt, false, false, nil)
}
// recordInitialMessage creates the initial user message and updates session state after launch.
// autoStart marks the message as having been created by an automated trigger
// (workflow auto-start, PR/issue watch, Jira/Linear integration) so cleanup
@@ -954,6 +982,26 @@ func (s *Service) ResumeTaskSession(ctx context.Context, taskID, sessionID strin
// Preserve persisted task/session state; resume should not mutate state/columns.
execution.SessionState = v1.TaskSessionState(session.State)
// Backfill the initial user message when a prior failed launch never got
// to recordInitialMessage. Without this, the resume can succeed and the
// agent starts replying, but the chat shows agent output with no user
// prompt above it.
//
// We use task.Description (the raw user input) rather than the
// workflow-effective prompt produced by applyWorkflowAndPlanMode. The
// effective prompt may carry a plan-mode prefix or be templated through a
// workflow step, but reconstructing the exact prompt the original launch
// sent to the agent is brittle (workflow state may have advanced since).
// Surfacing the raw description is intentionally conservative: it shows
// what the user actually typed, which is what they expect to see in chat.
if task, taskErr := s.repo.GetTask(resumeCtx, taskID); taskErr != nil {
s.logger.Warn("resume: failed to load task for initial message backfill",
zap.String("task_id", taskID),
zap.Error(taskErr))
} else if task != nil {
s.backfillInitialUserMessageIfMissing(resumeCtx, taskID, sessionID, task.Description)
}
s.logger.Debug("task session resumed and ready for input",
zap.String("task_id", taskID),
zap.String("session_id", sessionID))
@@ -473,6 +473,113 @@ func (m *mockMessageCreator) AppendThinkingMessage(context.Context, string, stri
}
func (m *mockMessageCreator) InvalidateModelCache(string) {}
// --- backfillInitialUserMessageIfMissing ---
func TestBackfillInitialUserMessageIfMissing_RecordsWhenSessionEmpty(t *testing.T) {
ctx := context.Background()
repo := setupTestRepo(t)
seedTaskAndSession(t, repo, "task1", "session1", models.TaskSessionStateFailed)
mc := &mockMessageCreator{}
svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo())
svc.messageCreator = mc
// Session has zero messages — backfill should record the prompt.
svc.backfillInitialUserMessageIfMissing(ctx, "task1", "session1", "original prompt")
if len(mc.userMessages) != 1 {
t.Fatalf("expected 1 user message recorded, got %d", len(mc.userMessages))
}
if mc.userMessages[0].content != "original prompt" {
t.Errorf("content = %q, want %q", mc.userMessages[0].content, "original prompt")
}
}
func TestBackfillInitialUserMessageIfMissing_SkipsWhenUserMessageExists(t *testing.T) {
ctx := context.Background()
repo := setupTestRepo(t)
seedTaskAndSession(t, repo, "task1", "session1", models.TaskSessionStateFailed)
// Seed an existing user message — the backfill must be a no-op so a
// successful prior launch isn't duplicated on a subsequent resume.
if err := repo.CreateTurn(ctx, &models.Turn{ID: "turn1", TaskSessionID: "session1", TaskID: "task1", StartedAt: time.Now().UTC()}); err != nil {
t.Fatalf("create turn: %v", err)
}
if err := repo.CreateMessage(ctx, &models.Message{
ID: "msg1",
TaskSessionID: "session1",
TaskID: "task1",
TurnID: "turn1",
AuthorType: models.MessageAuthorUser,
Content: "user already sent this",
CreatedAt: time.Now().UTC(),
}); err != nil {
t.Fatalf("create message: %v", err)
}
mc := &mockMessageCreator{}
svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo())
svc.messageCreator = mc
svc.backfillInitialUserMessageIfMissing(ctx, "task1", "session1", "would be a duplicate")
if len(mc.userMessages) != 0 {
t.Fatalf("expected no user message recorded (one already exists), got %d", len(mc.userMessages))
}
}
// TestBackfillInitialUserMessageIfMissing_SkipsWhenAgentMessageExists covers
// the regression where a partial prior run produced agent output but never
// recorded the initial user message. Recording the user message now with
// CreatedAt=time.Now() would place it at the bottom of the chat (after the
// agent messages), which is worse than leaving the chat alone.
func TestBackfillInitialUserMessageIfMissing_SkipsWhenAgentMessageExists(t *testing.T) {
ctx := context.Background()
repo := setupTestRepo(t)
seedTaskAndSession(t, repo, "task1", "session1", models.TaskSessionStateFailed)
if err := repo.CreateTurn(ctx, &models.Turn{ID: "turn1", TaskSessionID: "session1", TaskID: "task1", StartedAt: time.Now().UTC()}); err != nil {
t.Fatalf("create turn: %v", err)
}
if err := repo.CreateMessage(ctx, &models.Message{
ID: "agent-msg-1",
TaskSessionID: "session1",
TaskID: "task1",
TurnID: "turn1",
AuthorType: models.MessageAuthorAgent,
Content: "agent partial output from a prior run",
CreatedAt: time.Now().UTC(),
}); err != nil {
t.Fatalf("create agent message: %v", err)
}
mc := &mockMessageCreator{}
svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo())
svc.messageCreator = mc
svc.backfillInitialUserMessageIfMissing(ctx, "task1", "session1", "the original prompt")
if len(mc.userMessages) != 0 {
t.Fatalf("expected no backfill when agent messages exist, got %d", len(mc.userMessages))
}
}
func TestBackfillInitialUserMessageIfMissing_SkipsEmptyPrompt(t *testing.T) {
ctx := context.Background()
repo := setupTestRepo(t)
seedTaskAndSession(t, repo, "task1", "session1", models.TaskSessionStateFailed)
mc := &mockMessageCreator{}
svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo())
svc.messageCreator = mc
svc.backfillInitialUserMessageIfMissing(ctx, "task1", "session1", "")
if len(mc.userMessages) != 0 {
t.Fatalf("expected no user message for empty prompt, got %d", len(mc.userMessages))
}
}
func TestRecordInitialMessage_DoesNotChangeSessionState(t *testing.T) {
ctx := context.Background()
repo := setupTestRepo(t)
@@ -107,7 +107,7 @@ func TestOptimize_StartsJobAndSucceeds(t *testing.T) {
}
func TestHandleVacuum_Returns202WithJobID(t *testing.T) {
svc, _, _, _ := newTestService(t)
svc, tracker, _, _ := newTestService(t)
gin.SetMode(gin.TestMode)
r := gin.New()
r.POST("/vacuum", HandleVacuum(svc))
@@ -119,10 +119,15 @@ func TestHandleVacuum_Returns202WithJobID(t *testing.T) {
if !contains(w.Body.String(), `"job_id"`) {
t.Errorf("body missing job_id: %s", w.Body.String())
}
// Make sure the spawned job finishes before the test exits so the temp dir
// cleanup does not race with VACUUM still writing the SQLite journal.
for _, j := range tracker.List() {
waitForState(t, tracker, j.ID, jobs.StateSucceeded)
}
}
func TestHandleOptimize_Returns202WithJobID(t *testing.T) {
svc, _, _, _ := newTestService(t)
svc, tracker, _, _ := newTestService(t)
gin.SetMode(gin.TestMode)
r := gin.New()
r.POST("/optimize", HandleOptimize(svc))
@@ -134,4 +139,8 @@ func TestHandleOptimize_Returns202WithJobID(t *testing.T) {
if !contains(w.Body.String(), `"job_id"`) {
t.Errorf("body missing job_id: %s", w.Body.String())
}
// Same cleanup-race guard as TestHandleVacuum_Returns202WithJobID.
for _, j := range tracker.List() {
waitForState(t, tracker, j.ID, jobs.StateSucceeded)
}
}