diff --git a/.github/workflows/backend-tests.yml b/.github/workflows/backend-tests.yml index f8aaad994e..58d0e73c1e 100644 --- a/.github/workflows/backend-tests.yml +++ b/.github/workflows/backend-tests.yml @@ -28,7 +28,7 @@ concurrency: group: ${{ github.workflow }}-${{ github.ref }} cancel-in-progress: true -# Least-privilege default. The Linux `test` job and the Windows `test-windows` +# Least-privilege default. The Linux static/test jobs and the Windows `test-windows` # job only read source and run tests — no PR/issue writes, no release ops. # `packages: read` is required for the Linux job to pull the kandev-ci # container image from GHCR (currently public; explicit grant keeps the @@ -39,8 +39,8 @@ permissions: packages: read jobs: - test: - name: Run Backend Tests + static_checks: + name: Backend Static Checks runs-on: ubuntu-latest # Pre-baked image bundles Go, golangci-lint, go-licenses, Docker CLI, Node, # pnpm, and the Playwright base — see .github/docker/ci-base/Dockerfile. @@ -58,6 +58,7 @@ jobs: uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 # v6 with: fetch-depth: 0 + persist-credentials: false # actions/checkout under a container job marks the workspace as a safe # directory automatically, but subsequent `git` invocations from steps @@ -69,19 +70,16 @@ jobs: run: git config --global --add safe.directory "$GITHUB_WORKSPACE" - name: Cache Go modules - # Homelab self-hosted runners (kept for future re-enablement; the - # `container:` above forces the job onto a github-hosted runner today, - # so this condition is dormant). If homelab is reactivated, split this - # job into a matrix where the homelab variant omits `container:`. - if: ${{ contains(runner.name, 'homelab') }} uses: actions/cache@0057852bfaa89a56745cba8c7296529d2fc39830 # v4 with: path: | ~/go/pkg/mod ~/.cache/go-build ~/.cache/golangci-lint - key: go-${{ runner.os }}-${{ hashFiles('apps/backend/go.sum') }} - restore-keys: go-${{ runner.os }}- + key: go-static-${{ runner.os }}-${{ hashFiles('apps/backend/go.mod', 'apps/backend/go.sum') }} + restore-keys: | + go-static-${{ runner.os }}- + go-${{ runner.os }}- - name: Build run: go build -v ./cmd/kandev @@ -144,28 +142,135 @@ jobs: shell: bash run: golangci-lint run ./... --new-from-rev="${BASE_SHA}" --timeout=5m + test_shards: + name: Backend Tests (${{ matrix.shard }}/2) + runs-on: ubuntu-latest + strategy: + fail-fast: false + matrix: + include: + - shard: 1 + index: 0 + - shard: 2 + index: 1 + # Pre-baked image bundles Go and jq, so test shards can start immediately + # without repeating tool installation. + container: ghcr.io/kdlbs/kandev-ci:build-latest + defaults: + run: + shell: bash + working-directory: apps/backend + + steps: + - name: Checkout code + uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 # v6 + with: + persist-credentials: false + + - name: Mark workspace safe for git + shell: bash + working-directory: ${{ github.workspace }} + run: git config --global --add safe.directory "$GITHUB_WORKSPACE" + + - name: Cache Go modules + uses: actions/cache@0057852bfaa89a56745cba8c7296529d2fc39830 # v4 + with: + path: | + ~/go/pkg/mod + ~/.cache/go-build + key: go-test-${{ matrix.shard }}-${{ runner.os }}-${{ hashFiles('apps/backend/go.mod', 'apps/backend/go.sum') }} + restore-keys: | + go-test-${{ matrix.shard }}-${{ runner.os }}- + go-${{ runner.os }}- + - name: Run tests + env: + SHARD_INDEX: ${{ matrix.index }} + SHARD_NUMBER: ${{ matrix.shard }} run: | - set -o pipefail - go test -v -race -coverprofile=coverage.out -covermode=atomic -json ./... 2>&1 | tee test-results.json + set -euo pipefail + + package_list="$(go list ./...)" + if [[ -z "${package_list}" ]]; then + echo "go list returned no backend packages" + exit 1 + fi + + mapfile -t all_packages <<<"${package_list}" + shard_packages=() + for package in "${all_packages[@]}"; do + # Stable hashing keeps package assignments unchanged when another + # package is added or removed. + checksum="$(cksum <<<"${package}")" + checksum="${checksum%% *}" + if (( checksum % 2 == SHARD_INDEX )); then + shard_packages+=("${package}") + fi + done + + if (( ${#shard_packages[@]} == 0 )); then + echo "Shard ${SHARD_NUMBER}/2 selected no packages" + exit 1 + fi + + echo "Running ${#shard_packages[@]} packages in shard ${SHARD_NUMBER}/2" + # Keep complete JSON for reports/artifacts while limiting live output + # to compiler/process lines and package-level pass/fail summaries. + go test \ + -race \ + -coverprofile="coverage-${SHARD_NUMBER}.out" \ + -covermode=atomic \ + -json \ + "${shard_packages[@]}" \ + 2>&1 \ + | tee "test-results-${SHARD_NUMBER}.json" \ + | jq --unbuffered -Rr ' + . as $line + | (try fromjson catch null) as $event + | if $event == null then + $line + elif ($event.Test == null and ($event.Action == "pass" or $event.Action == "fail")) then + "\($event.Action): \($event.Package) (\($event.Elapsed // 0)s)" + else + empty + end + ' - name: Generate test report uses: robherley/go-test-action@3856e4e57831b3b6d6d8d89f91747e5200c533f7 # v0 if: always() with: - fromJSONFile: apps/backend/test-results.json + fromJSONFile: apps/backend/test-results-${{ matrix.shard }}.json omit: successful - name: Upload test artifacts uses: actions/upload-artifact@330a01c490aca151604b8cf639adc76d48f6c5d4 # v5 if: always() with: - name: test-results + name: backend-test-results-${{ matrix.shard }} path: | - apps/backend/coverage.out - apps/backend/test-results.json + apps/backend/coverage-${{ matrix.shard }}.out + apps/backend/test-results-${{ matrix.shard }}.json retention-days: 30 + test: + name: Run Backend Tests + runs-on: ubuntu-latest + needs: [static_checks, test_shards] + if: ${{ always() }} + steps: + - name: Require backend checks to pass + env: + STATIC_CHECKS_RESULT: ${{ needs.static_checks.result }} + TEST_SHARDS_RESULT: ${{ needs.test_shards.result }} + run: | + set -euo pipefail + if [[ "${STATIC_CHECKS_RESULT}" != "success" || "${TEST_SHARDS_RESULT}" != "success" ]]; then + echo "Static checks: ${STATIC_CHECKS_RESULT}" + echo "Test shards: ${TEST_SHARDS_RESULT}" + exit 1 + fi + postgres-boot: name: Backend Postgres runs-on: ubuntu-latest @@ -248,6 +353,8 @@ jobs: steps: - name: Checkout code uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 # v6 + with: + persist-credentials: false - name: Set up Go uses: actions/setup-go@924ae3a1cded613372ab5595356fb5720e22ba16 # v6 diff --git a/apps/backend/internal/integration/mcp_stop_task_test.go b/apps/backend/internal/integration/mcp_stop_task_test.go index 4ec298730e..e7611a2e77 100644 --- a/apps/backend/internal/integration/mcp_stop_task_test.go +++ b/apps/backend/internal/integration/mcp_stop_task_test.go @@ -204,6 +204,22 @@ func TestMCPStopTask_DirectParentStopsLongRunningChild(t *testing.T) { } }() + initialToolCallPersisted := make(chan struct{}) + var initialToolCallOnce sync.Once + messageSubscription, err := ts.EventBus.Subscribe(events.MessageAdded, func(_ context.Context, event *bus.Event) error { + data, ok := event.Data.(map[string]interface{}) + if ok && data["task_id"] == child.ID && data["type"] == string(models.MessageTypeToolCall) { + initialToolCallOnce.Do(func() { close(initialToolCallPersisted) }) + } + return nil + }) + require.NoError(t, err) + defer func() { + if err := messageSubscription.Unsubscribe(); err != nil { + t.Errorf("unsubscribe message observer: %v", err) + } + }() + launch, err := orchestratorSvc.LaunchSession(context.Background(), &orchestrator.LaunchSessionRequest{ TaskID: child.ID, Intent: orchestrator.IntentStart, AgentProfileID: "integration-agent", Prompt: "Keep working until explicitly stopped.", @@ -212,6 +228,7 @@ func TestMCPStopTask_DirectParentStopsLongRunningChild(t *testing.T) { require.True(t, launch.Success) require.NotEmpty(t, launch.SessionID) mcpStopAwaitSignal(t, running, 2*time.Second, "child RUNNING state") + mcpStopAwaitSignal(t, initialToolCallPersisted, 2*time.Second, "initial tool-call persistence") sessionBefore, err := ts.TaskRepo.GetTaskSession(context.Background(), launch.SessionID) require.NoError(t, err) diff --git a/apps/backend/internal/mcp/handlers/handlers.go b/apps/backend/internal/mcp/handlers/handlers.go index d03e8047f0..dd83d2ce1b 100644 --- a/apps/backend/internal/mcp/handlers/handlers.go +++ b/apps/backend/internal/mcp/handlers/handlers.go @@ -3103,9 +3103,7 @@ func (h *Handlers) setSessionRunning(ctx context.Context, taskID, sessionID stri h.logger.Warn("failed to update task state to IN_PROGRESS", zap.String("task_id", taskID), zap.Error(err)) - return - } - if !taskStateChanged { + } else if !taskStateChanged { h.logger.Debug("skipping stale clarification resume after session state changed", zap.String("task_id", taskID), zap.String("session_id", sessionID)) @@ -3113,34 +3111,51 @@ func (h *Handlers) setSessionRunning(ctx context.Context, taskID, sessionID stri } } - // Publish session state changed event - if h.eventBus != nil { - eventData := map[string]any{ - "task_id": taskID, - "session_id": sessionID, - "new_state": string(models.TaskSessionStateRunning), - } - if !updatedAt.IsZero() { - eventData["updated_at"] = updatedAt.UTC().Format(time.RFC3339Nano) - } else if persistedUpdatedAt, ok := h.sessionUpdatedAtForStateEvent(ctx, sessionID); ok { - eventData["updated_at"] = persistedUpdatedAt - } else { - h.logger.Warn("skipping session state_changed publish; could not load authoritative updated_at", - zap.String("session_id", sessionID)) - return - } - _ = h.eventBus.Publish(ctx, events.TaskSessionStateChanged, bus.NewEvent( + // Publish session state changed event after any task event for this task. + if h.eventBus == nil { + return + } + eventData := map[string]any{ + "task_id": taskID, + "session_id": sessionID, + "new_state": string(models.TaskSessionStateRunning), + } + if !updatedAt.IsZero() { + eventData["updated_at"] = updatedAt.UTC().Format(time.RFC3339Nano) + } else if persistedUpdatedAt, ok := h.sessionUpdatedAtForStateEvent(ctx, sessionID); ok { + eventData["updated_at"] = persistedUpdatedAt + } else { + h.logger.Warn("skipping session state_changed publish; could not load authoritative updated_at", + zap.String("session_id", sessionID)) + return + } + publish := func(publicationCtx context.Context) { + _ = h.eventBus.Publish(publicationCtx, events.TaskSessionStateChanged, bus.NewEvent( events.TaskSessionStateChanged, "mcp-handlers", eventData, )) } + if h.taskSvc != nil && taskID != "" { + h.taskSvc.PublishAfterTaskEvents(ctx, taskID, events.TaskSessionStateChanged, publish) + return + } + publish(ctx) } func (h *Handlers) setTaskInProgressForClarification( ctx context.Context, taskID, sessionID string, ) (bool, error) { + if h.taskSvc != nil { + return h.taskSvc.UpdateTaskStateIfSessionState( + ctx, + taskID, + sessionID, + models.TaskSessionStateRunning, + v1.TaskStateInProgress, + ) + } if updater, ok := h.taskRepo.(sessionOwnedTaskStateUpdater); ok { _, updated, err := updater.UpdateTaskStateIfSessionState( ctx, diff --git a/apps/backend/internal/mcp/handlers/handlers_test.go b/apps/backend/internal/mcp/handlers/handlers_test.go index 33224c7152..bbb7356dc8 100644 --- a/apps/backend/internal/mcp/handlers/handlers_test.go +++ b/apps/backend/internal/mcp/handlers/handlers_test.go @@ -15,6 +15,7 @@ import ( "github.com/kandev/kandev/internal/clarification" "github.com/kandev/kandev/internal/common/logger" "github.com/kandev/kandev/internal/db" + "github.com/kandev/kandev/internal/events" "github.com/kandev/kandev/internal/events/bus" "github.com/kandev/kandev/internal/orchestrator" "github.com/kandev/kandev/internal/orchestrator/executor" @@ -410,6 +411,182 @@ func TestSessionStateEventsIncludeUpdatedAt(t *testing.T) { } } +func TestSetSessionRunning_PublishesTaskStateBeforeSession(t *testing.T) { + svc, repo, eventBus := newTestTaskServiceWithEventBus(t) + const taskID = "task-clarification-order" + const sessionID = "session-clarification-order" + seedMCPHandlerSession(t, repo, taskID, sessionID, models.TaskSessionStateWaitingForInput) + require.NoError(t, repo.UpdateTaskState(context.Background(), taskID, v1.TaskStateReview)) + + var published []string + sub, err := eventBus.Subscribe(">", func(_ context.Context, event *bus.Event) error { + switch event.Type { + case events.TaskStateChanged, events.TaskSessionStateChanged: + published = append(published, event.Type) + } + return nil + }) + require.NoError(t, err) + t.Cleanup(func() { _ = sub.Unsubscribe() }) + + h := &Handlers{ + taskSvc: svc, + sessionRepo: repo, + taskRepo: repo, + eventBus: eventBus, + logger: testLogger(t).WithFields(), + } + h.setSessionRunning(context.Background(), taskID, sessionID) + + require.Equal(t, []string{events.TaskStateChanged, events.TaskSessionStateChanged}, published) + task, err := repo.GetTask(context.Background(), taskID) + require.NoError(t, err) + assert.Equal(t, v1.TaskStateInProgress, task.State) +} + +type failingClarificationTaskRepository struct { + repository.TaskRepository + err error +} + +func (r failingClarificationTaskRepository) UpdateTaskStateIfSessionState( + context.Context, + string, + string, + models.TaskSessionState, + v1.TaskState, +) (v1.TaskState, bool, error) { + return v1.TaskStateReview, false, r.err +} + +func newClarificationTaskService( + t *testing.T, + repo *sqliterepo.Repository, + tasks repository.TaskRepository, + eventBus *bus.MemoryEventBus, +) *service.Service { + t.Helper() + return service.NewService(service.Repos{ + Workspaces: repo, + Tasks: tasks, + TaskRepos: repo, + Workflows: repo, + Messages: repo, + Turns: repo, + Sessions: repo, + GitSnapshots: repo, + RepoEntities: repo, + Executors: repo, + Environments: repo, + Reviews: repo, + }, eventBus, testLogger(t), service.RepositoryDiscoveryConfig{}) +} + +func TestSetSessionRunning_PreservesSessionEventOnTaskServiceError(t *testing.T) { + _, repo, eventBus := newTestTaskServiceWithEventBus(t) + const taskID = "task-clarification-service-error" + const sessionID = "session-clarification-service-error" + seedMCPHandlerSession(t, repo, taskID, sessionID, models.TaskSessionStateWaitingForInput) + require.NoError(t, repo.UpdateTaskState(context.Background(), taskID, v1.TaskStateReview)) + + failingSvc := newClarificationTaskService(t, repo, failingClarificationTaskRepository{ + TaskRepository: repo, + err: errors.New("task state write failed"), + }, eventBus) + var published []string + sub, err := eventBus.Subscribe(">", func(_ context.Context, event *bus.Event) error { + published = append(published, event.Type) + return nil + }) + require.NoError(t, err) + t.Cleanup(func() { _ = sub.Unsubscribe() }) + + h := &Handlers{ + taskSvc: failingSvc, + sessionRepo: repo, + taskRepo: repo, + eventBus: eventBus, + logger: testLogger(t).WithFields(), + } + h.setSessionRunning(context.Background(), taskID, sessionID) + + require.Equal(t, []string{events.TaskSessionStateChanged}, published) + session, err := repo.GetTaskSession(context.Background(), sessionID) + require.NoError(t, err) + assert.Equal(t, models.TaskSessionStateRunning, session.State) +} + +func TestSetSessionRunning_QueuesSessionAfterBusyTaskPublication(t *testing.T) { + svc, repo, eventBus := newTestTaskServiceWithEventBus(t) + const taskID = "task-clarification-busy-queue" + const sessionID = "session-clarification-busy-queue" + seedMCPHandlerSession(t, repo, taskID, sessionID, models.TaskSessionStateWaitingForInput) + require.NoError(t, repo.UpdateTaskState(context.Background(), taskID, v1.TaskStateReview)) + + entered := make(chan struct{}) + release := make(chan struct{}) + var mu sync.Mutex + var published []string + first := true + sub, err := eventBus.Subscribe(">", func(_ context.Context, event *bus.Event) error { + mu.Lock() + published = append(published, event.Type) + block := first + first = false + mu.Unlock() + if block { + close(entered) + <-release + } + return nil + }) + require.NoError(t, err) + t.Cleanup(func() { _ = sub.Unsubscribe() }) + + ordinaryDone := make(chan struct{}) + go func() { + svc.PublishTaskUpdated(context.Background(), &models.Task{ + ID: taskID, + WorkspaceID: "ws-state-event", + Title: "ordinary", + WorkflowID: "wf-state-event", + WorkflowStepID: "step-state-event", + }) + close(ordinaryDone) + }() + <-entered + + h := &Handlers{ + taskSvc: svc, + sessionRepo: repo, + taskRepo: repo, + eventBus: eventBus, + logger: testLogger(t).WithFields(), + } + resumeDone := make(chan struct{}) + go func() { + h.setSessionRunning(context.Background(), taskID, sessionID) + close(resumeDone) + }() + select { + case <-resumeDone: + case <-time.After(time.Second): + t.Fatal("clarification resume waited for the busy task publication") + } + + close(release) + <-ordinaryDone + + mu.Lock() + got := append([]string(nil), published...) + mu.Unlock() + require.Equal(t, []string{ + events.TaskUpdated, + events.TaskStateChanged, + events.TaskSessionStateChanged, + }, got) +} + func TestHandleCreateTask_SubtaskMissingDescription(t *testing.T) { h := &Handlers{} msg := makeWSMessage(t, ws.ActionMCPCreateTask, map[string]interface{}{ diff --git a/apps/backend/internal/task/service/service_events.go b/apps/backend/internal/task/service/service_events.go index bea68160bf..3dc00bbb90 100644 --- a/apps/backend/internal/task/service/service_events.go +++ b/apps/backend/internal/task/service/service_events.go @@ -82,6 +82,24 @@ func (s *Service) PublishTaskStateChanged(ctx context.Context, task *models.Task s.publishTaskEvent(ctx, events.TaskStateChanged, task, &oldState) } +// PublishAfterTaskEvents appends a task-scoped publication behind any task +// events already being drained. Reentrant callers append without waiting, +// preserving FIFO order without deadlocking the active publication. +func (s *Service) PublishAfterTaskEvents( + ctx context.Context, + taskID, eventType string, + publish func(context.Context), +) { + if publish == nil { + return + } + if taskID == "" { + publish(ctx) + return + } + s.enqueueTaskPublication(ctx, taskID, eventType, publish) +} + // PublishTaskDeleted publishes a task.deleted event for the given task. // Used by cascade-delete callers (HandoffService.DeleteTaskTree) that // bypass Service.DeleteTask and therefore would otherwise leave WS diff --git a/apps/web/e2e/tests/chat/clarification.spec.ts b/apps/web/e2e/tests/chat/clarification.spec.ts index 909489a156..aa940f28ac 100644 --- a/apps/web/e2e/tests/chat/clarification.spec.ts +++ b/apps/web/e2e/tests/chat/clarification.spec.ts @@ -6,6 +6,7 @@ import type { ApiClient } from "../../helpers/api-client"; import { seedClarificationSession } from "../../helpers/clarification"; import { SessionPage } from "../../pages/session-page"; import { KanbanPage } from "../../pages/kanban-page"; +import { SidebarFilterPopoverPage } from "../../pages/sidebar-filter-popover"; /** * Seed a task + session with a clarification scenario and navigate to the session page. @@ -27,6 +28,12 @@ const PLAN_WITH_CLARIFICATION_SCRIPT = [ 'e2e:mcp:kandev:ask_user_question_kandev({"questions":[{"id":"db","prompt":"Which database should we use?","options":[{"label":"PostgreSQL","description":"Relational"},{"label":"SQLite","description":"Embedded"}]},{"id":"language","prompt":"Which language should we use?","options":[{"label":"Go","description":"Compiled"},{"label":"TypeScript","description":"Web"}]},{"id":"deploy","prompt":"How should we deploy?","options":[{"label":"Docker","description":"Containerized"},{"label":"Bare metal","description":"Direct"}]}]})', ].join("\n"); +const CLARIFICATION_WITH_POST_ANSWER_DELAY_SCRIPT = [ + 'e2e:mcp:kandev:ask_user_question_kandev({"questions":[{"id":"db","prompt":"Which database should we use?","options":[{"label":"PostgreSQL","description":"Relational"},{"label":"SQLite","description":"Embedded"}]}]})', + "e2e:delay(5000)", + 'e2e:message("The clarification answer resumed the active turn.")', +].join("\n"); + // Exercises the regular task-create dialog (New Task in the sidebar); run with office off. useRegularMode(); @@ -82,6 +89,61 @@ test.describe("Clarification flow", () => { await expect(session.chat).toContainText(/You answered|selected_option/); }); + test("moves answered task from Review to In progress without reload", async ({ + testPage, + apiClient, + seedData, + }) => { + const task = await apiClient.createTaskWithAgent( + seedData.workspaceId, + "Clarification State Group", + seedData.agentProfileId, + { + description: CLARIFICATION_WITH_POST_ANSWER_DELAY_SCRIPT, + workflow_id: seedData.workflowId, + workflow_step_id: seedData.startStepId, + repository_ids: [seedData.repositoryId], + }, + ); + if (!task.session_id) throw new Error("createTaskWithAgent did not return a session_id"); + + // The mock agent reaches the clarification call asynchronously. Wait for + // the owning session to park before opening the page, then explicitly + // establish the Review state that the answer should move it out of. + await expect + .poll( + async () => + (await apiClient.listTaskSessions(task.id)).sessions.find((s) => s.id === task.session_id) + ?.state, + { timeout: 30_000 }, + ) + .toBe("WAITING_FOR_INPUT"); + await apiClient.updateTaskState(task.id, "REVIEW"); + await expect.poll(async () => (await apiClient.getTask(task.id)).state).toBe("REVIEW"); + + await testPage.goto(`/t/${task.id}`); + const session = new SessionPage(testPage); + await session.waitForLoad(); + await expect(session.clarificationOverlay()).toBeVisible({ timeout: 30_000 }); + + const filters = new SidebarFilterPopoverPage(testPage); + await filters.open(); + await filters.setGroup("State"); + await filters.close(); + + const reviewGroup = session.sidebar.locator( + '[data-testid="sidebar-group-header"][data-group-key="REVIEW"]', + ); + await expect(reviewGroup).toBeVisible(); + + await session.clarificationOption("PostgreSQL").click(); + + await expect( + session.sidebar.locator('[data-testid="sidebar-group-header"][data-group-key="IN_PROGRESS"]'), + ).toBeVisible({ timeout: 10_000 }); + await expect(reviewGroup).toHaveCount(0); + }); + test("custom answer accepts multiple lines via Shift+Enter", async ({ testPage, apiClient, diff --git a/docs/decisions/2026-07-30-runtime-task-state-before-running-event.md b/docs/decisions/2026-07-30-runtime-task-state-before-running-event.md index 14afdddc68..14e7bc648a 100644 --- a/docs/decisions/2026-07-30-runtime-task-state-before-running-event.md +++ b/docs/decisions/2026-07-30-runtime-task-state-before-running-event.md @@ -20,8 +20,11 @@ before publishing `session.state_changed(RUNNING)`. The reconciliation runs only after the session compare-and-set succeeds and before its event publication. If it changes the task, the existing -`task.state_changed` event is therefore published first. Repeated stream events -for an already-`RUNNING` session retain the existing no-write fast path. +`task.state_changed` event is therefore published first. The follow-on session +event is appended to the same per-task publication FIFO, so an already-draining +task publication cannot be overtaken and the handler never waits reentrantly on +its own queue. Repeated stream events for an already-`RUNNING` session retain the +existing no-write fast path. Archive, terminal-session, clarification, cancellation, and Office guards remain authoritative. The executor-success reconciliation remains as a healing fallback. diff --git a/docs/plans/clarification-task-state-publication/plan.md b/docs/plans/clarification-task-state-publication/plan.md new file mode 100644 index 0000000000..28ff30a46b --- /dev/null +++ b/docs/plans/clarification-task-state-publication/plan.md @@ -0,0 +1,141 @@ +--- +spec: docs/specs/tasks/runtime-state-publication-order.md +created: 2026-08-02 +status: complete +--- + +# Implementation Plan: Clarification Task-State Publication + +## Overview + +Repair the direct `ask_user_question_kandev` answer path so its persisted task +transition is published before the already-correct `RUNNING` session event. +The confirmed root cause is that `setTaskInProgressForClarification` writes +through the raw SQLite task repository; the database reaches `IN_PROGRESS`, but +that repository does not own `task.state_changed`, so connected clients remain +on `REVIEW` until boot hydration reloads the row. Review follow-up also +requires preserving that order when another task publication is already +draining and reporting the persisted session transition if task reconciliation +fails. + +The fix reuses the task service's existing session-state-guarded update and +event publisher. It does not change task/session schemas, WebSocket payloads, +frontend merge rules, sidebar grouping, or clarification UI. + +## Backend + +### Publish the guarded clarification transition + +- Update `setTaskInProgressForClarification` in + `apps/backend/internal/mcp/handlers/handlers.go` to use the already-wired task + service for the production guarded `REVIEW -> IN_PROGRESS` transition. +- Reuse `task.Service.UpdateTaskStateIfSessionState` with expected session state + `RUNNING`. This keeps cancellation/archive guards and emits the canonical + rich `task.state_changed` payload before the handler publishes + `session.state_changed(RUNNING)`. Enqueue the session event through the same + per-task FIFO so a busy queue cannot reverse the order and reentrant + subscribers do not deadlock. +- If guarded task reconciliation returns an error, log it and still publish the + authoritative `session.state_changed(RUNNING)` event. Retain the early return + for a clean `taskStateChanged == false` stale-state guard. +- Preserve the existing repository fallback for narrow handler tests or + alternate construction where no task service is available. +- Do not infer task state from session state in the frontend or add a second + event publisher in the MCP handler. + +## Frontend + +No frontend production change is planned. The desktop sidebar and mobile task +drawer already share the task store and State grouping. Once the missing task +event arrives, both consume the existing authoritative `task.state` update. + +Mobile parity is state-only: there is no composition, navigation, scrolling, +safe-area, pointer, or touch-target change. Existing mobile clarification +coverage continues to exercise answering a question; the backend ordering test +proves the shared state event, so no duplicate mobile layout test is required. + +## Tests + +- **What:** answering a clarification from `REVIEW` / `WAITING_FOR_INPUT` + persists `IN_PROGRESS` and publishes `task.state_changed` before + `task_session.state_changed(RUNNING)`. + **File:** `apps/backend/internal/mcp/handlers/handlers_test.go`. + **How:** use the real SQLite repository, real task service, and shared memory + event bus; record lifecycle events through one wildcard subscription and + assert durable state plus exact event order. The regression test must first + fail with only the session event on the current implementation. +- **What:** a pre-existing task publication cannot be overtaken by the + clarification task/session pair, and a task-service reconciliation error + still emits the session event. + **File:** `apps/backend/internal/mcp/handlers/handlers_test.go`. + **How:** block one task publication with a channel barrier, resume the + clarification concurrently, and assert the final FIFO order; inject a task + repository error and assert the session event remains observable. +- **What:** the existing clarification cancellation races remain fail-closed. + **File:** `apps/backend/internal/mcp/handlers/clarification_pause_test.go`. + **How:** rerun the focused coordinator-stop race cases alongside the new test, + including the package with `-race`. + +## E2E Tests + +- **Scenario:** **GIVEN** the sidebar is grouped by State and a task is in + `Review` on an open clarification, **WHEN** the user answers while the mock + agent intentionally remains active, **THEN** the live sidebar moves the task + to `In progress` without reload. +- **File:** `apps/web/e2e/tests/chat/clarification.spec.ts`. +- **What to verify:** use a test-local mock-agent script with a bounded post-answer + delay, select State grouping through the existing sidebar page object, and + assert the `REVIEW` group is replaced by the `IN_PROGRESS` group immediately + after the answer. No fixed wait is used in the Playwright assertion. + +## Verification Results + +- Backend targeted tests passed: `go test -count=1 -run + 'Test(SetSessionRunning_PublishesTaskStateBeforeSession|SessionStateEventsIncludeUpdatedAt|HandleAskUserQuestion_CoordinatorStopWinsRunningTransition|HandleAskUserQuestion_CoordinatorStopWinsAfterRunningTransition)$' + ./internal/mcp/handlers` — 7 passed. +- Backend race coverage passed: `go test -race -count=1 -run + 'Test(SetSessionRunning_PublishesTaskStateBeforeSession|HandleAskUserQuestion_CoordinatorStopWinsRunningTransition|HandleAskUserQuestion_CoordinatorStopWinsAfterRunningTransition)$' + ./internal/mcp/handlers` — 3 passed. +- Focused E2E passed: `pnpm e2e:run tests/chat/clarification.spec.ts -- --grep + 'moves answered task from Review to In progress without reload'` — 1 passed. +- The E2E setup waits for the mock session's `WAITING_FOR_INPUT` state and + establishes `REVIEW` before navigation, keeping the pre-answer sidebar group + deterministic without adding production hooks or fixed assertion waits. +- Review follow-up targeted tests passed in normal and race modes for the + busy-queue ordering and task-service-error paths. The changed-file + `golangci-lint` check passed with no issues. + +## Implementation Waves And Parallel Candidates + +Wave 1: + +- [x] [Task 01: Publish clarification task state](task-01-publish-clarification-task-state.md) — completed + +Wave 2: + +- [x] [Task 02: Prove live sidebar regrouping](task-02-prove-live-sidebar-regrouping.md) — completed + +Both tasks are sequential. The E2E proof depends on the backend event repair; +the waves do not authorize subagents. + +Wave 3: + +- [x] [Task 03: Address ordered publication review findings](task-03-address-publication-review.md) — completed + +Task 03 is sequential with the backend implementation and documentation +corrections. + +## Risks + +- The task event is intentionally added before the existing session event; + consumers already depend on this ordering contract for other runtime starts. +- The task service performs canonical event enrichment reads. A publication + failure remains logged by the service and must not fabricate frontend state. +- The E2E mock agent must remain active long enough to observe `IN_PROGRESS` + without turning the assertion into a timing sleep. + +## Out of Scope + +- Changing the clarification waiting transition to `REVIEW`. +- Changing sidebar grouping or deriving task state from session events. +- Changing Office task-state ownership, schemas, event names, or payloads. diff --git a/docs/plans/clarification-task-state-publication/task-01-publish-clarification-task-state.md b/docs/plans/clarification-task-state-publication/task-01-publish-clarification-task-state.md new file mode 100644 index 0000000000..0e6d6f843d --- /dev/null +++ b/docs/plans/clarification-task-state-publication/task-01-publish-clarification-task-state.md @@ -0,0 +1,77 @@ +--- +id: "01-publish-clarification-task-state" +title: "Publish clarification task state" +status: completed +wave: 1 +depends_on: [] +plan: "plan.md" +spec: "../../specs/tasks/runtime-state-publication-order.md" +--- + +# Task 01: Publish clarification task state + +## Acceptance + +- An answered clarification persists an eligible task as `IN_PROGRESS` and + publishes `task.state_changed` before + `task_session.state_changed(RUNNING)`. +- The update remains guarded by the owning session's authoritative `RUNNING` + state, so existing coordinator-stop races cannot revive a cancelled task. +- No new event, payload field, frontend-derived state, or schema is introduced. + +## Verification + +```bash +(cd apps/backend && go test -count=1 \ + -run 'Test(SetSessionRunning_PublishesTaskStateBeforeSession|SessionStateEventsIncludeUpdatedAt|HandleAskUserQuestion_CoordinatorStopWinsRunningTransition|HandleAskUserQuestion_CoordinatorStopWinsAfterRunningTransition)$' \ + ./internal/mcp/handlers) +(cd apps/backend && go test -race -count=1 \ + -run 'Test(SetSessionRunning_PublishesTaskStateBeforeSession|HandleAskUserQuestion_CoordinatorStopWinsRunningTransition|HandleAskUserQuestion_CoordinatorStopWinsAfterRunningTransition)$' \ + ./internal/mcp/handlers) +``` + +## Files likely touched + +- `apps/backend/internal/mcp/handlers/handlers.go` +- `apps/backend/internal/mcp/handlers/handlers_test.go` +- `apps/backend/internal/mcp/handlers/clarification_pause_test.go` only if the + existing race fixture needs a non-behavioral adjustment for the canonical + task-service path + +## Dependencies + +None. + +## Parallelism + +Sequential. Production behavior and its regression test share the same handler +lifecycle seam. + +## Inputs + +- Spec `What`, `State machine`, `Failure modes`, and clarification scenario. +- Plan section `Publish the guarded clarification transition`. +- `task.Service.UpdateTaskStateIfSessionState` as the canonical guarded writer + and event publisher. +- Existing clarification coordinator-stop race tests. + +## Output contract + +Report the RED failure, GREEN results, files changed, exact commands and +outcomes, remaining risks, and synchronized task/plan status. + +## Results + +- RED: `cd apps/backend && go test ./internal/mcp/handlers -run + '^TestSetSessionRunning_PublishesTaskStateBeforeSession$' -count=1` failed with + only `task_session.state_changed` observed. +- GREEN: `cd apps/backend && go test -count=1 -run + 'Test(SetSessionRunning_PublishesTaskStateBeforeSession|SessionStateEventsIncludeUpdatedAt|HandleAskUserQuestion_CoordinatorStopWinsRunningTransition|HandleAskUserQuestion_CoordinatorStopWinsAfterRunningTransition)$' + ./internal/mcp/handlers` — 7 passed. +- GREEN: `cd apps/backend && go test -race -count=1 -run + 'Test(SetSessionRunning_PublishesTaskStateBeforeSession|HandleAskUserQuestion_CoordinatorStopWinsRunningTransition|HandleAskUserQuestion_CoordinatorStopWinsAfterRunningTransition)$' + ./internal/mcp/handlers` — 3 passed. +- Files changed: `apps/backend/internal/mcp/handlers/handlers.go` and + `apps/backend/internal/mcp/handlers/handlers_test.go`. +- Temporary repro test was removed; no generated artifacts or external side + effects remain. diff --git a/docs/plans/clarification-task-state-publication/task-02-prove-live-sidebar-regrouping.md b/docs/plans/clarification-task-state-publication/task-02-prove-live-sidebar-regrouping.md new file mode 100644 index 0000000000..df906e5244 --- /dev/null +++ b/docs/plans/clarification-task-state-publication/task-02-prove-live-sidebar-regrouping.md @@ -0,0 +1,69 @@ +--- +id: "02-prove-live-sidebar-regrouping" +title: "Prove live sidebar regrouping" +status: completed +wave: 2 +depends_on: ["01-publish-clarification-task-state"] +plan: "plan.md" +spec: "../../specs/tasks/runtime-state-publication-order.md" +--- + +# Task 02: Prove live sidebar regrouping + +## Acceptance + +- A State-grouped desktop sidebar moves an answered clarification task from + `Review` to `In progress` while the same agent turn remains active, without a + reload. +- The test uses existing sidebar and clarification controls and adds no + frontend production behavior or test-only production hooks. +- Mobile parity remains covered by the shared task store plus the existing + mobile clarification-answer flow; no responsive composition changes. + +## Verification + +```bash +(cd apps && pnpm install --frozen-lockfile) +(cd apps/web && pnpm e2e:run tests/chat/clarification.spec.ts -- \ + --grep 'moves answered task from Review to In progress without reload' +) +``` + +## Files likely touched + +- `apps/web/e2e/tests/chat/clarification.spec.ts` + +## Dependencies + +- Task 01: the backend must publish the live task-state event. + +## Parallelism + +Sequential. The test is expected to stay red until Task 01 lands. + +## Inputs + +- Spec clarification-answer and State-grouped sidebar scenarios. +- Plan `E2E Tests` and mobile-parity note. +- `apps/web/e2e/helpers/clarification.ts` and + `apps/web/e2e/pages/sidebar-filter-popover.ts` as existing patterns. + +## Output contract + +Report the failing pre-fix behavior if captured, the final Playwright result and +test count, generated failure/evidence artifacts, cleanup, and synchronized +task/plan status. + +## Results + +- Focused Playwright verification passed: `cd apps/web && pnpm e2e:run + tests/chat/clarification.spec.ts -- --grep 'moves answered task from Review to + In progress without reload'` — 1 passed. +- The test waits for the mock session to reach `WAITING_FOR_INPUT`, sets the + persisted task state to `REVIEW`, then verifies the State-grouped sidebar + moves the task to `IN_PROGRESS` immediately after answering, without reload. +- An earlier attempt exposed nondeterministic setup timing (the task was still + `IN_PROGRESS` before the mock clarification parked); the final setup polls + the session state and makes the pre-answer state explicit. +- The E2E runner cleaned its generated test-results, blob-report, and shard + artifacts; no generated files remain in the worktree. diff --git a/docs/plans/clarification-task-state-publication/task-03-address-publication-review.md b/docs/plans/clarification-task-state-publication/task-03-address-publication-review.md new file mode 100644 index 0000000000..8e1424475d --- /dev/null +++ b/docs/plans/clarification-task-state-publication/task-03-address-publication-review.md @@ -0,0 +1,55 @@ +--- +id: "03-address-publication-review" +title: "Address ordered publication review findings" +status: completed +wave: 3 +depends_on: ["01-publish-clarification-task-state"] +plan: "plan.md" +spec: "../../specs/tasks/runtime-state-publication-order.md" +--- + +# Task 03: Address ordered publication review findings + +## Acceptance + +- A clarification task-state event and its `RUNNING` session event remain in + order when another publication for the same task is already draining. +- A task-service reconciliation error does not suppress the authoritative + `session.state_changed(RUNNING)` event; a clean stale guard still suppresses + the pair. +- Validation commands in Task 01 and Task 02 can be pasted and run from the + repository root without accumulating `cd` state. + +## Files + +- `apps/backend/internal/mcp/handlers/handlers.go` +- `apps/backend/internal/mcp/handlers/handlers_test.go` +- `apps/backend/internal/task/service/service_events.go` +- `docs/specs/tasks/runtime-state-publication-order.md` +- `docs/decisions/2026-07-30-runtime-task-state-before-running-event.md` +- `docs/plans/clarification-task-state-publication/plan.md` +- `docs/plans/clarification-task-state-publication/task-01-publish-clarification-task-state.md` +- `docs/plans/clarification-task-state-publication/task-02-prove-live-sidebar-regrouping.md` + +## Verification + +```bash +(cd apps/backend && go test -count=1 -run 'Test(SetSessionRunning_PublishesTaskStateBeforeSession|SetSessionRunning_PreservesSessionEventOnTaskServiceError|SetSessionRunning_QueuesSessionAfterBusyTaskPublication)$' ./internal/mcp/handlers) +(cd apps/backend && go test -race -count=1 -run 'Test(SetSessionRunning_PublishesTaskStateBeforeSession|SetSessionRunning_PreservesSessionEventOnTaskServiceError|SetSessionRunning_QueuesSessionAfterBusyTaskPublication)$' ./internal/mcp/handlers) +``` + +## Parallelism + +Sequential. The handler implementation and both regression tests share the same +publication lifecycle seam. + +## Results + +- RED: the busy-queue regression observed `task.updated`, then + `task_session.state_changed`, then `task.state_changed`; the task-service + error regression emitted no session event. +- GREEN: normal and race-focused handler tests passed for all three + `setSessionRunning` scenarios. +- GREEN: full `internal/mcp/handlers` and `internal/task/service` packages + passed in normal and race modes. +- GREEN: changed-file `golangci-lint` passed with no issues. diff --git a/docs/plans/parent-child-task-stop/task-05-integration-regression.md b/docs/plans/parent-child-task-stop/task-05-integration-regression.md index e77622d3d7..6c44e7a7f1 100644 --- a/docs/plans/parent-child-task-stop/task-05-integration-regression.md +++ b/docs/plans/parent-child-task-stop/task-05-integration-regression.md @@ -3,7 +3,13 @@ id: "05-integration-regression" title: "Integrated stop regression" status: done wave: 5 -depends_on: ["01-execution-stop-semantics", "02-coordinator-stop-operation", "03-mcp-stop-handler", "04-task-mcp-tool"] +depends_on: + [ + "01-execution-stop-semantics", + "02-coordinator-stop-operation", + "03-mcp-stop-handler", + "04-task-mcp-tool", + ] plan: "plan.md" spec: "../../specs/tasks/parent-child-task-stop.md" --- @@ -58,5 +64,11 @@ files changed, tests run, blockers, and any unverified process-cleanup risks. - Verified no replacement prompt or queued message is created, runtime stop completes after release, and a repeat returns `not_running` without mutation. - Passed the focused test, 20 repeated runs, and the focused race-detector run. +- After CI exposed a `RUNNING`-versus-stream ordering race, added a deterministic + `message.added` barrier for the persisted initial tool call before capturing + the pre-stop message baseline. The strict no-new-message assertion remains; + no sleep, retry, or relaxed expectation was added. +- Passed 300 race-detector repetitions across `-cpu=1,2,4`, the complete + integration package, and the full 133-package shard-2 race test. - Remaining lifecycle boundary: the test proves the simulated runtime reaches stopped status, not operating-system cleanup for every production provider. diff --git a/docs/specs/tasks/runtime-state-publication-order.md b/docs/specs/tasks/runtime-state-publication-order.md index 44ea899b71..14ac4f318e 100644 --- a/docs/specs/tasks/runtime-state-publication-order.md +++ b/docs/specs/tasks/runtime-state-publication-order.md @@ -19,8 +19,14 @@ though the task's workflow step and runtime are already active. - For a non-Office, unarchived task whose session transitions into `RUNNING`, Kandev reconciles the persisted task state to `IN_PROGRESS` before publishing that session's `session.state_changed` event. +- The same ordering applies when an answered `ask_user_question_kandev` call + resumes the still-open agent turn directly, without launching a replacement + prompt or waiting for a later runtime stream event. - When reconciliation changes the task state, observers receive `task.state_changed` before `session.state_changed` announces `RUNNING`. +- The task and session publications share the task's existing per-task FIFO, so + a session event cannot overtake an earlier task event when another + task-scoped publication is already draining. - The WebSocket gateway consumes both lifecycle notifications through one ordered NATS-style subscription, preserving that order when the event bus is remote; the in-memory event bus supports the same wildcard semantics. @@ -61,6 +67,10 @@ The ordering contract is defined by - When a prompt or runtime transition successfully changes an owning session to `RUNNING`, the same lifecycle operation reconciles the task to `IN_PROGRESS` before exposing the new session state to clients. +- When a clarification answer wakes a `WAITING_FOR_INPUT` session inside the + open MCP call, the clarification lifecycle performs the guarded task + reconciliation and publishes its task event before publishing the + `RUNNING` session event. - When the turn settles and no sibling session is working, existing behavior returns the task to `REVIEW`. @@ -72,6 +82,9 @@ The ordering contract is defined by persisted, Kandev logs the error and still reports the truthful session state. The existing executor-success reconciliation remains available to heal the task state. +- If the task publication queue is already draining, the task-state and + session-state events append in lifecycle order without waiting reentrantly on + the active publication. - Office task status remains controlled by the Office lifecycle and is not promoted by this runtime reconciliation. @@ -81,6 +94,12 @@ The ordering contract is defined by `WAITING_FOR_INPUT`, **WHEN** the session accepts a new prompt and transitions to `RUNNING`, **THEN** `tasks.state` is `IN_PROGRESS` and `task.state_changed` is published before `session.state_changed(RUNNING)`. +- **GIVEN** a non-Office task in `REVIEW` whose owning session is + `WAITING_FOR_INPUT` inside an open `ask_user_question_kandev` call, **WHEN** + the user answers and the same agent turn resumes, **THEN** the persisted task + becomes `IN_PROGRESS`, WebSocket observers receive `task.state_changed` + before `session.state_changed(RUNNING)`, and the State-grouped sidebar moves + the task out of `Review` without a page reload. - **GIVEN** the sidebar is grouped by State, **WHEN** it observes a session transition to `RUNNING`, **THEN** the task is not rendered with a running spinner inside the `Review` group.