536cdf08a1
Execution agents were spawning blind — no input files, no knowledge of what predecessor tasks accomplished. This adds three capabilities: 1. summary column on tasks table — completeTask() reads the finishing agent's result.message and stores it on the task record 2. dispatchNext() gathers full initiative context (initiative, phase, sibling tasks, pages) and passes it as inputContext so agents get .cw/input/task.md, initiative.md, phase.md, and context directories 3. context/tasks/*.md files now include the summary field in frontmatter so dependent agents can see what prior agents accomplished
581 lines
18 KiB
TypeScript
581 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,
|
|
TaskPendingApprovalEvent,
|
|
} from '../events/index.js';
|
|
import type { AgentManager, AgentResult } 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 { Task, Phase } from '../db/schema.js';
|
|
import type { PageForSerialization } from '../agent/content-serializer.js';
|
|
import type { DispatchManager, QueuedTask, DispatchResult } from './types.js';
|
|
import { phaseBranchName, taskBranchName, isPlanningCategory, generateInitiativeBranch } from '../git/branch-naming.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,
|
|
) {}
|
|
|
|
/**
|
|
* 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.
|
|
* If the task requires approval, sets status to 'pending_approval' instead.
|
|
* 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}`);
|
|
}
|
|
|
|
// Determine if approval is required
|
|
const requiresApproval = await this.taskRequiresApproval(task);
|
|
|
|
// Store agent result summary on the task for propagation to dependent tasks
|
|
await this.storeAgentSummary(taskId, agentId);
|
|
|
|
if (requiresApproval) {
|
|
// Set to pending_approval instead of completed
|
|
await this.taskRepository.update(taskId, { status: 'pending_approval' });
|
|
|
|
// Remove from queue
|
|
this.taskQueue.delete(taskId);
|
|
|
|
log.info({ taskId, category: task.category }, 'task pending approval');
|
|
|
|
// Emit TaskPendingApprovalEvent
|
|
const event: TaskPendingApprovalEvent = {
|
|
type: 'task:pending_approval',
|
|
timestamp: new Date(),
|
|
payload: {
|
|
taskId,
|
|
agentId: agentId ?? '',
|
|
category: task.category,
|
|
name: task.name,
|
|
},
|
|
};
|
|
this.eventBus.emit(event);
|
|
} else {
|
|
// Complete directly
|
|
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);
|
|
}
|
|
|
|
/**
|
|
* Approve a task that is pending approval.
|
|
* Sets status to 'completed' and emits completion event.
|
|
*/
|
|
async approveTask(taskId: string): Promise<void> {
|
|
const task = await this.taskRepository.findById(taskId);
|
|
if (!task) {
|
|
throw new Error(`Task not found: ${taskId}`);
|
|
}
|
|
|
|
if (task.status !== 'pending_approval') {
|
|
throw new Error(`Task ${taskId} is not pending approval (status: ${task.status})`);
|
|
}
|
|
|
|
// Complete the task
|
|
await this.taskRepository.update(taskId, { status: 'completed' });
|
|
|
|
log.info({ taskId }, 'task approved and completed');
|
|
|
|
// Emit TaskCompletedEvent
|
|
const event: TaskCompletedEvent = {
|
|
type: 'task:completed',
|
|
timestamp: new Date(),
|
|
payload: {
|
|
taskId,
|
|
agentId: '',
|
|
success: true,
|
|
message: 'Task approved',
|
|
},
|
|
};
|
|
this.eventBus.emit(event);
|
|
}
|
|
|
|
/**
|
|
* 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);
|
|
}
|
|
|
|
/**
|
|
* 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') {
|
|
// Merge 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
|
|
}
|
|
}
|
|
|
|
// 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)
|
|
const 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,
|
|
});
|
|
|
|
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 };
|
|
}
|
|
|
|
/**
|
|
* Determine if a task requires approval before being marked complete.
|
|
* Checks task-level override first, then falls back to initiative setting.
|
|
*/
|
|
private async taskRequiresApproval(task: Task): Promise<boolean> {
|
|
// Task-level override takes precedence
|
|
if (task.requiresApproval !== null) {
|
|
return task.requiresApproval;
|
|
}
|
|
|
|
// Fall back to initiative setting if we have initiative access
|
|
if (this.initiativeRepository && task.initiativeId) {
|
|
const initiative = await this.initiativeRepository.findById(task.initiativeId);
|
|
if (initiative) {
|
|
return initiative.mergeRequiresApproval;
|
|
}
|
|
}
|
|
|
|
// If task has a phaseId but no initiativeId, we could traverse up but for now default to false
|
|
// Default: no approval required
|
|
return false;
|
|
}
|
|
}
|