From c8cacc562758e36b4a367b1e0d52d00c50105914 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Carlos=20Flor=C3=AAncio?= Date: Wed, 16 Sep 2026 22:57:09 +0100 Subject: [PATCH] fix: wake MCP queued messages (#3716) (#3737) --- .../backend/internal/mcp/handlers/handlers.go | 4 + .../internal/mcp/handlers/handlers_test.go | 3 + .../handlers/message_task_readiness_test.go | 370 ++++++++++++++++++ .../mcp/handlers/message_task_test.go | 7 + .../orchestrator/resume_prompt_queue_test.go | 105 +++++ docs/plans/mcp-queued-message-wakeup/plan.md | 178 +++++++++ .../task-01-wake-mcp-queue.md | 139 +++++++ .../parent-child-message-interrupt.md | 88 +++++ 8 files changed, 894 insertions(+) create mode 100644 apps/backend/internal/mcp/handlers/message_task_readiness_test.go create mode 100644 docs/plans/mcp-queued-message-wakeup/plan.md create mode 100644 docs/plans/mcp-queued-message-wakeup/task-01-wake-mcp-queue.md create mode 100644 docs/specs/tasks/system-design/parent-child-message-interrupt.md diff --git a/apps/backend/internal/mcp/handlers/handlers.go b/apps/backend/internal/mcp/handlers/handlers.go index 2a3e2bf1b1..6a967467f2 100644 --- a/apps/backend/internal/mcp/handlers/handlers.go +++ b/apps/backend/internal/mcp/handlers/handlers.go @@ -189,6 +189,9 @@ type SessionLauncher interface { ProcessOnTurnStart(ctx context.Context, taskID, sessionID string) (orchestrator.ProcessOnTurnStartResult, error) QueueUserPrompt(ctx context.Context, taskID, sessionID, prompt, model string, planMode bool, attachments []v1.MessageAttachment, metadata map[string]interface{}, userMessageRecorded bool) error GetMessageQueue() *messagequeue.Service + // CheckQueueAdmissionReadiness rechecks automatic dispatch for the exact + // session incarnation admitted by a queue operation. + CheckQueueAdmissionReadiness(context.Context, messagequeue.QueueSessionIdentity) // QueueAndInterruptForPeerMessage atomically queues prompt for sessionID // then interrupts the session's in-flight turn to dispatch it right // away, bypassing FIFO order. Used only by queueThenInterruptTaskMessage @@ -3411,6 +3414,7 @@ func (h *Handlers) queueTaskMessage(ctx context.Context, taskID string, session } return taskMessageDispatchResult{}, fmt.Errorf("failed to queue message: %w", err) } + h.sessionLauncher.CheckQueueAdmissionReadiness(ctx, identity) h.publishQueueStatusEvent(ctx, identity, queue) return taskMessageDispatchResult{status: taskMessageStatusQueued, sessionID: session.ID, queuedEntryID: queued.ID}, nil } diff --git a/apps/backend/internal/mcp/handlers/handlers_test.go b/apps/backend/internal/mcp/handlers/handlers_test.go index 36afa1b06c..e32c1cc2ea 100644 --- a/apps/backend/internal/mcp/handlers/handlers_test.go +++ b/apps/backend/internal/mcp/handlers/handlers_test.go @@ -2196,6 +2196,9 @@ func (m *mockSessionLauncher) QueueUserPrompt(context.Context, string, string, s } func (m *mockSessionLauncher) GetMessageQueue() *messagequeue.Service { return nil } +func (m *mockSessionLauncher) CheckQueueAdmissionReadiness(context.Context, messagequeue.QueueSessionIdentity) { +} + // QueueAndInterruptForPeerMessage always reports a successful immediate // dispatch with a fake queued entry; tests exercising other outcomes // (failure, not-dispatched) use fakeOrchestrator in message_task_test.go diff --git a/apps/backend/internal/mcp/handlers/message_task_readiness_test.go b/apps/backend/internal/mcp/handlers/message_task_readiness_test.go new file mode 100644 index 0000000000..ba2a47cd58 --- /dev/null +++ b/apps/backend/internal/mcp/handlers/message_task_readiness_test.go @@ -0,0 +1,370 @@ +package handlers + +import ( + "context" + "encoding/json" + "sync" + "testing" + "time" + + "github.com/kandev/kandev/internal/events/bus" + "github.com/kandev/kandev/internal/orchestrator" + "github.com/kandev/kandev/internal/orchestrator/executor" + "github.com/kandev/kandev/internal/orchestrator/messagequeue" + "github.com/kandev/kandev/internal/orchestrator/scheduler" + "github.com/kandev/kandev/internal/task/models" + sqliterepo "github.com/kandev/kandev/internal/task/repository/sqlite" + "github.com/kandev/kandev/pkg/api/v1" + ws "github.com/kandev/kandev/pkg/websocket" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// @covers AC-TASKS-PARENT-CHILD-MESSAGE-INTERRUPT-001.1 +func TestMessageTaskReadiness_ReadyBeforeAdmission(t *testing.T) { + ctx := context.Background() + svc, repo := newTestTaskService(t) + sender, target, session := seedTaskWithSession(t, svc, repo, models.TaskSessionStateRunning) + h, orch := newMessageTaskHandler(t, svc, repo) + + identity, err := orch.queue.ResolveSessionIdentity(ctx, target.ID, session.ID) + require.NoError(t, err) + require.NoError(t, repo.UpdateTaskSessionState(ctx, session.ID, models.TaskSessionStateWaitingForInput, "")) + + result, err := h.dispatchTaskMessage( + ctx, + target.ID, + session, + "wake after stale state", + map[string]interface{}{"sender_task_id": sender.ID}, + false, + false, + ) + + require.NoError(t, err) + assert.Equal(t, taskMessageStatusQueued, result.status) + assert.Equal(t, 1, orch.queue.GetStatus(ctx, session.ID).Count) + require.Len(t, orch.readinessCalls, 1) + assert.Equal(t, identity, orch.readinessCalls[0]) +} + +// @covers AC-TASKS-PARENT-CHILD-MESSAGE-INTERRUPT-001.1 +func TestMessageTaskReadiness_PreparedBusySession(t *testing.T) { + ctx := context.Background() + svc, repo := newTestTaskService(t) + _, target, session := seedTaskWithSession(t, svc, repo, models.TaskSessionStateRunning) + h, orch := newMessageTaskHandler(t, svc, repo) + + identity, err := orch.queue.ResolveSessionIdentity(ctx, target.ID, session.ID) + require.NoError(t, err) + result, err := h.dispatchPreparedTaskMessage( + ctx, + target.ID, + session, + "prepared branch", + map[string]interface{}{"sender_task_id": "sender"}, + ) + + require.NoError(t, err) + assert.Equal(t, taskMessageStatusQueued, result.status) + assert.Equal(t, 1, orch.queue.GetStatus(ctx, session.ID).Count) + require.Len(t, orch.readinessCalls, 1) + assert.Equal(t, identity, orch.readinessCalls[0]) +} + +// @covers AC-TASKS-PARENT-CHILD-MESSAGE-INTERRUPT-001.1 +func TestMessageTaskReadiness_RejectedAdmission(t *testing.T) { + ctx := context.Background() + svc, repo := newTestTaskService(t) + _, target, session := seedTaskWithSession(t, svc, repo, models.TaskSessionStateRunning) + h, orch := newMessageTaskHandler(t, svc, repo) + orch.queue.SetMaxPerSession(1) + orch.queue.SetMergeEnabled(false) + + identity, err := orch.queue.ResolveSessionIdentity(ctx, target.ID, session.ID) + require.NoError(t, err) + _, err = orch.queue.QueueMessageWithMetadataForSession( + ctx, identity, "existing work", "", messagequeue.QueuedByAgent, false, nil, nil, + ) + require.NoError(t, err) + + _, err = h.dispatchTaskMessage( + ctx, + target.ID, + session, + "rejected work", + map[string]interface{}{"sender_task_id": "sender"}, + false, + false, + ) + + var queueErr *queueFullDispatchError + require.ErrorAs(t, err, &queueErr) + assert.Empty(t, orch.readinessCalls) + assert.Equal(t, 1, orch.queue.GetStatus(ctx, session.ID).Count) +} + +// @covers AC-TASKS-PARENT-CHILD-MESSAGE-INTERRUPT-001.1 +func TestMessageTaskReadiness_UnrelatedSender(t *testing.T) { + svc, repo := newTestTaskService(t) + sender, target, session := seedTaskWithSession(t, svc, repo, models.TaskSessionStateRunning) + h, orch := newMessageTaskHandler(t, svc, repo) + + msg := makeWSMessage(t, ws.ActionMCPMessageTask, senderPayload(target.ID, "queued peer message", sender.ID)) + resp, err := h.handleMessageTask(context.Background(), msg) + + require.NoError(t, err) + var payload map[string]interface{} + require.NoError(t, json.Unmarshal(resp.Payload, &payload)) + assert.Equal(t, taskMessageStatusQueued, payload[stopTaskStatusKey]) + assert.Empty(t, orch.interruptCalls) + require.Len(t, orch.readinessCalls, 1) + assert.Equal(t, session.ID, orch.readinessCalls[0].SessionID) + assert.Equal(t, target.ID, orch.readinessCalls[0].TaskID) + entry := orch.queue.GetStatus(context.Background(), session.ID).Entries[0] + assert.Equal(t, sender.ID, entry.Metadata["sender_task_id"]) +} + +// @covers AC-TASKS-PARENT-CHILD-MESSAGE-INTERRUPT-001.1 +func TestMessageTaskReadiness_Delivery(t *testing.T) { + ctx := context.Background() + svc, repo := newTestTaskService(t) + sender, target, session := seedTaskWithSession(t, svc, repo, models.TaskSessionStateRunning) + require.NoError(t, repo.UpdateTaskState(ctx, target.ID, v1.TaskStateInProgress)) + + accepted := make(chan mcpReadinessAcceptedPrompt, 1) + completed := make(chan struct{}) + queueRepo := &mcpReadinessQueueRepository{ + Repository: messagequeue.NewMemoryRepositoryWithAuthority( + func(ctx context.Context, taskID, sessionID string) (messagequeue.QueueSessionIdentity, error) { + current, err := repo.GetTaskSession(ctx, sessionID) + if err != nil { + return messagequeue.QueueSessionIdentity{}, err + } + if current == nil || current.TaskID != taskID || current.QueueIncarnationID == "" { + return messagequeue.QueueSessionIdentity{}, messagequeue.ErrSessionIdentityMismatch + } + return messagequeue.QueueSessionIdentity{ + TaskID: taskID, SessionID: sessionID, + SessionIncarnationID: current.QueueIncarnationID, + }, nil + }, + ), + beforeFirstResolve: func() { + require.NoError(t, repo.UpdateTaskSessionState(ctx, session.ID, models.TaskSessionStateWaitingForInput, "")) + require.NoError(t, repo.UpsertExecutorRunning(ctx, &models.ExecutorRunning{ + ID: "exec-row-" + session.ID, + SessionID: session.ID, + TaskID: target.ID, + Status: "running", + Resumable: true, + AgentExecutionID: "exec-" + session.ID, + })) + }, + completed: completed, + } + queue := messagequeue.NewService(queueRepo, messagequeue.DefaultMaxPerSession, testLogger(t)) + agentManager := &mcpReadinessAgentManager{accepted: accepted} + log := testLogger(t) + eventBus := bus.NewMemoryEventBus(log) + t.Cleanup(func() { eventBus.Close() }) + realOrchestrator := orchestrator.NewService( + orchestrator.DefaultServiceConfig(), + eventBus, + agentManager, + &mcpReadinessSchedulerTaskRepo{repo: repo}, + repo, + nil, + nil, + queue, + log, + ) + h := &Handlers{ + taskSvc: svc, + sessionRepo: repo, + sessionLauncher: realOrchestrator, + logger: log.WithFields(), + } + + resp, err := h.handleMessageTask(ctx, makeWSMessage( + t, + ws.ActionMCPMessageTask, + senderPayload(target.ID, "deliver through the orchestrator", sender.ID), + )) + require.NoError(t, err) + var payload map[string]interface{} + require.NoError(t, json.Unmarshal(resp.Payload, &payload)) + assert.Equal(t, taskMessageStatusQueued, payload[stopTaskStatusKey]) + + select { + case prompt := <-accepted: + assert.Equal(t, "exec-sess-1", prompt.ExecutionID) + assert.Contains(t, prompt.Prompt, "deliver through the orchestrator") + assert.Contains(t, prompt.Prompt, "") + case <-time.After(time.Second): + t.Fatal("timed out waiting for provider acceptance") + } + select { + case <-completed: + case <-time.After(time.Second): + t.Fatal("timed out waiting for queued delivery cleanup") + } + assert.Equal(t, 0, queue.GetStatus(ctx, session.ID).Count) +} + +type mcpReadinessAcceptedPrompt struct { + ExecutionID string + Prompt string +} + +type mcpReadinessAgentManager struct { + executor.AgentManagerClient + accepted chan<- mcpReadinessAcceptedPrompt +} + +func (m *mcpReadinessAgentManager) IsAgentRunningForSession(context.Context, string) bool { + return true +} + +func (m *mcpReadinessAgentManager) IsAgentReadyForPrompt(context.Context, string) bool { + return true +} + +func (m *mcpReadinessAgentManager) IsPassthroughSession(context.Context, string) bool { + return false +} + +func (m *mcpReadinessAgentManager) GetExecutionIDForSession(context.Context, string) (string, error) { + return "exec-sess-1", nil +} + +func (m *mcpReadinessAgentManager) PromptAgent( + _ context.Context, + _ string, + _ string, + _ []v1.MessageAttachment, + _ bool, +) (*executor.PromptResult, error) { + return &executor.PromptResult{}, nil +} + +func (m *mcpReadinessAgentManager) PromptAgentWithDispatchCallback( + ctx context.Context, + executionID, prompt string, + attachments []v1.MessageAttachment, + _ bool, + onDispatched func(), +) (*executor.PromptResult, error) { + if onDispatched != nil { + onDispatched() + } + select { + case m.accepted <- mcpReadinessAcceptedPrompt{ExecutionID: executionID, Prompt: prompt}: + case <-ctx.Done(): + return nil, ctx.Err() + } + return &executor.PromptResult{}, nil +} + +type mcpReadinessQueueRepository struct { + messagequeue.Repository + beforeFirstResolve func() + beforeResolveOnce sync.Once + completed chan<- struct{} + completedOnce sync.Once +} + +func (r *mcpReadinessQueueRepository) ResolveSessionIdentity( + ctx context.Context, + taskID, sessionID string, +) (messagequeue.QueueSessionIdentity, error) { + identity, err := r.Repository.ResolveSessionIdentity(ctx, taskID, sessionID) + if err == nil && r.beforeFirstResolve != nil { + r.beforeResolveOnce.Do(r.beforeFirstResolve) + } + return identity, err +} + +func (r *mcpReadinessQueueRepository) ListPendingQueueDispatches( + context.Context, +) ([]messagequeue.PendingQueueDispatch, error) { + return nil, nil +} + +func (r *mcpReadinessQueueRepository) MarkPendingQueueDispatchAccepted( + context.Context, + *messagequeue.QueuedMessage, +) error { + return nil +} + +func (r *mcpReadinessQueueRepository) DeletePendingQueueDispatch( + ctx context.Context, + message *messagequeue.QueuedMessage, +) error { + if delegate, ok := r.Repository.(interface { + DeletePendingQueueDispatch(context.Context, *messagequeue.QueuedMessage) error + }); ok { + if err := delegate.DeletePendingQueueDispatch(ctx, message); err != nil { + return err + } + } + if r.completed != nil { + r.completedOnce.Do(func() { close(r.completed) }) + } + return nil +} + +type mcpReadinessSchedulerTaskRepo struct { + repo *sqliterepo.Repository +} + +var _ scheduler.TaskRepository = (*mcpReadinessSchedulerTaskRepo)(nil) + +func (r *mcpReadinessSchedulerTaskRepo) GetTask( + ctx context.Context, + taskID string, +) (*v1.Task, error) { + task, err := r.repo.GetTask(ctx, taskID) + if err != nil || task == nil { + return nil, err + } + return task.ToAPI(), nil +} + +func (r *mcpReadinessSchedulerTaskRepo) UpdateTaskState( + ctx context.Context, + taskID string, + state v1.TaskState, +) error { + return r.repo.UpdateTaskState(ctx, taskID, state) +} + +func (r *mcpReadinessSchedulerTaskRepo) UpdateTaskStateIfCurrentIn( + ctx context.Context, + taskID string, + state v1.TaskState, + allowed []v1.TaskState, +) (bool, error) { + _, updated, err := r.repo.UpdateTaskStateIfCurrentIn(ctx, taskID, state, allowed) + return updated, err +} + +func (r *mcpReadinessSchedulerTaskRepo) UpdateTaskStateIfNotArchived( + ctx context.Context, + taskID string, + state v1.TaskState, +) (bool, error) { + _, updated, err := r.repo.UpdateTaskStateIfNotArchived(ctx, taskID, state) + return updated, err +} + +func (r *mcpReadinessSchedulerTaskRepo) UpdateTaskStateIfSessionState( + ctx context.Context, + taskID, sessionID string, + expectedSessionState models.TaskSessionState, + state v1.TaskState, +) (bool, error) { + _, updated, err := r.repo.UpdateTaskStateIfSessionState(ctx, taskID, sessionID, expectedSessionState, state) + return updated, err +} diff --git a/apps/backend/internal/mcp/handlers/message_task_test.go b/apps/backend/internal/mcp/handlers/message_task_test.go index fc02c625bd..aef5ee9926 100644 --- a/apps/backend/internal/mcp/handlers/message_task_test.go +++ b/apps/backend/internal/mcp/handlers/message_task_test.go @@ -40,6 +40,7 @@ type fakeOrchestrator struct { onTurnStart func(context.Context, string, string) error turnStartResult orchestrator.ProcessOnTurnStartResult interruptCalls []interruptCall + readinessCalls []messagequeue.QueueSessionIdentity launchCalls []*orchestrator.LaunchSessionRequest launchErr error launchFunc func(context.Context, *orchestrator.LaunchSessionRequest) (*orchestrator.LaunchSessionResponse, error) @@ -214,6 +215,12 @@ func (f *fakeOrchestrator) QueueUserPrompt(ctx context.Context, taskID, sessionI func (f *fakeOrchestrator) GetMessageQueue() *messagequeue.Service { return f.queue } +func (f *fakeOrchestrator) CheckQueueAdmissionReadiness(_ context.Context, identity messagequeue.QueueSessionIdentity) { + f.mu.Lock() + defer f.mu.Unlock() + f.readinessCalls = append(f.readinessCalls, identity) +} + // QueueAndInterruptForPeerMessage inserts prompt into the fake's real // message queue (so tests can assert on queue state via f.queue), then // reports the configured interruptErr, or (false, nil) when diff --git a/apps/backend/internal/orchestrator/resume_prompt_queue_test.go b/apps/backend/internal/orchestrator/resume_prompt_queue_test.go index dec93365ab..deae32a105 100644 --- a/apps/backend/internal/orchestrator/resume_prompt_queue_test.go +++ b/apps/backend/internal/orchestrator/resume_prompt_queue_test.go @@ -9,6 +9,7 @@ import ( "github.com/kandev/kandev/internal/orchestrator/executor" "github.com/kandev/kandev/internal/orchestrator/messagequeue" "github.com/kandev/kandev/internal/task/models" + v1 "github.com/kandev/kandev/pkg/api/v1" "github.com/stretchr/testify/require" ) @@ -140,5 +141,109 @@ func TestCheckQueueAdmissionReadinessDispatchesOneConcurrentHead(t *testing.T) { case <-time.After(time.Second): t.Fatal("timed out waiting for the winning queued dispatch") } + agentMgr.mu.Lock() + captured := append([]string(nil), agentMgr.capturedPrompts...) + agentMgr.mu.Unlock() + require.Equal(t, []string{"dispatch once"}, captured) require.Equal(t, 0, svc.messageQueue.GetStatus(ctx, "session-1").Count) } + +// @covers AC-TASKS-PARENT-CHILD-MESSAGE-INTERRUPT-001.1 +func TestCheckQueueAdmissionReadinessPreservesFIFOAndProviderAcceptance(t *testing.T) { + ctx := context.Background() + repo := setupTestRepo(t) + seedTaskAndSession(t, repo, "task-1", "session-1", models.TaskSessionStateWaitingForInput) + seedExecutorRunning(t, repo, "session-1", "task-1", "execution-1") + prompts := make(chan string, 2) + completed := make(chan struct{}, 2) + agentMgr := &mockAgentManager{ + isAgentRunning: true, + repoForExecutionLookup: repo, + promptAgentFunc: func(ctx context.Context, _ string, prompt string, _ []v1.MessageAttachment, _ bool) (*executor.PromptResult, error) { + session, err := repo.GetTaskSession(ctx, "session-1") + if err != nil { + return nil, err + } + session.State = models.TaskSessionStateWaitingForInput + if err := repo.UpdateTaskSession(ctx, session); err != nil { + return nil, err + } + prompts <- prompt + return &executor.PromptResult{}, nil + }, + } + svc := createTestServiceWithAgent(repo, newMockStepGetter(), newMockTaskRepo(), agentMgr) + svc.executor = executor.NewExecutor(agentMgr, repo, testLogger(), executor.ExecutorConfig{}) + svc.messageCreator = &mockMessageCreator{} + svc.messageQueue.SetAutoMergeEnabled(false) + svc.onQueuedMessageExecutionComplete = func() { completed <- struct{}{} } + identity, err := svc.messageQueue.ResolveSessionIdentity(ctx, "task-1", "session-1") + require.NoError(t, err) + for _, prompt := range []string{"first queued", "second queued"} { + _, err := svc.messageQueue.QueueMessageWithMetadataForSession( + ctx, identity, prompt, "", messagequeue.QueuedByUser, false, nil, nil, + ) + require.NoError(t, err) + } + + svc.CheckQueueAdmissionReadiness(ctx, identity) + gotPrompts := make([]string, 0, 2) + for range 2 { + select { + case prompt := <-prompts: + gotPrompts = append(gotPrompts, prompt) + case <-time.After(2 * time.Second): + t.Fatalf("timed out waiting for queued prompt; got=%q", gotPrompts) + } + } + for range 2 { + select { + case <-completed: + case <-time.After(2 * time.Second): + t.Fatal("timed out waiting for queued delivery cleanup") + } + } + + require.Equal(t, []string{"first queued", "second queued"}, gotPrompts) + require.Equal(t, 0, svc.messageQueue.GetStatus(ctx, "session-1").Count) +} + +// @covers AC-TASKS-PARENT-CHILD-MESSAGE-INTERRUPT-001.1 +func TestCheckQueueAdmissionReadinessDoesNotBypassClarification(t *testing.T) { + ctx := context.Background() + repo := setupTestRepo(t) + seedTaskAndSession(t, repo, "task-1", "session-1", models.TaskSessionStateWaitingForInput) + seedPendingClarificationMessage(t, repo, "task-1", "session-1") + svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo()) + identity, err := svc.messageQueue.ResolveSessionIdentity(ctx, "task-1", "session-1") + require.NoError(t, err) + _, err = svc.messageQueue.QueueMessageWithMetadataForSession( + ctx, identity, "during clarification", "", messagequeue.QueuedByUser, false, nil, nil, + ) + require.NoError(t, err) + + svc.CheckQueueAdmissionReadiness(ctx, identity) + require.Equal(t, 1, svc.messageQueue.GetStatus(ctx, "session-1").Count) +} + +// @covers AC-TASKS-PARENT-CHILD-MESSAGE-INTERRUPT-001.1 +func TestCheckQueueAdmissionReadinessDoesNotBypassWIPWait(t *testing.T) { + ctx := context.Background() + repo := setupTestRepo(t) + seedTaskAndSession(t, repo, "task-1", "session-1", models.TaskSessionStateWaitingForInput) + task, err := repo.GetTask(ctx, "task-1") + require.NoError(t, err) + task.QueuedForStepID = "step-1" + task.WIPAdmitted = false + require.NoError(t, repo.UpdateTask(ctx, task)) + svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo()) + identity, err := svc.messageQueue.ResolveSessionIdentity(ctx, "task-1", "session-1") + require.NoError(t, err) + _, err = svc.messageQueue.QueueMessageWithMetadataForSession( + ctx, identity, "during WIP wait", "", messagequeue.QueuedByUser, false, nil, nil, + ) + require.NoError(t, err) + + svc.CheckQueueAdmissionReadiness(ctx, identity) + require.Equal(t, 1, svc.messageQueue.GetStatus(ctx, "session-1").Count) +} diff --git a/docs/plans/mcp-queued-message-wakeup/plan.md b/docs/plans/mcp-queued-message-wakeup/plan.md new file mode 100644 index 0000000000..429caa234b --- /dev/null +++ b/docs/plans/mcp-queued-message-wakeup/plan.md @@ -0,0 +1,178 @@ +--- +created: 2026-09-16 +status: done +requirements: + - REQ-TASKS-PARENT-CHILD-MESSAGE-INTERRUPT-001 + - REQ-UI-MESSAGE-QUEUE-AUTOMATION-CONTROLS-001 +system_design: + - ../../specs/tasks/system-design/parent-child-message-interrupt.md + - ../../specs/ui/system-design/message-queue-automation-controls.md +legacy_specs: [] +--- + +# Implementation Plan: MCP Queued-message Wakeup + +## Overview + +Close the confirmed peer-message admission/readiness race reported in +[issue 3716](https://github.com/kdlbs/kandev/issues/3716). One sequential work order +wires the existing guarded readiness operation and verifies actual delivery. +The issue is assigned to `carlosflorencio`. + +This package covers a confirmed contributor to stranded messages. It does not +claim to explain every symptom in the issue or authorize closing it as fully fixed. + +## Evidence and root cause + +Source inspected at `4c6e80091d`: + +1. `internal/mcp/handlers/handlers.go:dispatchTaskMessage` selects the queue path + from a previously read `RUNNING` or `STARTING` session snapshot. +2. `queueTaskMessage` persists through `QueueMessageWithMetadataForSession`, + publishes queue status, and returns. It never requests readiness checking. +3. If the final ready event drains the empty queue before insertion, the now-idle + target has no subsequent event that must consume this new entry. +4. `internal/orchestrator/service.go:CheckQueueAdmissionReadiness` already covers + this ordering for browser admission. `QueueHandlers.wsQueueMessage` calls it; + MCP ordinary admission does not. + +A temporary `issue3716_probe_test.go` used the existing handler fixtures. It +captured a busy session, updated the authoritative database row to +`WAITING_FOR_INPUT`, then called `dispatchTaskMessage` with the stale snapshot. +The result was `queued`, one pending entry, no prompt call, and zero calls to a +readiness-capable launcher. The test expected one readiness call and failed there. +The probe was removed after diagnosis; no permanent tests or production code +were changed. This proves the missing wakeup seam, not a complete production +reproduction of the reporter's timeout. + +## Scope + +### In scope + +- Ordinary MCP peer-message admission after a busy snapshot becomes stale. +- The same admission path reached through `dispatchPreparedTaskMessage`. +- Required MCP-to-orchestrator wiring with the captured session incarnation. +- FIFO, sender metadata, Auto-run, and existing lifecycle guards. +- Handler-to-provider acceptance evidence using controlled test collaborators. + +### Out of scope and unresolved issue evidence + +- The reported 30-second transport timeout. There is no reporter trace or + reproduction here proving which operation blocked. Status snapshot/publication + occurs after persistence, but that ordering alone does not prove the timeout. +- Deferred-move delivery, same-step semantics, transition-ledger interpretation, + zombie turns, and the `workflow_id` schema observation in the comments. + Current immediate moves use `MoveTaskWithOptions` and entry overlays; they + must not receive an ordinary-message wakeup before their transition commits. +- New admission receipts, blind retry, startup scans, and historical data repair. +- UI, translations, public parameters, schemas, and new background workers. + +## Requirement conformance + +The pre-fix implementation violated existing behavior; no new requirement was needed. +Reuse `AC-TASKS-PARENT-CHILD-MESSAGE-INTERRUPT-001.1` and its queue/turn safety +contract. Preserve `AC-UI-MESSAGE-QUEUE-AUTOMATION-CONTROLS-001.3`, `.4`, and `.6` +for automatic FIFO dispatch and durable paused policy. + +The assumption check found no material product choice: unrelated senders may +queue but may not interrupt; Auto-run does not bypass lifecycle barriers. +The new task-system design documents the missing producer integration. + +## Technical approach + +Add `CheckQueueAdmissionReadiness(context.Context, messagequeue.QueueSessionIdentity)` +to the MCP `SessionLauncher` interface in `handlers.go`. Update its test doubles +and any required production adapter, discovered through compile checks. The +production orchestrator already implements the method. + +After successful ordinary queue insertion, invoke the checker outside admission +locks and before best-effort status publication. Pass the existing captured +identity; do not re-resolve the primary session or use the manual drain API. +Keep existing error behavior before admission and the accepted `queued` response +after admission. Keep explicit interrupt on its existing atomic operation. + +Reuse `CheckQueueAdmissionReadiness` and its automatic reservation rather than +duplicating eligibility logic. Do not change deferred-move helpers. Do not add +an optional type assertion that silently skips production readiness wiring. + +## Tests + +Add `internal/mcp/handlers/message_task_readiness_test.go` with these cases: + +- `TestMessageTaskReadiness_ReadyBeforeAdmission`: stale busy snapshot, real + idle row, successful admission and exactly one readiness request for its identity. +- `TestMessageTaskReadiness_PreparedBusySession`: the prepared-session queue + branch uses the same check. +- `TestMessageTaskReadiness_RejectedAdmission`: capacity or identity rejection + neither checks readiness nor changes accepted work. +- `TestMessageTaskReadiness_UnrelatedSender`: the full handler accepts queued + mode and never invokes interrupt; explicit interrupt remains forbidden. +- `TestMessageTaskReadiness_Delivery`: real queue and orchestrator with an + accepting fake executor; exactly one accepted prompt and preserved attribution. + +Extend `internal/orchestrator/resume_prompt_queue_test.go` for +ready-before-insert, insert-before-ready, concurrent ready/admission checks, +Auto-run OFF, a pending clarification, WIP wait, and incarnation replacement. +Assert provider acceptance and pending-entry ownership, not only queue count. +Use channel barriers, not sleeps. Keep older FIFO work ahead of new work. + +These cases cover the peer-message criterion and Auto-run `.3`/`.4`/`.6` above. +Existing browser readiness tests are a dependency, not a second implementation. + +## End-to-end evidence + +The backend handler-to-orchestrator-to-fake-executor delivery case proves the +agent-facing flow. It must enter through `handleMessageTask`, control the stale +read boundary, use authoritative session storage, and observe prompt acceptance. +No rendered UI changes or artificial browser scenario are required. + +## Work orders + +- [x] [Task 01: Wake eligible MCP queue admissions](task-01-wake-mcp-queue.md) (done) + +## Verification results + +Diagnosis command, from `apps/backend`: + +```bash +go test ./internal/mcp/handlers -run '^TestIssue3716StaleBusySnapshotRequestsReadiness$' -count=1 +``` + +Expected failure reproduced: readiness calls were `0`, expected `1`. The temporary +probe was then removed. The permanent regression test now passes after the +required readiness call was added. + +Implementation and verification checks passed: + +```text +(cd apps/backend && go test ./internal/mcp/handlers -run 'TestMessageTaskReadiness_' -count=1) +(cd apps/backend && go test -race ./internal/mcp/handlers ./internal/orchestrator ./internal/orchestrator/handlers ./internal/orchestrator/messagequeue -count=1) +(cd apps/backend && go test ./internal/mcp/... -run '^$') +(cd apps/backend && make build) +python3 scripts/list-docs.py validate +python3 scripts/lint-spec-files.py --all +git diff --check +``` + +The race run passed all four packages. The backend build passed; its only +output of note was the existing warning that macOS artifacts were left +unsigned because no codesign tool was installed. Catalog validation covered +281 decisions and 970 specifications. Assignment was verified through +`gh issue view`. + +Package checks passed: `python3 scripts/list-docs.py validate` (281 decisions, +970 specifications), `python3 scripts/lint-spec-files.py --all`, and +`git diff --check`. Assignment was verified through `gh issue view`. + +## Risks + +- Running a manual drain would silently enable paused queues. +- Checking a newly resolved identity could dispatch a replacement session. +- A test that only observes queue removal could mistake failed dispatch for delivery. +- A synchronous checker must not wait for the new turn to complete. +- This narrow repair cannot justify marking the timeout or move reports resolved. + +## Documentation impact + +Internal design and plan only. Public documentation remains unchanged because +this package restores an existing delivery contract and introduces no public API. diff --git a/docs/plans/mcp-queued-message-wakeup/task-01-wake-mcp-queue.md b/docs/plans/mcp-queued-message-wakeup/task-01-wake-mcp-queue.md new file mode 100644 index 0000000000..85e91e15af --- /dev/null +++ b/docs/plans/mcp-queued-message-wakeup/task-01-wake-mcp-queue.md @@ -0,0 +1,139 @@ +--- +id: "01-wake-mcp-queue" +title: "Wake eligible MCP queue admissions" +status: done +wave: 1 +depends_on: [] +plan: "plan.md" +requirements: + - REQ-TASKS-PARENT-CHILD-MESSAGE-INTERRUPT-001 + - REQ-UI-MESSAGE-QUEUE-AUTOMATION-CONTROLS-001 +acceptance_criteria: + - AC-TASKS-PARENT-CHILD-MESSAGE-INTERRUPT-001.1 + - AC-UI-MESSAGE-QUEUE-AUTOMATION-CONTROLS-001.3 + - AC-UI-MESSAGE-QUEUE-AUTOMATION-CONTROLS-001.4 + - AC-UI-MESSAGE-QUEUE-AUTOMATION-CONTROLS-001.6 +system_design: + - ../../specs/tasks/system-design/parent-child-message-interrupt.md + - ../../specs/ui/system-design/message-queue-automation-controls.md +--- + +# Task 01: Wake Eligible MCP Queue Admissions + +## Summary + +Connect ordinary MCP peer-message admission to the existing guarded automatic +readiness check. Prove that a completed readiness event cannot strand a later +admission based on an older busy-session snapshot. + +## In scope + +- Required readiness method on `SessionLauncher`, real wiring, and test doubles. +- Both callers of `queueTaskMessage`, unchanged captured identity, and metadata. +- The regression matrix and backend end-to-end case in [the plan](plan.md#tests). + +## Out of scope + +Timeout diagnosis, deferred moves, new retries, startup recovery, storage changes, +and frontend work. Do not claim these issue symptoms are fixed by this work order. + +## Acceptance + +1. After readiness wins before admission, the full MCP flow delivers exactly one + prompt without another turn event. Insertion-first and concurrent triggers + also deliver once, preserving FIFO and sender attribution. +2. Auto-run OFF, active clarification, WIP wait, and stale incarnation retain + their existing barriers. Rejected admission performs no wakeup. Non-parent + queued sends never cancel a turn and cannot acquire interrupt authority. +3. Committed admission still returns accepted `queued` when dispatch defers. + The operation does not wait for turn completion, use manual drain, or require + another browser event. Required collaborator wiring is verified by compilation. + +## Implementation and TDD + +First add `TestMessageTaskReadiness_ReadyBeforeAdmission` and reproduce the missing +check. Use the existing `newTestTaskService`, `seedTaskWithSession`, and +`newMessageTaskHandler` fixtures; hold the busy snapshot while the database +session becomes idle. Then add the full handler delivery case with a real +orchestrator and controlled executor. A callback-only assertion is insufficient +as final proof of delivery. + +Add the required method to `SessionLauncher` and invoke it after successful +`QueueMessageWithMetadataForSession`, outside its locks and before status +publication. Reuse the orchestrator implementation without adding eligibility +checks to the MCP layer. Keep dispatch errors separate from committed admission. + +Complete the plan's regression matrix, including a second FIFO entry and +concurrent readiness. Test the paused queue remains OFF. Use existing +`queue_user_prompt_fastpath_test.go` and `resume_prompt_queue_test.go` +patterns, but assert accepted prompts rather than queue-count reduction alone. + +## Verification + +Run from the repository root: + +```bash +(cd apps/backend && go test ./internal/mcp/handlers -run 'TestMessageTaskReadiness_' -count=1) +(cd apps/backend && go test -race ./internal/mcp/handlers ./internal/orchestrator ./internal/orchestrator/handlers ./internal/orchestrator/messagequeue -count=1) +(cd apps/backend && go test ./internal/mcp/... -run '^$') +(cd apps/backend && make build) +python3 scripts/list-docs.py validate +python3 scripts/lint-spec-files.py --all +git diff --check +``` + +All commands passed after implementation. The first command failed before the +production change with zero readiness calls instead of one, then passed after +the change. The race command passed the MCP handlers, orchestrator, +orchestrator handlers, and message queue packages. The compile-only MCP sweep, +catalog validation, specification lint, whitespace check, and backend build +also passed. The backend build reported only the existing unsigned macOS +artifact warning. + +## Files likely touched + +- `apps/backend/internal/mcp/handlers/handlers.go` +- `apps/backend/internal/mcp/handlers/message_task_readiness_test.go` (new) +- `apps/backend/internal/mcp/handlers/message_task_test.go` (existing launcher fake) +- `apps/backend/internal/orchestrator/resume_prompt_queue_test.go` +- Other MCP launcher test doubles only where the required method needs it. + +## Dependencies + +None. Reuse the existing `CheckQueueAdmissionReadiness` implementation in +`apps/backend/internal/orchestrator/service.go`. + +## Risks + +Preserve lock order and session incarnation. Do not treat a reservation as +provider acceptance. Avoid waiting for a full agent turn in the request path. +The confirmed defect may coexist with the issue's unconfirmed timeout cause. + +## Parallelism + +`sequential` + +## Inputs + +- [Peer dispatch design](../../specs/tasks/system-design/parent-child-message-interrupt.md) +- [Peer-message requirement](../../specs/tasks/requirements/parent-child-message-interrupt.md) +- [Auto-run requirement](../../specs/ui/requirements/message-queue-automation-controls.md) +- `apps/backend/internal/orchestrator/handlers/queue_handlers_admission_test.go` +- `apps/backend/internal/orchestrator/queue_user_prompt_fastpath_test.go` + +## Results + +Implemented the required MCP-to-orchestrator readiness seam. Ordinary +`queueTaskMessage` admissions now invoke `CheckQueueAdmissionReadiness` with +the captured queue session identity after successful insertion and before +queue-status publication. Explicit parent interrupt admissions retain their +atomic queue-and-interrupt path. + +Added regression coverage for stale busy snapshots, the prepared busy branch, +rejected capacity, unrelated senders, real handler-to-provider delivery, FIFO +delivery, concurrent readiness, Auto-run OFF, active clarification, WIP wait, +and replaced session incarnations. The handler delivery test observes provider +acceptance and queue cleanup, not only queue-count reduction. + +The design-turn temporary probe failed as expected before the production change +and was removed. No timeout, deferred-move, or zombie-turn claim is made. diff --git a/docs/specs/tasks/system-design/parent-child-message-interrupt.md b/docs/specs/tasks/system-design/parent-child-message-interrupt.md new file mode 100644 index 0000000000..677a29710a --- /dev/null +++ b/docs/specs/tasks/system-design/parent-child-message-interrupt.md @@ -0,0 +1,88 @@ +--- +status: current +system: tasks +requirements: + - REQ-TASKS-PARENT-CHILD-MESSAGE-INTERRUPT-001 +--- + +# Peer Message Dispatch System Design + +## Purpose and mapping + +The task system owns peer-message admission and dispatch. This design implements +[AC-TASKS-PARENT-CHILD-MESSAGE-INTERRUPT-001.1](../requirements/parent-child-message-interrupt.md) +through the routing, readiness, and failure boundaries below. It supplements the +migrated requirement's technical detail without changing its authorization rules. + +## Routing and authorization + +`internal/mcp/handlers.handleMessageTask` validates the sender, resolves the +recipient session, expands prompt references, and retains sender metadata. +Explicit session selection and fallback session selection remain pinned. + +`dispatchTaskMessage` routes busy sessions to durable queue admission. Idle +sessions use existing turn-start, prompt, and resume operations. An explicit +interrupt remains restricted to the target task's direct parent and uses +`QueueAndInterruptForPeerMessage`; ordinary queue admission must never invoke it. + +## Readiness after admission + +The session snapshot used for routing can become stale before queue insertion. +Every successful ordinary peer-message queue admission requests an automatic +readiness check with the exact admitted `QueueSessionIdentity`. + +Expose the orchestrator's existing `CheckQueueAdmissionReadiness` operation on +the MCP `SessionLauncher` collaborator. Make this capability required so that +production wiring cannot silently omit it. Invoke it after queue admission +releases its locks. Queue status publication remains a separate projection; +publishing `MessageQueueStatusChanged` does not request dispatch. + +The orchestrator owns all eligibility decisions. Reuse its identity-aware, +task-admission-aware drain and atomic automatic reservation. Retain Auto-run, +clarification, WIP admission, cancellation, steering, active-dispatch, and +session-incarnation guards. Never substitute the manual `DrainQueuedMessage` +operation, which enables Auto-run. + +If insertion wins first, the later readiness event sees the accepted work. +If readiness wins first, the admission check sees the now-promptable session. +Concurrent checks serialize through existing reservation and dispatch guards. +The check attempts the FIFO head, which can precede the newly admitted message. +Automatic merging retains its existing behavior and surviving entry identity. + +## Failure, persistence, and observability + +Queue admission success is independent of a subsequent deferred dispatch. +Retain the existing `queued` response for this admission path; it does not +promise provider acceptance. A skipped readiness check or dispatch failure must +not convert committed admission into rejection. Existing dispatch restoration +and queue status events remain responsible for pending work and its projection. + +Use the existing checker and dispatch lifecycle. Do not wait for agent turn +completion or add a scheduler, timer, detached worker, or new persistence layer. +The checker must preserve the accepted identity even if a session is replaced. +Existing structured dispatch logs must not include prompt content. + +This design does not guarantee a response after transport loss, make unidentified +MCP submissions idempotent, or repair historical stranded rows at startup. +Those are separate contracts. No schema, public tool parameter, or UI change is +required. + +## Dependencies and verification + +- [Server-owned Auto-run](../../../decisions/2026-08-16-server-owned-queue-auto-run.md) + defines automatic reservation policy, including guarded deferral. +- [Prompt generation ownership](../../../decisions/0035-version-agent-ready-events-by-prompt-generation.md) + defines turn-event ownership and per-session serialization. +- [Resume prompt queue](resume-prompt-queue.md) describes the existing browser + producer's admission/readiness pattern. MCP reuses its orchestrator operation. +- [Queue automation controls](../../ui/requirements/message-queue-automation-controls.md) + retains the independent Auto-run contract; no presentation change is proposed. + +Deterministic tests cover both readiness/insertion orders, one accepted provider +prompt under concurrent triggers, paused queues, stale identities, and blocked +admission. A handler-to-orchestrator test must prove delivery, not just a smaller +queue count or a recorded method call. + +## Implementation plans + +- [MCP queued-message wakeup](../../../plans/mcp-queued-message-wakeup/plan.md)