From 4b7df3dd41fb7a38a74000f3742180fb6f8214fe Mon Sep 17 00:00:00 2001
From: Kandev Agent <agent@kandev.dev>
Date: Mon, 1 Jun 2026 20:07:51 +0100
Subject: [PATCH 1/5] fix(orchestrator): recover orphaned workflow auto-start
 queue entries via watchdog
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

AFK workflows sometimes hang one step before the final MERGE step
because a workflow auto-start prompt gets queued but no inline drain
path picks it up. The recent fixes (#1087/#1096/#1160/#1163) each
closed one specific path inline; this adds a periodic watchdog as a
single self-healing safety net for the residual long-tail of races
(stale executor record, silent tryEnsureExecution failures, queues
written via paths that bypass scheduleAutoResumeForWorkflowQueue).

Every 60s, the watchdog scans queued_messages for queued_by='workflow'
entries older than 90s and, per session: drops the entries when the
session is terminal, skips when an active turn / cancel / reset is in
flight or the agent process is genuinely alive, and otherwise drives
tryEnsureExecution to fire the same ResumeSession → agent.boot_ready
→ handleAgentBootReady → drainQueuedMessageForPromptableSession
cascade #1163 introduced inline. Bypasses the in-memory execution
gate from scheduleAutoResumeForWorkflowQueue because the agentManager
probe is authoritative when the executor store lags a dying process.

Wired into Service.Start/Stop with explicit drain so goleak stays
green. Interval and orphan-age are tunable via env vars
KANDEV_WORKFLOW_QUEUE_WATCHDOG_INTERVAL_SECONDS and
KANDEV_WORKFLOW_QUEUE_WATCHDOG_ORPHAN_AGE_SECONDS.

Tests cover the full recover chain end-to-end (sweep → LaunchAgent →
boot_ready → drain), the skip paths (live agent, active turn), the
terminal-session drop path, and goleak-clean Start/Stop lifecycle.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
---
 .../orchestrator/event_handlers_workflow.go   |  85 +++++
 .../orchestrator/messagequeue/repository.go   |  11 +-
 .../messagequeue/repository_memory.go         |  19 ++
 .../messagequeue/repository_sqlite.go         |  24 ++
 .../messagequeue/repository_sqlite_test.go    |  52 +++
 .../orchestrator/messagequeue/service.go      |   8 +
 .../orchestrator/messagequeue/service_test.go |  43 +++
 .../messagequeue/service_test_helpers.go      |  24 ++
 apps/backend/internal/orchestrator/service.go |  18 ++
 .../orchestrator/workflow_queue_watchdog.go   | 127 ++++++++
 .../workflow_queue_watchdog_test.go           | 302 ++++++++++++++++++
 11 files changed, 712 insertions(+), 1 deletion(-)
 create mode 100644 apps/backend/internal/orchestrator/messagequeue/service_test_helpers.go
 create mode 100644 apps/backend/internal/orchestrator/workflow_queue_watchdog.go
 create mode 100644 apps/backend/internal/orchestrator/workflow_queue_watchdog_test.go

diff --git a/apps/backend/internal/orchestrator/event_handlers_workflow.go b/apps/backend/internal/orchestrator/event_handlers_workflow.go
index 76b04ffbf3..0d254d99ec 100644
--- a/apps/backend/internal/orchestrator/event_handlers_workflow.go
+++ b/apps/backend/internal/orchestrator/event_handlers_workflow.go
@@ -1539,6 +1539,91 @@ 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) {
+	session, err := s.repo.GetTaskSession(ctx, sessionID)
+	if err != nil || session == nil {
+		s.svcDropOrphanedWorkflowQueue(ctx, sessionID, "session_missing")
+		return
+	}
+	if isTerminalSessionState(session.State) {
+		s.svcDropOrphanedWorkflowQueue(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 {
+		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 != nil && 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.
+	go s.tryEnsureExecution(context.WithoutCancel(ctx), sessionID)
+}
+
+// svcDropOrphanedWorkflowQueue 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) svcDropOrphanedWorkflowQueue(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 d59251586b..d0b5cb9e55 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 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.
+	ListStaleByQueuedBy(ctx context.Context, queuedBy string, olderThan time.Time) ([]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 257e10e819..5193c21d48 100644
--- a/apps/backend/internal/orchestrator/messagequeue/repository_memory.go
+++ b/apps/backend/internal/orchestrator/messagequeue/repository_memory.go
@@ -98,6 +98,25 @@ func (r *memoryRepository) ListBySession(_ context.Context, sessionID string) ([
 	return out, nil
 }
 
+func (r *memoryRepository) ListStaleByQueuedBy(_ context.Context, queuedBy string, olderThan time.Time) ([]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) })
+	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 b8c9a7a0c2..a0a258f33e 100644
--- a/apps/backend/internal/orchestrator/messagequeue/repository_sqlite.go
+++ b/apps/backend/internal/orchestrator/messagequeue/repository_sqlite.go
@@ -208,6 +208,30 @@ 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) ([]QueuedMessage, error) {
+	rows, err := r.ro.QueryxContext(ctx, r.ro.Rebind(`
+		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
+	`), queuedBy, olderThan)
+	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 5f21600ee9..2304d19ddd 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,57 @@ 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)
+	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)
+	}
+
+	// Empty result when cutoff predates everything.
+	empty, err := repo.ListStaleByQueuedBy(ctx, QueuedByWorkflow, stale.Add(-2*time.Hour))
+	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 4dfbc994ee..717ad9017d 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,13 @@ func (s *Service) CancelAll(ctx context.Context, sessionID string) (int, error)
 	return n, nil
 }
 
