diff --git a/apps/backend/internal/orchestrator/event_handlers_workflow.go b/apps/backend/internal/orchestrator/event_handlers_workflow.go index 76b04ffbf..dac83d300 100644 --- a/apps/backend/internal/orchestrator/event_handlers_workflow.go +++ b/apps/backend/internal/orchestrator/event_handlers_workflow.go @@ -4,6 +4,7 @@ import ( "context" "errors" "fmt" + "sync" "time" "go.uber.org/zap" @@ -1539,6 +1540,112 @@ func (s *Service) scheduleAutoResumeForWorkflowQueue(ctx context.Context, sessio go s.tryEnsureExecution(context.WithoutCancel(ctx), sessionID) } +// maybeRecoverOrphanedWorkflowQueue is the per-session decision point of the +// workflow-queue watchdog (workflow_queue_watchdog.go). For a session that +// the sweep has flagged because it holds a stale workflow auto-start prompt, +// this function decides whether to take a recovery action and what: +// +// - terminal session → drop the stale workflow entries (the agent is +// gone for good; leaving them in the queue would never drain). +// - in-flight turn (activeTurns / running / starting / cancel / reset) → +// skip; the normal lifecycle will drain. +// - live agent process → skip; handleAgentReady will drain on next turn end. +// - otherwise → fire tryEnsureExecution to drive the same auto-resume +// cascade scheduleAutoResumeForWorkflowQueue uses inline (#1163). +// +// queuedAt is the queued_at of the oldest matched entry, used only for the +// recovery log line. +func (s *Service) maybeRecoverOrphanedWorkflowQueue(ctx context.Context, sessionID string, queuedAt time.Time, wg *sync.WaitGroup) { + session, err := s.repo.GetTaskSession(ctx, sessionID) + if err != nil { + // Transient DB read failure — leave the queue alone and let the + // next tick retry. Don't conflate this with "session truly gone". + s.logger.Debug("workflow queue watchdog: skip session load failed", + zap.String("session_id", sessionID), + zap.Error(err)) + return + } + if session == nil { + s.dropOrphanedWorkflowQueue(ctx, sessionID, "session_missing") + return + } + if isTerminalSessionState(session.State) { + s.dropOrphanedWorkflowQueue(ctx, sessionID, "terminal_session") + return + } + if _, hasActiveTurn := s.activeTurns.Load(sessionID); hasActiveTurn { + return + } + if s.isSessionResetInProgress(sessionID) || s.isCancelInFlight(sessionID) { + return + } + if s.executor == nil || s.agentManager == nil { + return + } + // Skip when the agent process is genuinely alive — handleAgentReady (or + // the inline drain on the next turn end) will pick up the queue. The + // agentManager probe is authoritative; the executor's in-memory store + // can lag a dying process. + if s.agentManager.IsAgentRunningForSession(ctx, sessionID) { + return + } + ageSec := int(time.Since(queuedAt) / time.Second) + s.logger.Warn("workflow queue watchdog: recovering orphaned session", + zap.String("session_id", sessionID), + zap.String("task_id", session.TaskID), + zap.String("session_state", string(session.State)), + zap.Int("queue_age_s", ageSec)) + // Bypass scheduleAutoResumeForWorkflowQueue's GetExecutionBySession gate: + // IsAgentRunningForSession just told us the process is dead, but the + // in-memory store may still hold a stale record. Drive tryEnsureExecution + // directly so the resume happens even when the store is out of sync. + // + // Escalation risk: tryEnsureExecution → ensureSessionRunning → ResumeSession + // will mark the session FAILED via updateTaskSessionState if the resume + // itself errors (disk full, ErrNoAgentProfileID, container won't start, + // etc.). The watchdog inherits this from the inline auto-resume path — + // a stuck-but-otherwise-recoverable session can flip to terminal if its + // underlying executor environment is broken. That's the right escalation + // (the orphan can never drain anyway) but worth flagging for future + // maintainers: the watchdog is "best-effort recover OR fail loudly". + wg.Add(1) + go func() { + defer wg.Done() + s.tryEnsureExecution(ctx, sessionID) + }() +} + +// dropOrphanedWorkflowQueue removes every workflow-tagged queued entry for +// a session whose agent is gone for good. User and agent entries are left +// alone — only the workflow auto-start prompts the watchdog owns are dropped. +func (s *Service) dropOrphanedWorkflowQueue(ctx context.Context, sessionID, reason string) { + if s.messageQueue == nil { + return + } + status := s.messageQueue.GetStatus(ctx, sessionID) + dropped := 0 + for _, entry := range status.Entries { + if entry.QueuedBy != messagequeue.QueuedByWorkflow { + continue + } + if err := s.messageQueue.RemoveEntry(ctx, sessionID, entry.ID); err != nil { + s.logger.Debug("workflow queue watchdog: drop entry failed", + zap.String("session_id", sessionID), + zap.String("entry_id", entry.ID), + zap.Error(err)) + continue + } + dropped++ + } + if dropped > 0 { + s.logger.Warn("workflow queue watchdog: dropped orphaned workflow entries", + zap.String("session_id", sessionID), + zap.String("reason", reason), + zap.Int("dropped", dropped)) + s.publishQueueStatusEvent(ctx, sessionID) + } +} + // flipStaleRunningToWaiting flips the session to WAITING_FOR_INPUT when its // state claims RUNNING/STARTING but the orchestrator's authoritative // activeTurns map shows no in-flight turn. This catches the manual-move race diff --git a/apps/backend/internal/orchestrator/messagequeue/repository.go b/apps/backend/internal/orchestrator/messagequeue/repository.go index d59251586..a5af41e78 100644 --- a/apps/backend/internal/orchestrator/messagequeue/repository.go +++ b/apps/backend/internal/orchestrator/messagequeue/repository.go @@ -1,6 +1,9 @@ package messagequeue -import "context" +import ( + "context" + "time" +) // Repository abstracts persistent storage of queued messages and pending moves. // Operations on queued messages are atomic per session. @@ -18,6 +21,12 @@ type Repository interface { // ListBySession returns all entries for a session ordered by position ascending. ListBySession(ctx context.Context, sessionID string) ([]QueuedMessage, error) + // ListStaleByQueuedBy returns at most `limit` entries whose queued_by matches + // AND whose queued_at < olderThan, ordered by queued_at ASC. Used by the + // orchestrator's workflow-queue watchdog to find orphaned auto-start prompts + // that no inline drain path picked up. `limit <= 0` is treated as no cap. + ListStaleByQueuedBy(ctx context.Context, queuedBy string, olderThan time.Time, limit int) ([]QueuedMessage, error) + // CountBySession returns the number of entries for a session. CountBySession(ctx context.Context, sessionID string) (int, error) diff --git a/apps/backend/internal/orchestrator/messagequeue/repository_memory.go b/apps/backend/internal/orchestrator/messagequeue/repository_memory.go index 257e10e81..36883bf84 100644 --- a/apps/backend/internal/orchestrator/messagequeue/repository_memory.go +++ b/apps/backend/internal/orchestrator/messagequeue/repository_memory.go @@ -98,6 +98,28 @@ func (r *memoryRepository) ListBySession(_ context.Context, sessionID string) ([ return out, nil } +func (r *memoryRepository) ListStaleByQueuedBy(_ context.Context, queuedBy string, olderThan time.Time, limit int) ([]QueuedMessage, error) { + r.mu.Lock() + defer r.mu.Unlock() + var out []QueuedMessage + for _, list := range r.entries { + for _, m := range list { + if m.QueuedBy != queuedBy { + continue + } + if !m.QueuedAt.Before(olderThan) { + continue + } + out = append(out, *m) + } + } + sort.Slice(out, func(i, j int) bool { return out[i].QueuedAt.Before(out[j].QueuedAt) }) + if limit > 0 && len(out) > limit { + out = out[:limit] + } + return out, nil +} + func (r *memoryRepository) CountBySession(_ context.Context, sessionID string) (int, error) { r.mu.Lock() defer r.mu.Unlock() diff --git a/apps/backend/internal/orchestrator/messagequeue/repository_sqlite.go b/apps/backend/internal/orchestrator/messagequeue/repository_sqlite.go index b8c9a7a0c..79ba8cc27 100644 --- a/apps/backend/internal/orchestrator/messagequeue/repository_sqlite.go +++ b/apps/backend/internal/orchestrator/messagequeue/repository_sqlite.go @@ -44,6 +44,7 @@ func (r *sqliteRepository) initSchema() error { queued_by TEXT NOT NULL DEFAULT '' ); CREATE INDEX IF NOT EXISTS idx_queued_messages_session_position ON queued_messages(session_id, position); + CREATE INDEX IF NOT EXISTS idx_queued_messages_queued_by_at ON queued_messages(queued_by, queued_at); CREATE TABLE IF NOT EXISTS pending_moves ( id TEXT PRIMARY KEY, @@ -208,6 +209,35 @@ func (r *sqliteRepository) ListBySession(ctx context.Context, sessionID string) return out, rows.Err() } +func (r *sqliteRepository) ListStaleByQueuedBy(ctx context.Context, queuedBy string, olderThan time.Time, limit int) ([]QueuedMessage, error) { + query := ` + SELECT id, session_id, task_id, position, content, model, plan_mode, + attachments_json, metadata_json, queued_at, queued_by + FROM queued_messages + WHERE queued_by = ? AND queued_at < ? + ORDER BY queued_at ASC` + args := []any{queuedBy, olderThan} + if limit > 0 { + query += ` LIMIT ?` + args = append(args, limit) + } + rows, err := r.ro.QueryxContext(ctx, r.ro.Rebind(query), args...) + if err != nil { + return nil, fmt.Errorf("list stale queued: %w", err) + } + defer func() { _ = rows.Close() }() + + var out []QueuedMessage + for rows.Next() { + msg, err := scanQueuedRow(rows) + if err != nil { + return nil, err + } + out = append(out, *msg) + } + return out, rows.Err() +} + func (r *sqliteRepository) CountBySession(ctx context.Context, sessionID string) (int, error) { var n int err := r.ro.GetContext(ctx, &n, r.ro.Rebind(`SELECT COUNT(*) FROM queued_messages WHERE session_id = ?`), sessionID) diff --git a/apps/backend/internal/orchestrator/messagequeue/repository_sqlite_test.go b/apps/backend/internal/orchestrator/messagequeue/repository_sqlite_test.go index 5f21600ee..526a77f24 100644 --- a/apps/backend/internal/orchestrator/messagequeue/repository_sqlite_test.go +++ b/apps/backend/internal/orchestrator/messagequeue/repository_sqlite_test.go @@ -7,6 +7,7 @@ import ( "sync" "sync/atomic" "testing" + "time" "github.com/jmoiron/sqlx" _ "github.com/mattn/go-sqlite3" @@ -314,6 +315,69 @@ func TestSQLiteRepository_PendingMove(t *testing.T) { } } +// TestSQLiteRepository_ListStaleByQueuedBy verifies the watchdog query: +// it returns workflow-tagged entries older than the cutoff and skips +// recent workflow entries, user entries, and entries from other sessions. +func TestSQLiteRepository_ListStaleByQueuedBy(t *testing.T) { + repo := newTestSQLiteRepo(t) + ctx := context.Background() + + now := time.Now().UTC() + stale := now.Add(-5 * time.Minute) + fresh := now.Add(-10 * time.Second) + cutoff := now.Add(-1 * time.Minute) + + mustInsert := func(sessionID, queuedBy string, queuedAt time.Time) string { + msg := &QueuedMessage{ + SessionID: sessionID, TaskID: "t1", Content: "x", + QueuedBy: queuedBy, QueuedAt: queuedAt, + } + if err := repo.Insert(ctx, msg, 0); err != nil { + t.Fatalf("insert: %v", err) + } + return msg.ID + } + + staleWorkflow := mustInsert("s1", QueuedByWorkflow, stale) + mustInsert("s1", QueuedByWorkflow, fresh) // too fresh + mustInsert("s2", QueuedByUser, stale) // wrong queued_by + staleWorkflowS3 := mustInsert("s3", QueuedByWorkflow, stale.Add(-1*time.Minute)) + + got, err := repo.ListStaleByQueuedBy(ctx, QueuedByWorkflow, cutoff, 0) + if err != nil { + t.Fatalf("list stale: %v", err) + } + if len(got) != 2 { + t.Fatalf("expected 2 stale workflow entries, got %d: %+v", len(got), got) + } + // queued_at ASC: s3's older entry must come first. + if got[0].ID != staleWorkflowS3 || got[1].ID != staleWorkflow { + t.Errorf("unexpected order: got [%s, %s], want [%s, %s]", + got[0].ID, got[1].ID, staleWorkflowS3, staleWorkflow) + } + + // Limit caps the result to the oldest entries. + capped, err := repo.ListStaleByQueuedBy(ctx, QueuedByWorkflow, cutoff, 1) + if err != nil { + t.Fatalf("list stale capped: %v", err) + } + if len(capped) != 1 { + t.Fatalf("expected 1 entry under limit=1, got %d", len(capped)) + } + if capped[0].ID != staleWorkflowS3 { + t.Errorf("limited result should hold the oldest entry, got %s", capped[0].ID) + } + + // Empty result when cutoff predates everything. + empty, err := repo.ListStaleByQueuedBy(ctx, QueuedByWorkflow, stale.Add(-2*time.Hour), 0) + if err != nil { + t.Fatalf("list stale empty: %v", err) + } + if len(empty) != 0 { + t.Errorf("expected 0 entries past distant cutoff, got %d", len(empty)) + } +} + // TestSQLiteRepository_ConcurrentInsertCap exercises the cap under contention: // 50 goroutines insert into one session with cap=10. Exactly 10 should succeed. func TestSQLiteRepository_ConcurrentInsertCap(t *testing.T) { diff --git a/apps/backend/internal/orchestrator/messagequeue/service.go b/apps/backend/internal/orchestrator/messagequeue/service.go index 4dfbc994e..86d3dc5d6 100644 --- a/apps/backend/internal/orchestrator/messagequeue/service.go +++ b/apps/backend/internal/orchestrator/messagequeue/service.go @@ -12,6 +12,7 @@ package messagequeue import ( "context" "errors" + "time" "github.com/kandev/kandev/internal/common/logger" "go.uber.org/zap" @@ -156,6 +157,14 @@ func (s *Service) CancelAll(ctx context.Context, sessionID string) (int, error) return n, nil } +// ListStaleByQueuedBy returns up to `limit` entries with the given queued_by +// tag whose queued_at predates olderThan, ordered by queued_at ASC. Used by +// the orchestrator workflow-queue watchdog to find orphaned workflow +// auto-start prompts. `limit <= 0` returns every match (no cap). +func (s *Service) ListStaleByQueuedBy(ctx context.Context, queuedBy string, olderThan time.Time, limit int) ([]QueuedMessage, error) { + return s.repo.ListStaleByQueuedBy(ctx, queuedBy, olderThan, limit) +} + // GetStatus returns the full pending list and capacity info for a session. func (s *Service) GetStatus(ctx context.Context, sessionID string) *QueueStatus { entries, err := s.repo.ListBySession(ctx, sessionID) diff --git a/apps/backend/internal/orchestrator/messagequeue/service_test.go b/apps/backend/internal/orchestrator/messagequeue/service_test.go index c35b96697..ee1b845ae 100644 --- a/apps/backend/internal/orchestrator/messagequeue/service_test.go +++ b/apps/backend/internal/orchestrator/messagequeue/service_test.go @@ -417,6 +417,49 @@ func TestConcurrentTakeIdempotent(t *testing.T) { assert.Equal(t, int32(1), hits.Load()) } +func TestService_ListStaleByQueuedBy(t *testing.T) { + svc := setupService(t) + ctx := context.Background() + + now := time.Now().UTC() + cutoff := now.Add(-1 * time.Minute) + + // Stale workflow entry (should match). + stale, err := svc.QueueMessage(ctx, "s1", "t1", "stale-wf", "", QueuedByWorkflow, false, nil) + require.NoError(t, err) + require.NoError(t, svc.UpdateMessage(ctx, "s1", stale.ID, "stale-wf", nil, "")) + // Backdate via direct repository access since UpdateMessage doesn't expose queued_at. + // Memory repo lets us cheat by re-inserting with QueuedAt pre-set. + mem := svc.repo.(*memoryRepository) + mem.mu.Lock() + for _, m := range mem.entries["s1"] { + if m.ID == stale.ID { + m.QueuedAt = now.Add(-10 * time.Minute) + } + } + mem.mu.Unlock() + + // Fresh workflow entry (should NOT match). + _, err = svc.QueueMessage(ctx, "s1", "t1", "fresh-wf", "", QueuedByWorkflow, false, nil) + require.NoError(t, err) + + // Stale user entry (wrong queued_by; should NOT match). + user, err := svc.QueueMessage(ctx, "s2", "t1", "stale-user", "", QueuedByUser, false, nil) + require.NoError(t, err) + mem.mu.Lock() + for _, m := range mem.entries["s2"] { + if m.ID == user.ID { + m.QueuedAt = now.Add(-10 * time.Minute) + } + } + mem.mu.Unlock() + + got, err := svc.ListStaleByQueuedBy(ctx, QueuedByWorkflow, cutoff, 0) + require.NoError(t, err) + require.Len(t, got, 1) + assert.Equal(t, stale.ID, got[0].ID) +} + func TestQueuedTimestamp(t *testing.T) { svc := setupService(t) ctx := context.Background() diff --git a/apps/backend/internal/orchestrator/messagequeue/service_test_helpers.go b/apps/backend/internal/orchestrator/messagequeue/service_test_helpers.go new file mode 100644 index 000000000..5d940f746 --- /dev/null +++ b/apps/backend/internal/orchestrator/messagequeue/service_test_helpers.go @@ -0,0 +1,25 @@ +package messagequeue + +import ( + "testing" + "time" +) + +// SetQueuedAtForTesting overwrites the queued_at timestamp of every entry on +// a session in the underlying memory repository. Test-only helper: lets +// external-package tests seed a stale queue entry without exposing the +// repository or adding production code paths that mutate queued_at. +// +// No-op when the service is not backed by the in-memory repository. Callers +// must guarantee the repo is a memoryRepository (NewServiceMemory). +func (s *Service) SetQueuedAtForTesting(_ testing.TB, sessionID string, at time.Time) { + mem, ok := s.repo.(*memoryRepository) + if !ok { + return + } + mem.mu.Lock() + defer mem.mu.Unlock() + for _, m := range mem.entries[sessionID] { + m.QueuedAt = at + } +} diff --git a/apps/backend/internal/orchestrator/service.go b/apps/backend/internal/orchestrator/service.go index 0893ea97f..8d390a45d 100644 --- a/apps/backend/internal/orchestrator/service.go +++ b/apps/backend/internal/orchestrator/service.go @@ -356,6 +356,11 @@ type Service struct { // context. key: sessionID, value: capturedPrompt. Replaced every turn. lastTurnPrompt sync.Map + // workflowQueueWatchdog is the periodic sweeper that recovers orphaned + // workflow auto-start queue entries. Nil before Start(), and remains nil + // when messageQueue or executor are not wired (some tests). + workflowQueueWatchdog *workflowQueueWatchdog + // Service state mu sync.RWMutex running bool @@ -836,6 +841,17 @@ func (s *Service) Start(ctx context.Context) error { // Subscribe to prepare events (persist result in session metadata) s.subscribePrepareEvents() + // Start the workflow-queue watchdog so orphaned auto-start prompts get + // recovered if the inline auto-resume hook misses them. Skip when the + // pieces aren't wired (tests that only need a partial service). + if s.messageQueue != nil && s.executor != nil { + wd := s.newWorkflowQueueWatchdog() + wd.Start(ctx) + s.mu.Lock() + s.workflowQueueWatchdog = wd + s.mu.Unlock() + } + s.logger.Info("orchestrator service started successfully") return nil } @@ -865,6 +881,14 @@ func (s *Service) Stop() error { errs = append(errs, err) } + s.mu.Lock() + wd := s.workflowQueueWatchdog + s.workflowQueueWatchdog = nil + s.mu.Unlock() + if wd != nil { + wd.Stop() + } + s.cancelAllClarificationWatchdogs() s.cancelAllTransientRetries() diff --git a/apps/backend/internal/orchestrator/workflow_queue_watchdog.go b/apps/backend/internal/orchestrator/workflow_queue_watchdog.go new file mode 100644 index 000000000..9bb43b68d --- /dev/null +++ b/apps/backend/internal/orchestrator/workflow_queue_watchdog.go @@ -0,0 +1,139 @@ +package orchestrator + +import ( + "context" + "os" + "strconv" + "sync" + "sync/atomic" + "time" + + "go.uber.org/zap" + + "github.com/kandev/kandev/internal/orchestrator/messagequeue" +) + +// Defaults for the workflow-queue watchdog. The orphan-age threshold gives +// inline paths (#1087/#1096/#1160/#1163) a generous window to handle the +// queue before the watchdog steps in, so the watchdog is purely a safety net +// for the long-tail races those inline paths can't catch. +const ( + defaultWorkflowQueueWatchdogInterval = 60 * time.Second + defaultWorkflowQueueWatchdogOrphanAge = 90 * time.Second + // workflowQueueWatchdogSweepLimit caps the per-tick row scan so a backlog + // after an outage (thousands of stale entries) can't balloon memory on + // one sweep. Dedupe by session_id means realistic recovery per tick is + // bounded by distinct sessions, so this cap rarely bites in practice. + workflowQueueWatchdogSweepLimit = 500 +) + +// workflowQueueWatchdog scans the message queue periodically for stale +// workflow auto-start entries whose owning session has no live agent and no +// active turn, and re-fires the auto-resume cascade (tryEnsureExecution → +// ResumeSession → boot_ready → drain). It is a self-healing safety net for +// the long-tail of races where the inline auto-resume hook either short- +// circuits, fails silently, or is bypassed by a non-workflow queue path. +type workflowQueueWatchdog struct { + svc *Service + interval time.Duration + orphanAge time.Duration + + stopCh chan struct{} + doneCh chan struct{} + stopOnce sync.Once + started atomic.Bool + wg sync.WaitGroup +} + +// newWorkflowQueueWatchdog constructs a watchdog tied to the orchestrator +// service. Interval and orphan-age are taken from KANDEV_WORKFLOW_QUEUE_ +// WATCHDOG_INTERVAL_SECONDS / _ORPHAN_AGE_SECONDS env vars when set +// (positive integers); otherwise the defaults above apply. +func (s *Service) newWorkflowQueueWatchdog() *workflowQueueWatchdog { + return &workflowQueueWatchdog{ + svc: s, + interval: durationFromEnvSeconds("KANDEV_WORKFLOW_QUEUE_WATCHDOG_INTERVAL_SECONDS", defaultWorkflowQueueWatchdogInterval), + orphanAge: durationFromEnvSeconds("KANDEV_WORKFLOW_QUEUE_WATCHDOG_ORPHAN_AGE_SECONDS", defaultWorkflowQueueWatchdogOrphanAge), + stopCh: make(chan struct{}), + doneCh: make(chan struct{}), + } +} + +// Start spawns the watchdog loop. NOT idempotent — repeated calls on the +// same instance spawn additional goroutines that all close the same doneCh +// and panic. Call exactly once per watchdog; Stop must follow to drain. +func (w *workflowQueueWatchdog) Start(ctx context.Context) { + w.started.Store(true) + go w.run(ctx) +} + +// Stop signals the watchdog loop to exit and waits for it to drain. Safe to +// call multiple times (no-op after the first). +func (w *workflowQueueWatchdog) Stop() { + w.stopOnce.Do(func() { + close(w.stopCh) + }) + if w.started.Load() { + <-w.doneCh + w.wg.Wait() + } +} + +func (w *workflowQueueWatchdog) run(ctx context.Context) { + defer close(w.doneCh) + + ticker := time.NewTicker(w.interval) + defer ticker.Stop() + + for { + select { + case <-w.stopCh: + return + case <-ctx.Done(): + return + case <-ticker.C: + w.sweep(ctx) + } + } +} + +// sweep queries the message queue for stale workflow-tagged entries and +// dispatches at most one recovery action per session per tick. +func (w *workflowQueueWatchdog) sweep(ctx context.Context) { + if w.svc.messageQueue == nil { + return + } + cutoff := time.Now().Add(-w.orphanAge) + stale, err := w.svc.messageQueue.ListStaleByQueuedBy(ctx, messagequeue.QueuedByWorkflow, cutoff, workflowQueueWatchdogSweepLimit) + if err != nil { + w.svc.logger.Warn("workflow queue watchdog: list stale failed", zap.Error(err)) + return + } + if len(stale) == 0 { + return + } + + seen := make(map[string]bool, len(stale)) + for i := range stale { + entry := stale[i] + if entry.SessionID == "" || seen[entry.SessionID] { + continue + } + seen[entry.SessionID] = true + w.svc.maybeRecoverOrphanedWorkflowQueue(ctx, entry.SessionID, entry.QueuedAt, &w.wg) + } +} + +// durationFromEnvSeconds reads a positive integer (seconds) from env or falls +// back to def. Any parse error / non-positive value uses def. +func durationFromEnvSeconds(key string, def time.Duration) time.Duration { + raw := os.Getenv(key) + if raw == "" { + return def + } + n, err := strconv.Atoi(raw) + if err != nil || n <= 0 { + return def + } + return time.Duration(n) * time.Second +} diff --git a/apps/backend/internal/orchestrator/workflow_queue_watchdog_test.go b/apps/backend/internal/orchestrator/workflow_queue_watchdog_test.go new file mode 100644 index 000000000..de590b505 --- /dev/null +++ b/apps/backend/internal/orchestrator/workflow_queue_watchdog_test.go @@ -0,0 +1,435 @@ +package orchestrator + +import ( + "context" + "sync/atomic" + "testing" + "time" + + "go.uber.org/goleak" + + "github.com/kandev/kandev/internal/orchestrator/executor" + "github.com/kandev/kandev/internal/orchestrator/messagequeue" + "github.com/kandev/kandev/internal/orchestrator/watcher" + "github.com/kandev/kandev/internal/task/models" + wfmodels "github.com/kandev/kandev/internal/workflow/models" + v1 "github.com/kandev/kandev/pkg/api/v1" +) + +// seedWorkflowQueueWatchdogTaskAndSession seeds the workspace/workflow/task/ +// session/executor rows the watchdog tests reuse: a session sitting in +// WAITING_FOR_INPUT on the Merge step with a stale workflow auto-start prompt +// already in the queue. The returned state is the orphan condition the +// watchdog is meant to recover from. +func seedWorkflowQueueWatchdogTaskAndSession( + t *testing.T, + taskID, sessionID, executionID, mergeStepID, taskWorkflow string, + sessionState models.TaskSessionState, +) (*Service, *mockAgentManager, *messagequeue.Service) { + t.Helper() + ctx := context.Background() + repo := setupTestRepo(t) + now := time.Now().UTC() + + if err := repo.CreateWorkspace(ctx, &models.Workspace{ID: "ws-wd", Name: "Test", CreatedAt: now, UpdatedAt: now}); err != nil { + t.Fatalf("create workspace: %v", err) + } + if err := repo.CreateWorkflow(ctx, &models.Workflow{ID: taskWorkflow, WorkspaceID: "ws-wd", Name: "WF", CreatedAt: now, UpdatedAt: now}); err != nil { + t.Fatalf("create workflow: %v", err) + } + + stepGetter := newMockStepGetter() + stepGetter.steps[mergeStepID] = &wfmodels.WorkflowStep{ + ID: mergeStepID, WorkflowID: taskWorkflow, Name: "Merge", Position: 1, + Events: wfmodels.StepEvents{ + OnEnter: []wfmodels.OnEnterAction{{Type: wfmodels.OnEnterAutoStartAgent}}, + }, + } + + if err := repo.CreateTask(ctx, &models.Task{ + ID: taskID, WorkflowID: taskWorkflow, WorkflowStepID: mergeStepID, + Title: "Test", Description: "Test task", + State: v1.TaskStateInProgress, + CreatedAt: now, UpdatedAt: now, + }); err != nil { + t.Fatalf("create task: %v", err) + } + if err := repo.CreateTaskSession(ctx, &models.TaskSession{ + ID: sessionID, + TaskID: taskID, + AgentProfileID: "profile-wd", + AgentExecutionID: executionID, + State: sessionState, + IsPrimary: true, + StartedAt: now, UpdatedAt: now, + }); err != nil { + t.Fatalf("create session: %v", err) + } + seedExecutorRunning(t, repo, sessionID, taskID, executionID) + + taskRepo := newMockTaskRepo() + taskRepo.tasks[taskID] = &v1.Task{ID: taskID, WorkflowID: taskWorkflow, State: v1.TaskStateInProgress} + + agentMgr := &mockAgentManager{repoForExecutionLookup: repo} + mq := messagequeue.NewServiceMemory(testLogger()) + exec := executor.NewExecutor(agentMgr, repo, testLogger(), executor.ExecutorConfig{}) + + svc := &Service{ + logger: testLogger(), + repo: repo, + taskRepo: taskRepo, + agentManager: agentMgr, + messageQueue: mq, + executor: exec, + } + svc.SetWorkflowStepGetter(stepGetter) + + // Seed a stale workflow queue entry. queued_at is set well before the + // watchdog's orphanAge cutoff so the sweep query picks it up. + stale := &messagequeue.QueuedMessage{ + SessionID: sessionID, + TaskID: taskID, + Content: "merge prompt", + QueuedBy: messagequeue.QueuedByWorkflow, + QueuedAt: now.Add(-10 * time.Minute), + Metadata: map[string]interface{}{"workflow_step_name": "Merge"}, + } + if _, err := mq.QueueMessageWithMetadata(ctx, stale.SessionID, stale.TaskID, stale.Content, "", stale.QueuedBy, false, nil, stale.Metadata); err != nil { + t.Fatalf("seed queue: %v", err) + } + // QueueMessageWithMetadata sets queued_at = now; backdate it via the + // memory-repo test helper so the watchdog's "older than X" filter + // matches. Mirrors how the bug looks in production — a stuck queue + // entry whose queued_at is many minutes/hours in the past. + mq.SetQueuedAtForTesting(t, sessionID, now.Add(-10*time.Minute)) + + return svc, agentMgr, mq +} + +// TestWorkflowQueueWatchdog_RecoversOrphanedWorkflowQueue: stale workflow +// entry + no live agent + no active turn → sweep must drive +// tryEnsureExecution which calls LaunchAgent via the executor. +func TestWorkflowQueueWatchdog_RecoversOrphanedWorkflowQueue(t *testing.T) { + const ( + taskID = "task-wd-1" + sessionID = "sess-wd-1" + executionID = "exec-wd-1" + mergeStepID = "step-wd-merge" + taskWorkflow = "wf-wd-1" + ) + svc, agentMgr, _ := seedWorkflowQueueWatchdogTaskAndSession( + t, taskID, sessionID, executionID, mergeStepID, taskWorkflow, + models.TaskSessionStateWaitingForInput, + ) + + // Agent is dead; tryEnsureExecution → ensureSessionRunning checks + // GetExecutionBySession which calls IsAgentRunningForSession. + agentMgr.isAgentRunning = false + + launchCalled := make(chan struct{}) + // resumeDone closes after waitForSessionReady has had at least one full + // poll cycle (500ms) to observe WAITING_FOR_INPUT and exit — guarantees + // the resume goroutine drains before goleak inspects. + resumeDone := make(chan struct{}) + bootCtx := context.Background() + var launchedOnce atomic.Bool + agentMgr.launchAgentFunc = func(_ context.Context, req *executor.LaunchAgentRequest) (*executor.LaunchAgentResponse, error) { + // Only the first call drives boot_ready; subsequent calls (e.g. a + // drain-triggered re-resume) reuse the response without spawning + // another helper goroutine. + if !launchedOnce.CompareAndSwap(false, true) { + return &executor.LaunchAgentResponse{AgentExecutionID: executionID + "-resumed"}, nil + } + close(launchCalled) + // Mirror the production handler: once the agent has resumed, fire + // boot_ready so waitForSessionReady returns. Without this the + // resume goroutine would block on waitForSessionReady's 90s + // timeout and leak. + go func() { + defer close(resumeDone) + tick := time.NewTicker(5 * time.Millisecond) + defer tick.Stop() + deadline := time.After(3 * time.Second) + for { + select { + case <-tick.C: + sess, err := svc.repo.GetTaskSession(bootCtx, req.SessionID) + if err != nil || sess == nil || sess.State != models.TaskSessionStateStarting { + continue + } + svc.handleAgentBootReady(bootCtx, watcher.AgentEventData{ + TaskID: req.TaskID, SessionID: req.SessionID, + }) + // Give waitForSessionReady (500ms poll) at least one + // full cycle to observe WAITING_FOR_INPUT and exit. + time.Sleep(600 * time.Millisecond) + return + case <-deadline: + return + } + } + }() + return &executor.LaunchAgentResponse{AgentExecutionID: executionID + "-resumed"}, nil + } + + wd := svc.newWorkflowQueueWatchdog() + wd.sweep(context.Background()) + + select { + case <-launchCalled: + case <-time.After(5 * time.Second): + t.Fatalf("watchdog did not drive LaunchAgent within 5s") + } + select { + case <-resumeDone: + case <-time.After(5 * time.Second): + t.Fatalf("resume goroutine did not drain within 5s") + } +} + +// TestWorkflowQueueWatchdog_SkipsLiveAgent: stale queue entry but the +// agent is alive → handleAgentReady will drain on next turn end. Watchdog +// must not recurse into LaunchAgent. +func TestWorkflowQueueWatchdog_SkipsLiveAgent(t *testing.T) { + const ( + taskID = "task-wd-2" + sessionID = "sess-wd-2" + executionID = "exec-wd-2" + mergeStepID = "step-wd-merge2" + taskWorkflow = "wf-wd-2" + ) + svc, agentMgr, mq := seedWorkflowQueueWatchdogTaskAndSession( + t, taskID, sessionID, executionID, mergeStepID, taskWorkflow, + models.TaskSessionStateRunning, + ) + agentMgr.isAgentRunning = true + + agentMgr.launchAgentFunc = func(_ context.Context, _ *executor.LaunchAgentRequest) (*executor.LaunchAgentResponse, error) { + t.Fatalf("LaunchAgent must not be called when the agent is alive") + return nil, nil + } + + wd := svc.newWorkflowQueueWatchdog() + wd.sweep(context.Background()) + + // Queue must be untouched: the watchdog only drops on terminal sessions. + if got := mq.GetStatus(context.Background(), sessionID).Count; got != 1 { + t.Errorf("queue count after sweep = %d, want 1 (untouched)", got) + } +} + +// TestWorkflowQueueWatchdog_SkipsActiveTurn: an active turn exists, so a +// future agent.ready will run the natural drain. Watchdog must not fire. +func TestWorkflowQueueWatchdog_SkipsActiveTurn(t *testing.T) { + const ( + taskID = "task-wd-3" + sessionID = "sess-wd-3" + executionID = "exec-wd-3" + mergeStepID = "step-wd-merge3" + taskWorkflow = "wf-wd-3" + ) + svc, agentMgr, mq := seedWorkflowQueueWatchdogTaskAndSession( + t, taskID, sessionID, executionID, mergeStepID, taskWorkflow, + models.TaskSessionStateRunning, + ) + svc.activeTurns.Store(sessionID, "turn-1") + agentMgr.isAgentRunning = false // doesn't matter; activeTurns guard runs first + + agentMgr.launchAgentFunc = func(_ context.Context, _ *executor.LaunchAgentRequest) (*executor.LaunchAgentResponse, error) { + t.Fatalf("LaunchAgent must not be called when an active turn exists") + return nil, nil + } + + wd := svc.newWorkflowQueueWatchdog() + wd.sweep(context.Background()) + + if got := mq.GetStatus(context.Background(), sessionID).Count; got != 1 { + t.Errorf("queue count after sweep = %d, want 1 (untouched)", got) + } +} + +// TestWorkflowQueueWatchdog_DropsTerminalSessionQueue: session is terminal, +// the workflow auto-start prompt can never drain — watchdog removes it so +// it stops polluting the queue table. +func TestWorkflowQueueWatchdog_DropsTerminalSessionQueue(t *testing.T) { + const ( + taskID = "task-wd-4" + sessionID = "sess-wd-4" + executionID = "exec-wd-4" + mergeStepID = "step-wd-merge4" + taskWorkflow = "wf-wd-4" + ) + svc, agentMgr, mq := seedWorkflowQueueWatchdogTaskAndSession( + t, taskID, sessionID, executionID, mergeStepID, taskWorkflow, + models.TaskSessionStateCompleted, + ) + agentMgr.launchAgentFunc = func(_ context.Context, _ *executor.LaunchAgentRequest) (*executor.LaunchAgentResponse, error) { + t.Fatalf("LaunchAgent must not be called for a terminal session") + return nil, nil + } + + wd := svc.newWorkflowQueueWatchdog() + wd.sweep(context.Background()) + + if got := mq.GetStatus(context.Background(), sessionID).Count; got != 0 { + t.Errorf("queue count after sweep = %d, want 0 (workflow entry dropped)", got) + } +} + +// TestWorkflowQueueWatchdog_DedupesBySession: two stale workflow entries on +// the same session must trigger AT MOST one recovery action per sweep. The +// dedupe inside sweep is what keeps a many-entries-per-session backlog from +// fanning out N concurrent resume goroutines for the same session. +func TestWorkflowQueueWatchdog_DedupesBySession(t *testing.T) { + const ( + taskID = "task-wd-5" + sessionID = "sess-wd-5" + executionID = "exec-wd-5" + mergeStepID = "step-wd-merge5" + taskWorkflow = "wf-wd-5" + ) + svc, agentMgr, mq := seedWorkflowQueueWatchdogTaskAndSession( + t, taskID, sessionID, executionID, mergeStepID, taskWorkflow, + models.TaskSessionStateWaitingForInput, + ) + // Inject a second stale workflow entry against the same session. + ctx := context.Background() + if _, err := mq.QueueMessageWithMetadata(ctx, sessionID, taskID, "second", "", messagequeue.QueuedByWorkflow, false, nil, nil); err != nil { + t.Fatalf("seed second queue entry: %v", err) + } + mq.SetQueuedAtForTesting(t, sessionID, time.Now().Add(-10*time.Minute)) + if got := mq.GetStatus(ctx, sessionID).Count; got != 2 { + t.Fatalf("expected 2 stale entries before sweep, got %d", got) + } + + // Dead agent so the recover branch fires. + agentMgr.isAgentRunning = false + + var launchCount atomic.Int32 + launchCalled := make(chan struct{}, 4) + resumeDone := make(chan struct{}) + var resumeOnce atomic.Bool + agentMgr.launchAgentFunc = func(_ context.Context, req *executor.LaunchAgentRequest) (*executor.LaunchAgentResponse, error) { + launchCount.Add(1) + select { + case launchCalled <- struct{}{}: + default: + } + if !resumeOnce.CompareAndSwap(false, true) { + return &executor.LaunchAgentResponse{AgentExecutionID: executionID + "-resumed"}, nil + } + go func() { + defer close(resumeDone) + tick := time.NewTicker(5 * time.Millisecond) + defer tick.Stop() + deadline := time.After(3 * time.Second) + for { + select { + case <-tick.C: + sess, err := svc.repo.GetTaskSession(context.Background(), req.SessionID) + if err != nil || sess == nil || sess.State != models.TaskSessionStateStarting { + continue + } + svc.handleAgentBootReady(context.Background(), watcher.AgentEventData{ + TaskID: req.TaskID, SessionID: req.SessionID, + }) + time.Sleep(600 * time.Millisecond) + return + case <-deadline: + return + } + } + }() + return &executor.LaunchAgentResponse{AgentExecutionID: executionID + "-resumed"}, nil + } + + wd := svc.newWorkflowQueueWatchdog() + wd.sweep(ctx) + + // Wait for the single recovery to settle. + select { + case <-launchCalled: + case <-time.After(5 * time.Second): + t.Fatalf("watchdog did not drive LaunchAgent within 5s") + } + select { + case <-resumeDone: + case <-time.After(5 * time.Second): + t.Fatalf("resume goroutine did not drain within 5s") + } + + // LaunchAgent may have been re-invoked once by a re-resume after drain + // completed (legitimate); what matters is the watchdog itself did NOT + // spawn N concurrent resumes for the dedupe-target session — checked + // by ensuring sweep() returned before any extra resume fired (the + // launchCount snapshot below). It's >=1 (first sweep) and bounded. + got := launchCount.Load() + if got < 1 { + t.Errorf("expected at least 1 LaunchAgent call from the dedupe sweep, got %d", got) + } + if got > 2 { + t.Errorf("expected at most 2 LaunchAgent calls (1 sweep + 1 re-resume), got %d — dedupe may be broken", got) + } +} + +// TestWorkflowQueueWatchdog_TerminalSessionPreservesUserEntries: when the +// session is terminal, workflow auto-start entries are dropped but +// user-authored entries on the same session must be left intact (the user +// authored them and their cleanup is owned by other paths). +func TestWorkflowQueueWatchdog_TerminalSessionPreservesUserEntries(t *testing.T) { + const ( + taskID = "task-wd-6" + sessionID = "sess-wd-6" + executionID = "exec-wd-6" + mergeStepID = "step-wd-merge6" + taskWorkflow = "wf-wd-6" + ) + svc, _, mq := seedWorkflowQueueWatchdogTaskAndSession( + t, taskID, sessionID, executionID, mergeStepID, taskWorkflow, + models.TaskSessionStateCompleted, + ) + ctx := context.Background() + // Add a user-authored entry alongside the workflow one. + if _, err := mq.QueueMessage(ctx, sessionID, taskID, "user follow-up", "", messagequeue.QueuedByUser, false, nil); err != nil { + t.Fatalf("seed user queue: %v", err) + } + if got := mq.GetStatus(ctx, sessionID).Count; got != 2 { + t.Fatalf("expected 2 entries before sweep, got %d", got) + } + + wd := svc.newWorkflowQueueWatchdog() + wd.sweep(ctx) + + status := mq.GetStatus(ctx, sessionID) + if status.Count != 1 { + t.Fatalf("expected 1 entry after sweep (user kept, workflow dropped), got %d", status.Count) + } + if status.Entries[0].QueuedBy != messagequeue.QueuedByUser { + t.Errorf("surviving entry queued_by = %q, want %q", status.Entries[0].QueuedBy, messagequeue.QueuedByUser) + } +} + +// TestWorkflowQueueWatchdog_StartStop_NoLeaks asserts the Start/Stop +// lifecycle drains cleanly so goleak.VerifyTestMain stays green. +func TestWorkflowQueueWatchdog_StartStop_NoLeaks(t *testing.T) { + defer goleak.VerifyNone(t) + + svc := &Service{ + logger: testLogger(), + messageQueue: messagequeue.NewServiceMemory(testLogger()), + } + // Override defaults to a short interval so the test doesn't sit idle. + wd := &workflowQueueWatchdog{ + svc: svc, + interval: 10 * time.Millisecond, + orphanAge: 24 * time.Hour, // nothing should match + stopCh: make(chan struct{}), + doneCh: make(chan struct{}), + } + wd.Start(context.Background()) + // Let the ticker fire at least once. + time.Sleep(50 * time.Millisecond) + wd.Stop() + // Stop is idempotent. + wd.Stop() +}