fix: publish clarification task state updates (#2174)

This commit is contained in:
Carlos Florêncio
2026-08-02 20:42:57 +01:00
committed by GitHub
parent 62d4522e0f
commit f51a83e4d3
13 changed files with 811 additions and 39 deletions
+123 -16
View File
@@ -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
@@ -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)
+35 -20
View File
@@ -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,
@@ -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{}{
@@ -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
@@ -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,
@@ -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.
@@ -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.
@@ -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.
@@ -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.
@@ -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.
@@ -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.
@@ -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.