diff --git a/apps/backend/config/prompts/kandev-context.md b/apps/backend/config/prompts/kandev-context.md index 0a0eaf9fd2..025fbde6fb 100644 --- a/apps/backend/config/prompts/kandev-context.md +++ b/apps/backend/config/prompts/kandev-context.md @@ -7,6 +7,7 @@ Use these IDs when calling tools that require task_id or session_id. Available tools: - ask_user_question_kandev: Ask the user one or more clarifying questions in a single tool call. Use this whenever you need user input before proceeding. Required params: questions (array of 1-4 question objects; each object has prompt (string) and options (array of 2-6 {label, description})). Optional: context (string). +- step_complete_kandev: Signal that every user-stated requirement for the CURRENT workflow step is satisfied. Call this as the LAST action of the step (after the final tool call / commit / answer). Idempotent — a second call within the same step is a no-op. Do NOT call when asking a question, mid-conversation, or on partial progress. Required params: summary (one-paragraph plain text). Optional: handoff, blockers. - create_task_plan_kandev: Save an implementation plan for the current task. Required params: task_id, content (markdown). Optional: title. - get_task_plan_kandev: Retrieve the current plan for a task (includes any user edits). Required params: task_id. - update_task_plan_kandev: Update an existing plan. Required params: task_id, content (markdown). Optional: title. diff --git a/apps/backend/internal/events/types.go b/apps/backend/internal/events/types.go index 22edce917a..2f28fab6e4 100644 --- a/apps/backend/internal/events/types.go +++ b/apps/backend/internal/events/types.go @@ -35,6 +35,12 @@ const ( WorkflowStepCreated = "workflow_step.created" WorkflowStepUpdated = "workflow_step.updated" WorkflowStepDeleted = "workflow_step.deleted" + // WorkflowStepCompletionSignaled fires when an agent (or the manual + // fallback button) signals that the current workflow step is complete + // for a given (task, session) — ADR 0015. The orchestrator subscriber + // reads the pending bag on TaskSession.Metadata and drives the + // transition. + WorkflowStepCompletionSignaled = "workflow.step_completion_signaled" ) // Event types for comments diff --git a/apps/backend/internal/mcp/handlers/handlers.go b/apps/backend/internal/mcp/handlers/handlers.go index 9e3348e332..1346b377c2 100644 --- a/apps/backend/internal/mcp/handlers/handlers.go +++ b/apps/backend/internal/mcp/handlers/handlers.go @@ -54,6 +54,10 @@ type MessageCreator interface { type SessionRepository interface { UpdateTaskSessionState(ctx context.Context, sessionID string, state models.TaskSessionState, errorMessage string) error GetTaskSession(ctx context.Context, id string) (*models.TaskSession, error) + // SetSessionMetadataKey is used by handleStepComplete (ADR 0015) to + // atomically write the pending-completion bag without clobbering other + // metadata keys. + SetSessionMetadataKey(ctx context.Context, sessionID, key string, value interface{}) error } // TaskRepository interface for updating task state. @@ -171,6 +175,7 @@ func (h *Handlers) RegisterHandlers(d *ws.Dispatcher) { d.RegisterFunc(ws.ActionMCPUpdateTask, h.handleUpdateTask) d.RegisterFunc(ws.ActionMCPAddBranchToTask, h.handleAddBranchToTask) d.RegisterFunc(ws.ActionMCPUpdateRepositoryBaseBranch, h.handleUpdateRepositoryBaseBranch) + d.RegisterFunc(ws.ActionMCPStepComplete, h.handleStepComplete) d.RegisterFunc(ws.ActionMCPMessageTask, h.handleMessageTask) d.RegisterFunc(ws.ActionMCPGetTaskConversation, h.handleGetTaskConversation) d.RegisterFunc(ws.ActionMCPAskUserQuestion, h.handleAskUserQuestion) @@ -179,7 +184,7 @@ func (h *Handlers) RegisterHandlers(d *ws.Dispatcher) { d.RegisterFunc(ws.ActionMCPUpdateTaskPlan, h.handleUpdateTaskPlan) d.RegisterFunc(ws.ActionMCPDeleteTaskPlan, h.handleDeleteTaskPlan) d.RegisterFunc(ws.ActionMCPClarificationTimeout, h.handleClarificationTimeout) - count := 15 + count := 17 // Config-mode handlers (registered when config deps are set) if h.workflowSvc != nil { @@ -900,6 +905,194 @@ func classifyAddBranchError(err error) string { return ws.ErrorCodeInternalError } +// handleStepComplete records the agent's explicit step-completion signal +// (ADR 0015). The handler: +// +// - Loads the session and the task to identify the current workflow step. +// - Dedupes: if a pending signal already exists for the same step, returns +// {accepted: false, reason: "already_signaled"} without overwriting. +// When the session is WAITING_FOR_INPUT, the bus event is re-published +// so a failed first-attempt publish can still drive the subscriber. +// - Otherwise writes a PendingStepCompletionSignal blob under +// TaskSession.Metadata[SessionMetaKeyPendingStepCompletion] via +// SetSessionMetadataKey (json_set — preserves other metadata keys). +// - Publishes events.WorkflowStepCompletionSignaled so the orchestrator +// subscriber can drive the on_turn_complete transition for steps with +// AutoAdvanceRequiresSignal=true. Steps that don't opt in ignore the +// signal entirely; the bag entry is cleared on the next turn start +// (no separate audit trail is persisted). +// +// Idempotency is intentionally lossy — a second call within the same step +// silently keeps the first signal's payload (summary/handoff/blockers). The +// orchestrator treats the first signal as authoritative. +func (h *Handlers) handleStepComplete(ctx context.Context, msg *ws.Message) (*ws.Message, error) { + var req struct { + TaskID string `json:"task_id"` + SessionID string `json:"session_id"` + Summary string `json:"summary"` + Handoff string `json:"handoff"` + Blockers string `json:"blockers"` + } + if err := json.Unmarshal(msg.Payload, &req); err != nil { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeBadRequest, "Invalid payload: "+err.Error(), nil) + } + if req.TaskID == "" || req.SessionID == "" { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "task_id and session_id are required", nil) + } + if strings.TrimSpace(req.Summary) == "" { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "summary is required", nil) + } + + session, task, errMsg, err := h.resolveStepCompleteTarget(ctx, msg, req.TaskID, req.SessionID) + if errMsg != nil { + return errMsg, err + } + + // Idempotency: if a pending signal exists for the current step, return + // without overwriting. A stale signal for a different step (left over + // from a transition that hasn't yet cleared the bag) is treated as + // absent and overwritten — the new step's signal supersedes. + // + // Re-publish on the dedup path when the session is WAITING_FOR_INPUT. + // Without this, an agent's retry after a publish failure short-circuits + // to `already_signaled` without firing the out-of-band subscriber, + // leaving the session stuck until the user replies. Publish is + // idempotent on the subscriber side (it re-checks bag + step), so a + // double-fire when the first publish actually landed is harmless. + if existing, ok := models.LoadPendingStepSignal(session.Metadata); ok && existing.StepID == task.WorkflowStepID { + if session.State == models.TaskSessionStateWaitingForInput { + if errMsg, err := h.publishStepCompletionEvent(ctx, msg, req.TaskID, req.SessionID, task.WorkflowStepID, existing); errMsg != nil { + return errMsg, err + } + } + return ws.NewResponse(msg.ID, msg.Action, map[string]interface{}{ + "accepted": false, + "reason": "already_signaled", + }) + } + + signal := models.PendingStepCompletionSignal{ + StepID: task.WorkflowStepID, + Source: models.StepCompletionSourceAgent, + Summary: strings.TrimSpace(req.Summary), + Handoff: strings.TrimSpace(req.Handoff), + Blockers: strings.TrimSpace(req.Blockers), + SignaledAt: time.Now().UTC(), + } + if err := h.sessionRepo.SetSessionMetadataKey(ctx, req.SessionID, models.SessionMetaKeyPendingStepCompletion, signal); err != nil { + h.logger.Error("failed to persist step-completion signal", + zap.String("task_id", req.TaskID), + zap.String("session_id", req.SessionID), + zap.Error(err)) + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, "failed to record signal", nil) + } + + if errMsg, err := h.publishStepCompletionEvent(ctx, msg, req.TaskID, req.SessionID, task.WorkflowStepID, signal); errMsg != nil { + return errMsg, err + } + + return ws.NewResponse(msg.ID, msg.Action, map[string]interface{}{ + "accepted": true, + "step_id": task.WorkflowStepID, + "signaled_at": signal.SignaledAt, + }) +} + +// resolveStepCompleteTarget loads the session + task the signal applies to +// and runs the up-front validation (ownership, terminal-state guard, +// workflow-step presence). Returns a populated session+task pair on success, +// or a ready-to-send WS error envelope (and its marshal error if any) on +// any failed precondition. +func (h *Handlers) resolveStepCompleteTarget( + ctx context.Context, msg *ws.Message, taskID, sessionID string, +) (*models.TaskSession, *models.Task, *ws.Message, error) { + session, err := h.sessionRepo.GetTaskSession(ctx, sessionID) + if err != nil { + // Session repo has no exported not-found sentinel; classify by + // substring and treat anything else as transient so the agent + // retries instead of abandoning the session. + if strings.Contains(err.Error(), "not found") { + errMsg, mErr := ws.NewError(msg.ID, msg.Action, ws.ErrorCodeNotFound, "session not found", nil) + return nil, nil, errMsg, mErr + } + h.logger.Error("step_complete: failed to load session", + zap.String("session_id", sessionID), zap.Error(err)) + errMsg, mErr := ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, "failed to load session", nil) + return nil, nil, errMsg, mErr + } + if session.TaskID != taskID { + errMsg, mErr := ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "session does not belong to task", nil) + return nil, nil, errMsg, mErr + } + // Terminal sessions cannot consume a signal: the orchestrator's + // out-of-band subscriber short-circuits on every state other than + // WAITING_FOR_INPUT, and no future turn-end will fire on a closed + // session. Reject up front so the agent gets a clear error instead of + // `accepted: true` followed by silent no-op. + switch session.State { + case models.TaskSessionStateCompleted, + models.TaskSessionStateFailed, + models.TaskSessionStateCancelled: + errMsg, mErr := ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, + "cannot signal completion for a terminal session (state: "+string(session.State)+")", nil) + return nil, nil, errMsg, mErr + } + + task, err := h.taskSvc.GetTask(ctx, taskID) + if err != nil { + // Task repo exports ErrTaskNotFound; anything else is a transient + // load failure that the agent should retry rather than interpret + // as "task gone". + if errors.Is(err, taskrepo.ErrTaskNotFound) { + errMsg, mErr := ws.NewError(msg.ID, msg.Action, ws.ErrorCodeNotFound, "task not found", nil) + return nil, nil, errMsg, mErr + } + h.logger.Error("step_complete: failed to load task", + zap.String("task_id", taskID), zap.Error(err)) + errMsg, mErr := ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, "failed to load task", nil) + return nil, nil, errMsg, mErr + } + if task.WorkflowStepID == "" { + errMsg, mErr := ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "task has no current workflow step", nil) + return nil, nil, errMsg, mErr + } + return session, task, nil, nil +} + +// publishStepCompletionEvent emits the bus event the orchestrator's +// out-of-band subscriber listens for. If publish fails the bag is already +// persisted but the subscriber will not fire — surface the error to the +// agent so it can retry. The handler-level idempotency guard guarantees the +// retry either succeeds end-to-end or short-circuits with `already_signaled` +// once the publish lands. Returns (nil, nil) on success or when no bus is wired. +func (h *Handlers) publishStepCompletionEvent( + ctx context.Context, msg *ws.Message, taskID, sessionID, stepID string, + signal models.PendingStepCompletionSignal, +) (*ws.Message, error) { + if h.eventBus == nil { + return nil, nil + } + if err := h.eventBus.Publish(ctx, events.WorkflowStepCompletionSignaled, bus.NewEvent( + events.WorkflowStepCompletionSignaled, "mcp-handlers", + map[string]interface{}{ + "task_id": taskID, + "session_id": sessionID, + "step_id": stepID, + "source": signal.Source, + "summary": signal.Summary, + "signaled_at": signal.SignaledAt, + }, + )); err != nil { + h.logger.Error("failed to publish step-completion signal", + zap.String("task_id", taskID), + zap.String("session_id", sessionID), + zap.Error(err)) + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, + "failed to notify orchestrator (signal persisted; retry)", nil) + } + return nil, nil +} + // handleMessageTask sends a prompt to an existing task on behalf of an agent // in another task. The MCP server (agentctl) injects the sender's task_id and // session_id into the payload; this handler validates the sender, looks up its diff --git a/apps/backend/internal/mcp/handlers/step_complete_test.go b/apps/backend/internal/mcp/handlers/step_complete_test.go new file mode 100644 index 0000000000..c734b9db0f --- /dev/null +++ b/apps/backend/internal/mcp/handlers/step_complete_test.go @@ -0,0 +1,274 @@ +package handlers + +import ( + "context" + "encoding/json" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/kandev/kandev/internal/events" + "github.com/kandev/kandev/internal/task/models" + sqliterepo "github.com/kandev/kandev/internal/task/repository/sqlite" + "github.com/kandev/kandev/internal/task/service" + v1 "github.com/kandev/kandev/pkg/api/v1" + ws "github.com/kandev/kandev/pkg/websocket" +) + +// seedStepCompleteTarget seeds a workspace, task (with WorkflowStepID), and +// session in the requested state. Used by every TestHandleStepComplete_* case +// that needs the precondition chain in `resolveStepCompleteTarget` to succeed. +func seedStepCompleteTarget(t *testing.T, repo *sqliterepo.Repository, taskID, sessionID, stepID string, state models.TaskSessionState) { + t.Helper() + ctx := context.Background() + now := time.Now().UTC() + require.NoError(t, repo.CreateWorkspace(ctx, &models.Workspace{ + ID: "ws-step-complete", + Name: "Step Complete", + CreatedAt: now, + UpdatedAt: now, + })) + require.NoError(t, repo.CreateTask(ctx, &models.Task{ + ID: taskID, + WorkspaceID: "ws-step-complete", + WorkflowStepID: stepID, + Title: "Step Complete Task", + State: v1.TaskStateInProgress, + CreatedAt: now, + UpdatedAt: now, + })) + require.NoError(t, repo.CreateTaskSession(ctx, &models.TaskSession{ + ID: sessionID, + TaskID: taskID, + State: state, + StartedAt: now, + UpdatedAt: now, + })) +} + +func newStepCompleteHandler(t *testing.T, taskSvc *service.Service, repo *sqliterepo.Repository, bus *mcpRecordingEventBus) *Handlers { + t.Helper() + return &Handlers{ + taskSvc: taskSvc, + sessionRepo: repo, + eventBus: bus, + logger: testLogger(t).WithFields(), + } +} + +// TestHandleStepComplete_MissingFields covers the input-validation branches +// that fail before any DB lookup. All three return ErrorCodeValidation. +func TestHandleStepComplete_MissingFields(t *testing.T) { + cases := []struct { + name string + payload map[string]interface{} + want string + }{ + { + name: "missing task_id", + payload: map[string]interface{}{"session_id": "s1", "summary": "done"}, + want: ws.ErrorCodeValidation, + }, + { + name: "missing session_id", + payload: map[string]interface{}{"task_id": "t1", "summary": "done"}, + want: ws.ErrorCodeValidation, + }, + { + name: "blank summary", + payload: map[string]interface{}{"task_id": "t1", "session_id": "s1", "summary": " "}, + want: ws.ErrorCodeValidation, + }, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + h := &Handlers{logger: testLogger(t).WithFields()} + msg := makeWSMessage(t, ws.ActionMCPStepComplete, tc.payload) + resp, err := h.handleStepComplete(context.Background(), msg) + require.NoError(t, err) + assertWSError(t, resp, tc.want) + }) + } +} + +// TestHandleStepComplete_SessionDoesNotBelongToTask verifies the ownership +// guard rejects requests where the session.TaskID doesn't match the request's +// task_id. +func TestHandleStepComplete_SessionDoesNotBelongToTask(t *testing.T) { + svc, repo := newTestTaskService(t) + seedStepCompleteTarget(t, repo, "task-owner", "session-owner", "step-1", models.TaskSessionStateRunning) + bus := &mcpRecordingEventBus{} + h := newStepCompleteHandler(t, svc, repo, bus) + + msg := makeWSMessage(t, ws.ActionMCPStepComplete, map[string]interface{}{ + "task_id": "task-OTHER", + "session_id": "session-owner", + "summary": "wrong task", + }) + resp, err := h.handleStepComplete(context.Background(), msg) + require.NoError(t, err) + assertWSError(t, resp, ws.ErrorCodeValidation) + assert.Empty(t, bus.events, "ownership rejection must not publish an event") +} + +// TestHandleStepComplete_TerminalSessionRejected covers the +// Completed/Failed/Cancelled guard: writing a signal to a terminal session +// would never be consumed (subscriber short-circuits on non-WAITING state, +// no future turn-end fires), so we reject up front instead of returning +// accepted=true followed by silent no-op. +func TestHandleStepComplete_TerminalSessionRejected(t *testing.T) { + for _, state := range []models.TaskSessionState{ + models.TaskSessionStateCompleted, + models.TaskSessionStateFailed, + models.TaskSessionStateCancelled, + } { + t.Run(string(state), func(t *testing.T) { + svc, repo := newTestTaskService(t) + seedStepCompleteTarget(t, repo, "task-term", "session-term", "step-1", state) + bus := &mcpRecordingEventBus{} + h := newStepCompleteHandler(t, svc, repo, bus) + + msg := makeWSMessage(t, ws.ActionMCPStepComplete, map[string]interface{}{ + "task_id": "task-term", + "session_id": "session-term", + "summary": "too late", + }) + resp, err := h.handleStepComplete(context.Background(), msg) + require.NoError(t, err) + assertWSError(t, resp, ws.ErrorCodeValidation) + assert.Empty(t, bus.events, "terminal-session rejection must not publish") + }) + } +} + +// TestHandleStepComplete_FirstCallAccepted covers the happy path: bag is +// written, event is published with the documented payload shape, and the +// response reports accepted=true with the persisted step_id + signaled_at. +func TestHandleStepComplete_FirstCallAccepted(t *testing.T) { + svc, repo := newTestTaskService(t) + seedStepCompleteTarget(t, repo, "task-first", "session-first", "step-1", models.TaskSessionStateRunning) + bus := &mcpRecordingEventBus{} + h := newStepCompleteHandler(t, svc, repo, bus) + + msg := makeWSMessage(t, ws.ActionMCPStepComplete, map[string]interface{}{ + "task_id": "task-first", + "session_id": "session-first", + "summary": "implementation finished", + "handoff": "tests next", + }) + resp, err := h.handleStepComplete(context.Background(), msg) + require.NoError(t, err) + require.NotNil(t, resp) + assert.Equal(t, ws.MessageTypeResponse, resp.Type) + + var payload map[string]interface{} + require.NoError(t, json.Unmarshal(resp.Payload, &payload)) + assert.Equal(t, true, payload["accepted"]) + assert.Equal(t, "step-1", payload["step_id"]) + // signaled_at is part of the documented response contract — pin its + // presence + RFC3339Nano shape so a future refactor can't silently + // drop or rename the field. + signaledAt, ok := payload["signaled_at"].(string) + require.True(t, ok, "expected signaled_at string in response payload") + _, parseErr := time.Parse(time.RFC3339Nano, signaledAt) + require.NoError(t, parseErr, "signaled_at must be RFC3339Nano") + + // Bag written under the canonical key. + session, err := repo.GetTaskSession(context.Background(), "session-first") + require.NoError(t, err) + bag, ok := models.LoadPendingStepSignal(session.Metadata) + require.True(t, ok, "expected bag entry to be persisted") + assert.Equal(t, "step-1", bag.StepID) + assert.Equal(t, models.StepCompletionSourceAgent, bag.Source) + assert.Equal(t, "implementation finished", bag.Summary) + assert.Equal(t, "tests next", bag.Handoff) + + // Bus event published with the public payload shape (no handoff/blockers + // on the wire — those live in the bag only). + require.Len(t, bus.events, 1, "expected one bus publish") + assert.Equal(t, events.WorkflowStepCompletionSignaled, bus.events[0].Type) + data, ok := bus.events[0].Data.(map[string]interface{}) + require.True(t, ok) + assert.Equal(t, "task-first", data["task_id"]) + assert.Equal(t, "session-first", data["session_id"]) + assert.Equal(t, "step-1", data["step_id"]) + assert.Equal(t, "implementation finished", data["summary"]) + _, hasHandoff := data["handoff"] + assert.False(t, hasHandoff, "handoff is bag-only, not on the wire") +} + +// TestHandleStepComplete_DedupRunningNoRepublish covers the +// `already_signaled` short-circuit while the session is still RUNNING. The +// inline turn-end path will pick up the bag — no re-publish is needed and +// none should fire (avoids a spurious second event for the subscriber). +func TestHandleStepComplete_DedupRunningNoRepublish(t *testing.T) { + ctx := context.Background() + svc, repo := newTestTaskService(t) + seedStepCompleteTarget(t, repo, "task-dup", "session-dup", "step-1", models.TaskSessionStateRunning) + // Pre-write the bag to simulate "first call already happened". + require.NoError(t, repo.SetSessionMetadataKey(ctx, "session-dup", models.SessionMetaKeyPendingStepCompletion, models.PendingStepCompletionSignal{ + StepID: "step-1", + Source: models.StepCompletionSourceAgent, + Summary: "first call", + SignaledAt: time.Now().UTC(), + })) + bus := &mcpRecordingEventBus{} + h := newStepCompleteHandler(t, svc, repo, bus) + + msg := makeWSMessage(t, ws.ActionMCPStepComplete, map[string]interface{}{ + "task_id": "task-dup", + "session_id": "session-dup", + "summary": "second call (same step)", + }) + resp, err := h.handleStepComplete(ctx, msg) + require.NoError(t, err) + require.NotNil(t, resp) + assert.Equal(t, ws.MessageTypeResponse, resp.Type) + + var payload map[string]interface{} + require.NoError(t, json.Unmarshal(resp.Payload, &payload)) + assert.Equal(t, false, payload["accepted"]) + assert.Equal(t, "already_signaled", payload["reason"]) + + assert.Empty(t, bus.events, "RUNNING dedup path must not re-publish (inline turn-end will consume the bag)") +} + +// TestHandleStepComplete_DedupWaitingRepublishes covers the retry-after- +// publish-failure path. When the session is WAITING_FOR_INPUT, the bag is +// already persisted but the orchestrator subscriber may have never fired +// (e.g., the first call's publish failed after the bag write). A retry +// MUST re-publish so the subscriber gets a chance to drive the transition, +// otherwise the session stays stuck until the user replies. +func TestHandleStepComplete_DedupWaitingRepublishes(t *testing.T) { + ctx := context.Background() + svc, repo := newTestTaskService(t) + seedStepCompleteTarget(t, repo, "task-retry", "session-retry", "step-1", models.TaskSessionStateWaitingForInput) + require.NoError(t, repo.SetSessionMetadataKey(ctx, "session-retry", models.SessionMetaKeyPendingStepCompletion, models.PendingStepCompletionSignal{ + StepID: "step-1", + Source: models.StepCompletionSourceAgent, + Summary: "first call", + SignaledAt: time.Now().UTC(), + })) + bus := &mcpRecordingEventBus{} + h := newStepCompleteHandler(t, svc, repo, bus) + + msg := makeWSMessage(t, ws.ActionMCPStepComplete, map[string]interface{}{ + "task_id": "task-retry", + "session_id": "session-retry", + "summary": "retry after publish failure", + }) + resp, err := h.handleStepComplete(ctx, msg) + require.NoError(t, err) + require.NotNil(t, resp) + assert.Equal(t, ws.MessageTypeResponse, resp.Type) + + var payload map[string]interface{} + require.NoError(t, json.Unmarshal(resp.Payload, &payload)) + assert.Equal(t, false, payload["accepted"]) + assert.Equal(t, "already_signaled", payload["reason"]) + + require.Len(t, bus.events, 1, "WAITING dedup must re-publish the bus event so the subscriber can drive the transition") + assert.Equal(t, events.WorkflowStepCompletionSignaled, bus.events[0].Type) +} diff --git a/apps/backend/internal/mcp/server/server.go b/apps/backend/internal/mcp/server/server.go index 5d99114d01..ba02d64751 100644 --- a/apps/backend/internal/mcp/server/server.go +++ b/apps/backend/internal/mcp/server/server.go @@ -6,6 +6,7 @@ import ( "context" "encoding/json" "fmt" + "strings" "sync" "time" @@ -375,6 +376,12 @@ func (s *Server) registerTools() { // (repository, branch) to. External mode has no such context. s.registerAddBranchToTaskTool() s.registerUpdateRepositoryBaseBranchTool() + count += 2 + // Task-mode only: ADR 0015 explicit step-completion signal. The + // tool targets the current (task, session, step) the MCP server + // was bound to, so it has no meaningful semantics outside a task + // session. + s.registerStepCompleteTool() count++ } s.logger.Info("registered MCP tools", @@ -685,6 +692,66 @@ func (s *Server) updateRepositoryBaseBranchHandler() server.ToolHandlerFunc { } } +// registerStepCompleteTool registers step_complete_kandev — the ADR 0015 +// explicit completion signal. The tool is bound to the current (task, session) +// and writes a pending-signal entry on the session's metadata bag; the +// orchestrator consumes that signal to drive the workflow's on_turn_complete +// transitions. Steps with `auto_advance_requires_signal=false` (the legacy +// default) ignore the signal entirely. +func (s *Server) registerStepCompleteTool() { + s.mcpServer.AddTool( + mcp.NewTool("step_complete_kandev", + mcp.WithDescription(`Signal that every user-stated requirement for the CURRENT workflow step is satisfied. + +WHEN TO CALL: +- All work for the current step is finished and the task is ready to move forward in the workflow. +- This is the LAST thing you do in the step — call it after the final tool call / commit / answer that completes the requested work. + +WHEN NOT TO CALL: +- You are about to ask the user a question (use ask_user_question_kandev instead and wait). +- The work is partially done or you ran into a blocker you couldn't resolve. +- You are mid-conversation and expect the user to reply with more direction. + +BEHAVIOUR: +- The call is idempotent within a step: subsequent calls return accepted=false with reason="already_signaled" and have no other effect. +- The call returns immediately. The workflow transition (if the step is configured to auto-advance) is driven asynchronously by the orchestrator on turn-end. +- If the user sends another message before the transition fires, the signal is cancelled and the conversation continues on the current step. Call again at the end of the new turn if appropriate. +- For steps that do NOT have auto-advance enabled, the call succeeds (accepted=true) but the workflow does not move automatically. The signal is discarded on the next turn start; there is no separate audit history to query later. + +The summary you provide is shown to the user in chat and may be forwarded to the next step's agent as a hand-off note.`), + mcp.WithString("summary", mcp.Required(), mcp.Description("One-paragraph plain-text summary of what was done in this step. Shown to the user.")), + mcp.WithString("handoff", mcp.Description("Optional context the next step's agent will need to pick up where you left off (decisions, open files, follow-ups).")), + mcp.WithString("blockers", mcp.Description("Optional list of known unresolved issues. Use sparingly — only when the step is complete in the sense that you cannot make further progress without input, not for normal partial work.")), + ), + s.wrapHandler("step_complete_kandev", s.stepCompleteHandler()), + ) +} + +func (s *Server) stepCompleteHandler() server.ToolHandlerFunc { + return func(ctx context.Context, req mcp.CallToolRequest) (*mcp.CallToolResult, error) { + if s.taskID == "" || s.sessionID == "" { + return mcp.NewToolResultError("step_complete_kandev requires a bound task and session"), nil + } + summary := strings.TrimSpace(req.GetString("summary", "")) + if summary == "" { + return mcp.NewToolResultError("summary is required"), nil + } + payload := map[string]interface{}{ + "task_id": s.taskID, + "session_id": s.sessionID, + "summary": summary, + "handoff": req.GetString("handoff", ""), + "blockers": req.GetString("blockers", ""), + } + var result map[string]interface{} + if err := s.backend.RequestPayload(ctx, ws.ActionMCPStepComplete, payload, &result); err != nil { + return mcp.NewToolResultError(err.Error()), nil + } + data, _ := json.MarshalIndent(result, "", " ") + return mcp.NewToolResultText(string(data)), nil + } +} + func (s *Server) registerInteractionTools() { s.mcpServer.AddTool( mcp.NewTool("ask_user_question_kandev", diff --git a/apps/backend/internal/mcp/server/server_test.go b/apps/backend/internal/mcp/server/server_test.go index 896daf6c2d..25141329f7 100644 --- a/apps/backend/internal/mcp/server/server_test.go +++ b/apps/backend/internal/mcp/server/server_test.go @@ -211,10 +211,11 @@ func TestServerModeTask_ToolCount(t *testing.T) { s := New(backend, "test-session", "test-task", 10005, log, "", false, ModeTask) tools := getRegisteredToolNames(s) // 13 kanban (incl. delete + archive task) + 1 add_branch_to_task + - // 1 update_repository_base_branch + 1 interaction + 4 plan + - // 1 related-tasks = 21. Task-document tools (list/get/write) are - // office-only. - assert.Equal(t, 21, len(tools)) + // 1 update_repository_base_branch + 1 step_complete (ADR 0015) + + // 1 interaction + 4 plan + 1 related-tasks = 22. Task-document tools + // (list/get/write) are office-only. + assert.Contains(t, tools, "step_complete_kandev", "ADR 0015 explicit-completion signal must be registered in task mode") + assert.Equal(t, 22, len(tools)) } func TestServerModeConfig_ToolCount(t *testing.T) { @@ -225,6 +226,7 @@ func TestServerModeConfig_ToolCount(t *testing.T) { s := New(backend, "test-session", "test-task", 10005, log, "", false, ModeConfig) tools := getRegisteredToolNames(s) // 12 workflow (incl. list_repositories + import_workflow) + 4 agent + 4 mcp + 5 executor + 6 task + 1 interaction = 32 + assert.NotContains(t, tools, "step_complete_kandev", "step_complete_kandev requires a live task session; must NOT register in config mode") assert.Equal(t, 32, len(tools)) } @@ -297,6 +299,7 @@ func TestServerModeOffice_ToolCount(t *testing.T) { tools := getRegisteredToolNames(s) // 4 plan + 1 interaction + 1 related-tasks + 3 task-documents = 9 // (delegate_task_kandev retired in favour of `agentctl kandev task create …`). + assert.NotContains(t, tools, "step_complete_kandev", "step_complete_kandev is kanban-task-only; office mode advances tasks via its own approval surface") assert.Equal(t, 9, len(tools)) } diff --git a/apps/backend/internal/orchestrator/event_handlers_step_completion.go b/apps/backend/internal/orchestrator/event_handlers_step_completion.go new file mode 100644 index 0000000000..096b13c524 --- /dev/null +++ b/apps/backend/internal/orchestrator/event_handlers_step_completion.go @@ -0,0 +1,171 @@ +package orchestrator + +import ( + "context" + + "go.uber.org/zap" + + "github.com/kandev/kandev/internal/events" + "github.com/kandev/kandev/internal/events/bus" + "github.com/kandev/kandev/internal/task/models" +) + +// subscribeStepCompletionEvents wires the ADR 0015 out-of-band subscriber +// for `step_complete_kandev` signals that arrive after the agent's turn +// already ended. Safe to call when the feature is gated off — the +// subscriber's gating check short-circuits on every event in that case. +func (s *Service) subscribeStepCompletionEvents() { + if s.eventBus == nil { + return + } + if _, err := s.eventBus.Subscribe(events.WorkflowStepCompletionSignaled, s.handleStepCompletionSignaled); err != nil { + s.logger.Error("failed to subscribe to workflow.step_completion_signaled events", zap.Error(err)) + } +} + +// handleStepCompletionSignaled adapts the bus.Subscribe callback signature +// (returns error) to onStepCompletionSignaled, which does its own logging +// and does not surface errors to the bus. +func (s *Service) handleStepCompletionSignaled(ctx context.Context, event *bus.Event) error { + s.onStepCompletionSignaled(ctx, event) + return nil +} + +// clearPendingStepSignal removes the pending bag entry from the session's +// metadata, both in-memory (so callers operating on the same struct see it +// gone) and in the DB (so a later reload doesn't resurrect a stale entry). +// Best-effort: on DB error the in-memory mutation still wins, since the +// orchestrator's read uses the in-memory copy for the rest of the turn. +func (s *Service) clearPendingStepSignal(ctx context.Context, session *models.TaskSession) { + if session == nil { + return + } + if session.Metadata != nil { + delete(session.Metadata, models.SessionMetaKeyPendingStepCompletion) + } + s.clearPendingStepSignalByID(ctx, session.ID) +} + +// clearPendingStepSignalByID issues only the DB write — skip when the +// caller has no in-memory session struct to update (or already discarded +// it). Same best-effort failure handling as clearPendingStepSignal. +func (s *Service) clearPendingStepSignalByID(ctx context.Context, sessionID string) { + if sessionID == "" { + return + } + if err := s.repo.SetSessionMetadataKey(ctx, sessionID, models.SessionMetaKeyPendingStepCompletion, nil); err != nil { + s.logger.Debug("clearPendingStepSignal: failed to persist nil bag entry", + zap.String("session_id", sessionID), zap.Error(err)) + } +} + +// onStepCompletionSignaled subscribes to events.WorkflowStepCompletionSignaled +// to handle the case where the agent's `step_complete_kandev` call lands +// AFTER the turn already ended — at that point processOnTurnCompleteViaEngine +// has already setSessionWaitingForInput. The subscriber re-triggers the +// transition pipeline so the gated step finally advances. +// +// Happy path (call lands before turn-end): the bag is already populated by +// the time processOnTurnCompleteViaEngine runs, the gating check passes, +// and the transition fires inline — the bus event arrives later and finds +// nothing to do (bag already cleared by the transition). Idempotent. +func (s *Service) onStepCompletionSignaled(ctx context.Context, event *bus.Event) { + taskID, sessionID, stepID, ok := parseStepCompletionEvent(event) + if !ok { + if event != nil && event.Data != nil { + if _, isMap := event.Data.(map[string]interface{}); !isMap { + s.logger.Warn("onStepCompletionSignaled: unexpected event payload type") + } + } + return + } + + session, err := s.repo.GetTaskSession(ctx, sessionID) + if err != nil { + s.logger.Warn("onStepCompletionSignaled: failed to load session", + zap.String("session_id", sessionID), zap.Error(err)) + return + } + // If the session is still running (turn hasn't ended yet) the inline + // turn-end check will pick the signal up — no out-of-band work needed. + // Only act on signals that arrive while the session is waiting. + if session.State != models.TaskSessionStateWaitingForInput { + return + } + + task, err := s.repo.GetTask(ctx, taskID) + if err != nil { + s.logger.Warn("onStepCompletionSignaled: failed to load task", + zap.String("task_id", taskID), zap.Error(err)) + return + } + if task.WorkflowStepID != stepID { + s.logger.Debug("onStepCompletionSignaled: signal stale (step changed)", + zap.String("signal_step", stepID), zap.String("current_step", task.WorkflowStepID)) + // Only clear the bag when its current contents are themselves + // stale (matching THIS subscriber's stepID). Re-load the session + // before checking: the local `session` snapshot was taken when + // the subscriber fired and may be older than a concurrent + // write from the new step's agent. Without the reload the + // in-memory check could see a stale signal that's already been + // cleared/replaced and erase a freshly-written valid bag entry. + latestSession, loadErr := s.repo.GetTaskSession(ctx, sessionID) + if loadErr != nil { + return + } + if existing, ok := models.LoadPendingStepSignal(latestSession.Metadata); ok && existing.StepID == stepID { + s.clearPendingStepSignal(ctx, latestSession) + } + return + } + + if !s.stepIsSignalGated(ctx, task.WorkflowStepID) { + return + } + + // Drive the transition via the engine path. It will re-read the bag + // and consume it through the same code path the inline turn-end uses. + s.processOnTurnCompleteViaEngine(ctx, taskID, session) +} + +// parseStepCompletionEvent extracts (task_id, session_id, step_id) from a +// step-completion bus event. Returns ok=false when the event is nil/empty, +// the payload isn't a map, or any required ID is missing. +func parseStepCompletionEvent(event *bus.Event) (taskID, sessionID, stepID string, ok bool) { + if event == nil || event.Data == nil { + return "", "", "", false + } + data, isMap := event.Data.(map[string]interface{}) + if !isMap { + return "", "", "", false + } + taskID = models.StringFromAny(data["task_id"]) + sessionID = models.StringFromAny(data["session_id"]) + stepID = models.StringFromAny(data["step_id"]) + if taskID == "" || sessionID == "" || stepID == "" { + return "", "", "", false + } + return taskID, sessionID, stepID, true +} + +// stepIsSignalGated reports whether the given workflow step opts in to +// the ADR-0015 explicit-signal gate. False on a missing step-getter or +// any lookup error — those land the subscriber on the safe side (don't +// auto-advance a step we can't classify). +func (s *Service) stepIsSignalGated(ctx context.Context, stepID string) bool { + if s.workflowStepGetter == nil { + return false + } + currentStep, err := s.workflowStepGetter.GetStep(ctx, stepID) + if err != nil || currentStep == nil { + s.logger.Debug("onStepCompletionSignaled: cannot load current step, skipping", + zap.String("step_id", stepID), zap.Error(err)) + return false + } + if !currentStep.AutoAdvanceRequiresSignal { + s.logger.Debug("onStepCompletionSignaled: step is not signal-gated, ignoring", + zap.String("step_id", currentStep.ID)) + return false + } + return true +} diff --git a/apps/backend/internal/orchestrator/event_handlers_step_completion_test.go b/apps/backend/internal/orchestrator/event_handlers_step_completion_test.go new file mode 100644 index 0000000000..1a0bef5ad8 --- /dev/null +++ b/apps/backend/internal/orchestrator/event_handlers_step_completion_test.go @@ -0,0 +1,359 @@ +package orchestrator + +import ( + "context" + "testing" + "time" + + "github.com/kandev/kandev/internal/events/bus" + "github.com/kandev/kandev/internal/task/models" + wfmodels "github.com/kandev/kandev/internal/workflow/models" +) + +// TestProcessOnTurnComplete_ExplicitSignalGating verifies the ADR 0015 +// gating: when AutoAdvanceRequiresSignal=true, turn-end without a matching +// pending signal must NOT transition. With the signal present, the +// transition fires as normal. +func TestProcessOnTurnComplete_ExplicitSignalGating(t *testing.T) { + ctx := context.Background() + + build := func(t *testing.T, withSignal bool, stepRequires bool) (svc *Service, taskID, sessionID string) { + t.Helper() + repo := setupTestRepo(t) + seedSession(t, repo, "t1", "s1", "step1") + + stepGetter := newMockStepGetter() + stepGetter.steps["step1"] = &wfmodels.WorkflowStep{ + ID: "step1", WorkflowID: "wf1", Name: "Step 1", Position: 0, + AutoAdvanceRequiresSignal: stepRequires, + Events: wfmodels.StepEvents{ + OnTurnComplete: []wfmodels.OnTurnCompleteAction{ + {Type: wfmodels.OnTurnCompleteMoveToNext}, + }, + }, + } + stepGetter.steps["step2"] = &wfmodels.WorkflowStep{ + ID: "step2", WorkflowID: "wf1", Name: "Step 2", Position: 1, + } + + svc = createTestService(repo, stepGetter, newMockTaskRepo()) + + if withSignal { + signal := models.PendingStepCompletionSignal{ + StepID: "step1", + Source: models.StepCompletionSourceAgent, + Summary: "all done", + SignaledAt: time.Now().UTC(), + } + if err := repo.SetSessionMetadataKey(ctx, "s1", models.SessionMetaKeyPendingStepCompletion, signal); err != nil { + t.Fatalf("seed pending signal: %v", err) + } + } + return svc, "t1", "s1" + } + + t.Run("step requires, no signal → no transition", func(t *testing.T) { + svc, taskID, sessionID := build(t, false, true) + task, _ := svc.repo.GetTask(ctx, taskID) + session, _ := svc.repo.GetTaskSession(ctx, sessionID) + if got := svc.processOnTurnComplete(ctx, task, session); got { + t.Errorf("expected gating to BLOCK transition, got transition=true") + } + updated, _ := svc.repo.GetTask(ctx, taskID) + if updated.WorkflowStepID != "step1" { + t.Errorf("expected to stay on step1, got %q", updated.WorkflowStepID) + } + }) + + t.Run("step requires, signal present → transition fires", func(t *testing.T) { + svc, taskID, sessionID := build(t, true, true) + task, _ := svc.repo.GetTask(ctx, taskID) + session, _ := svc.repo.GetTaskSession(ctx, sessionID) + if got := svc.processOnTurnComplete(ctx, task, session); !got { + t.Errorf("expected transition with pending signal, got transition=false") + } + updated, _ := svc.repo.GetTask(ctx, taskID) + if updated.WorkflowStepID != "step2" { + t.Errorf("expected to move to step2, got %q", updated.WorkflowStepID) + } + }) + + t.Run("step does not require → legacy behaviour", func(t *testing.T) { + svc, taskID, sessionID := build(t, false, false) + task, _ := svc.repo.GetTask(ctx, taskID) + session, _ := svc.repo.GetTaskSession(ctx, sessionID) + if got := svc.processOnTurnComplete(ctx, task, session); !got { + t.Errorf("expected transition (step does not require signal), got transition=false") + } + }) + + t.Run("step requires, signal for DIFFERENT step → still blocked", func(t *testing.T) { + svc, taskID, sessionID := build(t, false, true) + stale := models.PendingStepCompletionSignal{ + StepID: "step_old", // stale entry — doesn't match current step + Source: models.StepCompletionSourceAgent, + Summary: "stale", + SignaledAt: time.Now().UTC(), + } + if err := svc.repo.SetSessionMetadataKey(ctx, sessionID, models.SessionMetaKeyPendingStepCompletion, stale); err != nil { + t.Fatalf("seed stale signal: %v", err) + } + task, _ := svc.repo.GetTask(ctx, taskID) + session, _ := svc.repo.GetTaskSession(ctx, sessionID) + if got := svc.processOnTurnComplete(ctx, task, session); got { + t.Errorf("expected stale signal to be treated as absent, but got transition=true") + } + }) +} + +// TestLoadPendingStepSignal_RoundTrip verifies the bag survives JSON +// rehydration — important for the backend-restart path where the bag is +// read from the DB as map[string]interface{} rather than the typed struct. +func TestLoadPendingStepSignal_RoundTrip(t *testing.T) { + t.Run("typed struct", func(t *testing.T) { + now := time.Now().UTC().Truncate(time.Nanosecond) + want := models.PendingStepCompletionSignal{ + StepID: "step-1", Source: "agent", Summary: "ok", SignaledAt: now, + } + meta := map[string]interface{}{ + models.SessionMetaKeyPendingStepCompletion: want, + } + got, ok := models.LoadPendingStepSignal(meta) + if !ok || got.StepID != "step-1" || got.Source != "agent" { + t.Errorf("typed struct round-trip failed: ok=%v got=%+v", ok, got) + } + }) + + t.Run("json-rehydrated map", func(t *testing.T) { + meta := map[string]interface{}{ + models.SessionMetaKeyPendingStepCompletion: map[string]interface{}{ + "step_id": "step-2", + "source": "manual_fallback", + "summary": "user marked complete", + "signaled_at": "2026-06-04T12:00:00Z", + }, + } + got, ok := models.LoadPendingStepSignal(meta) + if !ok { + t.Fatal("expected models.LoadPendingStepSignal to recognise map shape") + } + if got.StepID != "step-2" || got.Source != "manual_fallback" || got.Summary != "user marked complete" { + t.Errorf("map round-trip mismatch: %+v", got) + } + }) + + t.Run("absent key returns false", func(t *testing.T) { + _, ok := models.LoadPendingStepSignal(map[string]interface{}{}) + if ok { + t.Error("expected ok=false on empty metadata") + } + }) + + t.Run("nil metadata returns false", func(t *testing.T) { + _, ok := models.LoadPendingStepSignal(nil) + if ok { + t.Error("expected ok=false on nil metadata") + } + }) +} + +// TestOnStepCompletionSignaled covers the out-of-band subscriber that +// drives a step transition when a `step_complete_kandev` signal arrives +// AFTER the turn has already ended. The three branches: +// +// - session still RUNNING (turn in flight): no-op, inline path will handle it. +// - session WAITING + step matches + step gated: re-runs transition pipeline. +// - signal stale (step has changed under us): clear the bag, no transition. +// - step not signal-gated: do not advance (signal is not a manual-advance trigger). +func TestOnStepCompletionSignaled(t *testing.T) { + ctx := context.Background() + + buildEvent := func(taskID, sessionID, stepID string) *bus.Event { + return bus.NewEvent("workflow.step_completion_signaled", "test", map[string]interface{}{ + "task_id": taskID, + "session_id": sessionID, + "step_id": stepID, + }) + } + + t.Run("session still RUNNING — subscriber is a no-op", func(t *testing.T) { + repo := setupTestRepo(t) + seedSession(t, repo, "t1", "s1", "step1") + // seedSession leaves the session in RUNNING; that's what we want. + + stepGetter := newMockStepGetter() + stepGetter.steps["step1"] = &wfmodels.WorkflowStep{ + ID: "step1", WorkflowID: "wf1", Name: "Step 1", Position: 0, + AutoAdvanceRequiresSignal: true, + Events: wfmodels.StepEvents{ + OnTurnComplete: []wfmodels.OnTurnCompleteAction{ + {Type: wfmodels.OnTurnCompleteMoveToNext}, + }, + }, + } + stepGetter.steps["step2"] = &wfmodels.WorkflowStep{ + ID: "step2", WorkflowID: "wf1", Name: "Step 2", Position: 1, + } + svc := createTestService(repo, stepGetter, newMockTaskRepo()) + + svc.onStepCompletionSignaled(ctx, buildEvent("t1", "s1", "step1")) + + updated, _ := repo.GetTask(ctx, "t1") + if updated.WorkflowStepID != "step1" { + t.Errorf("expected to stay on step1 (turn in flight), got %q", updated.WorkflowStepID) + } + }) + + t.Run("WAITING + matching step + gated → transition fires", func(t *testing.T) { + repo := setupTestRepo(t) + seedSession(t, repo, "t1", "s1", "step1") + // Flip session to WAITING_FOR_INPUT (the only state the subscriber acts on). + if err := repo.UpdateTaskSessionState(ctx, "s1", models.TaskSessionStateWaitingForInput, ""); err != nil { + t.Fatalf("flip session waiting: %v", err) + } + // Pre-write the signal in the bag — the subscriber re-runs the + // inline turn-end path, which reads the bag for gating. + signal := models.PendingStepCompletionSignal{ + StepID: "step1", + Source: models.StepCompletionSourceAgent, + Summary: "ok", + SignaledAt: time.Now().UTC(), + } + if err := repo.SetSessionMetadataKey(ctx, "s1", models.SessionMetaKeyPendingStepCompletion, signal); err != nil { + t.Fatalf("seed bag: %v", err) + } + + stepGetter := newMockStepGetter() + stepGetter.steps["step1"] = &wfmodels.WorkflowStep{ + ID: "step1", WorkflowID: "wf1", Name: "Step 1", Position: 0, + AutoAdvanceRequiresSignal: true, + Events: wfmodels.StepEvents{ + OnTurnComplete: []wfmodels.OnTurnCompleteAction{ + {Type: wfmodels.OnTurnCompleteMoveToNext}, + }, + }, + } + stepGetter.steps["step2"] = &wfmodels.WorkflowStep{ + ID: "step2", WorkflowID: "wf1", Name: "Step 2", Position: 1, + } + svc := createTestService(repo, stepGetter, newMockTaskRepo()) + + svc.onStepCompletionSignaled(ctx, buildEvent("t1", "s1", "step1")) + + updated, _ := repo.GetTask(ctx, "t1") + if updated.WorkflowStepID != "step2" { + t.Errorf("expected transition to step2, got %q", updated.WorkflowStepID) + } + }) + + t.Run("stale step → bag cleared, no transition", func(t *testing.T) { + repo := setupTestRepo(t) + seedSession(t, repo, "t1", "s1", "step_current") + if err := repo.UpdateTaskSessionState(ctx, "s1", models.TaskSessionStateWaitingForInput, ""); err != nil { + t.Fatalf("flip session waiting: %v", err) + } + // Stale signal: written when step was "step_old", but the task has + // already moved on to "step_current" via some other path. + stale := models.PendingStepCompletionSignal{ + StepID: "step_old", + Source: models.StepCompletionSourceAgent, + Summary: "stale", + SignaledAt: time.Now().UTC(), + } + if err := repo.SetSessionMetadataKey(ctx, "s1", models.SessionMetaKeyPendingStepCompletion, stale); err != nil { + t.Fatalf("seed stale signal: %v", err) + } + + stepGetter := newMockStepGetter() + stepGetter.steps["step_current"] = &wfmodels.WorkflowStep{ + ID: "step_current", WorkflowID: "wf1", Name: "Current", Position: 5, + } + svc := createTestService(repo, stepGetter, newMockTaskRepo()) + + svc.onStepCompletionSignaled(ctx, buildEvent("t1", "s1", "step_old")) + + updatedSession, _ := repo.GetTaskSession(ctx, "s1") + if _, hasBag := models.LoadPendingStepSignal(updatedSession.Metadata); hasBag { + t.Error("expected stale bag entry to be cleared") + } + updatedTask, _ := repo.GetTask(ctx, "t1") + if updatedTask.WorkflowStepID != "step_current" { + t.Errorf("expected no transition (stale signal), got %q", updatedTask.WorkflowStepID) + } + }) + + t.Run("stale event → valid bag for CURRENT step is preserved", func(t *testing.T) { + // Pins the negative side of the StepID guard in the subscriber's + // stale-step branch: a late step-A event must not erase a + // freshly-written step-B bag (which can happen when the session + // is reused across steps without auto_start_agent). A regression + // here would silently leave signal-gated steps stuck waiting. + repo := setupTestRepo(t) + seedSession(t, repo, "t1", "s1", "step_current") + if err := repo.UpdateTaskSessionState(ctx, "s1", models.TaskSessionStateWaitingForInput, ""); err != nil { + t.Fatalf("flip session waiting: %v", err) + } + // Bag holds a VALID signal for the current step (step_current). + valid := models.PendingStepCompletionSignal{ + StepID: "step_current", + Source: models.StepCompletionSourceAgent, + Summary: "valid current-step signal", + SignaledAt: time.Now().UTC(), + } + if err := repo.SetSessionMetadataKey(ctx, "s1", models.SessionMetaKeyPendingStepCompletion, valid); err != nil { + t.Fatalf("seed valid signal: %v", err) + } + + stepGetter := newMockStepGetter() + stepGetter.steps["step_current"] = &wfmodels.WorkflowStep{ + ID: "step_current", WorkflowID: "wf1", Name: "Current", Position: 5, + } + svc := createTestService(repo, stepGetter, newMockTaskRepo()) + + // Fire a STALE event (step_old != current step_current). The + // guard must see that the bag's StepID is "step_current" (not + // "step_old") and leave it alone. + svc.onStepCompletionSignaled(ctx, buildEvent("t1", "s1", "step_old")) + + updatedSession, _ := repo.GetTaskSession(ctx, "s1") + bag, hasBag := models.LoadPendingStepSignal(updatedSession.Metadata) + if !hasBag { + t.Fatal("expected valid bag to survive stale event") + } + if bag.StepID != "step_current" { + t.Errorf("expected bag StepID=step_current, got %q", bag.StepID) + } + }) + + t.Run("step not signal-gated → subscriber ignores", func(t *testing.T) { + repo := setupTestRepo(t) + seedSession(t, repo, "t1", "s1", "step1") + if err := repo.UpdateTaskSessionState(ctx, "s1", models.TaskSessionStateWaitingForInput, ""); err != nil { + t.Fatalf("flip session waiting: %v", err) + } + + // Step explicitly NOT gated on the signal — even though one was + // written and matches, the subscriber must not advance. + stepGetter := newMockStepGetter() + stepGetter.steps["step1"] = &wfmodels.WorkflowStep{ + ID: "step1", WorkflowID: "wf1", Name: "Step 1", Position: 0, + AutoAdvanceRequiresSignal: false, + Events: wfmodels.StepEvents{ + OnTurnComplete: []wfmodels.OnTurnCompleteAction{ + {Type: wfmodels.OnTurnCompleteMoveToNext}, + }, + }, + } + stepGetter.steps["step2"] = &wfmodels.WorkflowStep{ + ID: "step2", WorkflowID: "wf1", Name: "Step 2", Position: 1, + } + svc := createTestService(repo, stepGetter, newMockTaskRepo()) + + svc.onStepCompletionSignaled(ctx, buildEvent("t1", "s1", "step1")) + + updated, _ := repo.GetTask(ctx, "t1") + if updated.WorkflowStepID != "step1" { + t.Errorf("expected no transition for un-gated step, got %q", updated.WorkflowStepID) + } + }) +} diff --git a/apps/backend/internal/orchestrator/event_handlers_workflow.go b/apps/backend/internal/orchestrator/event_handlers_workflow.go index 76b04ffbf3..29f24da333 100644 --- a/apps/backend/internal/orchestrator/event_handlers_workflow.go +++ b/apps/backend/internal/orchestrator/event_handlers_workflow.go @@ -56,6 +56,20 @@ func (s *Service) processOnTurnComplete(ctx context.Context, task *models.Task, return false } + // ADR 0015 — explicit completion signal gating (legacy path mirror of + // processOnTurnCompleteViaEngine). Steps marked + // `auto_advance_requires_signal=true` wait for a step_complete_kandev + // signal before evaluating their transition actions. + if currentStep.AutoAdvanceRequiresSignal { + signal, has := models.LoadPendingStepSignal(session.Metadata) + if !has || signal.StepID != currentStep.ID { + s.logger.Debug("on_turn_complete gated on explicit signal (legacy path)", + zap.String("step_id", currentStep.ID)) + s.setSessionWaitingForInput(ctx, taskID, sessionID, session) + return false + } + } + // Process side-effect actions first, then find the first transition action transitionAction := s.processTurnCompleteActions(ctx, session, currentStep) @@ -194,6 +208,10 @@ func (s *Service) ProcessOnTurnStart(ctx context.Context, taskID, sessionID stri if err != nil { return fmt.Errorf("load session for on_turn_start: %w", err) } + // ADR 0015 — a fresh user message before the pending signal's + // transition has fired cancels the signal (re-open semantics). The + // user is continuing the conversation; this step is no longer "done". + s.clearPendingStepSignal(ctx, session) s.processOnTurnStartViaEngine(ctx, taskID, session) return nil } @@ -257,6 +275,15 @@ func (s *Service) executeStepTransition(ctx context.Context, taskID, sessionID s zap.Bool("trigger_on_enter", triggerOnEnter)) if triggerOnEnter { + // ADR 0015 — clear any pending completion-signal bag for the + // step we just left. Only on_turn_complete transitions trigger + // gating, so the triggerOnEnter=true branch is the only one + // that could have consumed a signal; on_turn_start moves leave + // the bag alone (it's still tied to an unsignaled step we have + // not left). The session struct isn't used after this point, + // so skip the extra GetTaskSession round-trip and write + // straight to the DB by session_id. + s.clearPendingStepSignalByID(ctx, sessionID) // Automated transitions always clear review: the agent just completed // a turn, so any pending review from a prior step is stale regardless // of whether the new step has auto_start_agent. @@ -1978,6 +2005,47 @@ func (s *Service) processOnTurnCompleteViaEngine(ctx context.Context, taskID str return false } + // ADR 0015 — explicit completion signal gating. Steps marked + // `auto_advance_requires_signal=true` only transition when the agent + // (or the manual fallback button) has written the pending bag entry. + // On gate-fail we set the session to WAITING_FOR_INPUT and bail — + // either the user replies (clearing the bag) or a later + // step_complete_kandev call triggers the out-of-band subscriber. + // + // Fail closed on step-load errors: a missing/broken step record must + // not silently bypass the gate (which would let a signal-required step + // auto-advance whenever the loader hiccups). Block the transition and + // let the next turn re-evaluate after the underlying error clears. + currentStep, stepErr := s.workflowStepGetter.GetStep(ctx, task.WorkflowStepID) + if stepErr != nil || currentStep == nil { + s.logger.Warn("on_turn_complete: failed to load current step for signal gating, blocking transition", + zap.String("task_id", taskID), + zap.String("session_id", session.ID), + zap.String("step_id", task.WorkflowStepID), + zap.Error(stepErr)) + s.setSessionWaitingForInput(ctx, taskID, session.ID, session) + return false + } + if currentStep.AutoAdvanceRequiresSignal { + signal, has := models.LoadPendingStepSignal(session.Metadata) + if !has || signal.StepID != task.WorkflowStepID { + s.logger.Debug("on_turn_complete gated on explicit signal (none received yet)", + zap.String("task_id", taskID), + zap.String("session_id", session.ID), + zap.String("step_id", task.WorkflowStepID)) + s.setSessionWaitingForInput(ctx, taskID, session.ID, session) + return false + } + s.logger.Info("on_turn_complete consuming explicit signal", + zap.String("task_id", taskID), + zap.String("session_id", session.ID), + zap.String("step_id", task.WorkflowStepID), + zap.String("source", signal.Source)) + // Bag is consumed once the transition executes (in + // applyEngineTransition's stamp + clear), so don't clear here — + // otherwise a failed transition would lose the signal. + } + state := s.buildMachineState(ctx, task, session) result, err := s.workflowEngine.HandleTrigger(ctx, engine.HandleInput{ TaskID: taskID, @@ -2061,6 +2129,13 @@ func (s *Service) applyEngineTransition( return false } + // ADR 0015 — a successful on_turn_complete transition consumes any + // pending step-completion signal for the source step. The bag must be + // cleared so the next step's gating starts from a clean slate. + if trigger == engine.TriggerOnTurnComplete { + s.clearPendingStepSignal(ctx, session) + } + if len(result.DataPatch) > 0 { if err := s.workflowStore.PersistData(ctx, session.ID, result.DataPatch); err != nil { s.logger.Warn("failed to persist workflow data patch", diff --git a/apps/backend/internal/orchestrator/service.go b/apps/backend/internal/orchestrator/service.go index 2124ae6d14..9cb0e3befa 100644 --- a/apps/backend/internal/orchestrator/service.go +++ b/apps/backend/internal/orchestrator/service.go @@ -864,6 +864,10 @@ func (s *Service) Start(ctx context.Context) error { // Subscribe to prepare events (persist result in session metadata) s.subscribePrepareEvents() + // Subscribe to ADR-0015 step-completion signals (out-of-band path: + // signal arrives after turn-end). + s.subscribeStepCompletionEvents() + s.logger.Info("orchestrator service started successfully") return nil } diff --git a/apps/backend/internal/task/models/models.go b/apps/backend/internal/task/models/models.go index a5a8626416..1cc1a314bf 100644 --- a/apps/backend/internal/task/models/models.go +++ b/apps/backend/internal/task/models/models.go @@ -76,6 +76,81 @@ const ( // SSR reload, alongside the in-memory re-apply on context reset. See issue #1183. const SessionMetaKeySessionMode = "session_mode" +// SessionMetaKeyPendingStepCompletion stores the agent's (or manual fallback's) +// step-complete signal under TaskSession.Metadata. ADR 0015: the orchestrator +// reads this on turn-end for steps with AutoAdvanceRequiresSignal=true and +// only fires on_turn_complete transitions when a matching signal is present. +// Cleared on successful transition, on user reply before transition, or any +// other step change. +const SessionMetaKeyPendingStepCompletion = "pending_step_completion_signal" + +// PendingStepCompletionSignal source values. +const ( + StepCompletionSourceAgent = "agent" + StepCompletionSourceManualFallback = "manual_fallback" +) + +// PendingStepCompletionSignal is the JSON shape persisted under +// TaskSession.Metadata[SessionMetaKeyPendingStepCompletion]. It records an +// agent-emitted (or user-emitted via the fallback button) completion signal +// that the orchestrator should consume to drive a workflow step transition. +// See ADR 0015 for the lifecycle (set → read → clear). +type PendingStepCompletionSignal struct { + StepID string `json:"step_id"` + Source string `json:"source"` + Summary string `json:"summary"` + Handoff string `json:"handoff,omitempty"` + Blockers string `json:"blockers,omitempty"` + SignaledAt time.Time `json:"signaled_at"` +} + +// LoadPendingStepSignal decodes the pending-completion bag entry from a +// session's metadata. Single source of truth shared by the MCP handler +// (write site, idempotency check) and the orchestrator (read site, gating). +// Survives both the in-process typed shape and the JSON-rehydrated +// `map[string]interface{}` shape produced when the row is loaded fresh from +// SQLite after a backend restart. +func LoadPendingStepSignal(metadata map[string]interface{}) (PendingStepCompletionSignal, bool) { + if metadata == nil { + return PendingStepCompletionSignal{}, false + } + raw, ok := metadata[SessionMetaKeyPendingStepCompletion] + if !ok || raw == nil { + return PendingStepCompletionSignal{}, false + } + switch v := raw.(type) { + case PendingStepCompletionSignal: + return v, true + case map[string]interface{}: + out := PendingStepCompletionSignal{ + StepID: StringFromAny(v["step_id"]), + Source: StringFromAny(v["source"]), + Summary: StringFromAny(v["summary"]), + Handoff: StringFromAny(v["handoff"]), + Blockers: StringFromAny(v["blockers"]), + } + if ts, ok := v["signaled_at"].(string); ok { + if parsed, err := time.Parse(time.RFC3339Nano, ts); err == nil { + out.SignaledAt = parsed + } + } + return out, out.StepID != "" + } + return PendingStepCompletionSignal{}, false +} + +// StringFromAny narrows an interface{} slot to a string, returning "" when +// the value is absent or of a different type. Used by both the +// PendingStepCompletionSignal map-shape decoder and the orchestrator's +// step-completion event-payload parser — single source of truth so the +// two decoders can't drift on what counts as a "missing string". +func StringFromAny(v interface{}) string { + if s, ok := v.(string); ok { + return s + } + return "" +} + // Task origin values for the Origin field. const ( TaskOriginManual = "manual" diff --git a/apps/backend/internal/workflow/controller/controller.go b/apps/backend/internal/workflow/controller/controller.go index dcf14081cb..cb407da7ce 100644 --- a/apps/backend/internal/workflow/controller/controller.go +++ b/apps/backend/internal/workflow/controller/controller.go @@ -94,15 +94,16 @@ func (c *Controller) CreateStepsFromTemplate(ctx context.Context, req CreateStep // CreateStepRequest is the request for creating a single workflow step. type CreateStepRequest struct { - WorkflowID string `json:"workflow_id"` - Name string `json:"name"` - Position int `json:"position"` - Color string `json:"color"` - Prompt string `json:"prompt,omitempty"` - Events *models.StepEvents `json:"events,omitempty"` - AllowManualMove bool `json:"allow_manual_move"` - IsStartStep *bool `json:"is_start_step,omitempty"` - ShowInCommandPanel *bool `json:"show_in_command_panel,omitempty"` + WorkflowID string `json:"workflow_id"` + Name string `json:"name"` + Position int `json:"position"` + Color string `json:"color"` + Prompt string `json:"prompt,omitempty"` + Events *models.StepEvents `json:"events,omitempty"` + AllowManualMove bool `json:"allow_manual_move"` + IsStartStep *bool `json:"is_start_step,omitempty"` + ShowInCommandPanel *bool `json:"show_in_command_panel,omitempty"` + AutoAdvanceRequiresSignal *bool `json:"auto_advance_requires_signal,omitempty"` } // CreateStep creates a new workflow step. @@ -126,6 +127,9 @@ func (c *Controller) CreateStep(ctx context.Context, req CreateStepRequest) (*Ge } else { step.ShowInCommandPanel = true // default to visible } + if req.AutoAdvanceRequiresSignal != nil { + step.AutoAdvanceRequiresSignal = *req.AutoAdvanceRequiresSignal + } if err := c.svc.CreateStep(ctx, step); err != nil { return nil, err } @@ -134,17 +138,18 @@ func (c *Controller) CreateStep(ctx context.Context, req CreateStepRequest) (*Ge // UpdateStepRequest is the request for updating a workflow step. type UpdateStepRequest struct { - ID string `json:"id"` - Name *string `json:"name,omitempty"` - Position *int `json:"position,omitempty"` - Color *string `json:"color,omitempty"` - Prompt *string `json:"prompt,omitempty"` - Events *models.StepEvents `json:"events,omitempty"` - AllowManualMove *bool `json:"allow_manual_move,omitempty"` - IsStartStep *bool `json:"is_start_step,omitempty"` - ShowInCommandPanel *bool `json:"show_in_command_panel,omitempty"` - AutoArchiveAfterHours *int `json:"auto_archive_after_hours,omitempty"` - AgentProfileID *string `json:"agent_profile_id,omitempty"` + ID string `json:"id"` + Name *string `json:"name,omitempty"` + Position *int `json:"position,omitempty"` + Color *string `json:"color,omitempty"` + Prompt *string `json:"prompt,omitempty"` + Events *models.StepEvents `json:"events,omitempty"` + AllowManualMove *bool `json:"allow_manual_move,omitempty"` + IsStartStep *bool `json:"is_start_step,omitempty"` + ShowInCommandPanel *bool `json:"show_in_command_panel,omitempty"` + AutoArchiveAfterHours *int `json:"auto_archive_after_hours,omitempty"` + AgentProfileID *string `json:"agent_profile_id,omitempty"` + AutoAdvanceRequiresSignal *bool `json:"auto_advance_requires_signal,omitempty"` } // UpdateStep updates an existing workflow step. @@ -183,6 +188,9 @@ func (c *Controller) UpdateStep(ctx context.Context, req UpdateStepRequest) (*Ge if req.AgentProfileID != nil { step.AgentProfileID = strings.TrimSpace(*req.AgentProfileID) } + if req.AutoAdvanceRequiresSignal != nil { + step.AutoAdvanceRequiresSignal = *req.AutoAdvanceRequiresSignal + } if err := c.svc.UpdateStep(ctx, step); err != nil { return nil, err } diff --git a/apps/backend/internal/workflow/models/models.go b/apps/backend/internal/workflow/models/models.go index f047300a65..ecf81e2ccd 100644 --- a/apps/backend/internal/workflow/models/models.go +++ b/apps/backend/internal/workflow/models/models.go @@ -169,6 +169,9 @@ type StepDefinition struct { // default + coordination workflows can declare their UX role // ("work", "review", "approval", "custom") in YAML. StageType StageType `json:"stage_type,omitempty"` + // AutoAdvanceRequiresSignal gates on_turn_complete transitions on an + // explicit `step_complete_kandev` MCP signal from the agent (ADR 0015). + AutoAdvanceRequiresSignal bool `json:"auto_advance_requires_signal,omitempty" yaml:"auto_advance_requires_signal,omitempty"` } // WorkflowStep represents a step in a workflow @@ -190,8 +193,14 @@ type WorkflowStep struct { // on it. Stored as TEXT in workflow_steps.stage_type, defaulting to // "custom" so existing rows remain unchanged. StageType StageType `json:"stage_type,omitempty"` - CreatedAt time.Time `json:"created_at"` - UpdatedAt time.Time `json:"updated_at"` + // AutoAdvanceRequiresSignal gates on_turn_complete transitions on an + // explicit `step_complete_kandev` MCP signal from the agent (ADR 0015). + // When true, bare turn-end does NOT trigger the step's transition + // actions; instead the orchestrator waits for the agent (or a manual + // UI fallback) to write the pending-signal bag on TaskSession.Metadata. + AutoAdvanceRequiresSignal bool `json:"auto_advance_requires_signal,omitempty"` + CreatedAt time.Time `json:"created_at"` + UpdatedAt time.Time `json:"updated_at"` } // HasOnEnterAction checks if the step has a specific on_enter action type. diff --git a/apps/backend/internal/workflow/repository/sqlite.go b/apps/backend/internal/workflow/repository/sqlite.go index 410dce9410..8b64feadb5 100644 --- a/apps/backend/internal/workflow/repository/sqlite.go +++ b/apps/backend/internal/workflow/repository/sqlite.go @@ -112,6 +112,10 @@ func (r *Repository) initSchema() error { // frontend ("work" | "review" | "approval" | "custom"). Backend code // MUST NOT branch on it. Idempotent ALTER; default keeps existing rows at "custom". r.migrate.Apply("workflow_steps.stage_type", `ALTER TABLE workflow_steps ADD COLUMN stage_type TEXT NOT NULL DEFAULT 'custom'`) + // ADR 0015 — gate auto-advance on an explicit `step_complete_kandev` + // MCP signal. Idempotent ALTER; existing rows keep today's behaviour + // (immediate transition on turn-end) until the column is set to 1. + r.migrate.Apply("workflow_steps.auto_advance_requires_signal", `ALTER TABLE workflow_steps ADD COLUMN auto_advance_requires_signal INTEGER NOT NULL DEFAULT 0`) // Phase 2 — multi-agent participation tables. Empty rows for a step // preserve today's single-agent behaviour, so existing kanban @@ -558,11 +562,11 @@ func (r *Repository) CreateStep(ctx context.Context, step *models.WorkflowStep) _, err = r.db.ExecContext(ctx, r.db.Rebind(` INSERT INTO workflow_steps ( id, workflow_id, name, position, color, - prompt, events, allow_manual_move, is_start_step, show_in_command_panel, auto_archive_after_hours, agent_profile_id, stage_type, created_at, updated_at - ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + prompt, events, allow_manual_move, is_start_step, show_in_command_panel, auto_archive_after_hours, agent_profile_id, stage_type, auto_advance_requires_signal, created_at, updated_at + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) `), step.ID, step.WorkflowID, step.Name, step.Position, step.Color, step.Prompt, string(eventsJSON), dialect.BoolToInt(step.AllowManualMove), - dialect.BoolToInt(step.IsStartStep), dialect.BoolToInt(step.ShowInCommandPanel), step.AutoArchiveAfterHours, step.AgentProfileID, normalizeStageType(step.StageType), step.CreatedAt, step.UpdatedAt) + dialect.BoolToInt(step.IsStartStep), dialect.BoolToInt(step.ShowInCommandPanel), step.AutoArchiveAfterHours, step.AgentProfileID, normalizeStageType(step.StageType), dialect.BoolToInt(step.AutoAdvanceRequiresSignal), step.CreatedAt, step.UpdatedAt) return err } @@ -583,12 +587,12 @@ func (r *Repository) scanStep(row interface { Scan(dest ...interface{}) error }) (*models.WorkflowStep, error) { step := &models.WorkflowStep{} - var allowManualMove, isStartStep, showInCommandPanel int + var allowManualMove, isStartStep, showInCommandPanel, autoAdvanceRequiresSignal int var autoArchiveAfterHours sql.NullInt64 var color, prompt, eventsJSON, agentProfileID, stageType sql.NullString err := row.Scan(&step.ID, &step.WorkflowID, &step.Name, &step.Position, &color, - &prompt, &eventsJSON, &allowManualMove, &isStartStep, &showInCommandPanel, &autoArchiveAfterHours, &agentProfileID, &stageType, &step.CreatedAt, &step.UpdatedAt) + &prompt, &eventsJSON, &allowManualMove, &isStartStep, &showInCommandPanel, &autoArchiveAfterHours, &agentProfileID, &stageType, &autoAdvanceRequiresSignal, &step.CreatedAt, &step.UpdatedAt) if err != nil { return nil, err @@ -597,6 +601,7 @@ func (r *Repository) scanStep(row interface { step.AllowManualMove = allowManualMove == 1 step.IsStartStep = isStartStep == 1 step.ShowInCommandPanel = showInCommandPanel == 1 + step.AutoAdvanceRequiresSignal = autoAdvanceRequiresSignal == 1 if autoArchiveAfterHours.Valid { step.AutoArchiveAfterHours = int(autoArchiveAfterHours.Int64) } @@ -623,7 +628,7 @@ func (r *Repository) scanStep(row interface { return step, nil } -const stepSelectColumns = `id, workflow_id, name, position, color, prompt, events, allow_manual_move, is_start_step, show_in_command_panel, auto_archive_after_hours, agent_profile_id, stage_type, created_at, updated_at` +const stepSelectColumns = `id, workflow_id, name, position, color, prompt, events, allow_manual_move, is_start_step, show_in_command_panel, auto_archive_after_hours, agent_profile_id, stage_type, auto_advance_requires_signal, created_at, updated_at` // GetStep retrieves a workflow step by ID. func (r *Repository) GetStep(ctx context.Context, id string) (*models.WorkflowStep, error) { @@ -655,11 +660,11 @@ func (r *Repository) UpdateStep(ctx context.Context, step *models.WorkflowStep) UPDATE workflow_steps SET name = ?, position = ?, color = ?, prompt = ?, events = ?, - allow_manual_move = ?, is_start_step = ?, show_in_command_panel = ?, auto_archive_after_hours = ?, agent_profile_id = ?, stage_type = ?, updated_at = ? + allow_manual_move = ?, is_start_step = ?, show_in_command_panel = ?, auto_archive_after_hours = ?, agent_profile_id = ?, stage_type = ?, auto_advance_requires_signal = ?, updated_at = ? WHERE id = ? `), step.Name, step.Position, step.Color, step.Prompt, string(eventsJSON), - dialect.BoolToInt(step.AllowManualMove), dialect.BoolToInt(step.IsStartStep), dialect.BoolToInt(step.ShowInCommandPanel), step.AutoArchiveAfterHours, step.AgentProfileID, normalizeStageType(step.StageType), step.UpdatedAt, step.ID) + dialect.BoolToInt(step.AllowManualMove), dialect.BoolToInt(step.IsStartStep), dialect.BoolToInt(step.ShowInCommandPanel), step.AutoArchiveAfterHours, step.AgentProfileID, normalizeStageType(step.StageType), dialect.BoolToInt(step.AutoAdvanceRequiresSignal), step.UpdatedAt, step.ID) if err != nil { return err } @@ -762,7 +767,7 @@ func (r *Repository) ListStepsByWorkflow(ctx context.Context, workflowID string) func (r *Repository) ListStepsByWorkspaceID(ctx context.Context, workspaceID string) ([]*models.WorkflowStep, error) { rows, err := r.ro.QueryContext(ctx, r.ro.Rebind(` SELECT ws.id, ws.workflow_id, ws.name, ws.position, ws.color, ws.prompt, ws.events, - ws.allow_manual_move, ws.is_start_step, ws.show_in_command_panel, ws.auto_archive_after_hours, ws.agent_profile_id, ws.stage_type, ws.created_at, ws.updated_at + ws.allow_manual_move, ws.is_start_step, ws.show_in_command_panel, ws.auto_archive_after_hours, ws.agent_profile_id, ws.stage_type, ws.auto_advance_requires_signal, ws.created_at, ws.updated_at FROM workflow_steps ws JOIN workflows w ON ws.workflow_id = w.id WHERE w.workspace_id = ? diff --git a/apps/backend/pkg/websocket/actions.go b/apps/backend/pkg/websocket/actions.go index f5186d6887..0a66a644ee 100644 --- a/apps/backend/pkg/websocket/actions.go +++ b/apps/backend/pkg/websocket/actions.go @@ -327,6 +327,7 @@ const ( ActionMCPUpdateTask = "mcp.update_task" ActionMCPAddBranchToTask = "mcp.add_branch_to_task" ActionMCPUpdateRepositoryBaseBranch = "mcp.update_repository_base_branch" + ActionMCPStepComplete = "mcp.step_complete" // ADR 0015: agent-emitted explicit completion signal ActionMCPAskUserQuestion = "mcp.ask_user_question" ActionMCPCreateTaskPlan = "mcp.create_task_plan" ActionMCPGetTaskPlan = "mcp.get_task_plan" diff --git a/apps/web/components/settings/workflow-pipeline-editor-step-actions.tsx b/apps/web/components/settings/workflow-pipeline-editor-step-actions.tsx index 18474aaa2f..b9569efca5 100644 --- a/apps/web/components/settings/workflow-pipeline-editor-step-actions.tsx +++ b/apps/web/components/settings/workflow-pipeline-editor-step-actions.tsx @@ -252,3 +252,13 @@ export function TurnCompleteSelect({ ); } + +// ADR 0015 explicit-completion-signal UI (the "Wait for agent completion +// signal" toggle) is deferred to a follow-up PR. The backend column ships +// here so workflow steps can opt in via the MCP +// `update_workflow_step_kandev` tool. The pipeline-editor toggle needs its +// own iteration to avoid an unrelated E2E flake in +// `workflow-agent-profile.spec.ts` caused by additional UI shifting the +// Radix select dropdown into a sibling tooltip's hover path. The workflow +// import/export YAML also needs to be extended to serialize this flag in +// the follow-up so the toggle setting round-trips through export. diff --git a/apps/web/lib/types/http.ts b/apps/web/lib/types/http.ts index 0953be4c38..0bb331bd6c 100644 --- a/apps/web/lib/types/http.ts +++ b/apps/web/lib/types/http.ts @@ -82,6 +82,14 @@ export type WorkflowStep = { * frontend uses it to choose presentation (review/approval styling, etc). */ stage_type?: "work" | "review" | "approval" | "custom"; + /** + * ADR 0015: gate on_turn_complete transitions on an explicit + * `step_complete_kandev` MCP signal from the agent. When true, the + * step's auto-advance only fires once the agent (or the manual + * fallback button) signals completion. Default false preserves + * legacy "any turn-end advances" behaviour. + */ + auto_advance_requires_signal?: boolean; created_at: string; updated_at: string; }; diff --git a/docs/decisions/0015-explicit-completion-signal-for-auto-advance.md b/docs/decisions/0015-explicit-completion-signal-for-auto-advance.md new file mode 100644 index 0000000000..edf43b7484 --- /dev/null +++ b/docs/decisions/0015-explicit-completion-signal-for-auto-advance.md @@ -0,0 +1,127 @@ +# 0015: Explicit completion signal for auto-advance + +**Status:** proposed +**Date:** 2026-06-04 +**Area:** backend, frontend + +## Context + +Today, a workflow step's `on_turn_complete` actions (most commonly `move_to_next`, i.e. "auto-advance") fire whenever the agent's turn ends. The orchestrator's only signal that work is "done" is that the agent has stopped streaming and the ACP turn is complete. That signal is unreliable as a proxy for "user requirements satisfied": + +- **Halts are overloaded.** An agent stops emitting for many reasons: it actually finished the requested work; it called `ask_user_question_kandev` (or any clarification mechanism) and is waiting on the user; it hit a provider rate limit or transient error and bailed; it terminated mid-thought due to a context budget; it shelled out and exited; a future agent CLI added a new halt class we don't know about yet. From the orchestrator's vantage point these are indistinguishable — all of them look like "turn complete." +- **The reasons evolve faster than our interpreter.** Each new agent type and each upstream CLI revision can introduce new termination modes. Treating every halt as completion guarantees false positives now and on each upgrade. +- **Auto-advance changes user-observed behavior.** Users expect the agent to behave the same whether auto-advance is on or off: ask questions and wait indefinitely, hold extended back-and-forth conversations, defer until the user replies. With auto-advance on, the very act of asking a question (which halts the turn) trips a step transition, dropping the conversation onto the next step before the human has answered. +- **Working around it via "auto-advance off + side effect" is worse.** Disabling the checkbox and re-implementing advance as a hidden behavior reintroduces the same ambiguity at a different layer and loses the explicit UI affordance. +- **Tasks driving themselves through `update_task_kandev` / `move_task_kandev` is an alternative we rejected.** Letting an agent mutate the very task it's running on creates ordering and state-machine hazards (lifecycle races against the in-flight turn, ambiguous "which step's events fire," reconciliation against `task_repositories` and session state). Even if we hardened it, it would not address the underlying interpretation problem — we'd still be guessing what a halt meant. + +## Decision + +**Auto-advance is gated on an explicit, agent-emitted completion signal.** When auto-advance is enabled for a step, the orchestrator no longer treats turn-end as the trigger for `on_turn_complete` transitions. Instead, transitions fire only when the agent calls a dedicated MCP tool — `step_complete_kandev` — that semantically means "all user-stated requirements for this step are satisfied." (Named `step_*`, not `task_*`, because the signal is scoped to one workflow step; a task spans many steps and may be signaled-complete multiple times in its lifetime.) Naked halts surface a UI affordance instead of advancing silently. + +### Mechanism + +1. **MCP tool.** A new `step_complete_kandev` tool is registered alongside the existing kandev MCP handlers (`internal/mcp/handlers/`, exposed via the same WS-backed action plumbing as `add_branch_to_task_kandev`). Signature: + + ```jsonc + { + "name": "step_complete_kandev", + "arguments": { + "summary": "string", // required: one-paragraph "what was done" + "handoff": "string?", // optional: context for the next step's agent + "blockers": "string?" // optional: known unresolved issues + } + } + ``` + + The tool is implemented as a small `ws.ActionMCPStepComplete`. The handler (`internal/mcp/handlers/handlers.go:handleStepComplete`) does two things in order: (a) writes a `PendingStepCompletionSignal` blob into `TaskSession.Metadata` and (b) emits a `workflow.step_completion_signaled` bus event carrying `{task_id, session_id, step_id, source, summary, signaled_at}` — `handoff` and `blockers` are persisted in the bag only, not on the wire. The MCP tool returns `{accepted: true}` once both the bag write and the publish succeed; on publish failure it returns an error so the agent can retry (the persisted bag is idempotent against the retry). The actual transition is driven by the orchestrator off the event, so the agent's call site stays decoupled from step-machine timing. + +2. **Storage — `TaskSession.Metadata` bag, no new table.** The pending signal is small, short-lived, and read in exactly one place (`processOnTurnComplete`); a dedicated table would be overkill. The MCP handler (write site) and the orchestrator (read site) share a single decoder — `models.LoadPendingStepSignal` in `internal/task/models` — that handles both the in-process typed shape and the JSON-rehydrated `map[string]interface{}` shape produced by reloading the row after a restart. Schema: + + ```go + // Persisted JSON under TaskSession.Metadata["pending_step_completion_signal"] + type PendingStepCompletionSignal struct { + StepID string `json:"step_id"` + Source string `json:"source"` // "agent" | "manual_fallback" + Summary string `json:"summary"` + Handoff string `json:"handoff,omitempty"` + Blockers string `json:"blockers,omitempty"` + SignaledAt time.Time `json:"signaled_at"` + } + ``` + + - **Set** on `step_complete_kandev` MCP call or fallback button click — written into `Metadata` and saved via the existing session-update path. + - **Read** by `processOnTurnComplete` and by the `onStepCompletionSignaled` subscriber. Stale entries whose `StepID` no longer matches the session's current step are treated as absent and cleared on read. + - **Cleared** on (a) successful transition, (b) user message arriving before the transition runs (re-open semantics), (c) any other step change. + - **Idempotent.** Repeated calls within the same step are no-ops: if `existing != nil && existing.StepID == currentStepID`, the MCP tool returns `{accepted: false, reason: "already_signaled"}` without overwriting. + - **Audit (deferred).** Persisting consumed-signal context on the transition row (`SessionStepHistory.Metadata["signal_source"]` ∈ {`"agent"`, `"manual_fallback"`} plus `Metadata["signal_summary"]`) is the planned audit shape but is not yet implemented. Today the bag is cleared on transition and the only history is the chat transcript. Cancelled signals (overwritten by user reply) are deliberately not retained. + - **Chat visibility (deferred).** Surfacing a `step_complete_signaled` system message in the existing message stream is planned but not yet emitted. Today the only user-visible cue is the next-step transition itself; the bag remains the only truth source. + +3. **System-prompt injection.** When `auto_advance_requires_signal` is true on the current workflow step, `internal/sysprompt/` prepends a short instruction block to the agent's system prompt at launch / resume time, in the same path that already injects MCP tool descriptions. The text is fixed, terse, and points at the tool by name: + + > When auto-advance is enabled, you MUST call `step_complete_kandev` once — and only once — when every user request and follow-up requirement for the current workflow step is satisfied. Do not call it on partial progress, on a question you are about to ask, or after a failure. Calling it is the signal that triggers the workflow transition; halting without calling it leaves the step paused for the user. + + The instruction is unconditional — agents that never encounter a signal-gated step will simply never need to act on it (the tool no-ops without changing visible state). + +4. **Orchestrator gating.** `processOnTurnComplete` (`internal/orchestrator/event_handlers_workflow.go`) splits into two paths: + + - **Step requires explicit signal** (auto-advance enabled and `auto_advance_requires_signal == true`): on bare turn-end, read `Metadata["pending_step_completion_signal"]`. If absent (or stale for a prior step), do **not** evaluate transition actions; call `setSessionWaitingForInput` and surface a "Completion pending" UI hint. If present and matches the current step, run the existing `processTurnCompleteActions` / `resolveTransitionTargetStep` / `executeStepTransition` pipeline and clear the bag entry. The same path is triggered out-of-band by a new bus subscriber `onStepCompletionSignaled` for the case where the signal lands *after* the turn already ended (signal-after-halt is the common shape). + - **Step does not require explicit signal** (legacy behavior): unchanged. + +5. **Halt-without-signal fallback.** When the agent halts without having called `step_complete_kandev`, the UI shows an inline "Mark complete & advance" action (rendered in the chat composer area, akin to the existing clarification overlay). One click writes the bag entry with `source: "manual_fallback"` and emits the same `workflow.step_completion_signaled` event. This is the safety net: false negatives (work done, agent forgot to call the tool) cost one user click; false positives (agent calls the tool prematurely) are caught because user can keep messaging and the next turn re-opens the step. The fallback is **disabled** during an active ask-clarification waiting window so the button doesn't compete with a pending user question. + +6. **Re-open semantics.** If the user sends a new message *after* the signal has been written *but before* the transition has executed, the orchestrator clears the bag entry, cancels the pending transition, and treats the message as continued conversation. If the user messages *after* the transition has already moved to the next step, the message lands on the new step as normal — the prior step is not re-opened. + +### Configuration surface + +- **Per-step:** the existing `auto_advance` UI checkbox stays; a new sub-toggle "Wait for agent completion signal" appears underneath when an `on_turn_complete` transition is configured. Single switch; no profile/global flag. +- **Workflow step model:** add `auto_advance_requires_signal bool` to `wfmodels.WorkflowStep`. Migration adds the column with default `false`; existing steps keep legacy "any turn-end advances" behaviour until the user opts a step in. + +## Consequences + +- **Trades false positives for false negatives.** Today: silent wrong advances are the failure mode. New: stuck steps are the failure mode, recoverable in one click. We judge stuck > wrong: stuck is visible and fixable; wrong-advance corrupts downstream step inputs. +- **Compliance burden on the agent.** Whether the agent obeys the injected instruction is a function of (a) the model's instruction-following, and (b) the agent CLI's handling of system-prompt prepends. Both vary across Claude, Codex, Cursor, OpenCode. The fallback button is what makes this acceptable; without it, the design would brick non-Claude agents. +- **Adds a new MCP tool.** Increases the kandev MCP surface area; tool listing payload grows by one entry. The tool itself is small, but new agents and new CLIs need to be validated against it the same way ADR 0014's strategies were. +- **`step_complete_kandev` is fire-and-forget once it returns.** The MCP call only returns after the bag write + bus publish complete, but the actual transition runs asynchronously in the bus subscriber. An agent that calls the tool then immediately continues working may have trailing tokens race with the transition. Mitigation today is behavioural: the tool description in `kandev-context.md` instructs the agent to "call this as the LAST action of the step." The manual fallback button and re-open semantics are the safety nets if the agent gets it wrong. +- **Idempotency.** Multiple calls within the same step are deduped at the bag-write site (first call wins, subsequent calls return `{accepted: false, reason: "already_signaled"}` without error). Cross-step re-entry is allowed — a fresh step's `currentStepID` no longer matches the stale bag entry, so the next call writes through. +- **No new schema for the signal itself.** Storage piggybacks on the existing `task_sessions.metadata` JSON column. Audit of *consumed* signals piggybacks on the existing `session_step_history.metadata` JSON column. Cancelled signals (overwritten by a user reply) are not retained — chat already surfaces what the user needs. +- **Telemetry (deferred).** The plan is two `expvar` counters under `workflow_*` — `step_completion_signal_received_total` and `step_completion_signal_fallback_used_total` — whose ratio (fallback-used / received ≤ 10% per agent type) becomes the headline "is the explicit-signal flow working" metric. Neither counter exists yet; ship in a follow-up alongside the manual fallback button so both halves of the ratio can be measured at once. +- **No data-model coupling to `task_repositories` or `task_prs`.** The signal is per-session, per-step; it does not interact with the multi-branch work in ADR 0013. + +## Alternatives considered + +- **Naming: `task_complete_kandev`.** Rejected: signal is per-workflow-step, not per-task. A task may pass through `triage → in-progress → review → done`, with `step_complete_kandev` called at the end of each. `task_*` would either misname the per-step signal or imply a separate "the whole task is done" semantic we don't have. +- **Heuristic: classify turn-end reasons.** Pattern-match on the ACP stop reason, presence of recent `ask_user_question_kandev` calls, etc. Rejected: re-implements the interpretation problem we're trying to escape, and breaks on every new agent CLI revision. +- **Dedicated `session_step_completion_signals` table.** Considered (typed columns, partial UNIQUE index for idempotency, `source` column for SQL-aggregated telemetry, full cancelled-signal audit trail). Rejected as overkill: at most one pending signal per session at a time, lifetime measured in seconds, single reader. The `TaskSession.Metadata` bag covers idempotency (existence check), audit (lifted onto the existing `SessionStepHistory.Metadata` on transition), telemetry (in-process `expvar`), and chat visibility (system message) with zero new schema. +- **Two-signal: require both halt and explicit tool call.** Effectively what we do, except we let the user provide the second signal via the fallback button when the agent doesn't. Pure "agent must do both" guarantees the stuck-task failure mode without the escape hatch. +- **Agent self-mutating via `update_task_kandev` / `move_task_kandev`.** Lets the agent move its own task between steps. Rejected up front: state-machine races (orchestrator step-transition pipeline vs in-flight turn), ambiguous event semantics (does `on_exit` for the old step fire mid-turn?), and conflicts with the existing `workflowStepGetter`-driven transition path. Even if hardened, it doesn't fix the halt-interpretation problem; the agent still has to *decide* when to call it, which is the same decision as calling `task_complete_kandev` — but on a tool that has many other side effects. +- **Disable auto-advance entirely, reintroduce as silent side-effect.** Rejected: hides a feature the user explicitly enabled via the checkbox; loses the UI affordance; doesn't reduce ambiguity, just moves it. +- **Inject the system prompt always, gate only the transition.** Considered. Mild win (signal exists everywhere; checkbox only gates whether kandev acts on it), but pollutes prompts for users who don't use auto-advance and burns a small amount of every agent's context. Deferred — easy to flip later by changing the injection predicate without schema changes. +- **Require a `summary` arg and surface it in the UI.** Kept (above) — the summary is cheap, the next-step agent benefits from a handoff blurb, and it gives the user a one-line "what changed" without scrubbing the transcript. + +## Open questions + +- Do we want a per-agent-type override for the system-prompt text? Codex and Cursor have noticeably different instruction-following profiles; a one-size-fits-all prompt may underperform on the weaker ones. +- Should the fallback button be exposed as a keyboard shortcut, or kept click-only to avoid accidental advances? +- Long-running passthrough sessions (ADR 0014) — verify the MCP tool is reachable via `connectMCPStream` for every passthrough-capable CLI before depending on this flow in passthrough mode. +- **Late-signal re-close race.** `ProcessOnTurnStart` clears the bag when the user replies, but the persisted signal carries only `step_id` / `signaled_at` — not a monotonic turn or user-reply marker. A `step_complete_kandev` call that was in-flight when the user replied can repopulate the bag *after* the clear, and the next turn-end will then auto-advance, violating the re-open guarantee. Today the window is narrow (the tool call has to land after the user reply but before the next turn ends) and the worst case is one wrong advance recoverable by the user dragging the task back. Proper fix is to tie each signal to either `session.LastUserReplyAt` or an active-turn ID and reject signals older than the latest user reply at the write site. Deferred — needs either a new session column or a session-scoped sequence number. + +## Files this will touch (estimate) + +Backend: +- `apps/backend/internal/mcp/handlers/handlers.go` *(new `step_complete_kandev`)* +- `apps/backend/internal/mcp/server/server.go` *(register tool)* +- `apps/backend/pkg/websocket/actions.go` *(new `ActionMCPStepComplete`)* +- `apps/backend/internal/task/models/models.go` *(new `PendingStepCompletionSignal` type; metadata key constant)* +- `apps/backend/internal/orchestrator/event_handlers_workflow.go` *(gating in `processOnTurnComplete`; read/clear bag entry)* +- `apps/backend/internal/orchestrator/event_handlers_step_completion.go` *(new — `onStepCompletionSignaled` subscriber: write bag + drive transition)* +- `apps/backend/config/prompts/kandev-context.md` *(unconditional tool description)* +- `apps/backend/internal/workflow/models/` *(`AutoAdvanceRequiresSignal bool`; `SessionStepHistory.Metadata` key constants for `signal_source` / `signal_summary` — deferred)* +- `apps/backend/internal/workflow/repository/sqlite/` *(idempotent migration for `workflow_steps.auto_advance_requires_signal` only — no new tables)* + +Frontend: +- `apps/web/components/settings/workflow-pipeline-editor-step-actions.tsx` *(sub-toggle UI)* +- `apps/web/components/task/chat/` *(fallback "Mark complete & advance" affordance)* +- `apps/web/lib/types/http.ts` *(extend `WorkflowStep`)* +- `apps/web/lib/state/slices/` *(transitional state for "completion pending")* + +Spec: `docs/specs/workflow/explicit-completion-signal/spec.md` *(to be written via `/spec` once this ADR is accepted)* diff --git a/docs/decisions/INDEX.md b/docs/decisions/INDEX.md index 6086b11575..d13a1a043c 100644 --- a/docs/decisions/INDEX.md +++ b/docs/decisions/INDEX.md @@ -20,3 +20,4 @@ Read individual ADRs for full context. Create new ones via `/record decision` or | 0012 | [Service-only UI self-update](0012-service-only-self-update.md) | accepted | backend, frontend, cli | 2026-05-29 | | 0013 | [Multi-branch task support — N (repo, branch) pairs per task](0013-multi-branch-tasks.md) | accepted | backend, frontend | 2026-06-01 | | 0014 | [Per-CLI MCP server injection for passthrough mode](0014-passthrough-mcp-injection-strategies.md) | accepted | backend | 2026-05-29 | +| 0015 | [Explicit completion signal for auto-advance](0015-explicit-completion-signal-for-auto-advance.md) | proposed | backend, frontend | 2026-06-04 |