+// ListStaleByQueuedBy returns entries with the given queued_by tag whose
+// queued_at predates olderThan. Used by the orchestrator workflow-queue
+// watchdog to find orphaned workflow auto-start prompts.
+func (s *Service) ListStaleByQueuedBy(ctx context.Context, queuedBy string, olderThan time.Time) ([]QueuedMessage, error) {
+	return s.repo.ListStaleByQueuedBy(ctx, queuedBy, olderThan)
+}
+
 // 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 c35b966978..a350cd4cb5 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)
+	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 0000000000..f21a00f33d
--- /dev/null
+++ b/apps/backend/internal/orchestrator/messagequeue/service_test_helpers.go
@@ -0,0 +1,24 @@
+package messagequeue
+
+import (
+	"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(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 0893ea97f8..93c019f644 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,14 @@ 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 {
+		s.workflowQueueWatchdog = s.newWorkflowQueueWatchdog()
+		s.workflowQueueWatchdog.Start(ctx)
+	}
+
 	s.logger.Info("orchestrator service started successfully")
 	return nil
 }
@@ -865,6 +878,11 @@ func (s *Service) Stop() error {
 		errs = append(errs, err)
 	}
 
+	if s.workflowQueueWatchdog != nil {
+		s.workflowQueueWatchdog.Stop()
+		s.workflowQueueWatchdog = nil
+	}
+
 	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 0000000000..a55527442c
--- /dev/null
+++ b/apps/backend/internal/orchestrator/workflow_queue_watchdog.go
@@ -0,0 +1,127 @@
+package orchestrator
+
+import (
+	"context"
+	"os"
+	"strconv"
+	"sync"
+	"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
+)
+
+// 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
+}
+
+// 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. Idempotent: callers that invoke Start more
+// than once on the same instance will spawn additional goroutines, so this
+// must be called exactly once per watchdog. Stop must be called to drain.
+func (w *workflowQueueWatchdog) Start(ctx context.Context) {
+	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)
+	})
+	<-w.doneCh
+}
+
+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)
+	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)
+	}
+}
+
+// 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 0000000000..ad20f8ff28
--- /dev/null
+++ b/apps/backend/internal/orchestrator/workflow_queue_watchdog_test.go
@@ -0,0 +1,302 @@
+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(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_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()
+}

