50aea7e6f1
Task dispatch computed baseBranch as the phase branch name but never ensured it existed in the git clone. When phases weren't dispatched through the PhaseDispatchManager (which creates branches), the git worktree add failed with "fatal: invalid reference". Now DefaultDispatchManager calls ensureBranch for both the initiative and phase branches before spawning, matching what PhaseDispatchManager already does.
555 lines
18 KiB
TypeScript
555 lines
18 KiB
TypeScript
/**
|
|
* Default Dispatch Manager - Adapter Implementation
|
|
*
|
|
* Implements DispatchManager interface with in-memory queue
|
|
* and dependency-ordered dispatch.
|
|
*
|
|
* This is the ADAPTER for the DispatchManager PORT.
|
|
*/
|
|
|
|
import type {
|
|
EventBus,
|
|
TaskQueuedEvent,
|
|
TaskCompletedEvent,
|
|
TaskBlockedEvent,
|
|
TaskDispatchedEvent,
|
|
} from '../events/index.js';
|
|
import type { AgentManager, AgentResult, AgentInfo } from '../agent/types.js';
|
|
import type { TaskRepository } from '../db/repositories/task-repository.js';
|
|
import type { MessageRepository } from '../db/repositories/message-repository.js';
|
|
import type { AgentRepository } from '../db/repositories/agent-repository.js';
|
|
import type { InitiativeRepository } from '../db/repositories/initiative-repository.js';
|
|
import type { PhaseRepository } from '../db/repositories/phase-repository.js';
|
|
import type { PageRepository } from '../db/repositories/page-repository.js';
|
|
import type { ProjectRepository } from '../db/repositories/project-repository.js';
|
|
import type { Task, Phase } from '../db/schema.js';
|
|
import type { PageForSerialization } from '../agent/content-serializer.js';
|
|
import type { BranchManager } from '../git/branch-manager.js';
|
|
import type { DispatchManager, QueuedTask, DispatchResult } from './types.js';
|
|
import { phaseBranchName, taskBranchName, isPlanningCategory, generateInitiativeBranch } from '../git/branch-naming.js';
|
|
import { ensureProjectClone } from '../git/project-clones.js';
|
|
import { buildExecutePrompt } from '../agent/prompts/index.js';
|
|
import { createModuleLogger } from '../logger/index.js';
|
|
|
|
const log = createModuleLogger('dispatch');
|
|
|
|
// =============================================================================
|
|
// Internal Types
|
|
// =============================================================================
|
|
|
|
/**
|
|
* Internal representation of a blocked task.
|
|
*/
|
|
interface BlockedTask {
|
|
taskId: string;
|
|
reason: string;
|
|
}
|
|
|
|
// =============================================================================
|
|
// DefaultDispatchManager Implementation
|
|
// =============================================================================
|
|
|
|
/**
|
|
* In-memory implementation of DispatchManager.
|
|
*
|
|
* Uses Map for queue management and checks task_dependencies table
|
|
* for dependency resolution.
|
|
*/
|
|
export class DefaultDispatchManager implements DispatchManager {
|
|
/** Internal queue of tasks pending dispatch */
|
|
private taskQueue: Map<string, QueuedTask> = new Map();
|
|
|
|
/** Blocked tasks with their reasons */
|
|
private blockedTasks: Map<string, BlockedTask> = new Map();
|
|
|
|
constructor(
|
|
private taskRepository: TaskRepository,
|
|
private messageRepository: MessageRepository,
|
|
private agentManager: AgentManager,
|
|
private eventBus: EventBus,
|
|
private initiativeRepository?: InitiativeRepository,
|
|
private phaseRepository?: PhaseRepository,
|
|
private agentRepository?: AgentRepository,
|
|
private pageRepository?: PageRepository,
|
|
private projectRepository?: ProjectRepository,
|
|
private branchManager?: BranchManager,
|
|
private workspaceRoot?: string,
|
|
) {}
|
|
|
|
/**
|
|
* Queue a task for dispatch.
|
|
* Fetches task dependencies and adds to internal queue.
|
|
* Checkpoint tasks are queued but won't auto-dispatch.
|
|
*/
|
|
async queue(taskId: string): Promise<void> {
|
|
// Fetch task to verify it exists and get priority
|
|
const task = await this.taskRepository.findById(taskId);
|
|
if (!task) {
|
|
throw new Error(`Task not found: ${taskId}`);
|
|
}
|
|
|
|
// Get dependencies for this task from the repository
|
|
const dependsOn = await this.taskRepository.getDependencies(taskId);
|
|
|
|
const queuedTask: QueuedTask = {
|
|
taskId,
|
|
priority: task.priority,
|
|
queuedAt: new Date(),
|
|
dependsOn,
|
|
};
|
|
|
|
this.taskQueue.set(taskId, queuedTask);
|
|
|
|
log.info({ taskId, priority: task.priority, isCheckpoint: this.isCheckpointTask(task) }, 'task queued');
|
|
|
|
// Emit TaskQueuedEvent
|
|
const event: TaskQueuedEvent = {
|
|
type: 'task:queued',
|
|
timestamp: new Date(),
|
|
payload: {
|
|
taskId,
|
|
priority: task.priority,
|
|
dependsOn,
|
|
},
|
|
};
|
|
this.eventBus.emit(event);
|
|
}
|
|
|
|
/**
|
|
* Get next dispatchable task.
|
|
* Returns task with all dependencies complete, highest priority first.
|
|
* Checkpoint tasks are excluded (require human action).
|
|
*/
|
|
async getNextDispatchable(): Promise<QueuedTask | null> {
|
|
const queuedTasks = Array.from(this.taskQueue.values());
|
|
|
|
if (queuedTasks.length === 0) {
|
|
return null;
|
|
}
|
|
|
|
// Filter to only tasks with all dependencies complete and not checkpoint tasks
|
|
const readyTasks: QueuedTask[] = [];
|
|
|
|
log.debug({ queueSize: queuedTasks.length }, 'evaluating dispatchable tasks');
|
|
|
|
for (const qt of queuedTasks) {
|
|
// Check dependencies
|
|
const allDepsComplete = await this.areAllDependenciesComplete(qt.dependsOn);
|
|
if (!allDepsComplete) {
|
|
continue;
|
|
}
|
|
|
|
// Check if this is a checkpoint task (requires human action)
|
|
const task = await this.taskRepository.findById(qt.taskId);
|
|
if (task && this.isCheckpointTask(task)) {
|
|
log.debug({ taskId: qt.taskId, type: task.type }, 'skipping checkpoint task');
|
|
continue;
|
|
}
|
|
|
|
// Skip planning-category tasks (handled by architect flow)
|
|
if (task && isPlanningCategory(task.category)) {
|
|
log.debug({ taskId: qt.taskId, category: task.category }, 'skipping planning-category task');
|
|
continue;
|
|
}
|
|
|
|
readyTasks.push(qt);
|
|
}
|
|
|
|
log.debug({ queueSize: queuedTasks.length, readyCount: readyTasks.length }, 'dispatchable evaluation complete');
|
|
|
|
if (readyTasks.length === 0) {
|
|
return null;
|
|
}
|
|
|
|
// Sort by priority (high > medium > low), then by queuedAt (oldest first)
|
|
const priorityOrder: Record<string, number> = { high: 0, medium: 1, low: 2 };
|
|
|
|
readyTasks.sort((a, b) => {
|
|
const priorityDiff = priorityOrder[a.priority] - priorityOrder[b.priority];
|
|
if (priorityDiff !== 0) {
|
|
return priorityDiff;
|
|
}
|
|
return a.queuedAt.getTime() - b.queuedAt.getTime();
|
|
});
|
|
|
|
return readyTasks[0];
|
|
}
|
|
|
|
/**
|
|
* Mark a task as complete.
|
|
* Updates task status and removes from queue.
|
|
*
|
|
* @param taskId - ID of the task to complete
|
|
* @param agentId - Optional ID of the agent that completed the task
|
|
*/
|
|
async completeTask(taskId: string, agentId?: string): Promise<void> {
|
|
const task = await this.taskRepository.findById(taskId);
|
|
if (!task) {
|
|
throw new Error(`Task not found: ${taskId}`);
|
|
}
|
|
|
|
// Store agent result summary on the task for propagation to dependent tasks
|
|
await this.storeAgentSummary(taskId, agentId);
|
|
|
|
await this.taskRepository.update(taskId, { status: 'completed' });
|
|
|
|
// Remove from queue
|
|
this.taskQueue.delete(taskId);
|
|
|
|
log.info({ taskId }, 'task completed');
|
|
|
|
// Emit TaskCompletedEvent
|
|
const event: TaskCompletedEvent = {
|
|
type: 'task:completed',
|
|
timestamp: new Date(),
|
|
payload: {
|
|
taskId,
|
|
agentId: agentId ?? '',
|
|
success: true,
|
|
message: 'Task completed',
|
|
},
|
|
};
|
|
this.eventBus.emit(event);
|
|
|
|
// Also remove from blocked if it was there
|
|
this.blockedTasks.delete(taskId);
|
|
}
|
|
|
|
/**
|
|
* Mark a task as blocked.
|
|
* Updates task status and records block reason.
|
|
*/
|
|
async blockTask(taskId: string, reason: string): Promise<void> {
|
|
// Update task status to 'blocked'
|
|
await this.taskRepository.update(taskId, { status: 'blocked' });
|
|
|
|
// Record in blocked map
|
|
this.blockedTasks.set(taskId, { taskId, reason });
|
|
|
|
log.warn({ taskId, reason }, 'task blocked');
|
|
|
|
// Remove from queue (blocked tasks aren't dispatchable)
|
|
this.taskQueue.delete(taskId);
|
|
|
|
// Emit TaskBlockedEvent
|
|
const event: TaskBlockedEvent = {
|
|
type: 'task:blocked',
|
|
timestamp: new Date(),
|
|
payload: {
|
|
taskId,
|
|
reason,
|
|
},
|
|
};
|
|
this.eventBus.emit(event);
|
|
}
|
|
|
|
/**
|
|
* Retry a blocked task.
|
|
* Resets status to pending, clears block state, and re-queues for dispatch.
|
|
*/
|
|
async retryBlockedTask(taskId: string): Promise<void> {
|
|
const task = await this.taskRepository.findById(taskId);
|
|
if (!task) throw new Error(`Task not found: ${taskId}`);
|
|
if (task.status !== 'blocked') throw new Error(`Task ${taskId} is not blocked (status: ${task.status})`);
|
|
|
|
// Clear blocked state
|
|
this.blockedTasks.delete(taskId);
|
|
|
|
// Reset DB status to pending
|
|
await this.taskRepository.update(taskId, { status: 'pending' });
|
|
|
|
log.info({ taskId }, 'retrying blocked task');
|
|
|
|
// Re-queue for dispatch
|
|
await this.queue(taskId);
|
|
}
|
|
|
|
/**
|
|
* Dispatch next available task to an agent.
|
|
*/
|
|
async dispatchNext(): Promise<DispatchResult> {
|
|
// Get next dispatchable task
|
|
const nextTask = await this.getNextDispatchable();
|
|
|
|
if (!nextTask) {
|
|
log.debug('no dispatchable tasks');
|
|
return {
|
|
success: false,
|
|
taskId: '',
|
|
reason: 'No dispatchable tasks',
|
|
};
|
|
}
|
|
|
|
// Find available agent (status='idle')
|
|
const agents = await this.agentManager.list();
|
|
const idleAgent = agents.find((a) => a.status === 'idle');
|
|
|
|
if (!idleAgent) {
|
|
log.debug('no available agents');
|
|
return {
|
|
success: false,
|
|
taskId: nextTask.taskId,
|
|
reason: 'No available agents',
|
|
};
|
|
}
|
|
|
|
// Get task details
|
|
const task = await this.taskRepository.findById(nextTask.taskId);
|
|
if (!task) {
|
|
return {
|
|
success: false,
|
|
taskId: nextTask.taskId,
|
|
reason: 'Task not found',
|
|
};
|
|
}
|
|
|
|
// Compute branch info for branch-aware spawning
|
|
let baseBranch: string | undefined;
|
|
let branchName: string | undefined;
|
|
|
|
if (task.initiativeId && this.initiativeRepository) {
|
|
try {
|
|
if (isPlanningCategory(task.category)) {
|
|
// Planning tasks run on project default branches — no initiative branch needed.
|
|
// baseBranch and branchName remain undefined; ProcessManager uses per-project defaults.
|
|
} else if (task.phaseId && this.phaseRepository) {
|
|
// Execution task — ensure initiative has a branch
|
|
const initiative = await this.initiativeRepository.findById(task.initiativeId);
|
|
if (initiative) {
|
|
let initBranch = initiative.branch;
|
|
if (!initBranch) {
|
|
initBranch = generateInitiativeBranch(initiative.name);
|
|
await this.initiativeRepository.update(initiative.id, { branch: initBranch });
|
|
}
|
|
|
|
const phase = await this.phaseRepository.findById(task.phaseId);
|
|
if (phase) {
|
|
if (task.category === 'merge' || task.category === 'review') {
|
|
// Merge/review tasks work directly on the phase branch
|
|
baseBranch = initBranch;
|
|
branchName = phaseBranchName(initBranch, phase.name);
|
|
} else {
|
|
baseBranch = phaseBranchName(initBranch, phase.name);
|
|
branchName = taskBranchName(initBranch, task.id);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
} catch {
|
|
// Non-fatal: fall back to default branching
|
|
}
|
|
|
|
// Ensure branches exist in project clones before spawning worktrees
|
|
if (baseBranch && this.projectRepository && this.branchManager && this.workspaceRoot) {
|
|
try {
|
|
const initiative = await this.initiativeRepository.findById(task.initiativeId);
|
|
const initBranch = initiative?.branch;
|
|
if (initBranch) {
|
|
const projects = await this.projectRepository.findProjectsByInitiativeId(task.initiativeId);
|
|
for (const project of projects) {
|
|
const clonePath = await ensureProjectClone(project, this.workspaceRoot);
|
|
// Ensure initiative branch exists (from project defaultBranch)
|
|
await this.branchManager.ensureBranch(clonePath, initBranch, project.defaultBranch);
|
|
// Ensure phase branch exists (from initiative branch)
|
|
const phBranch = phaseBranchName(initBranch, (await this.phaseRepository?.findById(task.phaseId!))?.name ?? '');
|
|
if (phBranch) {
|
|
await this.branchManager.ensureBranch(clonePath, phBranch, initBranch);
|
|
}
|
|
}
|
|
}
|
|
} catch (err) {
|
|
log.warn({ taskId: task.id, err }, 'failed to ensure branches for task dispatch');
|
|
}
|
|
}
|
|
}
|
|
|
|
// Gather initiative context for the agent's input files
|
|
let inputContext: import('../agent/types.js').AgentInputContext | undefined;
|
|
if (task.initiativeId) {
|
|
try {
|
|
const initiative = await this.initiativeRepository?.findById(task.initiativeId);
|
|
const phase = task.phaseId ? await this.phaseRepository?.findById(task.phaseId) : undefined;
|
|
const context = await this.gatherInitiativeContext(task.initiativeId);
|
|
|
|
inputContext = {
|
|
initiative: initiative ?? undefined,
|
|
task,
|
|
phase: phase ?? undefined,
|
|
phases: context.phases.length > 0 ? context.phases : undefined,
|
|
tasks: context.tasks.length > 0 ? context.tasks : undefined,
|
|
pages: context.pages.length > 0 ? context.pages : undefined,
|
|
};
|
|
} catch (err) {
|
|
log.warn({ taskId: task.id, err }, 'failed to gather initiative context for dispatch');
|
|
}
|
|
}
|
|
|
|
// Spawn agent with task (alias auto-generated by agent manager)
|
|
let agent: AgentInfo;
|
|
try {
|
|
agent = await this.agentManager.spawn({
|
|
taskId: nextTask.taskId,
|
|
initiativeId: task.initiativeId ?? undefined,
|
|
phaseId: task.phaseId ?? undefined,
|
|
prompt: buildExecutePrompt(task.description || task.name),
|
|
baseBranch,
|
|
branchName,
|
|
inputContext,
|
|
});
|
|
} catch (err) {
|
|
const reason = `Spawn failed: ${err instanceof Error ? err.message : String(err)}`;
|
|
log.error({ taskId: nextTask.taskId, err: reason }, 'agent spawn failed, blocking task');
|
|
await this.blockTask(nextTask.taskId, reason);
|
|
return { success: false, taskId: nextTask.taskId, reason };
|
|
}
|
|
|
|
log.info({ taskId: nextTask.taskId, agentId: agent.id }, 'task dispatched');
|
|
|
|
// Update task status to 'in_progress'
|
|
await this.taskRepository.update(nextTask.taskId, { status: 'in_progress' });
|
|
|
|
// Remove from queue (now being worked on)
|
|
this.taskQueue.delete(nextTask.taskId);
|
|
|
|
// Emit TaskDispatchedEvent
|
|
const event: TaskDispatchedEvent = {
|
|
type: 'task:dispatched',
|
|
timestamp: new Date(),
|
|
payload: {
|
|
taskId: nextTask.taskId,
|
|
agentId: agent.id,
|
|
agentName: agent.name,
|
|
},
|
|
};
|
|
this.eventBus.emit(event);
|
|
|
|
return {
|
|
success: true,
|
|
taskId: nextTask.taskId,
|
|
agentId: agent.id,
|
|
};
|
|
}
|
|
|
|
/**
|
|
* Get current queue state.
|
|
*/
|
|
async getQueueState(): Promise<{
|
|
queued: QueuedTask[];
|
|
ready: QueuedTask[];
|
|
blocked: Array<{ taskId: string; reason: string }>;
|
|
}> {
|
|
const allQueued = Array.from(this.taskQueue.values());
|
|
|
|
// Determine which are ready
|
|
const ready: QueuedTask[] = [];
|
|
for (const qt of allQueued) {
|
|
const allDepsComplete = await this.areAllDependenciesComplete(qt.dependsOn);
|
|
if (allDepsComplete) {
|
|
ready.push(qt);
|
|
}
|
|
}
|
|
|
|
return {
|
|
queued: allQueued,
|
|
ready,
|
|
blocked: Array.from(this.blockedTasks.values()),
|
|
};
|
|
}
|
|
|
|
// =============================================================================
|
|
// Private Helpers
|
|
// =============================================================================
|
|
|
|
/**
|
|
* Check if all dependencies are complete.
|
|
*/
|
|
private async areAllDependenciesComplete(dependsOn: string[]): Promise<boolean> {
|
|
if (dependsOn.length === 0) {
|
|
return true;
|
|
}
|
|
|
|
for (const depTaskId of dependsOn) {
|
|
const depTask = await this.taskRepository.findById(depTaskId);
|
|
if (!depTask || depTask.status !== 'completed') {
|
|
return false;
|
|
}
|
|
}
|
|
|
|
return true;
|
|
}
|
|
|
|
/**
|
|
* Check if a task is a checkpoint task.
|
|
* Checkpoint tasks require human action and don't auto-dispatch.
|
|
*/
|
|
private isCheckpointTask(task: Task): boolean {
|
|
return task.type.startsWith('checkpoint:');
|
|
}
|
|
|
|
/**
|
|
* Store the completing agent's result summary on the task record.
|
|
*/
|
|
private async storeAgentSummary(taskId: string, agentId?: string): Promise<void> {
|
|
if (!agentId || !this.agentRepository) return;
|
|
try {
|
|
const agentRecord = await this.agentRepository.findById(agentId);
|
|
if (agentRecord?.result) {
|
|
const result: AgentResult = JSON.parse(agentRecord.result);
|
|
if (result.message) {
|
|
await this.taskRepository.update(taskId, { summary: result.message });
|
|
}
|
|
}
|
|
} catch (err) {
|
|
log.warn({ taskId, agentId, err }, 'failed to store agent summary on task');
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Gather initiative context for passing to execution agents.
|
|
* Reuses the same pattern as architect.ts gatherInitiativeContext.
|
|
*/
|
|
private async gatherInitiativeContext(initiativeId: string): Promise<{
|
|
phases: Array<Phase & { dependsOn?: string[] }>;
|
|
tasks: Task[];
|
|
pages: PageForSerialization[];
|
|
}> {
|
|
const [rawPhases, deps, initiativeTasks, pages] = await Promise.all([
|
|
this.phaseRepository?.findByInitiativeId(initiativeId) ?? [],
|
|
this.phaseRepository?.findDependenciesByInitiativeId(initiativeId) ?? [],
|
|
this.taskRepository.findByInitiativeId(initiativeId),
|
|
this.pageRepository?.findByInitiativeId(initiativeId) ?? [],
|
|
]);
|
|
|
|
// Merge dependencies into each phase as a dependsOn array
|
|
const depsByPhase = new Map<string, string[]>();
|
|
for (const dep of deps) {
|
|
const arr = depsByPhase.get(dep.phaseId) ?? [];
|
|
arr.push(dep.dependsOnPhaseId);
|
|
depsByPhase.set(dep.phaseId, arr);
|
|
}
|
|
const phases = rawPhases.map((ph) => ({
|
|
...ph,
|
|
dependsOn: depsByPhase.get(ph.id) ?? [],
|
|
}));
|
|
|
|
// Collect tasks from all phases (some tasks only have phaseId, not initiativeId)
|
|
const taskIds = new Set(initiativeTasks.map((t) => t.id));
|
|
const allTasks = [...initiativeTasks];
|
|
for (const ph of rawPhases) {
|
|
const phaseTasks = await this.taskRepository.findByPhaseId(ph.id);
|
|
for (const t of phaseTasks) {
|
|
if (!taskIds.has(t.id)) {
|
|
taskIds.add(t.id);
|
|
allTasks.push(t);
|
|
}
|
|
}
|
|
}
|
|
|
|
// Only include implementation tasks — planning tasks are irrelevant noise
|
|
const implementationTasks = allTasks.filter(t => !isPlanningCategory(t.category));
|
|
|
|
return { phases, tasks: implementationTasks, pages };
|
|
}
|
|
|
|
}
|