DOS-825: reclaim wedged todo dispatches - #1
Conversation
|
Caution Review failedAn error occurred during the review process. Please try again later. WalkthroughTodo dispatch recovery now selects eligible todo issues, revalidates them, and re-enqueues failed work during runtime sweeper ticks. Integration tests cover agent, squad, offline-runtime, and WIP-cap paths. Task availability notification tolerates a missing cache, and the views TypeScript configuration defines its project root. ChangesTodo dispatch recovery
Views build configuration
Estimated code review effort: 4 (Complex) | ~45 minutes Sequence Diagram(s)sequenceDiagram
participant RuntimeSweeper
participant CandidateQuery
participant TaskService
participant IssueDatabase
RuntimeSweeper->>CandidateQuery: list capped eligible todo dispatches
CandidateQuery-->>RuntimeSweeper: return reclaim candidates
RuntimeSweeper->>TaskService: RecoverTodoDispatch(issueID)
TaskService->>IssueDatabase: lock and revalidate candidate
IssueDatabase-->>TaskService: return eligible route
TaskService-->>RuntimeSweeper: enqueue fresh task and return route result
Suggested reviewers: Poem
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches🧪 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 |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 3
🤖 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 `@server/cmd/server/runtime_sweeper_test.go`:
- Around line 755-800: Update TestTodoDispatchReclaimSkipsOfflineRuntime and the
related WIP-cap test to select an active agent with an online runtime instead of
relying on findIntegrationAgentRuntime’s fixture. Explicitly set
max_concurrent_tasks to the intended deterministic value for the WIP scenario,
and capture and restore both the agent cap and runtime’s original status in
cleanup. Apply the same fixture setup and restoration to the additional test
range referenced by the review.
In `@server/cmd/server/runtime_sweeper.go`:
- Around line 328-364: Make the candidate validation and recovery flow atomic
around the dispatch loop using a transaction and an issue-level lock. Before
enqueueing, revalidate the issue’s current assignment, target runtime
availability, WIP capacity, and active-task state under that lock, then perform
a recovery-specific enqueue that does not cancel newly created tasks; replace
the unconditional RerunIssue call while preserving the existing skip and failure
accounting.
In `@server/pkg/db/queries/agent.sql`:
- Around line 575-591: The candidate query must apply eligibility filtering
before the batch limit so ineligible candidates cannot consume the processing
cap and starve later reclaimable issues. Update the query around the candidates
selection, joins, and ORDER BY/LIMIT to exclude archived agents, runtime-less or
offline runtimes, and agents at the work-in-progress task cap before applying
LIMIT `@max_per_tick`.
🪄 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: ASSERTIVE
Plan: Pro
Run ID: 40017f53-a62d-49c4-b340-65eecc291a30
⛔ Files ignored due to path filters (1)
server/pkg/db/generated/agent.sql.gois excluded by!**/generated/**
📒 Files selected for processing (4)
server/cmd/server/runtime_sweeper.goserver/cmd/server/runtime_sweeper_test.goserver/internal/service/task.goserver/pkg/db/queries/agent.sql
There was a problem hiding this comment.
Actionable comments posted: 2
🤖 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 `@server/internal/service/task.go`:
- Around line 1742-1750: Remove the explicit captureTaskQueued call from the
recovery enqueue flow near the task dispatch logging. Keep
NotifyTaskEnqueued(ctx, task) as the single path recording the queued event,
while preserving the slog logging and task event broadcast.
In `@server/pkg/db/queries/agent.sql`:
- Around line 651-659: Update the agent selection/enqueue flow around the
target_running_tasks capacity check to serialize per-agent work-in-progress
decisions: lock the selected target agent before evaluating capacity, then
recount its queued tasks in a separate statement immediately before insertion.
Ensure concurrent recoveries targeting the same agent cannot both pass the cap
check and enqueue beyond max_concurrent_tasks.
🪄 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: ASSERTIVE
Plan: Pro
Run ID: c36914c9-4217-4b6a-89c0-de469568a527
⛔ Files ignored due to path filters (1)
server/pkg/db/generated/agent.sql.gois excluded by!**/generated/**
📒 Files selected for processing (4)
server/cmd/server/runtime_sweeper.goserver/cmd/server/runtime_sweeper_test.goserver/internal/service/task.goserver/pkg/db/queries/agent.sql
| s.captureTaskQueued(ctx, task) | ||
| slog.Info("todo dispatch recovery task enqueued", | ||
| "task_id", util.UUIDToString(task.ID), | ||
| "issue_id", util.UUIDToString(task.IssueID), | ||
| "agent_id", util.UUIDToString(task.AgentID), | ||
| "is_squad_route", isSquadRoute, | ||
| ) | ||
| s.broadcastTaskEvent(ctx, protocol.EventTaskQueued, task) | ||
| s.NotifyTaskEnqueued(ctx, task) |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟡 Minor | ⚡ Quick win
Avoid recording the recovery enqueue twice.
NotifyTaskEnqueued already calls captureTaskQueued, so Line 1742 duplicates the recovery analytics/metrics event.
Proposed fix
- s.captureTaskQueued(ctx, task)
slog.Info("todo dispatch recovery task enqueued",📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| s.captureTaskQueued(ctx, task) | |
| slog.Info("todo dispatch recovery task enqueued", | |
| "task_id", util.UUIDToString(task.ID), | |
| "issue_id", util.UUIDToString(task.IssueID), | |
| "agent_id", util.UUIDToString(task.AgentID), | |
| "is_squad_route", isSquadRoute, | |
| ) | |
| s.broadcastTaskEvent(ctx, protocol.EventTaskQueued, task) | |
| s.NotifyTaskEnqueued(ctx, task) | |
| slog.Info("todo dispatch recovery task enqueued", | |
| "task_id", util.UUIDToString(task.ID), | |
| "issue_id", util.UUIDToString(task.IssueID), | |
| "agent_id", util.UUIDToString(task.AgentID), | |
| "is_squad_route", isSquadRoute, | |
| ) | |
| s.broadcastTaskEvent(ctx, protocol.EventTaskQueued, task) | |
| s.NotifyTaskEnqueued(ctx, task) |
🤖 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 `@server/internal/service/task.go` around lines 1742 - 1750, Remove the
explicit captureTaskQueued call from the recovery enqueue flow near the task
dispatch logging. Keep NotifyTaskEnqueued(ctx, task) as the single path
recording the queued event, while preserving the slog logging and task event
broadcast.
| CROSS JOIN LATERAL ( | ||
| SELECT count(*)::int AS target_running_tasks | ||
| FROM agent_task_queue running | ||
| WHERE running.agent_id = candidate.target_agent_id | ||
| AND running.status IN ('dispatched', 'running', 'waiting_local_directory') | ||
| ) running | ||
| WHERE a.archived_at IS NULL | ||
| AND ar.status = 'online' | ||
| AND running.target_running_tasks < a.max_concurrent_tasks; |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | 🏗️ Heavy lift
Serialize the agent-level WIP check before enqueueing.
The issue lock does not serialize recoveries for different issues targeting the same agent. Two transactions can both observe target_running_tasks < max_concurrent_tasks and enqueue, exceeding the cap. Lock the target agent first, then recount capacity in a subsequent statement before insertion.
🤖 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 `@server/pkg/db/queries/agent.sql` around lines 651 - 659, Update the agent
selection/enqueue flow around the target_running_tasks capacity check to
serialize per-agent work-in-progress decisions: lock the selected target agent
before evaluating capacity, then recount its queued tasks in a separate
statement immediately before insertion. Ensure concurrent recoveries targeting
the same agent cannot both pass the cap check and enqueue beyond
max_concurrent_tasks.
There was a problem hiding this comment.
Actionable comments posted: 1
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
server/cmd/server/runtime_sweeper_test.go (1)
895-903: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winAdd regression coverage for the cooldown and breaker.
Parameterize the failure age/count, then verify that a recent failure is not reclaimed and five failures transition the issue to
blockedwithout enqueueing. These are the new retry-control branches but are currently untested.🤖 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 `@server/cmd/server/runtime_sweeper_test.go` around lines 895 - 903, Extend insertFailedTask to accept configurable failure age and use it in the INSERT instead of the fixed 16-minute interval. Add regression tests covering that a recent failure is not reclaimed, and that five failures transition the issue to blocked without enqueueing another task, exercising the cooldown and breaker paths.
🤖 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 `@server/cmd/server/runtime_sweeper.go`:
- Around line 343-355: The breaker decision and status update around
RecoverTodoDispatch must run under the issue lock, using fresh issue/task state
rather than the stale FailedTasksCount snapshot. Move the attempt check and
blocked transition into the locked recovery transaction; record breaker_tripped
only after a successful status update, and increment Failed when that update
fails.
---
Outside diff comments:
In `@server/cmd/server/runtime_sweeper_test.go`:
- Around line 895-903: Extend insertFailedTask to accept configurable failure
age and use it in the INSERT instead of the fixed 16-minute interval. Add
regression tests covering that a recent failure is not reclaimed, and that five
failures transition the issue to blocked without enqueueing another task,
exercising the cooldown and breaker paths.
🪄 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: ASSERTIVE
Plan: Pro
Run ID: 8b0b168f-97cc-40ee-a095-20634d92e22b
⛔ Files ignored due to path filters (1)
server/pkg/db/generated/agent.sql.gois excluded by!**/generated/**
📒 Files selected for processing (5)
packages/views/tsconfig.jsonserver/cmd/server/runtime_sweeper.goserver/cmd/server/runtime_sweeper_test.goserver/internal/service/task.goserver/pkg/db/queries/agent.sql
| if c.FailedTasksCount >= todoDispatchReclaimMaxAttempts { | ||
| stats.skip("breaker_tripped") | ||
| _, blockErr := queries.UpdateIssueStatus(ctx, db.UpdateIssueStatusParams{ | ||
| ID: issueID, | ||
| Status: "blocked", | ||
| WorkspaceID: c.WorkspaceID, | ||
| }) | ||
| if blockErr != nil { | ||
| slog.Warn("todo dispatch reclaim: failed to block issue", "issue_id", issueKey, "error", blockErr) | ||
| } else { | ||
| slog.Info("todo dispatch reclaim: issue tripped breaker and blocked", "issue_id", issueKey, "failed_tasks", c.FailedTasksCount) | ||
| } | ||
| continue |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | 🏗️ Heavy lift
Make the breaker transition part of the locked recovery transaction.
FailedTasksCount is a stale list snapshot. Before this update, the issue can be completed, reassigned, or receive a new active task, yet this code still overwrites its status with blocked. Move the attempt check and status transition behind the issue lock used by RecoverTodoDispatch. Only record breaker_tripped after a successful update; count update failures in Failed.
🤖 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 `@server/cmd/server/runtime_sweeper.go` around lines 343 - 355, The breaker
decision and status update around RecoverTodoDispatch must run under the issue
lock, using fresh issue/task state rather than the stale FailedTasksCount
snapshot. Move the attempt check and blocked transition into the locked recovery
transaction; record breaker_tripped only after a successful status update, and
increment Failed when that update fails.
dev-qa's blocking review asked for tests covering both the cooldown gate and the per-issue attempt bound; the attempt bound (breaker) was implemented but never actually got test coverage despite an earlier comment claiming it did. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Co-authored-by: multica-agent <github@multica.ai>
1 participant
Closes DOS-825
Summary
Tests
Summary by CodeRabbit