fix(orchestrator): add workflow-queue watchdog to recover orphaned auto-start prompts

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

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)
 

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()

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)

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) {

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)

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()

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
+	}
+}