From 1b46c5aac491e35dd5806b2faefc7f5da400e641 Mon Sep 17 00:00:00 2001
From: Kandev Agent <agent@kandev.dev>
Date: Mon, 1 Jun 2026 20:11:14 +0100
Subject: [PATCH 2/5] fix(orchestrator): harden workflow queue watchdog
 decision branch
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

- Distinguish transient DB read failure from a genuinely missing
  session in maybeRecoverOrphanedWorkflowQueue. The previous "drop
  on err != nil || session == nil" path would have silently dropped
  workflow entries on a momentary repo blip; now a load error logs
  and returns so the next tick retries.
- Require agentManager non-nil alongside executor before firing
  tryEnsureExecution. The IsAgentRunningForSession gate already
  guarded against nil, but the subsequent resume cascade dereferences
  agentManager through the executor and would have nil-deref in any
  exotic test wiring that builds the watchdog without it.
- Rename svcDropOrphanedWorkflowQueue → dropOrphanedWorkflowQueue
  (redundant svc prefix on a *Service method).
- Tighten the Start() docstring on the watchdog: it is explicitly NOT
  idempotent (the prior comment said "Idempotent" then immediately
  said "call exactly once").

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
---
 .../orchestrator/event_handlers_workflow.go   | 22 +++++++++++++------
 .../orchestrator/workflow_queue_watchdog.go   |  6 ++---
 2 files changed, 18 insertions(+), 10 deletions(-)

