test: stabilize E2E cleanup and responsive workflow flows (#3100)

apps/backend/internal/agent/docker/client.go

@@ -15,6 +15,7 @@ import (
 	"sync"
 	"time"
 
+	"github.com/containerd/errdefs"
 	"github.com/kandev/kandev/internal/agent/runtime/activity"
 	"github.com/kandev/kandev/internal/common/config"
 	"github.com/kandev/kandev/internal/common/logger"
@@ -389,6 +390,12 @@ func (c *Client) StopContainer(ctx context.Context, containerID string, timeout
 		Timeout: &timeoutSeconds,
 	})
 	if err != nil {
+		if errdefs.IsNotFound(err) {
+			// Container teardown is intentionally idempotent. A container can
+			// disappear between the task resource snapshot and this stop call
+			// (for example, after an explicit environment reset).
+			return nil
+		}
 		c.logger.Error("Failed to stop container", zap.String("container_id", containerID), zap.Error(err))
 		return fmt.Errorf("failed to stop container %s: %w", containerID, err)
 	}
@@ -409,6 +416,11 @@ func (c *Client) RemoveContainer(ctx context.Context, containerID string, force
 		RemoveVolumes: true,
 	})
 	if err != nil {
+		if errdefs.IsNotFound(err) {
+			// Remove is a convergence operation. Another cleanup owner may have
+			// removed the container already, which is the desired end state.
+			return nil
+		}
 		c.logger.Error("Failed to remove container", zap.String("container_id", containerID), zap.Error(err))
 		return fmt.Errorf("failed to remove container %s: %w", containerID, err)
 	}
@@ -433,6 +445,11 @@ func (c *Client) KillContainer(ctx context.Context, containerID string, signal s
 
 	_, err := c.cli.ContainerKill(ctx, containerID, client.ContainerKillOptions{Signal: signal})
 	if err != nil {
+		if errdefs.IsNotFound(err) {
+			// A forced cleanup may race with Docker auto-removal or another
+			// teardown pass. Missing already means the container is stopped.
+			return nil
+		}
 		c.logger.Error("Failed to kill container", zap.String("container_id", containerID), zap.Error(err))
 		return fmt.Errorf("failed to kill container %s: %w", containerID, err)
 	}

apps/backend/internal/agent/docker/client_test.go

@@ -1,11 +1,20 @@
 package docker
 
 import (
+	"context"
 	"fmt"
+	"net/http"
+	"net/http/httptest"
 	"net/netip"
+	"strings"
 	"testing"
+	"time"
 
+	"github.com/kandev/kandev/internal/common/config"
+	"github.com/kandev/kandev/internal/common/logger"
 	"github.com/moby/moby/api/types/network"
+	"github.com/stretchr/testify/require"
+	"go.uber.org/zap"
 )
 
 func TestNormalizeDockerHostIP(t *testing.T) {
@@ -130,3 +139,30 @@ func TestParseHostPort(t *testing.T) {
 		}
 	}
 }
+
+func TestContainerTeardownTreatsMissingContainersAsSuccess(t *testing.T) {
+	dockerDaemon := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.Method == http.MethodHead && r.URL.Path == "/_ping" {
+			w.WriteHeader(http.StatusOK)
+			return
+		}
+		w.Header().Set("Content-Type", "application/json")
+		w.WriteHeader(http.StatusNotFound)
+		_, _ = w.Write([]byte(`{"message":"No such container: missing-container"}`))
+	}))
+	t.Cleanup(dockerDaemon.Close)
+
+	log, err := logger.NewFromZap(zap.NewNop())
+	if err != nil {
+		t.Fatalf("new logger: %v", err)
+	}
+	client, err := NewClient(config.DockerConfig{
+		Host:       "tcp://" + strings.TrimPrefix(dockerDaemon.URL, "http://"),
+		APIVersion: "1.44",
+	}, log)
+	require.NoError(t, err)
+
+	require.NoError(t, client.StopContainer(context.Background(), "missing-container", time.Second))
+	require.NoError(t, client.KillContainer(context.Background(), "missing-container", "SIGKILL"))
+	require.NoError(t, client.RemoveContainer(context.Background(), "missing-container", true))
+}

apps/backend/internal/backendapp/e2e_reset.go

