feat(workflow): explicit step-completion signal for auto-advance (#1276)
* feat(workflow): explicit step-completion signal for auto-advance Adds `step_complete_kandev` MCP tool + `WorkflowStep.auto_advance_requires_signal` column so workflow steps can gate their `on_turn_complete` transitions on an agent-emitted completion signal instead of inferring "done" from any turn-end. See docs/decisions/0015-explicit-completion-signal-for-auto-advance.md for the motivation, alternatives considered, and trade-offs. * fix(workflow): address PR review feedback for explicit completion signal - Dedup `LoadPendingStepSignal` into `internal/task/models` so the MCP handler (write site) and the orchestrator (read site) share one decoder. - Gate the out-of-band subscriber on `WorkflowStep.AutoAdvanceRequiresSignal` so `step_complete_kandev` is not an unintended manual-advance path for signalless steps. - Fail closed on step-load errors in `processOnTurnCompleteViaEngine`: a loader hiccup must block the transition rather than silently bypass the gate. - Reject `step_complete_kandev` for terminal sessions (Completed / Failed / Cancelled) so the agent gets a clear error instead of `accepted: true` followed by a no-op. - Surface bus-publish failures back to the agent instead of swallowing them — the idempotency guard makes the retry safe and prevents the session from getting stuck. - New `TestOnStepCompletionSignaled` covers the subscriber paths (RUNNING / WAITING+match / stale step / not signal-gated). - ADR doc fixes: payload field list now matches the wire shape, writer ownership is documented, the unsubstantiated 250ms-debounce claim is removed, and the late-signal re-close race is recorded as a deferred open question. - Drop the new toggle's HelpTip in favour of an inline description — the shadcn tooltip portal was intercepting clicks on the agent-profile dropdown options in the workflow settings E2E. * fix(workflow): round-2 review fixes for explicit completion signal - Split `handleStepComplete` into `resolveStepCompleteTarget` + `publishStepCompletionEvent` helpers so the entrypoint clears funlen (88 → under 80 lines) and the two halves are independently testable. - Classify session/task load errors instead of returning `NotFound` for every failure: use the exported `taskrepo.ErrTaskNotFound` sentinel for task lookups, substring-match for the session repo (no sentinel today), and surface anything else as `InternalError` with the raw error logged but not leaked to the agent — so a transient DB hiccup triggers a retry rather than session abandonment. - Pin `step_complete_kandev` explicitly in the mode-specific tool-count tests: `Contains` in `ModeTask`, `NotContains` in `ModeConfig` and `ModeOffice`, so accidental dereg or cross-mode registration is caught by the diff instead of by a still-equal count. * fix(workflow): round-3 review fixes + defer pipeline-editor toggle - Defer the "Wait for agent completion signal" sub-toggle to a follow-up PR. Adding it to the step config card shifts the layout enough that the agent profile select's Radix popper opens into a sibling tooltip's hover path, breaking workflow-agent-profile.spec.ts on shard 7. The backend column ships in this PR — workflows can opt in via the YAML import or the MCP `update_workflow_step_kandev` tool — and the UI affordance gets its own iteration where the test interference can be investigated without blocking the rest of the change. - Re-publish the bus event on the `already_signaled` short-circuit when the session is WAITING_FOR_INPUT. Without this, an agent's retry after the first call's publish failed would short-circuit on the persisted bag, never fire the subscriber, and leave the session stuck. - handleStepComplete docstring: drop the non-existent feature-flag claim and the "audit entry" wording — gating is purely `WorkflowStep.AutoAdvanceRequiresSignal` and the bag is cleared on the next turn start with no separate audit trail. - `step_complete_kandev` tool description: same fix — clarify that the call succeeds for non-gated steps but the signal is discarded on the next turn start (no audit history to query later). * test(workflow): cover handleStepComplete + reword defer-comment - Add `TestHandleStepComplete_*` (6 subtests, all passing) covering the uncovered handler paths flagged by Claude/cubic on round 3: missing-fields validation, session-does-not-belong-to-task ownership guard, terminal-session rejection (Completed/Failed/Cancelled), first-call happy path (bag persisted + bus payload shape pinned), dedup while RUNNING (no re-publish), and dedup while WAITING_FOR_INPUT (the retry-after-publish-failure path — must re-publish so the subscriber fires). - Reword the deferred-toggle comment to drop the YAML-export claim: the workflow export model doesn't yet serialize `auto_advance_requires_signal`, so MCP `update_workflow_step_kandev` is the only configuration path that round-trips today. Note the follow-up needs to extend the export schema alongside the UI toggle. * test(workflow): pin signaled_at + drop redundant GetTaskSession on bag-clear - Assert `signaled_at` is present and parses as RFC3339Nano in `TestHandleStepComplete_FirstCallAccepted`. The field is part of the handler's documented response contract; pinning it catches accidental drop/rename regressions. - Add `clearPendingStepSignalByID(ctx, sessionID)` as a thin DB-only variant of `clearPendingStepSignal` and use it from `executeStepTransition`, where the session struct isn't read after the clear. Avoids an extra `GetTaskSession` round-trip per transition. `clearPendingStepSignal(session)` still exists for the in-memory-mutation case (subscriber's stale-step path). * fix(workflow): subscriber stale-clear guard + ADR/comment drift fixes - onStepCompletionSignaled's stale-step path now reads the bag and only clears it when its current StepID matches the (stale) event's stepID. Previously the unconditional clear could destroy a valid signal that the new step's agent had just written when sessions are reused across steps without auto_start_agent. (Greptile P1.) - Drop the "feature flag" phrase from WorkflowStep.AutoAdvanceRequiresSignal docstring — leftover from an earlier draft. The gating is purely the boolean field; the global flag was removed two rounds ago. - Merge the two adjacent `if triggerOnEnter` blocks in executeStepTransition so the full triggerOnEnter=true path reads top- to-bottom in one place. (CodeRabbit + cubic.) - ADR 0015 Mechanism section: re-label the SessionStepHistory.Metadata audit annotation, the `step_complete_signaled` system message, and the expvar telemetry counters as deferred. They're planned shapes, not current behaviour — calling them "Audit" / "Chat visibility" / "Telemetry" in present tense was misleading for future readers. * fix(workflow): drop subscriber complexity + reload session on stale-clear - Extract `parseStepCompletionEvent` and `stepIsSignalGated` helpers from `onStepCompletionSignaled`. Cyclomatic complexity goes 17 → under 15 (cyclop on new code, CI lint regression after the round-6 stale-clear guard). - The stale-step branch now re-loads the session from the DB before deciding whether to clear the bag. The previous round-6 fix checked the local `session.Metadata` snapshot, which can be older than a concurrent write from the new step's agent — that check could see a stale signal that had since been overwritten with a valid one and erase the new bag entry. The narrower load-on-clear is cheap and closes the cubic P1 hole. (The broader race in `processOnTurnCompleteViaEngine` reading session.Metadata is the monotonic-marker open question in ADR 0015; out of scope here.) * test(workflow): pin stale-event guard's preserve-current-bag side + dedup StringFromAny - Add `TestOnStepCompletionSignaled` subtest "stale event → valid bag for CURRENT step is preserved" — fires a stale step-A event while the bag holds a valid step-B signal, asserts the bag still contains StepID=step_b after the subscriber runs. Pins the negative side of the round-6 `existing.StepID == stepID` guard, so a regression that drops or inverts the guard (and would silently leave signal-gated steps stuck) fails the test instead of slipping through. - Promote `pendingSignalString` to exported `models.StringFromAny` and reuse it from the orchestrator subscriber's `parseStepCompletionEvent`. Eliminates the byte-for-byte duplicate helper across the two packages; the bag-shape decoder and the bus- event decoder now share one source of truth for "missing string". --------- Co-authored-by: Kandev Agent <agent@kandev.dev>
This commit is contained in:
João Salavisa
@@ -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.
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
@@ -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",
|
||||
|
||||
@@ -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))
|
||||
}
|
||||
|
||||
|
||||
@@ -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
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
})
|
||||
}
|
||||
@@ -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",
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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"
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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 = ?
|
||||
|
||||
@@ -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"
|
||||
|
||||
@@ -252,3 +252,13 @@ export function TurnCompleteSelect({
|
||||
</div>
|
||||
);
|
||||
}
|
||||
|
||||
// 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.
|
||||
|
||||
@@ -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;
|
||||
};
|
||||
|
||||
@@ -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)*
|
||||
@@ -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 |
|
||||
|
||||
Reference in New Issue
Block a user