fix(mcp): disambiguate message_task task-not-found vs no-session error (#1186)

apps/backend/internal/agent/runtime/agentctl/workspace_stream.go

@@ -33,7 +33,13 @@ type WorkspaceStream struct {
 	inputCh   chan types.WorkspaceStreamMessage
 	closeCh   chan struct{}
 	closeOnce sync.Once
-	logger    *logger.Logger
+	// wg tracks the read + write goroutines so Wait() can block until they have
+	// fully unwound. Close()/Done() only signal shutdown; the read loop may
+	// still be returning from a blocked ReadJSON when they fire, so callers
+	// that need a true drain barrier (StreamManager shutdown, leak-sensitive
+	// tests) must call Wait after Close/Done.
+	wg     sync.WaitGroup
+	logger *logger.Logger
 }
 
 // StreamWorkspace opens a unified WebSocket connection for all workspace events
@@ -64,8 +70,19 @@ func (c *Client) StreamWorkspace(ctx context.Context, callbacks WorkspaceStreamC
 		logger:  c.logger,
 	}
 
-	go c.readWorkspaceStream(conn, stream, callbacks)
-	go stream.writeLoop(conn)
+	// Track both goroutines on the per-stream wg so WorkspaceStream.Wait can
+	// block until they have fully unwound. The workspace read loop only invokes
+	// data callbacks (shell/git/process) and self-closes on exit — it never
+	// re-enters manager teardown — so draining it is side-effect-free.
+	stream.wg.Add(2)
+	go func() {
+		defer stream.wg.Done()
+		c.readWorkspaceStream(conn, stream, callbacks)
+	}()
+	go func() {
+		defer stream.wg.Done()
+		stream.writeLoop(conn)
+	}()
 
 	return stream, nil
 }
@@ -175,6 +192,14 @@ func (ws *WorkspaceStream) Done() <-chan struct{} {
 	return ws.closeCh
 }
 
+// Wait blocks until the stream's read and write goroutines have fully exited.
+// Done() only reports that shutdown was *requested* (closeCh closed); Wait
+// reports that the goroutines have actually drained. Call after Close (or after
+// Done fires) to make teardown a true barrier for leak detection.
+func (ws *WorkspaceStream) Wait() {
+	ws.wg.Wait()
+}
+
 // CloseWorkspaceStream closes the workspace stream connection
 func (c *Client) CloseWorkspaceStream() {
 	c.mu.Lock()

apps/backend/internal/agent/runtime/lifecycle/streams.go

@@ -366,6 +366,11 @@ func (sm *StreamManager) connectWorkspaceStream(execution *AgentExecution, ready
 		case <-sm.stopCh:
 			ws.Close()
 		}
+		// Block until the stream's read/write goroutines have fully unwound
+		// before returning. Done()/Close only signal shutdown, so without this
+		// the StreamManager's wg releases while a blocked websocket read is
+		// still draining — stranding a goroutine that leak detection catches.
+		ws.Wait()
 		execution.ClearWorkspaceStream(ws)
 		return
 	}

apps/backend/internal/mcp/handlers/config_handlers_test.go