@@ -26,10 +26,12 @@ import (
 
 const (
 	// errKey is the JSON field used for error responses from the E2E endpoints.
-	errKey            = "error"
-	statusKey         = "status"
-	e2eResetSourceKey = "source"
-	e2eResetTypeKey   = "type"
+	errKey                     = "error"
+	statusKey                  = "status"
+	e2eResetSourceKey          = "source"
+	e2eResetTypeKey            = "type"
+	e2eTaskCleanupWaitTimeout  = 30 * time.Second
+	e2eTaskCleanupPollInterval = 25 * time.Millisecond
 )
 
 // registerE2EResetRoutes registers the E2E test-only endpoints.
@@ -107,6 +109,16 @@ func handleE2EReset(
 
 		ctx := c.Request.Context()
 
+		// Capture every task before deleting review watches or automations. Those
+		// owners delete their tasks as part of their own cleanup, so a later task
+		// list cannot discover the resource-cleanup jobs they leave behind.
+		taskIDsForCleanup, err := listE2ETaskIDs(ctx, repo.DB(), workspaceID)
+		if err != nil {
+			log.Error("e2e reset: failed to capture task IDs", zap.Error(err))
+			c.JSON(http.StatusInternalServerError, gin.H{errKey: err.Error()})
+			return
+		}
+
 		// Wipe routing state so the office-routing-* specs don't leak
 		// degraded health rows / route attempts / parked runs between
 		// each other. Office tables live in the same SQLite db so the
@@ -251,6 +263,7 @@ func handleE2EReset(
 			return
 		}
 		var deletedTasks int64
+		deletedTaskIDs := append([]string(nil), taskIDsForCleanup...)
 		for _, t := range tasks {
 			if err := taskSvc.DeleteTask(ctx, t.ID); err != nil {
 				// Abort: leaving an undeleted task with its workflow gone
@@ -261,6 +274,19 @@ func handleE2EReset(
 				return
 			}
 			deletedTasks++
+			if !containsE2ETaskID(deletedTaskIDs, t.ID) {
+				deletedTaskIDs = append(deletedTaskIDs, t.ID)
+			}
+		}
+
+		// DeleteTask removes the task row synchronously but stops agents and
+		// removes worktrees asynchronously. The next test reuses the worker's
+		// repository, so returning before those jobs finish lets an old cleanup
+		// race with the next test's repository setup and file-tree read.
+		if err := waitForE2ETaskCleanup(ctx, repo.DB(), deletedTaskIDs); err != nil {
+			log.Error("e2e reset: task cleanup did not finish", zap.Error(err))
+			c.JSON(http.StatusInternalServerError, gin.H{errKey: err.Error()})
+			return
 		}
 
 		deletedWorkflows, err := repo.DeleteWorkflowsByWorkspace(ctx, workspaceID, keepWorkflowIDs)
@@ -281,6 +307,162 @@ func handleE2EReset(
 	}
 }
 
+func waitForE2ETaskCleanup(ctx context.Context, database *sql.DB, taskIDs []string) error {
+	if database == nil || len(taskIDs) == 0 {
+		return nil
+	}
+	return waitForE2ETaskCleanupWithReader(
+		ctx,
+		taskIDs,
+		e2eTaskCleanupPollInterval,
+		func(ctx context.Context, ids []string) ([]e2eTaskCleanupStatus, error) {
+			return queryE2ETaskCleanupStatuses(ctx, database, ids)
+		},
+	)
+}
+
+type e2eTaskCleanupStatus struct {
+	taskID    string
+	state     taskmodels.TaskResourceCleanupState
+	lastError string
+}
+
+type e2eTaskCleanupStatusReader func(context.Context, []string) ([]e2eTaskCleanupStatus, error)
+
+func waitForE2ETaskCleanupWithReader(
+	ctx context.Context,
+	taskIDs []string,
+	pollInterval time.Duration,
+	readStatuses e2eTaskCleanupStatusReader,
+) error {
+	if len(taskIDs) == 0 {
+		return nil
+	}
+
+	waitCtx, cancel := context.WithTimeout(ctx, e2eTaskCleanupWaitTimeout)
+	defer cancel()
+	for {
+		statuses, err := readStatuses(waitCtx, taskIDs)
+		if err != nil {
+			return fmt.Errorf("query task cleanup status: %w", err)
+		}
+		activeTaskIDs, err := activeE2ETaskCleanupIDs(statuses)
+		if err != nil {
+			return err
+		}
+		if len(activeTaskIDs) == 0 {
+			return nil
+		}
+
+		timer := time.NewTimer(pollInterval)
+		select {
+		case <-waitCtx.Done():
+			timer.Stop()
+			return fmt.Errorf("wait for task cleanup (%s): %w", strings.Join(activeTaskIDs, ", "), waitCtx.Err())
+		case <-timer.C:
+		}
+	}
+}
+
+func queryE2ETaskCleanupStatuses(
+	ctx context.Context,
+	database *sql.DB,
+	taskIDs []string,
+) ([]e2eTaskCleanupStatus, error) {
+	const taskIDBatchSize = 100
+	statuses := make([]e2eTaskCleanupStatus, 0)
+	for start := 0; start < len(taskIDs); start += taskIDBatchSize {
+		end := start + taskIDBatchSize
+		if end > len(taskIDs) {
+			end = len(taskIDs)
+		}
+		batch := taskIDs[start:end]
+		placeholders := strings.TrimSuffix(strings.Repeat("?,", len(batch)), ",")
+		query := `SELECT task_id, state, last_error
+			FROM task_resource_cleanup_jobs
+			WHERE task_id IN (` + placeholders + `)
+			  AND state IN (?, ?, ?, ?, ?)`
+		args := make([]any, 0, len(batch)+5)
+		for _, taskID := range batch {
+			args = append(args, taskID)
+		}
+		args = append(args,
+			taskmodels.TaskResourceCleanupStatePrepared,
+			taskmodels.TaskResourceCleanupStatePending,
+			taskmodels.TaskResourceCleanupStateRunning,
+			taskmodels.TaskResourceCleanupStateRetryWait,
+			taskmodels.TaskResourceCleanupStateFailed,
+		)
+
+		rows, err := database.QueryContext(ctx, query, args...)
+		if err != nil {
+			return nil, err
+		}
+		for rows.Next() {
+			var status e2eTaskCleanupStatus
+			if err := rows.Scan(&status.taskID, &status.state, &status.lastError); err != nil {
+				_ = rows.Close()
+				return nil, err
+			}
+			statuses = append(statuses, status)
+		}
+		rowsErr := rows.Err()
+		_ = rows.Close()
+		if rowsErr != nil {
+			return nil, rowsErr
+		}
+	}
+	return statuses, nil
+}
+
+func activeE2ETaskCleanupIDs(statuses []e2eTaskCleanupStatus) ([]string, error) {
+	activeTaskIDs := make([]string, 0, len(statuses))
+	seen := make(map[string]struct{}, len(statuses))
+	for _, status := range statuses {
+		if status.state == taskmodels.TaskResourceCleanupStateFailed {
+			if status.lastError == "" {
+				return nil, fmt.Errorf("task cleanup failed for %s", status.taskID)
+			}
+			return nil, fmt.Errorf("task cleanup failed for %s: %s", status.taskID, status.lastError)
+		}
+		if _, ok := seen[status.taskID]; ok {
+			continue
+		}
+		seen[status.taskID] = struct{}{}
+		activeTaskIDs = append(activeTaskIDs, status.taskID)
+	}
+	return activeTaskIDs, nil
+}
+
+func listE2ETaskIDs(ctx context.Context, database *sql.DB, workspaceID string) ([]string, error) {
+	rows, err := database.QueryContext(ctx, `SELECT id FROM tasks WHERE workspace_id = ?`, workspaceID)
+	if err != nil {
+		return nil, err
+	}
+	defer func() { _ = rows.Close() }()
+	ids := make([]string, 0)
+	for rows.Next() {
+		var id string
+		if err := rows.Scan(&id); err != nil {
+			return nil, err
+		}
+		ids = append(ids, id)
+	}
+	if err := rows.Err(); err != nil {
+		return nil, err
+	}
+	return ids, nil
+}
+
+func containsE2ETaskID(taskIDs []string, taskID string) bool {
+	for _, id := range taskIDs {
+		if id == taskID {
+			return true
+		}
+	}
+	return false
+}
+
 func resetGitHubAppRegistrationsForE2E(
 	ctx context.Context,
 	service *github.Service,

apps/backend/internal/backendapp/e2e_reset_test.go

@@ -2,12 +2,17 @@ package backendapp
 
 import (
 	"context"
+	"errors"
 	"path/filepath"
+	"strconv"
+	"strings"
 	"testing"
+	"time"
 
 	"github.com/jmoiron/sqlx"
 
 	"github.com/kandev/kandev/internal/db"
+	taskmodels "github.com/kandev/kandev/internal/task/models"
 )
 
 func TestE2EResetDeletesWorkspaceGitHubAuthentication(t *testing.T) {
@@ -59,6 +64,136 @@ func TestE2EResetDeletesWorkspaceGitHubAuthentication(t *testing.T) {
 	}
 }
 
+func TestWaitForE2ETaskCleanupWithReader(t *testing.T) {
+	t.Run("waits for a running job to finish", func(t *testing.T) {
+		firstRead := make(chan struct{})
+		releaseFirstRead := make(chan struct{})
+		result := make(chan error, 1)
+		readCount := 0
+		go func() {
+			result <- waitForE2ETaskCleanupWithReader(
+				context.Background(),
+				[]string{"task-1"},
+				0,
+				func(context.Context, []string) ([]e2eTaskCleanupStatus, error) {
+					readCount++
+					if readCount == 1 {
+						close(firstRead)
+						<-releaseFirstRead
+						return []e2eTaskCleanupStatus{{
+							taskID: "task-1",
+							state:  taskmodels.TaskResourceCleanupStateRunning,
+						}}, nil
+					}
+					return nil, nil
+				},
+			)
+		}()
+
+		<-firstRead
+		close(releaseFirstRead)
+		if err := <-result; err != nil {
+			t.Fatalf("waitForE2ETaskCleanupWithReader: %v", err)
+		}
+		if readCount != 2 {
+			t.Fatalf("status reads = %d, want 2", readCount)
+		}
+	})
+
+	t.Run("returns failed job error", func(t *testing.T) {
+		err := waitForE2ETaskCleanupWithReader(
+			context.Background(),
+			[]string{"task-1"},
+			0,
+			func(context.Context, []string) ([]e2eTaskCleanupStatus, error) {
+				return []e2eTaskCleanupStatus{{
+					taskID:    "task-1",
+					state:     taskmodels.TaskResourceCleanupStateFailed,
+					lastError: "worktree is busy",
+				}}, nil
+			},
+		)
+		if err == nil || !strings.Contains(err.Error(), "task cleanup failed for task-1: worktree is busy") {
+			t.Fatalf("failed cleanup error = %v", err)
+		}
+	})
+
+	t.Run("returns context deadline when cleanup does not finish", func(t *testing.T) {
+		ctx, cancel := context.WithTimeout(context.Background(), 0)
+		defer cancel()
+		err := waitForE2ETaskCleanupWithReader(
+			ctx,
+			[]string{"task-1"},
+			time.Hour,
+			func(context.Context, []string) ([]e2eTaskCleanupStatus, error) {
+				return []e2eTaskCleanupStatus{{
+					taskID: "task-1",
+					state:  taskmodels.TaskResourceCleanupStateRunning,
+				}}, nil
+			},
+		)
+		if !errors.Is(err, context.DeadlineExceeded) {
+			t.Fatalf("cleanup timeout error = %v, want context deadline exceeded", err)
+		}
+	})
+}
+
+func TestQueryE2ETaskCleanupStatusesBatchesTaskIDs(t *testing.T) {
+	raw, err := db.OpenSQLite(filepath.Join(t.TempDir(), "e2e-reset-cleanup.db"))
+	if err != nil {
+		t.Fatalf("open database: %v", err)
+	}
+	database := sqlx.NewDb(raw, "sqlite3")
+	t.Cleanup(func() { _ = database.Close() })
+	if _, err := database.Exec(`
+		CREATE TABLE task_resource_cleanup_jobs (
+			task_id TEXT NOT NULL,
+			state TEXT NOT NULL,
+			last_error TEXT NOT NULL DEFAULT ''
+		);
+		INSERT INTO task_resource_cleanup_jobs(task_id, state) VALUES ('task-204', 'running');
+	`); err != nil {
+		t.Fatalf("seed cleanup job: %v", err)
+	}
+
+	taskIDs := make([]string, 205)
+	for i := range taskIDs {
+		taskIDs[i] = "task-" + strconv.Itoa(i)
+	}
+	statuses, err := queryE2ETaskCleanupStatuses(context.Background(), database.DB, taskIDs)
+	if err != nil {
+		t.Fatalf("queryE2ETaskCleanupStatuses: %v", err)
+	}
+	if len(statuses) != 1 || statuses[0].taskID != "task-204" || statuses[0].state != taskmodels.TaskResourceCleanupStateRunning {
+		t.Fatalf("cleanup statuses = %+v, want task-204 running", statuses)
+	}
+}
+
+func TestListE2ETaskIDsIncludesAutomationOwnedTasks(t *testing.T) {
+	raw, err := db.OpenSQLite(filepath.Join(t.TempDir(), "e2e-reset-tasks.db"))
+	if err != nil {
+		t.Fatalf("open database: %v", err)
+	}
+	database := sqlx.NewDb(raw, "sqlite3")
+	t.Cleanup(func() { _ = database.Close() })
+	if _, err := database.Exec(`
+		CREATE TABLE tasks (id TEXT NOT NULL, workspace_id TEXT NOT NULL, origin TEXT NOT NULL);
+		INSERT INTO tasks VALUES ('ordinary-task', 'ws-1', ''),
+			('automation-task', 'ws-1', 'automation_run'),
+			('other-workspace-task', 'ws-2', '');
+	`); err != nil {
+		t.Fatalf("seed tasks: %v", err)
+	}
+
+	ids, err := listE2ETaskIDs(context.Background(), database.DB, "ws-1")
+	if err != nil {
+		t.Fatalf("listE2ETaskIDs: %v", err)
+	}
+	if len(ids) != 2 || !containsE2ETaskID(ids, "ordinary-task") || !containsE2ETaskID(ids, "automation-task") {
+		t.Fatalf("task IDs = %v, want both ws-1 tasks", ids)
+	}
+}
+
 func assertWorkspaceRows(t *testing.T, database *sqlx.DB, table, workspaceID string, want int) {
 	t.Helper()
 	var got int

apps/web/e2e/pages/session-page.ts

@@ -1344,6 +1344,46 @@ export class SessionPage {
     await expect(this.files).not.toBeVisible({ timeout: 5_000 });
   }
 
+  /**
+   * Move the task to a workflow step through whichever stepper presentation
+   * the responsive top bar selected. Narrow layouts expose the target in the
+   * compact disclosure instead of rendering every step in the top bar.
+   */
+  async moveToWorkflowStep(step: { id: string; name: string }): Promise<void> {
+    const fullStep = this.page
+      .locator(`[data-testid=${JSON.stringify(`workflow-step-${step.name}`)}]:visible`)
+      .first();
+    const compactStepper = this.page.getByTestId("workflow-stepper-minimal");
+    let presentation: "full" | "compact" | undefined;
+    await expect
+      .poll(
+        async () => {
+          if (await fullStep.isVisible()) {
+            presentation = "full";
+            return true;
+          }
+          if (await compactStepper.isVisible()) {
+            presentation = "compact";
+            return true;
+          }
+          return false;
+        },
+        { timeout: 10_000, message: `Waiting for workflow step presentation for ${step.name}` },
+      )
+      .toBe(true);
+
+    if (presentation === "full") {
+      await fullStep.hover();
+      await this.page.getByRole("button", { name: "Move here", exact: true }).click();
+      return;
+    }
+
+    await compactStepper.click();
+    const moveButton = this.page.getByTestId(`workflow-step-disclosure-move-${step.id}`);
+    await expect(moveButton).toBeVisible();
+    await moveButton.click();
+  }
+
   /**
    * Assert the layout is in the default (non-maximized) state:
    * chat, terminal, files, and sidebar are all visible, and layout fills the viewport.

apps/web/e2e/tests/session/maximize-sidebar-switch.spec.ts

@@ -97,6 +97,9 @@ test.describe("Maximize survives sidebar task switch", () => {
     await sessionA.waitForLoad();
     await sessionA.waitForChatIdle({ timeout: 30_000 });
     // Task B starts with the default layout — sanity check before the bug.
+    // Git updates can focus Changes while the task settles. Select Files
+    // before asserting the default right-column panel.
+    await sessionA.clickTab("Files");
     await sessionA.expectDefaultLayout();
 
     // Switch back to Task A via the sidebar.

apps/web/e2e/tests/session/session-layout.spec.ts

@@ -47,6 +47,9 @@ test.describe("Session layout", () => {
     const session = await seedTaskWithSession(testPage, apiClient, seedData, "Maximize Test");
 
     // Default layout: all panels visible
+    // Git updates can focus Changes while the task settles. Select Files
+    // before asserting the default right-column panel.
+    await session.clickTab("Files");
     await session.expectDefaultLayout();
 
     // Type a command in the terminal

apps/web/e2e/tests/ssh/hostkey-rotation.spec.ts

@@ -1,5 +1,6 @@
 import { test, expect } from "../../fixtures/ssh-test-base";
 import { regenerateHostKey } from "../../helpers/ssh";
+import { waitForSessionDone } from "../../helpers/session";
 
 /**
  * Host-key rotation: regenerate the container's host key and assert
@@ -148,5 +149,13 @@ test.describe("ssh executor — host key rotation", () => {
         timeout: 60_000,
       })
       .toBe("ssh");
+
+    if (!task.session_id) throw new Error("SSH re-trust task did not return a session_id");
+    await waitForSessionDone(
+      apiClient,
+      task.id,
+      task.session_id,
+      "SSH re-trust task should finish before teardown",
+    );
   });
 });

apps/web/e2e/tests/ssh/test-endpoint.spec.ts

@@ -1,5 +1,6 @@
 import { test, expect } from "../../fixtures/ssh-test-base";
 import { execInContainer } from "../../helpers/ssh";
+import { waitForSessionDone } from "../../helpers/session";
 
 /**
  * HTTP contract for POST /api/v1/ssh/test. The endpoint dials the host with a
@@ -161,6 +162,14 @@ test.describe("ssh test-endpoint contract", () => {
         { message: "SSH agentctl should be cached after the initial launch", timeout: 60_000 },
       )
       .toBe("cached");
+
+    if (!task.session_id) throw new Error("SSH cache priming task did not return a session_id");
+    await waitForSessionDone(
+      apiClient,
+      task.id,
+      task.session_id,
+      "SSH cache priming task should finish before teardown",
+    );
   });
 
   // F6 — WS dispatcher returns the same shape as the HTTP route. Doesn't

apps/web/e2e/tests/task/file-tree-drag-drop.spec.ts

@@ -29,12 +29,24 @@ async function setupTask(
   taskTitle: string,
 ) {
   const profile = await createStandardProfile(apiClient, profileName);
-  await apiClient.createTaskWithAgent(seedData.workspaceId, taskTitle, profile.id, {
+  const task = await apiClient.createTaskWithAgent(seedData.workspaceId, taskTitle, profile.id, {
     description: "/e2e:simple-message",
     workflow_id: seedData.workflowId,
     workflow_step_id: seedData.startStepId,
     repository_ids: [seedData.repositoryId],
   });
+
+  // The task API returns before local workspace preparation finishes. Wait
+  // for the environment's durable ready state before the first tree request;
+  // otherwise the tree can legitimately snapshot the repository while the
+  // agent session is still being attached to it.
+  await expect
+    .poll(async () => (await apiClient.getTaskEnvironment(task.id))?.status ?? null, {
+      timeout: 30_000,
+      message: `Waiting for ${taskTitle} task environment to be ready`,
+    })
+    .toBe("ready");
+
   const session = await openTaskSession(testPage, taskTitle);
   await session.clickTab("Files");
   return session;

apps/web/e2e/tests/workflow/workflow-steps.spec.ts

@@ -97,8 +97,7 @@ test.describe("Workflow steps", () => {
     await session.waitForLoad();
     await expect(session.stepper).toBeVisible();
 
-    await session.stepper.getByTestId(`workflow-step-${targetStep.name}`).hover();
-    await testPage.getByRole("button", { name: "Move here" }).click();
+    await session.moveToWorkflowStep(targetStep);
 
     const moveError = testPage.getByTestId("task-move-error-banner");
     await expect(moveError).toBeVisible();