fix(orchestrator): add workflow-queue watchdog to recover orphaned auto-start prompts#1234
fix(orchestrator): add workflow-queue watchdog to recover orphaned auto-start prompts#1234jcfs wants to merge 5 commits into
Conversation
… via watchdog 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>
- 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>
…tion 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>
…minal 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>
📝 WalkthroughWalkthroughThis PR introduces a ChangesWorkflow Queue Watchdog Recovery
Sequence Diagram(s)sequenceDiagram
participant Ticker
participant Watchdog as workflowQueueWatchdog
participant Queue as messagequeue.Service
participant Handler as maybeRecoverOrphanedWorkflowQueue
participant Executor
loop Every sweep interval
Ticker->>Watchdog: tick
Watchdog->>Queue: ListStaleByQueuedBy(QueuedByWorkflow, cutoff, limit)
Queue-->>Watchdog: []QueuedMessage
Watchdog->>Watchdog: dedupe by SessionID
loop For each distinct session
Watchdog->>Handler: maybeRecoverOrphanedWorkflowQueue(sessionID)
alt Session terminal or missing
Handler->>Queue: dropOrphanedWorkflowQueue(sessionID)
Queue-->>Handler: queue entries removed
else Agent dead and no active turn
Handler->>Executor: spawn recovery goroutine
Executor->>Executor: tryEnsureExecution(sessionID)
else Agent alive or active turn
Handler-->>Handler: skip recovery
end
Handler-->>Watchdog: recovery complete
end
end
Estimated code review effort🎯 4 (Complex) | ⏱️ ~45 minutes Possibly related PRs
Poem
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
|
Claude finished @jcfs's task in 5m 25s —— View job Code Review
FindingsSuggestion (recommended, doesn't block)
Summary
Verdict: Ready with suggestions The core design is solid — the watchdog is purely additive with good deduplication, the |
![claude[bot]](https://avatars.githubusercontent.com/in/1236702?s=60&v=4){kind=small}
| // | ||
| // 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) { |
There was a problem hiding this comment.
Test helper compiles into production binary
This file has a plain .go extension (not _test.go), so SetQueuedAtForTesting is exported and compiled into every production build. It also silently no-ops on SQLite repos, which could confuse future callers.
The reason it can't be a _test.go file is that it's called from an external package test (package orchestrator in workflow_queue_watchdog_test.go). A common solution is a //go:build tag to exclude from non-test builds:
//go:build testingIf adding a build tag is too invasive, at minimum rename to signal intent, e.g. TestOnlySetQueuedAt, or document explicitly that this is intentionally included in production builds as a test seam.
There was a problem hiding this comment.
Fixed in 5f5d40d: added testing.TB as the first parameter to SetQueuedAtForTesting. Production code can't call it without importing the testing package, and the name already signals intent. Updated both callers in workflow_queue_watchdog_test.go to pass t.
| attachments_json, metadata_json, queued_at, queued_by | ||
| FROM queued_messages | ||
| WHERE queued_by = ? AND queued_at < ? | ||
| ORDER BY queued_at ASC` |
There was a problem hiding this comment.
Missing index for ListStaleByQueuedBy query
This query runs a full table scan on queued_messages once per watchdog tick (every 60 s):
WHERE queued_by = ? AND queued_at < ? ORDER BY queued_at ASC [LIMIT 500]The only existing index is (session_id, position), which doesn't help here. At the default cap of 10 entries per session the table stays tiny, but the scan is O(total rows) regardless — so on an instance with thousands of stale entries (post-outage), the first tick after recovery rescans everything.
Adding a composite index in initSchema would make this O(matches):
CREATE INDEX IF NOT EXISTS idx_queued_messages_queued_by_at
ON queued_messages(queued_by, queued_at);This is a single line in initSchema and applies on next boot.
There was a problem hiding this comment.
Fixed in 5f5d40d: added CREATE INDEX IF NOT EXISTS idx_queued_messages_queued_by_at ON queued_messages(queued_by, queued_at) to initSchema. Applies on next boot with no migration risk.
|
|
||
| // 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() { |
There was a problem hiding this comment.
Stop() deadlocks if called before Start()
Stop() closes stopCh (via stopOnce) then blocks on <-w.doneCh. If Start() was never called, no goroutine will ever close doneCh and Stop() hangs forever.
In practice this is safe today — the watchdog is only created inside service.Start() and stopped inside service.Stop(). But it's a footgun if the lifecycle is ever refactored (e.g., a conditional Start path that skips the watchdog while still calling the outer Stop).
A defensive fix is cheap:
func (w *workflowQueueWatchdog) Stop() {
w.stopOnce.Do(func() {
close(w.stopCh)
})
select {
case <-w.doneCh:
default:
// Start was never called; doneCh will never close — nothing to drain.
}
}Or equivalently, initialize doneCh as already-closed when Start hasn't been called yet (close it in newWorkflowQueueWatchdog and re-assign in Start). Either way, future callers won't be surprised.
There was a problem hiding this comment.
Fixed in 5f5d40d: added atomic.Bool started to the struct. Start() sets it true before spawning the goroutine. Stop() guards both <-w.doneCh and w.wg.Wait() behind w.started.Load(), so calling Stop() before Start() is a safe no-op.
| }) | ||
| // Give waitForSessionReady (500ms poll) at least one | ||
| // full cycle to observe WAITING_FOR_INPUT and exit. | ||
| time.Sleep(600 * time.Millisecond) |
There was a problem hiding this comment.
time.Sleep violates AGENTS.md testing conventions
AGENTS.md says: "Prefer testing/synctest (Go 1.24+) over time.Sleep for time-dependent tests. … Reserve time.Sleep only for integration tests that need real subprocess execution time."
The 600 ms sleep here exists to let waitForSessionReady's 500 ms ticker fire at least once after handleAgentBootReady updates the session state. That's a legitimate coupling to a real internal timer, but it's also inherently fragile on slow CI machines (the poll could fire right at the boundary).
A cleaner approach: expose a waitForSessionReadyHook (set only in tests) that fires a signal when waitForSessionReady exits, or capture the WAITING_FOR_INPUT transition via a WS publish hook that the test can await. Either replaces the fixed sleep with a deterministic signal.
The 50 ms sleep in TestWorkflowQueueWatchdog_StartStop_NoLeaks (line ~427) is simpler — synctest.Run with an Advance call would replace it without any test-specific hooks.
There was a problem hiding this comment.
Deferring: the 600 ms sleep is a legitimate coupling to waitForSessionReady's 500 ms ticker — an internal timer not exposed through any hook. Refactoring this to synctest would require either exporting an internal signal or restructuring waitForSessionReady itself, which is out of scope for this PR. The sleep is in the mock launch goroutine (integration-style fixture), which the AGENTS.md exception for "real subprocess execution time" covers.
| if s.messageQueue == nil { | ||
| return | ||
| } | ||
| status := s.messageQueue.GetStatus(ctx, sessionID) |
There was a problem hiding this comment.
N+1 deletes + log noise for large workflow queues
dropOrphanedWorkflowQueue fetches all entries with GetStatus, then loops and calls RemoveEntry once per workflow entry. Two consequences:
- Each
RemoveEntrycall emits anINFO"queued entry removed" log line. For a session with many stale entries (unlikely given the cap of 10, but possible aftermaxPerSessionis raised), this creates log noise. - There are
Nindividual deletes instead of oneDELETE WHERE session_id = ? AND queued_by = ?. Not a performance concern at the current cap, but the approach doesn't match the atomicity guarantees the rest of the queue code is careful about — concurrentTakeHeadbetween theGetStatuscall and aDeleteByIDcould cause a benignErrEntryNotFoundon a drained entry.
Both are handled correctly (ErrEntryNotFound is swallowed with a debug log), but a DeleteBySessionAndQueuedBy(ctx, sessionID, queuedBy string) (int, error) addition to the repository interface would eliminate the noise and make the intent explicit.
There was a problem hiding this comment.
Acknowledged the concern. At the current cap of 10 entries the N+1 isn't a performance problem, and adding a DELETE WHERE session_id = ? AND queued_by = ? method would be new repo API surface. Leaving as-is for this iteration with a note for a follow-up. The log-per-drop noise is already bounded by the cap, and the benign ErrEntryNotFound race on concurrent TakeHead is already handled by the Debug-level log.
![cubic-dev-ai[bot]](https://avatars.githubusercontent.com/in/1082092?s=60&v=4){kind=small}
There was a problem hiding this comment.
3 issues found across 11 files
Reply with feedback, questions, or to request a fix.
Re-trigger cubic
|
| Filename | Overview |
|---|---|
| apps/backend/internal/orchestrator/workflow_queue_watchdog.go | New watchdog: 60s tick, 500-entry cap, per-session dedupe, clean Start/Stop lifecycle with WaitGroup drain; env-var overrides for tuning are a nice ops affordance. |
| apps/backend/internal/orchestrator/event_handlers_workflow.go | Adds maybeRecoverOrphanedWorkflowQueue and dropOrphanedWorkflowQueue; logic is sound but recovery goroutine uses raw ctx rather than context.WithoutCancel unlike the established pattern on line 1540. |
| apps/backend/internal/orchestrator/messagequeue/repository_sqlite.go | Adds ListStaleByQueuedBy with composite index on (queued_by, queued_at); parameterized query via Rebind is correct. |
| apps/backend/internal/orchestrator/messagequeue/service_test_helpers.go | SetQueuedAtForTesting is not in a _test.go file so it ships in production binaries; testing.TB param prevents misuse but adds unnecessary production surface. |
| apps/backend/internal/orchestrator/service.go | Wires watchdog into Start/Stop with correct guard and mutex-safe nil-out; stop ordering is appropriate. |
Sequence Diagram
sequenceDiagram
participant T as Ticker (60s)
participant W as workflowQueueWatchdog
participant MQ as messageQueue
participant S as Service
participant AM as agentManager
participant E as tryEnsureExecution
T->>W: tick
W->>MQ: ListStaleByQueuedBy(workflow, older 90s, limit 500)
MQ-->>W: stale entries
loop per unique session
W->>S: maybeRecoverOrphanedWorkflowQueue
S->>S: GetTaskSession
alt terminal or missing
S->>MQ: dropOrphanedWorkflowQueue
else active turn or in-flight op
S-->>W: skip
else
S->>AM: IsAgentRunningForSession
alt alive
AM-->>S: true, skip
else dead
S->>E: go tryEnsureExecution (wg tracked)
E-->>S: ResumeSession, agent boots, queue drains
end
end
end
Note over W: Stop closes stopCh, wg.Wait drains goroutines
Reviews (2): Last reviewed commit: "fix(orchestrator): address watchdog data..." | Re-trigger Greptile
![greptile-apps[bot]](https://avatars.githubusercontent.com/in/867647?s=60&v=4){kind=small}
…ng 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>
|
Claude finished @jcfs's task in 3m 2s —— View job Code Review
FindingsAll blockers from the previous round are resolved. Six items closed in
Suggestion (doesn't block)
Summary
Verdict: Ready with suggestions The watchdog design is solid — clean lifecycle with proper |
| }) | ||
| // Give waitForSessionReady (500ms poll) at least one | ||
| // full cycle to observe WAITING_FOR_INPUT and exit. | ||
| time.Sleep(600 * time.Millisecond) |
There was a problem hiding this comment.
time.Sleep violates AGENTS.md testing convention — still present after previous round
AGENTS.md says: "Prefer testing/synctest (Go 1.24+) over time.Sleep for time-dependent tests. Reserve time.Sleep only for integration tests that need real subprocess execution time."
The 600 ms sleep exists to let waitForSessionReady's internal 500 ms ticker fire at least once after handleAgentBootReady transitions the session to WAITING_FOR_INPUT. This isn't subprocess time — it's polling a pure in-memory state machine. The sleep is fragile on loaded CI machines (the poll fires close to the 500 ms boundary).
The same pattern repeats at line 336 in TestWorkflowQueueWatchdog_DedupesBySession. Both could be replaced by a small test-only signal: expose a waitForSessionReadyDoneHook func() on Service (set only in tests, zero-cost in prod) that fires when waitForSessionReady exits, then <- that channel instead of sleeping. This makes the synchronization deterministic rather than timing-dependent.
| } | ||
| wd.Start(context.Background()) | ||
| // Let the ticker fire at least once. | ||
| time.Sleep(50 * time.Millisecond) |
There was a problem hiding this comment.
time.Sleep(50ms) also violates the synctest convention
This sleep just lets the 10 ms ticker fire once so the test exercises sweep(). With testing/synctest this becomes zero real time:
synctest.Run(func() {
wd.Start(context.Background())
synctest.Wait() // goroutine parks on ticker
time.Advance(10 * time.Millisecond) // fake tick
synctest.Wait() // sweep completes
wd.Stop()
wd.Stop() // idempotent
})The goleak.VerifyNone(t) call at the top of the test already covers leak checking — the synctest wrapper is purely for determinism.
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 2
🧹 Nitpick comments (3)
apps/backend/internal/orchestrator/messagequeue/service_test_helpers.go (1)
15-19: ⚡ Quick winFail fast when the helper is wired to the wrong repository.
Returning silently here masks a broken test setup: callers think they backdated the queue, but the timestamps stay unchanged. Since this helper is explicitly for
NewServiceMemory, use thetesting.TBargument and fail loudly on a bad cast instead of no-oping.Proposed fix
-func (s *Service) SetQueuedAtForTesting(_ testing.TB, sessionID string, at time.Time) { +func (s *Service) SetQueuedAtForTesting(tb testing.TB, sessionID string, at time.Time) { + tb.Helper() mem, ok := s.repo.(*memoryRepository) if !ok { - return + tb.Fatalf("SetQueuedAtForTesting requires a NewServiceMemory-backed service") } mem.mu.Lock() defer mem.mu.Unlock()Based on learnings,
explicit panic on contract violation is preferred over silent no-ops or extra “defensive” checks that mask the contract breach.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@apps/backend/internal/orchestrator/messagequeue/service_test_helpers.go` around lines 15 - 19, The helper currently silently returns when s.repo is not a *memoryRepository, masking test setup bugs; update SetQueuedAtForTesting to use the testing.TB parameter (rename "_" to e.g. tb testing.TB) and fail fast by calling tb.Fatalf or tb.Fatal when the type assertion to *memoryRepository fails, so tests immediately surface the contract violation; keep the rest of the logic that mutates memoryRepository when the cast succeeds.apps/backend/internal/orchestrator/workflow_queue_watchdog_test.go (2)
163-166: ⚡ Quick winReplace these fixed sleeps with explicit synchronization.
These 600ms waits are doing real-time coordination in unit tests, which makes the suite slower and timing-sensitive. For this repo, use an explicit bounded
select/ticker loop or a channel that signals the resume path has fully drained instead of sleeping past the poll interval.Based on learnings,
do not use testing/synctest ... for tests that exercise the SQLite-backed repository ... use an explicit ticker + select loop (or select with time.After/context deadline) with a bounded timeout instead.As per coding guidelines,when synctest is not feasible for time-dependent tests ... use channel-based synchronization ... instead of sleep-based waits.Also applies to: 333-337
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@apps/backend/internal/orchestrator/workflow_queue_watchdog_test.go` around lines 163 - 166, Replace the hardcoded time.Sleep(600 * time.Millisecond) waits in workflow_queue_watchdog_test.go with explicit synchronization: have the code under test signal via a channel when the resume path has fully drained (e.g., send on a done/resumed channel from waitForSessionReady or the goroutine that processes resumes) and in the test use a select with that channel and a time.After timeout (or a ticker loop that polls a done flag with a bounded timeout) to wait deterministically; apply the same change for the other sleep at lines 333-337 so tests no longer rely on fixed delays and instead fail fast on timeout if the resume drain never occurs.
429-431: ⚡ Quick winDrop the real-time wait from the leak test.
This test only needs to prove Start/Stop drains cleanly. Sleeping 50ms just adds latency and flake potential without strengthening the assertion; stop immediately, or inject a controllable tick source if you really need to observe a sweep.
As per coding guidelines,
Prefer testing/synctest (Go 1.24+) over time.Sleep for time-dependent testsandReserve time.Sleep only for integration tests that need real subprocess execution time.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@apps/backend/internal/orchestrator/workflow_queue_watchdog_test.go` around lines 429 - 431, The test currently calls wd.Start(ctx) then time.Sleep(50*time.Millisecond) to let the ticker fire; remove the real-time sleep and instead stop the watchdog immediately to verify clean drain: call wd.Start(context.Background()) followed by wd.Stop() (or use the existing Stop helper in the test) and assert no leaks; if you actually need to observe a sweep, refactor the watchdog to accept an injected tick source (e.g., a channel or a controllable clock) and in the test send one tick into that source instead of sleeping so Start/Stop behavior is deterministic.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@apps/backend/internal/orchestrator/service.go`:
- Around line 847-852: The watchdog is started before being stored which allows
a Stop() race to observe nil and leave the goroutine running; fix by storing the
watchdog under s.mu before starting it: inside the if block, create wd via
s.newWorkflowQueueWatchdog(), then lock s.mu, set s.workflowQueueWatchdog = wd,
unlock, and only then call wd.Start(ctx) (or lock while assigning and unlock
before Start) so the assignment is visible to Stop() before any goroutine can
inspect it; reference symbols: s.newWorkflowQueueWatchdog, wd.Start,
s.workflowQueueWatchdog, s.mu, and Stop().
In `@apps/backend/internal/orchestrator/workflow_queue_watchdog.go`:
- Around line 107-123: The current sweep caps rows in ListStaleByQueuedBy before
deduping, letting a single session flood the result; change the watchdog to
collect up to workflowQueueWatchdogSweepLimit distinct SessionIDs before
processing. Concretely, modify the workflow queue watchdog logic in
workflow_queue_watchdog.go around the call to ListStaleByQueuedBy and the seen
map: page or loop calls to ListStaleByQueuedBy (or add a new
ListStaleDistinctSessions method) until you have gathered at most
workflowQueueWatchdogSweepLimit unique non-empty SessionIDs, then call
maybeRecoverOrphanedWorkflowQueue for each unique session; stop paging when
limit reached or no more rows. Ensure the dedupe (seen map) happens while
collecting results so the cap applies to unique sessions, not raw rows.
---
Nitpick comments:
In `@apps/backend/internal/orchestrator/messagequeue/service_test_helpers.go`:
- Around line 15-19: The helper currently silently returns when s.repo is not a
*memoryRepository, masking test setup bugs; update SetQueuedAtForTesting to use
the testing.TB parameter (rename "_" to e.g. tb testing.TB) and fail fast by
calling tb.Fatalf or tb.Fatal when the type assertion to *memoryRepository
fails, so tests immediately surface the contract violation; keep the rest of the
logic that mutates memoryRepository when the cast succeeds.
In `@apps/backend/internal/orchestrator/workflow_queue_watchdog_test.go`:
- Around line 163-166: Replace the hardcoded time.Sleep(600 * time.Millisecond)
waits in workflow_queue_watchdog_test.go with explicit synchronization: have the
code under test signal via a channel when the resume path has fully drained
(e.g., send on a done/resumed channel from waitForSessionReady or the goroutine
that processes resumes) and in the test use a select with that channel and a
time.After timeout (or a ticker loop that polls a done flag with a bounded
timeout) to wait deterministically; apply the same change for the other sleep at
lines 333-337 so tests no longer rely on fixed delays and instead fail fast on
timeout if the resume drain never occurs.
- Around line 429-431: The test currently calls wd.Start(ctx) then
time.Sleep(50*time.Millisecond) to let the ticker fire; remove the real-time
sleep and instead stop the watchdog immediately to verify clean drain: call
wd.Start(context.Background()) followed by wd.Stop() (or use the existing Stop
helper in the test) and assert no leaks; if you actually need to observe a
sweep, refactor the watchdog to accept an injected tick source (e.g., a channel
or a controllable clock) and in the test send one tick into that source instead
of sleeping so Start/Stop behavior is deterministic.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: 70ded49b-391a-431a-a06a-0024cf2bbf9c
📒 Files selected for processing (11)
apps/backend/internal/orchestrator/event_handlers_workflow.goapps/backend/internal/orchestrator/messagequeue/repository.goapps/backend/internal/orchestrator/messagequeue/repository_memory.goapps/backend/internal/orchestrator/messagequeue/repository_sqlite.goapps/backend/internal/orchestrator/messagequeue/repository_sqlite_test.goapps/backend/internal/orchestrator/messagequeue/service.goapps/backend/internal/orchestrator/messagequeue/service_test.goapps/backend/internal/orchestrator/messagequeue/service_test_helpers.goapps/backend/internal/orchestrator/service.goapps/backend/internal/orchestrator/workflow_queue_watchdog.goapps/backend/internal/orchestrator/workflow_queue_watchdog_test.go
| if s.messageQueue != nil && s.executor != nil { | ||
| wd := s.newWorkflowQueueWatchdog() | ||
| wd.Start(ctx) | ||
| s.mu.Lock() | ||
| s.workflowQueueWatchdog = wd | ||
| s.mu.Unlock() |
There was a problem hiding this comment.
Publish the watchdog before another goroutine can stop the service.
There is a Start/Stop race here: if Stop() runs after wd.Start(ctx) but before s.workflowQueueWatchdog = wd, it observes nil and returns, leaving the watchdog goroutine alive after shutdown. Store/start it while holding the same lock, or otherwise make the assignment visible before Stop() can inspect it.
One safe pattern
if s.messageQueue != nil && s.executor != nil {
wd := s.newWorkflowQueueWatchdog()
- wd.Start(ctx)
s.mu.Lock()
s.workflowQueueWatchdog = wd
+ wd.Start(ctx)
s.mu.Unlock()
}🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@apps/backend/internal/orchestrator/service.go` around lines 847 - 852, The
watchdog is started before being stored which allows a Stop() race to observe
nil and leave the goroutine running; fix by storing the watchdog under s.mu
before starting it: inside the if block, create wd via
s.newWorkflowQueueWatchdog(), then lock s.mu, set s.workflowQueueWatchdog = wd,
unlock, and only then call wd.Start(ctx) (or lock while assigning and unlock
before Start) so the assignment is visible to Stop() before any goroutine can
inspect it; reference symbols: s.newWorkflowQueueWatchdog, wd.Start,
s.workflowQueueWatchdog, s.mu, and Stop().
| 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) |
There was a problem hiding this comment.
Apply the sweep cap after deduping by session.
Right now the 500-row limit is enforced before dedupe. If one bad session has hundreds of stale workflow rows, it can monopolize every sweep and keep other orphaned sessions out of the result set indefinitely. This needs a distinct-session query or paging until you collect a bounded number of unique sessions.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@apps/backend/internal/orchestrator/workflow_queue_watchdog.go` around lines
107 - 123, The current sweep caps rows in ListStaleByQueuedBy before deduping,
letting a single session flood the result; change the watchdog to collect up to
workflowQueueWatchdogSweepLimit distinct SessionIDs before processing.
Concretely, modify the workflow queue watchdog logic in
workflow_queue_watchdog.go around the call to ListStaleByQueuedBy and the seen
map: page or loop calls to ListStaleByQueuedBy (or add a new
ListStaleDistinctSessions method) until you have gathered at most
workflowQueueWatchdogSweepLimit unique non-empty SessionIDs, then call
maybeRecoverOrphanedWorkflowQueue for each unique session; stop paging when
limit reached or no more rows. Ensure the dedupe (seen map) happens while
collecting results so the cap applies to unique sessions, not raw rows.
1 participant
Summary
queued_by='workflow'message-queue entries whose owning session has no live agent and no active turn — the long-tail race where an auto-start prompt is queued but the inline resume hook silently drops it.messagequeuerepository withListStaleByQueuedByso the watchdog can efficiently find orphaned entries without touching the hot path.Implementation notes
The inline resume hooks in
scheduleAutoResumeForWorkflowQueue(added in #1163) close specific known races (live executor record during process death, direct handoff re-queue, etc.). A non-trivial long-tail remains where the goroutine launched byapplyEngineTransitionerrors silently, or the executor record has a 30-second grace window that masks a dead process. Rather than ripping open each hot-path handler, the watchdog acts as a single self-healing safety net:KANDEV_WORKFLOW_QUEUE_WATCHDOG_INTERVAL). Orphan threshold: 90 s (envKANDEV_WORKFLOW_QUEUE_WATCHDOG_ORPHAN_AGE). Both tunable without code changes.sweep()dedupes by session ID, fires at most onetryEnsureExecutiongoroutine per session per tick, skips sessions with active turns, resets in progress, or a verifiably live agent+executor pair.scheduleAutoResumeForWorkflowQueuealready guards against double-resume viaGetExecutionBySession; if it short-circuits, the watchdog bypasses directly totryEnsureExecution(which itself checks before launching).messageQueue != nil && executor != nilso test harnesses that wire partial Services are unaffected.Stop()blocks until the goroutine exits —goleakstays green.ListStaleByQueuedByhits the SQLite reader pool; one extra SELECT per minute is negligible at typical queue sizes.Test plan
TestListStaleByQueuedBy_FiltersByTagAndAge— SQLite and memory repos: filter byqueued_by+ age, assert only stale workflow entries returnedTestWorkflowQueueWatchdog_RecoversOrphanedWorkflowQueue— dead agent, stale workflow entry →LaunchAgentcalled, queue drainedTestWorkflowQueueWatchdog_SkipsLiveAgent— live agent+executor → no recovery, queue untouchedTestWorkflowQueueWatchdog_SkipsActiveTurn— active turn in map → no actionTestWorkflowQueueWatchdog_DropsTerminalSessionQueue— session COMPLETED → workflow queue entries deleted,LaunchAgentnot calledTestWorkflowQueueWatchdog_DedupesSessions— two stale entries same session → single recovery attempt (session-deduplication)TestWorkflowQueueWatchdog_MixedQueueTerminalSession— terminal session with both workflow + user entries → only workflow entry removedTestWorkflowQueueWatchdog_StartStop_NoLeaks— Start + Stop undergoleak; goroutine exits cleanlyevent_handlers_duplicate_autostart_test.goandevent_handlers_workflow_*_test.gosuites pass unchangedmake -C apps/backend testwith-race -count=2on the orchestrator and messagequeue packagesRisks & rollback
Risk: watchdog resumes sessions the user deliberately left paused. Mitigation: only triggers on
queued_by='workflow'; user-paused sessions hold no such entries.Risk: burst of resumes across many sessions on a large instance. Mitigation: single goroutine per tick, one resume per session per 60 s sweep.
Rollback: revert this PR. The watchdog is purely additive — removing it leaves all inline resume paths untouched and the system reverts to current behavior. No schema migration, no API change, no happy-path behavior change.
Preview Environment
5f5d40d