diff --git a/apps/backend/internal/orchestrator/event_handlers_workflow.go b/apps/backend/internal/orchestrator/event_handlers_workflow.go
index 0d254d99ec..135e846804 100644
--- a/apps/backend/internal/orchestrator/event_handlers_workflow.go
+++ b/apps/backend/internal/orchestrator/event_handlers_workflow.go
@@ -1556,12 +1556,20 @@ func (s *Service) scheduleAutoResumeForWorkflowQueue(ctx context.Context, sessio
 // recovery log line.
 func (s *Service) maybeRecoverOrphanedWorkflowQueue(ctx context.Context, sessionID string, queuedAt time.Time) {
 	session, err := s.repo.GetTaskSession(ctx, sessionID)
-	if err != nil || session == nil {
-		s.svcDropOrphanedWorkflowQueue(ctx, sessionID, "session_missing")
+	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.svcDropOrphanedWorkflowQueue(ctx, sessionID, "terminal_session")
+		s.dropOrphanedWorkflowQueue(ctx, sessionID, "terminal_session")
 		return
 	}
 	if _, hasActiveTurn := s.activeTurns.Load(sessionID); hasActiveTurn {
@@ -1570,14 +1578,14 @@ func (s *Service) maybeRecoverOrphanedWorkflowQueue(ctx context.Context, session
 	if s.isSessionResetInProgress(sessionID) || s.isCancelInFlight(sessionID) {
 		return
 	}
-	if s.executor == nil {
+	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 != nil && s.agentManager.IsAgentRunningForSession(ctx, sessionID) {
+	if s.agentManager.IsAgentRunningForSession(ctx, sessionID) {
 		return
 	}
 	ageSec := int(time.Since(queuedAt) / time.Second)
@@ -1593,10 +1601,10 @@ func (s *Service) maybeRecoverOrphanedWorkflowQueue(ctx context.Context, session
 	go s.tryEnsureExecution(context.WithoutCancel(ctx), sessionID)
 }
 
-// svcDropOrphanedWorkflowQueue removes every workflow-tagged queued entry for
+// 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) svcDropOrphanedWorkflowQueue(ctx context.Context, sessionID, reason string) {
+func (s *Service) dropOrphanedWorkflowQueue(ctx context.Context, sessionID, reason string) {
 	if s.messageQueue == nil {
 		return
 	}
diff --git a/apps/backend/internal/orchestrator/workflow_queue_watchdog.go b/apps/backend/internal/orchestrator/workflow_queue_watchdog.go
index a55527442c..59c2096183 100644
--- a/apps/backend/internal/orchestrator/workflow_queue_watchdog.go
+++ b/apps/backend/internal/orchestrator/workflow_queue_watchdog.go
@@ -51,9 +51,9 @@ func (s *Service) newWorkflowQueueWatchdog() *workflowQueueWatchdog {
 	}
 }
 
-// Start spawns the watchdog loop. Idempotent: callers that invoke Start more
-// than once on the same instance will spawn additional goroutines, so this
-// must be called exactly once per watchdog. Stop must be called to drain.
+// 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) {
 	go w.run(ctx)
 }

From f31a0d929fb842e1692cf60fb26cce8132feeb9a Mon Sep 17 00:00:00 2001
From: Kandev Agent <agent@kandev.dev>
Date: Mon, 1 Jun 2026 20:19:17 +0100
Subject: [PATCH 3/5] fix(messagequeue): cap watchdog stale-scan and document
 FAILED escalation
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Independent-review pass on the workflow queue watchdog flagged two
defensive improvements:

1. Cap ListStaleByQueuedBy result set. After a backend outage the
   queued_messages table could hold thousands of stale workflow
   entries; the previous unbounded SELECT would load all of them on
   the next tick. Add a `limit int` parameter to the Repository /
   Service signature (limit<=0 keeps "no cap" semantics) and have the
   watchdog pass workflowQueueWatchdogSweepLimit=500 per tick. The
   memory repo honors the same cap. Dedupe by session_id inside sweep
   means realistic per-tick recovery throughput is bounded by distinct
   sessions, so the cap rarely bites in practice — it just prevents
   pathological backlog floods.

2. Document the FAILED-escalation side-effect on
   maybeRecoverOrphanedWorkflowQueue. tryEnsureExecution →
   ensureSessionRunning marks the session FAILED via
   updateTaskSessionState on any non-AlreadyRunning ResumeSession
   error. The watchdog inherits this from the inline auto-resume path
   (#1163), but because the watchdog fires from a wider trigger surface
   (any orphaned workflow queue, not just freshly-queued ones), a stuck
   session with a broken executor environment can flip to terminal.
   Add a comment so the next maintainer knows the watchdog is
   "best-effort recover OR fail loudly".

Tests updated to pass the new limit parameter; new repository test
asserts the LIMIT clause caps and orders correctly.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
---
 .../orchestrator/event_handlers_workflow.go      |  9 +++++++++
 .../orchestrator/messagequeue/repository.go      | 10 +++++-----
 .../messagequeue/repository_memory.go            |  5 ++++-
 .../messagequeue/repository_sqlite.go            | 13 +++++++++----
 .../messagequeue/repository_sqlite_test.go       | 16 ++++++++++++++--
 .../orchestrator/messagequeue/service.go         | 11 ++++++-----
 .../orchestrator/messagequeue/service_test.go    |  2 +-
 .../orchestrator/workflow_queue_watchdog.go      |  7 ++++++-
 8 files changed, 54 insertions(+), 19 deletions(-)

diff --git a/apps/backend/internal/orchestrator/event_handlers_workflow.go b/apps/backend/internal/orchestrator/event_handlers_workflow.go
index 135e846804..d332c66beb 100644
--- a/apps/backend/internal/orchestrator/event_handlers_workflow.go
+++ b/apps/backend/internal/orchestrator/event_handlers_workflow.go
@@ -1598,6 +1598,15 @@ func (s *Service) maybeRecoverOrphanedWorkflowQueue(ctx context.Context, session
 	// 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".
 	go s.tryEnsureExecution(context.WithoutCancel(ctx), sessionID)
 }
 
diff --git a/apps/backend/internal/orchestrator/messagequeue/repository.go b/apps/backend/internal/orchestrator/messagequeue/repository.go
index d0b5cb9e55..a5af41e781 100644
--- a/apps/backend/internal/orchestrator/messagequeue/repository.go
+++ b/apps/backend/internal/orchestrator/messagequeue/repository.go
@@ -21,11 +21,11 @@ type Repository interface {
 	// ListBySession returns all entries for a session ordered by position ascending.
 	ListBySession(ctx context.Context, sessionID string) ([]QueuedMessage, error)
 
-	// ListStaleByQueuedBy returns 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.
-	ListStaleByQueuedBy(ctx context.Context, queuedBy string, olderThan time.Time) ([]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 5193c21d48..36883bf841 100644
--- a/apps/backend/internal/orchestrator/messagequeue/repository_memory.go
+++ b/apps/backend/internal/orchestrator/messagequeue/repository_memory.go
@@ -98,7 +98,7 @@ func (r *memoryRepository) ListBySession(_ context.Context, sessionID string) ([
 	return out, nil
 }
 
-func (r *memoryRepository) ListStaleByQueuedBy(_ context.Context, queuedBy string, olderThan time.Time) ([]QueuedMessage, error) {
+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
@@ -114,6 +114,9 @@ func (r *memoryRepository) ListStaleByQueuedBy(_ context.Context, queuedBy strin
 		}
 	}
 	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
 }
 
diff --git a/apps/backend/internal/orchestrator/messagequeue/repository_sqlite.go b/apps/backend/internal/orchestrator/messagequeue/repository_sqlite.go
index a0a258f33e..7249f2dbd2 100644
--- a/apps/backend/internal/orchestrator/messagequeue/repository_sqlite.go
+++ b/apps/backend/internal/orchestrator/messagequeue/repository_sqlite.go
@@ -208,14 +208,19 @@ 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) ([]QueuedMessage, error) {
-	rows, err := r.ro.QueryxContext(ctx, r.ro.Rebind(`
+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
-	`), queuedBy, olderThan)
+		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)
 	}
diff --git a/apps/backend/internal/orchestrator/messagequeue/repository_sqlite_test.go b/apps/backend/internal/orchestrator/messagequeue/repository_sqlite_test.go
index 2304d19ddd..526a77f249 100644
--- a/apps/backend/internal/orchestrator/messagequeue/repository_sqlite_test.go
+++ b/apps/backend/internal/orchestrator/messagequeue/repository_sqlite_test.go
@@ -343,7 +343,7 @@ func TestSQLiteRepository_ListStaleByQueuedBy(t *testing.T) {
 	mustInsert("s2", QueuedByUser, stale)     // wrong queued_by
 	staleWorkflowS3 := mustInsert("s3", QueuedByWorkflow, stale.Add(-1*time.Minute))
 
-	got, err := repo.ListStaleByQueuedBy(ctx, QueuedByWorkflow, cutoff)
+	got, err := repo.ListStaleByQueuedBy(ctx, QueuedByWorkflow, cutoff, 0)
 	if err != nil {
 		t.Fatalf("list stale: %v", err)
 	}
@@ -356,8 +356,20 @@ func TestSQLiteRepository_ListStaleByQueuedBy(t *testing.T) {
 			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))
+	empty, err := repo.ListStaleByQueuedBy(ctx, QueuedByWorkflow, stale.Add(-2*time.Hour), 0)
 	if err != nil {
 		t.Fatalf("list stale empty: %v", err)
 	}
diff --git a/apps/backend/internal/orchestrator/messagequeue/service.go b/apps/backend/internal/orchestrator/messagequeue/service.go
index 717ad9017d..86d3dc5d6f 100644
--- a/apps/backend/internal/orchestrator/messagequeue/service.go
+++ b/apps/backend/internal/orchestrator/messagequeue/service.go
@@ -157,11 +157,12 @@ func (s *Service) CancelAll(ctx context.Context, sessionID string) (int, error)
 	return n, nil
 }
 
-// ListStaleByQueuedBy returns entries with the given queued_by tag whose
-// queued_at predates olderThan. Used by the orchestrator workflow-queue
-// watchdog to find orphaned workflow auto-start prompts.
-func (s *Service) ListStaleByQueuedBy(ctx context.Context, queuedBy string, olderThan time.Time) ([]QueuedMessage, error) {
-	return s.repo.ListStaleByQueuedBy(ctx, queuedBy, olderThan)
+// 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.
diff --git a/apps/backend/internal/orchestrator/messagequeue/service_test.go b/apps/backend/internal/orchestrator/messagequeue/service_test.go
index a350cd4cb5..ee1b845aeb 100644
--- a/apps/backend/internal/orchestrator/messagequeue/service_test.go
+++ b/apps/backend/internal/orchestrator/messagequeue/service_test.go
@@ -454,7 +454,7 @@ func TestService_ListStaleByQueuedBy(t *testing.T) {
 	}
 	mem.mu.Unlock()
 
-	got, err := svc.ListStaleByQueuedBy(ctx, QueuedByWorkflow, cutoff)
+	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)
diff --git a/apps/backend/internal/orchestrator/workflow_queue_watchdog.go b/apps/backend/internal/orchestrator/workflow_queue_watchdog.go
index 59c2096183..6f9ba8c8df 100644
--- a/apps/backend/internal/orchestrator/workflow_queue_watchdog.go
+++ b/apps/backend/internal/orchestrator/workflow_queue_watchdog.go
@@ -19,6 +19,11 @@ import (
 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
@@ -92,7 +97,7 @@ func (w *workflowQueueWatchdog) sweep(ctx context.Context) {
 		return
 	}
 	cutoff := time.Now().Add(-w.orphanAge)
-	stale, err := w.svc.messageQueue.ListStaleByQueuedBy(ctx, messagequeue.QueuedByWorkflow, cutoff)
+	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

From 83588b4399f9f6622e6f9d527a120ff5214cc664 Mon Sep 17 00:00:00 2001
From: Kandev Agent <agent@kandev.dev>
Date: Mon, 1 Jun 2026 20:22:27 +0100
Subject: [PATCH 4/5] test(orchestrator): cover watchdog session-dedupe and
 mixed-queue terminal drop

QA pass found two plan edge cases without test coverage:

- Dedupe by session_id: multiple stale workflow entries on the same
  session must trigger at most one recovery action per sweep, otherwise
  a backlog would fan out N concurrent resume goroutines for the same
  session. Add TestWorkflowQueueWatchdog_DedupesBySession exercising
  this with two stale entries.
- Mixed queue on terminal session: workflow entries dropped, user
  entries preserved. The drop loop already guards on QueuedBy but no
  test asserted the preservation half. Add
  TestWorkflowQueueWatchdog_TerminalSessionPreservesUserEntries.

Pure test additions; no production code changes.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
---
 .../workflow_queue_watchdog_test.go           | 129 ++++++++++++++++++
 1 file changed, 129 insertions(+)

diff --git a/apps/backend/internal/orchestrator/workflow_queue_watchdog_test.go b/apps/backend/internal/orchestrator/workflow_queue_watchdog_test.go
index ad20f8ff28..57928e9121 100644
--- a/apps/backend/internal/orchestrator/workflow_queue_watchdog_test.go
+++ b/apps/backend/internal/orchestrator/workflow_queue_watchdog_test.go
@@ -276,6 +276,135 @@ func TestWorkflowQueueWatchdog_DropsTerminalSessionQueue(t *testing.T) {
 	}
 }
 
+// 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(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.
+	if got := launchCount.Load(); got < 1 {
+		t.Errorf("expected at least 1 LaunchAgent call from the dedupe sweep, got %d", 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) {

From 5f5d40d44968bee74b650aec8f080d98a633eb36 Mon Sep 17 00:00:00 2001
From: Kandev Agent <agent@kandev.dev>
Date: Mon, 1 Jun 2026 20:54:24 +0100
Subject: [PATCH 5/5] fix(orchestrator): address watchdog data race, goroutine
 drain, missing index

- Fix data race on s.workflowQueueWatchdog: assign under s.mu in Start(),
  read/nil under s.mu in Stop() (cubic P1)
- Add sync.WaitGroup to workflowQueueWatchdog; Stop() drains in-flight
  recovery goroutines via wg.Wait() after doneCh closes (greptile P2)
- Add atomic.Bool started guard so Stop() is safe if Start() never called
- Remove context.WithoutCancel so recovery goroutines are cancellable on shutdown
- Add composite index idx_queued_messages_queued_by_at for ListStaleByQueuedBy
- Gate SetQueuedAtForTesting behind testing.TB to prevent non-test call
- Tighten dedupe test: assert LaunchAgent count is both >= 1 and <= 2

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
---
 .../orchestrator/event_handlers_workflow.go      |  9 +++++++--
 .../messagequeue/repository_sqlite.go            |  1 +
 .../messagequeue/service_test_helpers.go         |  3 ++-
 apps/backend/internal/orchestrator/service.go    | 16 +++++++++++-----
 .../orchestrator/workflow_queue_watchdog.go      | 11 +++++++++--
 .../orchestrator/workflow_queue_watchdog_test.go | 10 +++++++---
 6 files changed, 37 insertions(+), 13 deletions(-)

diff --git a/apps/backend/internal/orchestrator/event_handlers_workflow.go b/apps/backend/internal/orchestrator/event_handlers_workflow.go
index d332c66beb..dac83d300b 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"
@@ -1554,7 +1555,7 @@ func (s *Service) scheduleAutoResumeForWorkflowQueue(ctx context.Context, sessio
 //
 // 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) {
+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
@@ -1607,7 +1608,11 @@ func (s *Service) maybeRecoverOrphanedWorkflowQueue(ctx context.Context, session
 	// 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".
-	go s.tryEnsureExecution(context.WithoutCancel(ctx), sessionID)
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		s.tryEnsureExecution(ctx, sessionID)
+	}()
 }
 
 // dropOrphanedWorkflowQueue removes every workflow-tagged queued entry for
diff --git a/apps/backend/internal/orchestrator/messagequeue/repository_sqlite.go b/apps/backend/internal/orchestrator/messagequeue/repository_sqlite.go
index 7249f2dbd2..79ba8cc27c 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,
diff --git a/apps/backend/internal/orchestrator/messagequeue/service_test_helpers.go b/apps/backend/internal/orchestrator/messagequeue/service_test_helpers.go
index f21a00f33d..5d940f746f 100644
--- a/apps/backend/internal/orchestrator/messagequeue/service_test_helpers.go
+++ b/apps/backend/internal/orchestrator/messagequeue/service_test_helpers.go
@@ -1,6 +1,7 @@
 package messagequeue
 
 import (
+	"testing"
 	"time"
 )
 
@@ -11,7 +12,7 @@ import (
 //
 // 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(sessionID string, at time.Time) {
+func (s *Service) SetQueuedAtForTesting(_ testing.TB, sessionID string, at time.Time) {
 	mem, ok := s.repo.(*memoryRepository)
 	if !ok {
 		return
diff --git a/apps/backend/internal/orchestrator/service.go b/apps/backend/internal/orchestrator/service.go
index 93c019f644..8d390a45dc 100644
--- a/apps/backend/internal/orchestrator/service.go
+++ b/apps/backend/internal/orchestrator/service.go
@@ -845,8 +845,11 @@ func (s *Service) Start(ctx context.Context) error {
 	// 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 {
-		s.workflowQueueWatchdog = s.newWorkflowQueueWatchdog()
-		s.workflowQueueWatchdog.Start(ctx)
+		wd := s.newWorkflowQueueWatchdog()
+		wd.Start(ctx)
+		s.mu.Lock()
+		s.workflowQueueWatchdog = wd
+		s.mu.Unlock()
 	}
 
 	s.logger.Info("orchestrator service started successfully")
@@ -878,9 +881,12 @@ func (s *Service) Stop() error {
 		errs = append(errs, err)
 	}
 
-	if s.workflowQueueWatchdog != nil {
-		s.workflowQueueWatchdog.Stop()
-		s.workflowQueueWatchdog = nil
+	s.mu.Lock()
+	wd := s.workflowQueueWatchdog
+	s.workflowQueueWatchdog = nil
+	s.mu.Unlock()
+	if wd != nil {
+		wd.Stop()
 	}
 
 	s.cancelAllClarificationWatchdogs()
diff --git a/apps/backend/internal/orchestrator/workflow_queue_watchdog.go b/apps/backend/internal/orchestrator/workflow_queue_watchdog.go
index 6f9ba8c8df..9bb43b68d4 100644
--- a/apps/backend/internal/orchestrator/workflow_queue_watchdog.go
+++ b/apps/backend/internal/orchestrator/workflow_queue_watchdog.go
@@ -5,6 +5,7 @@ import (
 	"os"
 	"strconv"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"go.uber.org/zap"
@@ -40,6 +41,8 @@ type workflowQueueWatchdog struct {
 	stopCh   chan struct{}
 	doneCh   chan struct{}
 	stopOnce sync.Once
+	started  atomic.Bool
+	wg       sync.WaitGroup
 }
 
 // newWorkflowQueueWatchdog constructs a watchdog tied to the orchestrator
@@ -60,6 +63,7 @@ func (s *Service) newWorkflowQueueWatchdog() *workflowQueueWatchdog {
 // 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)
 }
 
@@ -69,7 +73,10 @@ func (w *workflowQueueWatchdog) Stop() {
 	w.stopOnce.Do(func() {
 		close(w.stopCh)
 	})
-	<-w.doneCh
+	if w.started.Load() {
+		<-w.doneCh
+		w.wg.Wait()
+	}
 }
 
 func (w *workflowQueueWatchdog) run(ctx context.Context) {
@@ -113,7 +120,7 @@ func (w *workflowQueueWatchdog) sweep(ctx context.Context) {
 			continue
 		}
 		seen[entry.SessionID] = true
-		w.svc.maybeRecoverOrphanedWorkflowQueue(ctx, entry.SessionID, entry.QueuedAt)
+		w.svc.maybeRecoverOrphanedWorkflowQueue(ctx, entry.SessionID, entry.QueuedAt, &w.wg)
 	}
 }
 
diff --git a/apps/backend/internal/orchestrator/workflow_queue_watchdog_test.go b/apps/backend/internal/orchestrator/workflow_queue_watchdog_test.go
index 57928e9121..de590b5051 100644
--- a/apps/backend/internal/orchestrator/workflow_queue_watchdog_test.go
+++ b/apps/backend/internal/orchestrator/workflow_queue_watchdog_test.go
@@ -101,7 +101,7 @@ func seedWorkflowQueueWatchdogTaskAndSession(
 	// 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(sessionID, now.Add(-10*time.Minute))
+	mq.SetQueuedAtForTesting(t, sessionID, now.Add(-10*time.Minute))
 
 	return svc, agentMgr, mq
 }
@@ -297,7 +297,7 @@ func TestWorkflowQueueWatchdog_DedupesBySession(t *testing.T) {
 	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(sessionID, time.Now().Add(-10*time.Minute))
+	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)
 	}
@@ -363,9 +363,13 @@ func TestWorkflowQueueWatchdog_DedupesBySession(t *testing.T) {
 	// 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.
-	if got := launchCount.Load(); got < 1 {
+	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