97c1f689e4
* 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>
257 lines
8.2 KiB
Go
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)
|
|
}
|