Files
kandev/apps/backend/internal/workflow/controller/controller.go
T
João Salavisa 97c1f689e4 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>
2026-06-05 18:32:27 +01:00

257 lines
8.2 KiB
Go

package controller
import (
"context"
"strings"
"github.com/kandev/kandev/internal/workflow/models"
"github.com/kandev/kandev/internal/workflow/service"
)
// Controller handles workflow-related requests
type Controller struct {
svc *service.Service
}
// NewController creates a new workflow controller
func NewController(svc *service.Service) *Controller {
return &Controller{svc: svc}
}
// Template responses
type ListTemplatesResponse struct {
Templates []*models.WorkflowTemplate `json:"templates"`
}
type GetTemplateResponse struct {
Template *models.WorkflowTemplate `json:"template"`
}
func (c *Controller) ListTemplates(ctx context.Context) (*ListTemplatesResponse, error) {
templates, err := c.svc.ListTemplates(ctx)
if err != nil {
return nil, err
}
return &ListTemplatesResponse{Templates: templates}, nil
}
func (c *Controller) GetTemplate(ctx context.Context, id string) (*GetTemplateResponse, error) {
template, err := c.svc.GetTemplate(ctx, id)
if err != nil {
return nil, err
}
return &GetTemplateResponse{Template: template}, nil
}
// Step responses
type ListStepsRequest struct {
WorkflowID string `json:"workflow_id"`
}
type ListStepsResponse struct {
Steps []*models.WorkflowStep `json:"steps"`
}
type GetStepResponse struct {
Step *models.WorkflowStep `json:"step"`
}
type CreateStepsFromTemplateRequest struct {
WorkflowID string `json:"workflow_id"`
TemplateID string `json:"template_id"`
}
func (c *Controller) ListStepsByWorkflow(ctx context.Context, req ListStepsRequest) (*ListStepsResponse, error) {
steps, err := c.svc.ListStepsByWorkflow(ctx, req.WorkflowID)
if err != nil {
return nil, err
}
return &ListStepsResponse{Steps: steps}, nil
}
// ListStepsByWorkspace returns all workflow steps for all workflows in a workspace.
func (c *Controller) ListStepsByWorkspace(ctx context.Context, workspaceID string) (*ListStepsResponse, error) {
steps, err := c.svc.ListStepsByWorkspaceID(ctx, workspaceID)
if err != nil {
return nil, err
}
return &ListStepsResponse{Steps: steps}, nil
}
func (c *Controller) GetStep(ctx context.Context, id string) (*GetStepResponse, error) {
step, err := c.svc.GetStep(ctx, id)
if err != nil {
return nil, err
}
return &GetStepResponse{Step: step}, nil
}
func (c *Controller) CreateStepsFromTemplate(ctx context.Context, req CreateStepsFromTemplateRequest) error {
return c.svc.CreateStepsFromTemplate(ctx, req.WorkflowID, req.TemplateID)
}
// 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"`
AutoAdvanceRequiresSignal *bool `json:"auto_advance_requires_signal,omitempty"`
}
// CreateStep creates a new workflow step.
func (c *Controller) CreateStep(ctx context.Context, req CreateStepRequest) (*GetStepResponse, error) {
step := &models.WorkflowStep{
WorkflowID: req.WorkflowID,
Name: req.Name,
Position: req.Position,
Color: req.Color,
Prompt: req.Prompt,
AllowManualMove: req.AllowManualMove,
}
if req.Events != nil {
step.Events = *req.Events
}
if req.IsStartStep != nil {
step.IsStartStep = *req.IsStartStep
}
if req.ShowInCommandPanel != nil {
step.ShowInCommandPanel = *req.ShowInCommandPanel
} 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
}
return &GetStepResponse{Step: step}, nil
}
// 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"`
AutoAdvanceRequiresSignal *bool `json:"auto_advance_requires_signal,omitempty"`
}
// UpdateStep updates an existing workflow step.
func (c *Controller) UpdateStep(ctx context.Context, req UpdateStepRequest) (*GetStepResponse, error) {
step, err := c.svc.GetStep(ctx, req.ID)
if err != nil {
return nil, err
}
if req.Name != nil {
step.Name = *req.Name
}
if req.Position != nil {
step.Position = *req.Position
}
if req.Color != nil {
step.Color = *req.Color
}
if req.Prompt != nil {
step.Prompt = *req.Prompt
}
if req.Events != nil {
step.Events = *req.Events
}
if req.AllowManualMove != nil {
step.AllowManualMove = *req.AllowManualMove
}
if req.IsStartStep != nil {
step.IsStartStep = *req.IsStartStep
}
if req.ShowInCommandPanel != nil {
step.ShowInCommandPanel = *req.ShowInCommandPanel
}
if req.AutoArchiveAfterHours != nil {
step.AutoArchiveAfterHours = *req.AutoArchiveAfterHours
}
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
}
return &GetStepResponse{Step: step}, nil
}
// DeleteStep deletes a workflow step.
func (c *Controller) DeleteStep(ctx context.Context, id string) error {
return c.svc.DeleteStep(ctx, id)
}
// ReorderStepsRequest is the request for reordering workflow steps.
type ReorderStepsRequest struct {
WorkflowID string `json:"workflow_id"`
StepIDs []string `json:"step_ids"`
}
// ReorderSteps reorders workflow steps for a workflow.
func (c *Controller) ReorderSteps(ctx context.Context, req ReorderStepsRequest) error {
return c.svc.ReorderSteps(ctx, req.WorkflowID, req.StepIDs)
}
// History responses
type ListHistoryRequest struct {
SessionID string `json:"session_id"`
}
type ListHistoryResponse struct {
History []*models.SessionStepHistory `json:"history"`
}
func (c *Controller) ListHistoryBySession(ctx context.Context, req ListHistoryRequest) (*ListHistoryResponse, error) {
history, err := c.svc.ListHistoryBySession(ctx, req.SessionID)
if err != nil {
return nil, err
}
return &ListHistoryResponse{History: history}, nil
}
// Export/Import types and methods
// ImportWorkflowsRequest carries import data.
type ImportWorkflowsRequest struct {
WorkspaceID string `json:"workspace_id"`
Data *models.WorkflowExport `json:"data"`
}
// ExportWorkflow exports a single workflow.
func (c *Controller) ExportWorkflow(ctx context.Context, workflowID string) (*models.WorkflowExport, error) {
return c.svc.ExportWorkflow(ctx, workflowID)
}
// ExportWorkflows exports workflows for a workspace. A nil workflowIDs exports
// every workflow; a non-nil slice restricts the export to that set of IDs.
func (c *Controller) ExportWorkflows(ctx context.Context, workspaceID string, workflowIDs []string) (*models.WorkflowExport, error) {
return c.svc.ExportWorkflows(ctx, workspaceID, workflowIDs)
}
// ImportWorkflows imports workflows into a workspace.
func (c *Controller) ImportWorkflows(ctx context.Context, req ImportWorkflowsRequest) (*service.ImportResult, error) {
return c.svc.ImportWorkflows(ctx, req.WorkspaceID, req.Data)
}