@@ -7,6 +7,8 @@ import (
 
 	"github.com/kandev/kandev/internal/common/logger"
 	"github.com/kandev/kandev/internal/orchestrator/messagequeue"
+	"github.com/kandev/kandev/internal/task/models"
+	"github.com/kandev/kandev/internal/task/service"
 	ws "github.com/kandev/kandev/pkg/websocket"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -342,6 +344,34 @@ func TestHandleMoveTask_InvalidPayload(t *testing.T) {
 	assertWSError(t, resp, ws.ErrorCodeBadRequest)
 }
 
+// TestLookupSession_NoPrimarySession_ReturnsNilNil pins the contract that a
+// task with no primary session resolves to (nil, nil) — a legitimate "empty"
+// state that lets handleMoveTask fall through to the idle-move path — rather
+// than a backend error. Regression guard: lookupSession originally detected
+// this case by substring-matching the repository's error message; when the
+// repository switched to a wrapped taskrepo.ErrNoPrimarySession sentinel the
+// substring stopped matching, so sessionless task moves were wrongly rejected
+// as internal errors. Classifying via errors.Is keeps the two decoupled.
+func TestLookupSession_NoPrimarySession_ReturnsNilNil(t *testing.T) {
+	svc, repo := newTestTaskService(t)
+	ctx := context.Background()
+	require.NoError(t, repo.CreateWorkspace(ctx, &models.Workspace{ID: "ws-1", Name: "Test"}))
+	require.NoError(t, repo.CreateWorkflow(ctx, &models.Workflow{ID: "wf-1", WorkspaceID: "ws-1", Name: "Board"}))
+	// Task created without an agent → no primary session row.
+	task, err := svc.CreateTask(ctx, &service.CreateTaskRequest{
+		WorkspaceID: "ws-1",
+		WorkflowID:  "wf-1",
+		Title:       "Sessionless task",
+	})
+	require.NoError(t, err)
+
+	h := &Handlers{taskSvc: svc, logger: testLogger(t).WithFields()}
+
+	session, lookupErr := h.lookupSession(ctx, task.ID)
+	require.NoError(t, lookupErr, "no-primary-session must not surface as a backend error")
+	assert.Nil(t, session, "no-primary-session must resolve to a nil session, not a value")
+}
+
 // recordingMessageQueuer captures QueueMessage calls for assertion.
 type recordingMessageQueuer struct {
 	calls []messagequeue.QueuedMessage

apps/backend/internal/mcp/handlers/config_task_handlers.go

@@ -3,12 +3,13 @@ package handlers
 import (
 	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
-	"strings"
 
 	"github.com/kandev/kandev/internal/orchestrator/messagequeue"
 	"github.com/kandev/kandev/internal/task/dto"
 	"github.com/kandev/kandev/internal/task/models"
+	taskrepo "github.com/kandev/kandev/internal/task/repository/sqlite"
 	v1 "github.com/kandev/kandev/pkg/api/v1"
 	ws "github.com/kandev/kandev/pkg/websocket"
 	"go.uber.org/zap"
@@ -176,20 +177,18 @@ func (h *Handlers) synthesizeMovedTaskDTO(ctx context.Context, taskID, workflowI
 //   - (session, nil) — task has a primary session.
 //   - (nil, nil)     — task has no primary session yet (legitimate "empty"
 //     state — task was created but no agent has been launched). The
-//     repository signals this with a "no primary session found for task:"
-//     error string; we treat it as a not-found rather than a failure so the
-//     caller can fall through to the idle-move path instead of rejecting the
-//     request.
+//     repository signals this with the taskrepo.ErrNoPrimarySession sentinel;
+//     we treat it as a not-found rather than a failure so the caller can fall
+//     through to the idle-move path instead of rejecting the request.
 //   - (nil, err)     — real backend lookup failure (DB error, etc.). The
 //     caller should map this to an internal error rather than collapsing it
 //     into "no session" downstream.
 func (h *Handlers) lookupSession(ctx context.Context, taskID string) (*models.TaskSession, error) {
 	session, err := h.taskSvc.GetPrimarySession(ctx, taskID)
 	if err != nil {
-		// "no primary session found for task: <id>" is the repo's not-found
-		// signal — see scanTaskSession's noRowsErr formatting. Match by
-		// substring; there's no exported sentinel to errors.Is against.
-		if strings.Contains(err.Error(), "no primary session found for task") {
+		// Classify the repo's not-found signal via the typed sentinel rather
+		// than substring-matching the formatted message, which is brittle.
+		if errors.Is(err, taskrepo.ErrNoPrimarySession) {
 			return nil, nil
 		}
 		return nil, err

apps/backend/internal/mcp/handlers/handlers.go

@@ -21,6 +21,7 @@ import (
 	"github.com/kandev/kandev/internal/orchestrator/messagequeue"
 	"github.com/kandev/kandev/internal/task/dto"
 	"github.com/kandev/kandev/internal/task/models"
+	taskrepo "github.com/kandev/kandev/internal/task/repository/sqlite"
 	"github.com/kandev/kandev/internal/task/service"
 	workflowctrl "github.com/kandev/kandev/internal/workflow/controller"
 	workflowsvc "github.com/kandev/kandev/internal/workflow/service"
@@ -768,12 +769,28 @@ func (h *Handlers) handleMessageTask(ctx context.Context, msg *ws.Message) (*ws.
 		return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeNotFound, "sender task not found", nil)
 	}
 
+	// Verify the target task exists before looking up its session, so a bad
+	// task_id (e.g. a truncated UUID prefix) reports "task not found" instead
+	// of the misleading "no primary session" error from the session lookup.
+	// This is purely an existence check — GetTask returns a wrapped
+	// ErrTaskNotFound on no-rows, never (nil, nil).
+	if _, err := h.taskSvc.GetTask(ctx, req.TaskID); err != nil {
+		if errors.Is(err, taskrepo.ErrTaskNotFound) {
+			return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeNotFound,
+				"target task not found: "+req.TaskID+" (pass the full task UUID, not a truncated prefix)", nil)
+		}
+		return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError,
+			"failed to look up target task: "+err.Error(), nil)
+	}
+
 	session, err := h.taskSvc.GetPrimarySession(ctx, req.TaskID)
 	if err != nil {
-		return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeNotFound, "task not found or has no session: "+err.Error(), nil)
-	}
-	if session == nil {
-		return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeNotFound, "task has no active session — use create_task_kandev to start one", nil)
+		if errors.Is(err, taskrepo.ErrNoPrimarySession) {
+			return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeNotFound,
+				"target task exists but has no active session", nil)
+		}
+		return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError,
+			"failed to get session for task: "+err.Error(), nil)
 	}
 
 	wrappedPrompt, senderMeta := wrapAgentMessage(req.Prompt, senderTask, req.SenderSessionID)

apps/backend/internal/mcp/handlers/message_task_test.go

@@ -393,6 +393,40 @@ func TestHandleMessageTask_NoPrimarySession_Rejects(t *testing.T) {
 	resp, err := h.handleMessageTask(ctx, msg)
 	require.NoError(t, err)
 	assertWSError(t, resp, ws.ErrorCodeNotFound)
+
+	// The task exists but has no session — must report "no active session", not
+	// the generic "task not found" from the task-existence check.
+	payload := string(resp.Payload)
+	assert.Contains(t, payload, "no active session")
+	assert.NotContains(t, payload, "task not found")
+}
+
+func TestHandleMessageTask_NonexistentTask_ReportsTaskNotFound(t *testing.T) {
+	svc, repo := newTestTaskService(t)
+	ctx := context.Background()
+	require.NoError(t, repo.CreateWorkspace(ctx, &models.Workspace{ID: "ws-1", Name: "Test"}))
+	require.NoError(t, repo.CreateWorkflow(ctx, &models.Workflow{ID: "wf-1", WorkspaceID: "ws-1", Name: "Board"}))
+	// Only the sender exists; the target task_id below was never created (mimics
+	// passing a truncated UUID prefix).
+	sender, err := svc.CreateTask(ctx, &service.CreateTaskRequest{
+		WorkspaceID: "ws-1",
+		WorkflowID:  "wf-1",
+		Title:       "Sender task",
+	})
+	require.NoError(t, err)
+
+	h, _ := newMessageTaskHandler(t, svc)
+
+	msg := makeWSMessage(t, ws.ActionMCPMessageTask,
+		senderPayload("00000000-0000-0000-0000-000000000000", "hello", sender.ID))
+	resp, err := h.handleMessageTask(ctx, msg)
+	require.NoError(t, err)
+	assertWSError(t, resp, ws.ErrorCodeNotFound)
+
+	// Must surface the task-not-found error, NOT the misleading no-session error.
+	payload := string(resp.Payload)
+	assert.Contains(t, payload, "target task not found")
+	assert.NotContains(t, payload, "no session")
 }
 
 // --- sender attribution validation ---

apps/backend/internal/mcp/server/server.go

@@ -445,7 +445,7 @@ Behaviour by session state:
 - Failed/cancelled: an error is returned (use create_task_kandev to start fresh).
 
 Returns the dispatch status: "queued", "sent", or "started".`),
-			mcp.WithString("task_id", mcp.Required(), mcp.Description("The target task ID")),
+			mcp.WithString("task_id", mcp.Required(), mcp.Description("The target task's full UUID (not a truncated prefix)")),
 			mcp.WithString("prompt", mcp.Required(), mcp.Description("The message to deliver to the task's agent")),
 		),
 		s.wrapHandler("message_task_kandev", s.messageTaskHandler()),

apps/backend/internal/task/handlers/errors.go

@@ -44,6 +44,7 @@ func isNotFound(err error) bool {
 		return false
 	}
 	if errors.Is(err, taskrepo.ErrTaskNotFound) ||
+		errors.Is(err, taskrepo.ErrNoPrimarySession) ||
 		errors.Is(err, service.ErrDocumentNotFound) ||
 		errors.Is(err, service.ErrTaskPlanNotFound) ||
 		errors.Is(err, service.ErrRevisionNotFound) {

apps/backend/internal/task/handlers/errors_test.go

@@ -17,10 +17,11 @@ import (
 // status mapping survives downstream wrap/unwrap changes.
 func TestErrorsAreClassifiable(t *testing.T) {
 	notFoundSentinels := map[string]error{
-		"taskrepo.ErrTaskNotFound":    taskrepo.ErrTaskNotFound,
-		"service.ErrDocumentNotFound": service.ErrDocumentNotFound,
-		"service.ErrTaskPlanNotFound": service.ErrTaskPlanNotFound,
-		"service.ErrRevisionNotFound": service.ErrRevisionNotFound,
+		"taskrepo.ErrTaskNotFound":     taskrepo.ErrTaskNotFound,
+		"taskrepo.ErrNoPrimarySession": taskrepo.ErrNoPrimarySession,
+		"service.ErrDocumentNotFound":  service.ErrDocumentNotFound,
+		"service.ErrTaskPlanNotFound":  service.ErrTaskPlanNotFound,
+		"service.ErrRevisionNotFound":  service.ErrRevisionNotFound,
 	}
 	for name, sentinel := range notFoundSentinels {
 		t.Run("isNotFound recognizes "+name, func(t *testing.T) {

apps/backend/internal/task/repository/sqlite/errors.go

@@ -8,6 +8,11 @@ import "errors"
 // formatted message, which includes the task id and is therefore brittle.
 var ErrTaskNotFound = errors.New("task not found")
 
+// ErrNoPrimarySession is returned by GetPrimarySessionByTaskID when the task
+// has no primary session row. Callers should use errors.Is to distinguish this
+// "not found" case from genuine backend/DB errors.
+var ErrNoPrimarySession = errors.New("no primary session")
+
 // ErrOfficeSessionRaceConflict is returned by CreateTaskSession when the
 // insert violates the uniq_office_task_session partial unique index — i.e.
 // two callers raced past their SELECT-then-INSERT for the same

apps/backend/internal/task/repository/sqlite/errors_test.go

@@ -5,6 +5,7 @@ import (
 	"errors"
 	"testing"
 
+	"github.com/kandev/kandev/internal/task/models"
 	v1 "github.com/kandev/kandev/pkg/api/v1"
 )
 
@@ -44,4 +45,26 @@ func TestErrorsAreClassifiable(t *testing.T) {
 			t.Errorf("UpdateTaskState error not classifiable as ErrTaskNotFound: %v", err)
 		}
 	})
+
+	t.Run("GetPrimarySessionByTaskID returns ErrNoPrimarySession for task with no session", func(t *testing.T) {
+		// Create a task without a session so the no-rows path is exercised.
+		taskID := "task-no-session"
+		err := repo.CreateTask(ctx, &models.Task{
+			ID:          taskID,
+			WorkspaceID: "ws-err-test",
+			WorkflowID:  "wf-err-test",
+			Title:       "Sessionless task",
+			State:       v1.TaskStateCreated,
+		})
+		if err != nil {
+			t.Fatalf("CreateTask: %v", err)
+		}
+		_, err = repo.GetPrimarySessionByTaskID(ctx, taskID)
+		if err == nil {
+			t.Fatal("expected GetPrimarySessionByTaskID to fail for task with no session")
+		}
+		if !errors.Is(err, ErrNoPrimarySession) {
+			t.Errorf("GetPrimarySessionByTaskID error not classifiable as ErrNoPrimarySession: %v", err)
+		}
+	})
 }

apps/backend/internal/task/repository/sqlite/session.go

@@ -404,6 +404,11 @@ func (r *Repository) GetTaskSessionByTaskAndAgent(ctx context.Context, taskID, a
 // to detect "no row found" so the caller gets nil, nil instead of an error.
 const sessionNotFoundMsg = "task_sessions: no matching row"
 
+// noPrimarySessionSentinel is the no-rows marker passed to scanTaskSession by
+// GetPrimarySessionByTaskID. Matching this exact string lets the function wrap
+// the result with ErrNoPrimarySession so callers can use errors.Is.
+const noPrimarySessionSentinel = "task_sessions: no primary session row"
+
 // ListNonTerminalSessionsByAgentInstance returns every office task_session row
 // for the given agent_profile_id whose state is NOT terminal
 // (CREATED / STARTING / RUNNING / IDLE / WAITING_FOR_INPUT). Used by the
@@ -1071,12 +1076,18 @@ func (r *Repository) DeleteTaskSessionWorktreesBySession(ctx context.Context, se
 	return err
 }
 
-// GetPrimarySessionByTaskID retrieves the primary session for a task
+// GetPrimarySessionByTaskID retrieves the primary session for a task.
+// Returns ErrNoPrimarySession (wrapped) when the task has no primary session
+// row; callers should use errors.Is to distinguish this from real DB errors.
 func (r *Repository) GetPrimarySessionByTaskID(ctx context.Context, taskID string) (*models.TaskSession, error) {
 	row := r.ro.QueryRowContext(ctx, r.ro.Rebind(
 		`SELECT `+taskSessionSelectCols+` `+taskSessionFromClause+` WHERE ts.task_id = ? AND ts.is_primary = 1 LIMIT 1`,
 	), taskID)
-	return r.scanTaskSession(ctx, row, fmt.Sprintf("no primary session found for task: %s", taskID))
+	session, err := r.scanTaskSession(ctx, row, noPrimarySessionSentinel)
+	if err != nil && err.Error() == noPrimarySessionSentinel {
+		return nil, fmt.Errorf("%w: %s", ErrNoPrimarySession, taskID)
+	}
+	return session, err
 }
 
 // GetPrimarySessionIDsByTaskIDs returns a map of task ID to primary session ID for the given task IDs.