fix(queue): enforce max parallel tasks and auto-refresh UI (#1594)
* fix: queue system - enforce max parallel tasks and auto-refresh UI Fixes two issues with the queue auto-promotion system: - Max parallel tasks setting not being respected (e.g., 3 tasks moved to in_progress when max is 2) - UI not updating automatically after queue promotion (requires manual refresh button press) Changes: 1. Track ALL processed tasks (not just failed ones) to prevent duplicate promotions 2. Mark task as processed BEFORE calling persistTaskStatus to prevent race conditions 3. Count only tasks promoted in this call (promotedInThisCall) against maxParallelTasks 4. Add comprehensive debug logging to diagnose the issue Co-Authored-By: Claude <noreply@anthropic.com> * fix: replace dynamic require with static import for profile-utils The dynamic require('./claude-profile/profile-utils') in migrateCorruptedEmails() doesn't work with Vite's bundling, causing "Cannot find module" errors at runtime. Fixed by adding getEmailFromConfigDir to the static imports at the top of the file and using it directly instead of requiring it dynamically. Co-Authored-By: Claude <noreply@anthropic.com> * fix: use API profile environment variables for task title generation Task title generation was failing when an API profile was active because it only used Claude OAuth profile environment variables (CLAUDE_CODE_OAUTH_TOKEN), not the API profile vars (ANTHROPIC_AUTH_TOKEN, ANTHROPIC_BASE_URL, etc.). This fixes it by: 1. Adding getAPIProfileEnv() to fetch API profile env vars 2. Adding getOAuthModeClearVars() to clear stale ANTHROPIC_* vars when in OAuth mode 3. Including all three env sources in the spawn() call This matches the pattern used in agent-queue.ts for spawning agent processes. Fixes error: "Your account does not have access to Claude Code. Please run /login." when using API profiles. Co-Authored-By: Claude <noreply@anthropic.com> Signed-off-by: Joshua Riley <joshua@joshuariley.co.uk> * fix(queue): correctly enforce max parallel tasks limit Fixed two bugs in the processQueue() function: 1. Capacity check was incorrect - it only checked `promotedInThisCall` instead of `initialInProgress.length + promotedInThisCall`. This meant if there were already tasks in progress, the queue would over-promote. For example, with maxParallelTasks=2 and 1 task already in progress, it would promote 2 more tasks (total 3) instead of 1. 2. UI wasn't refreshing after queue promotion - added onRefresh() call after tasks are promoted to ensure the UI reflects all backend changes. This fixes the issue where users had to manually click refresh. Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com> * fix: clean up diagnostic logging and remove unused variables 1. Remove unused variables `previousStatus` and `statusChanged` in task-store.ts updateTaskStatus - they were never used after declaration 2. Replace console.warn/console.log with debugLog in KanbanBoard.tsx processQueue function (7 locations) - diagnostic logs should be gated behind DEBUG=true to avoid cluttering production console 3. Replace console.warn with debugLog in task-store.ts updateTaskStatus function - consistent with the file's existing use of debugLog 4. Replace console.warn with debugLog in task-store.ts persistTaskStatus function - matches the established logging pattern in the file Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com> --------- Signed-off-by: Joshua Riley <joshua@joshuariley.co.uk> Co-authored-by: Joshua Riley <joshua@joshuariley.co.uk> Co-authored-by: Claude <noreply@anthropic.com>
This commit is contained in:
Andy
+2
-1
@@ -173,4 +173,5 @@ OPUS_ANALYSIS_AND_IDEAS.md
|
||||
.security-key
|
||||
/shared_docs
|
||||
logs/security/
|
||||
Agents.md
|
||||
Agents.md
|
||||
packages/
|
||||
|
||||
@@ -28,6 +28,7 @@ import { TaskCard } from './TaskCard';
|
||||
import { SortableTaskCard } from './SortableTaskCard';
|
||||
import { QueueSettingsModal } from './QueueSettingsModal';
|
||||
import { TASK_STATUS_COLUMNS, TASK_STATUS_LABELS } from '../../shared/constants';
|
||||
import { debugLog } from '../../shared/utils/debug-logger';
|
||||
import { cn } from '../lib/utils';
|
||||
import { persistTaskStatus, forceCompleteTask, archiveTasks, useTaskStore } from '../stores/task-store';
|
||||
import { updateProjectSettings, useProjectStore } from '../stores/project-store';
|
||||
@@ -1010,29 +1011,74 @@ export function KanbanBoard({ tasks, onTaskClick, onNewTaskClick, onRefresh, isR
|
||||
isProcessingQueueRef.current = true;
|
||||
|
||||
try {
|
||||
// Track tasks we've already attempted to promote (to avoid infinite retries)
|
||||
const attemptedTaskIds = new Set<string>();
|
||||
// Track tasks we've already processed in this call to prevent duplicates
|
||||
// This is critical because store updates happen synchronously but we need to ensure
|
||||
// we never process the same task twice, even if there are timing issues
|
||||
const processedTaskIds = new Set<string>();
|
||||
let consecutiveFailures = 0;
|
||||
const MAX_CONSECUTIVE_FAILURES = 10; // Safety limit to prevent infinite loop
|
||||
|
||||
// Track promotions in this call to enforce max parallel tasks limit
|
||||
let promotedInThisCall = 0;
|
||||
|
||||
// Log initial state
|
||||
const initialTasks = useTaskStore.getState().tasks;
|
||||
const initialInProgress = initialTasks.filter((t) => t.status === 'in_progress' && !t.metadata?.archivedAt);
|
||||
const initialQueued = initialTasks.filter((t) => t.status === 'queue' && !t.metadata?.archivedAt);
|
||||
debugLog(`[Queue] === PROCESS QUEUE START ===`, {
|
||||
maxParallelTasks,
|
||||
initialInProgressCount: initialInProgress.length,
|
||||
initialInProgressIds: initialInProgress.map(t => t.id),
|
||||
initialQueuedCount: initialQueued.length,
|
||||
initialQueuedIds: initialQueued.map(t => t.id),
|
||||
projectId
|
||||
});
|
||||
|
||||
// Loop until capacity is full or queue is empty
|
||||
let iteration = 0;
|
||||
while (true) {
|
||||
// Get CURRENT state from store to ensure accuracy
|
||||
const currentTasks = useTaskStore.getState().tasks;
|
||||
const inProgressCount = currentTasks.filter((t) =>
|
||||
t.status === 'in_progress' && !t.metadata?.archivedAt
|
||||
).length;
|
||||
const queuedTasks = currentTasks.filter((t) =>
|
||||
t.status === 'queue' && !t.metadata?.archivedAt && !attemptedTaskIds.has(t.id)
|
||||
iteration++;
|
||||
// Calculate total in-progress count: tasks that were already in progress + tasks promoted in this call
|
||||
const totalInProgressCount = initialInProgress.length + promotedInThisCall;
|
||||
|
||||
debugLog(`[Queue] --- Iteration ${iteration} ---`, {
|
||||
initialInProgressCount: initialInProgress.length,
|
||||
promotedInThisCall,
|
||||
totalInProgressCount,
|
||||
capacityCheck: totalInProgressCount >= maxParallelTasks,
|
||||
processedCount: processedTaskIds.size
|
||||
});
|
||||
|
||||
// Stop if no capacity (initial in-progress + promoted in this call)
|
||||
if (totalInProgressCount >= maxParallelTasks) {
|
||||
debugLog(`[Queue] Capacity reached (${totalInProgressCount}/${maxParallelTasks}), stopping queue processing`);
|
||||
break;
|
||||
}
|
||||
|
||||
// Get CURRENT state from store to find queued tasks
|
||||
const latestTasks = useTaskStore.getState().tasks;
|
||||
const latestInProgress = latestTasks.filter((t) => t.status === 'in_progress' && !t.metadata?.archivedAt);
|
||||
const queuedTasks = latestTasks.filter((t) =>
|
||||
t.status === 'queue' && !t.metadata?.archivedAt && !processedTaskIds.has(t.id)
|
||||
);
|
||||
|
||||
// Stop if no capacity, no queued tasks, or too many consecutive failures
|
||||
if (inProgressCount >= maxParallelTasks || queuedTasks.length === 0) {
|
||||
debugLog(`[Queue] Current store state:`, {
|
||||
totalTasks: latestTasks.length,
|
||||
inProgressCount: latestInProgress.length,
|
||||
inProgressIds: latestInProgress.map(t => t.id),
|
||||
queuedCount: queuedTasks.length,
|
||||
queuedIds: queuedTasks.map(t => t.id),
|
||||
processedIds: Array.from(processedTaskIds)
|
||||
});
|
||||
|
||||
// Stop if no queued tasks or too many consecutive failures
|
||||
if (queuedTasks.length === 0) {
|
||||
debugLog('[Queue] No more queued tasks to process');
|
||||
break;
|
||||
}
|
||||
|
||||
if (consecutiveFailures >= MAX_CONSECUTIVE_FAILURES) {
|
||||
console.warn(`[Queue] Stopping queue processing after ${MAX_CONSECUTIVE_FAILURES} consecutive failures`);
|
||||
debugLog(`[Queue] Stopping queue processing after ${MAX_CONSECUTIVE_FAILURES} consecutive failures`);
|
||||
break;
|
||||
}
|
||||
|
||||
@@ -1043,28 +1089,62 @@ export function KanbanBoard({ tasks, onTaskClick, onNewTaskClick, onRefresh, isR
|
||||
return dateA - dateB; // Ascending order (oldest first)
|
||||
})[0];
|
||||
|
||||
console.log(`[Queue] Auto-promoting task ${nextTask.id} from Queue to In Progress (${inProgressCount + 1}/${maxParallelTasks})`);
|
||||
debugLog(`[Queue] Selected task for promotion:`, {
|
||||
id: nextTask.id,
|
||||
currentStatus: nextTask.status,
|
||||
title: nextTask.title?.substring(0, 50)
|
||||
});
|
||||
|
||||
// Mark task as processed BEFORE attempting promotion to prevent duplicates
|
||||
processedTaskIds.add(nextTask.id);
|
||||
|
||||
debugLog(`[Queue] Promoting task ${nextTask.id} (${promotedInThisCall + 1}/${maxParallelTasks})`);
|
||||
const result = await persistTaskStatus(nextTask.id, 'in_progress');
|
||||
|
||||
// Check store state after promotion
|
||||
const afterPromoteTasks = useTaskStore.getState().tasks;
|
||||
const afterPromoteInProgress = afterPromoteTasks.filter((t) => t.status === 'in_progress' && !t.metadata?.archivedAt);
|
||||
const afterPromoteQueued = afterPromoteTasks.filter((t) => t.status === 'queue' && !t.metadata?.archivedAt);
|
||||
|
||||
debugLog(`[Queue] After promotion attempt:`, {
|
||||
resultSuccess: result.success,
|
||||
promotedInThisCall,
|
||||
inProgressCount: afterPromoteInProgress.length,
|
||||
inProgressIds: afterPromoteInProgress.map(t => t.id),
|
||||
queuedCount: afterPromoteQueued.length,
|
||||
queuedIds: afterPromoteQueued.map(t => t.id)
|
||||
});
|
||||
|
||||
if (result.success) {
|
||||
// Increment our local promotion counter
|
||||
promotedInThisCall++;
|
||||
// Reset consecutive failures on success
|
||||
consecutiveFailures = 0;
|
||||
} else {
|
||||
// If promotion failed, log error, mark as attempted, and skip to next task
|
||||
// If promotion failed, log error and continue to next task
|
||||
console.error(`[Queue] Failed to promote task ${nextTask.id} to In Progress:`, result.error);
|
||||
attemptedTaskIds.add(nextTask.id);
|
||||
consecutiveFailures++;
|
||||
}
|
||||
}
|
||||
|
||||
// Log if we had failed tasks
|
||||
if (attemptedTaskIds.size > 0) {
|
||||
console.warn(`[Queue] Skipped ${attemptedTaskIds.size} task(s) that failed to promote`);
|
||||
// Log summary
|
||||
debugLog(`[Queue] === PROCESS QUEUE COMPLETE ===`, {
|
||||
totalIterations: iteration,
|
||||
tasksProcessed: processedTaskIds.size,
|
||||
tasksPromoted: promotedInThisCall,
|
||||
processedIds: Array.from(processedTaskIds)
|
||||
});
|
||||
|
||||
// Trigger UI refresh if tasks were promoted to ensure UI reflects all changes
|
||||
// This handles the case where store updates are batched/delayed via IPC events
|
||||
if (promotedInThisCall > 0 && onRefresh) {
|
||||
debugLog('[Queue] Triggering UI refresh after queue promotion');
|
||||
onRefresh();
|
||||
}
|
||||
} finally {
|
||||
isProcessingQueueRef.current = false;
|
||||
}
|
||||
}, [maxParallelTasks]);
|
||||
}, [maxParallelTasks, projectId, onRefresh]);
|
||||
|
||||
// Register task status change listener for queue auto-promotion
|
||||
// This ensures processQueue() is called whenever a task leaves in_progress
|
||||
@@ -1073,7 +1153,7 @@ export function KanbanBoard({ tasks, onTaskClick, onNewTaskClick, onRefresh, isR
|
||||
(taskId, oldStatus, newStatus) => {
|
||||
// When a task leaves in_progress (e.g., goes to human_review), process the queue
|
||||
if (oldStatus === 'in_progress' && newStatus !== 'in_progress') {
|
||||
console.log(`[Queue] Task ${taskId} left in_progress, processing queue to fill slot`);
|
||||
debugLog(`[Queue] Task ${taskId} left in_progress, processing queue to fill slot`);
|
||||
processQueue();
|
||||
}
|
||||
}
|
||||
|
||||
@@ -206,49 +206,50 @@ export const useTaskStore = create<TaskState>((set, get) => ({
|
||||
const state = get();
|
||||
const index = findTaskIndex(state.tasks, taskId);
|
||||
if (index === -1) {
|
||||
debugLog('[updateTaskStatus] Task not found:', taskId);
|
||||
console.warn('[updateTaskStatus] Task not found:', taskId);
|
||||
return;
|
||||
}
|
||||
const oldTask = state.tasks[index];
|
||||
const oldStatus = oldTask.status;
|
||||
|
||||
// Skip if status is the same
|
||||
if (oldStatus === status) return;
|
||||
if (oldStatus === status) {
|
||||
debugLog('[updateTaskStatus] Status unchanged, skipping:', { taskId, status });
|
||||
return;
|
||||
}
|
||||
|
||||
debugLog('[updateTaskStatus] START:', {
|
||||
taskId,
|
||||
oldStatus,
|
||||
newStatus: status,
|
||||
allInProgress: state.tasks.filter(t => t.status === 'in_progress' && !t.metadata?.archivedAt).map(t => t.id)
|
||||
});
|
||||
|
||||
// Perform the state update
|
||||
set((state) => {
|
||||
return {
|
||||
tasks: updateTaskAtIndex(state.tasks, index, (t) => {
|
||||
// Determine execution progress based on status transition
|
||||
let executionProgress = t.executionProgress;
|
||||
const updatedTasks = updateTaskAtIndex(state.tasks, index, (t) => {
|
||||
// Determine execution progress based on status transition
|
||||
let executionProgress = t.executionProgress;
|
||||
|
||||
// Track status transition for debugging flip-flop issues
|
||||
const previousStatus = t.status;
|
||||
const statusChanged = previousStatus !== status;
|
||||
if (status === 'backlog') {
|
||||
// When status goes to backlog, reset execution progress to idle
|
||||
// This ensures the planning/coding animation stops when task is stopped
|
||||
executionProgress = { phase: 'idle' as ExecutionPhase, phaseProgress: 0, overallProgress: 0 };
|
||||
} else if (status === 'in_progress' && !t.executionProgress?.phase) {
|
||||
// When starting a task and no phase is set yet, default to planning
|
||||
// This prevents the "no active phase" UI state during startup race condition
|
||||
executionProgress = { phase: 'planning' as ExecutionPhase, phaseProgress: 0, overallProgress: 0 };
|
||||
}
|
||||
|
||||
if (status === 'backlog') {
|
||||
// When status goes to backlog, reset execution progress to idle
|
||||
// This ensures the planning/coding animation stops when task is stopped
|
||||
executionProgress = { phase: 'idle' as ExecutionPhase, phaseProgress: 0, overallProgress: 0 };
|
||||
} else if (status === 'in_progress' && !t.executionProgress?.phase) {
|
||||
// When starting a task and no phase is set yet, default to planning
|
||||
// This prevents the "no active phase" UI state during startup race condition
|
||||
executionProgress = { phase: 'planning' as ExecutionPhase, phaseProgress: 0, overallProgress: 0 };
|
||||
}
|
||||
return { ...t, status, executionProgress, updatedAt: new Date() };
|
||||
});
|
||||
|
||||
// Log status transitions to help diagnose flip-flop issues
|
||||
debugLog('[updateTaskStatus] Status transition:', {
|
||||
taskId,
|
||||
previousStatus,
|
||||
newStatus: status,
|
||||
statusChanged,
|
||||
currentPhase: t.executionProgress?.phase,
|
||||
newPhase: executionProgress?.phase
|
||||
});
|
||||
debugLog('[updateTaskStatus] AFTER set():', {
|
||||
taskId,
|
||||
allInProgress: updatedTasks.filter((t: Task) => t.status === 'in_progress' && !t.metadata?.archivedAt).map(t => t.id)
|
||||
});
|
||||
|
||||
return { ...t, status, executionProgress, updatedAt: new Date() };
|
||||
})
|
||||
};
|
||||
return { tasks: updatedTasks };
|
||||
});
|
||||
|
||||
// Notify listeners after state update (schedule after current tick)
|
||||
@@ -796,7 +797,17 @@ export async function persistTaskStatus(
|
||||
}
|
||||
|
||||
// Only update local state after backend confirms success
|
||||
debugLog(`[persistTaskStatus] BEFORE store.updateTaskStatus:`, {
|
||||
taskId,
|
||||
newStatus: status,
|
||||
currentStoreStatus: store.tasks.find(t => t.id === taskId)?.status
|
||||
});
|
||||
store.updateTaskStatus(taskId, status);
|
||||
debugLog(`[persistTaskStatus] AFTER store.updateTaskStatus:`, {
|
||||
taskId,
|
||||
updatedStoreStatus: store.tasks.find(t => t.id === taskId)?.status,
|
||||
allInProgress: store.tasks.filter(t => t.status === 'in_progress' && !t.metadata?.archivedAt).map(t => t.id)
|
||||
});
|
||||
return { success: true };
|
||||
} catch (error) {
|
||||
console.error('Error persisting task status:', error);
|
||||
|
||||
Reference in New Issue
Block a user