diff --git a/apps/backend/internal/agent/runtime/agentctl/workspace_stream.go b/apps/backend/internal/agent/runtime/agentctl/workspace_stream.go index 2ca4f01fdd..7619fdf40d 100644 --- a/apps/backend/internal/agent/runtime/agentctl/workspace_stream.go +++ b/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() diff --git a/apps/backend/internal/agent/runtime/lifecycle/streams.go b/apps/backend/internal/agent/runtime/lifecycle/streams.go index 8592f881c5..67f1692d4c 100644 --- a/apps/backend/internal/agent/runtime/lifecycle/streams.go +++ b/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 } diff --git a/apps/backend/internal/mcp/handlers/config_handlers_test.go b/apps/backend/internal/mcp/handlers/config_handlers_test.go index 377da09c63..63b7f30c94 100644 --- a/apps/backend/internal/mcp/handlers/config_handlers_test.go +++ b/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 diff --git a/apps/backend/internal/mcp/handlers/config_task_handlers.go b/apps/backend/internal/mcp/handlers/config_task_handlers.go index f9021aac77..879e8aa008 100644 --- a/apps/backend/internal/mcp/handlers/config_task_handlers.go +++ b/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: " 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 diff --git a/apps/backend/internal/mcp/handlers/handlers.go b/apps/backend/internal/mcp/handlers/handlers.go index e3014d3b3e..a5468774b7 100644 --- a/apps/backend/internal/mcp/handlers/handlers.go +++ b/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) diff --git a/apps/backend/internal/mcp/handlers/message_task_test.go b/apps/backend/internal/mcp/handlers/message_task_test.go index 9957b2329c..11b23d3965 100644 --- a/apps/backend/internal/mcp/handlers/message_task_test.go +++ b/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 --- diff --git a/apps/backend/internal/mcp/server/server.go b/apps/backend/internal/mcp/server/server.go index 0e66b4447e..d9adcc4d3c 100644 --- a/apps/backend/internal/mcp/server/server.go +++ b/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()), diff --git a/apps/backend/internal/task/handlers/errors.go b/apps/backend/internal/task/handlers/errors.go index 88106cf764..62947af125 100644 --- a/apps/backend/internal/task/handlers/errors.go +++ b/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) { diff --git a/apps/backend/internal/task/handlers/errors_test.go b/apps/backend/internal/task/handlers/errors_test.go index 3d07dfb4b1..52b28196d0 100644 --- a/apps/backend/internal/task/handlers/errors_test.go +++ b/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) { diff --git a/apps/backend/internal/task/repository/sqlite/errors.go b/apps/backend/internal/task/repository/sqlite/errors.go index 637c7fd7d0..da4ea5f2fd 100644 --- a/apps/backend/internal/task/repository/sqlite/errors.go +++ b/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 diff --git a/apps/backend/internal/task/repository/sqlite/errors_test.go b/apps/backend/internal/task/repository/sqlite/errors_test.go index fcd58c6dc4..ae127f0e3e 100644 --- a/apps/backend/internal/task/repository/sqlite/errors_test.go +++ b/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) + } + }) } diff --git a/apps/backend/internal/task/repository/sqlite/session.go b/apps/backend/internal/task/repository/sqlite/session.go index e44455ca6c..bde88a25b4 100644 --- a/apps/backend/internal/task/repository/sqlite/session.go +++ b/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.