Enables parallel agents to communicate through a CLI-based conversation mechanism coordinated via tRPC. Agents can ask questions to peers and receive answers, with target resolution by agent ID, task ID, or phase ID.
4.5 KiB
4.5 KiB
Dispatch & Events
src/dispatch/ — Task and phase dispatch queues. src/events/ — Typed event bus.
Event Bus
src/events/ — Typed pub/sub system for inter-module communication.
Architecture
- Port:
EventBusinterface withemit(event)andon(type, handler) - Adapter:
TypedEventBususing Node.jsEventEmitter - All events implement
BaseEvent { type, timestamp, payload }
Event Types (52)
| Category | Events | Count |
|---|---|---|
| Agent | agent:spawned, agent:stopped, agent:crashed, agent:resumed, agent:account_switched, agent:deleted, agent:waiting, agent:output |
8 |
| Task | task:queued, task:dispatched, task:completed, task:blocked, task:pending_approval |
5 |
| Phase | phase:queued, phase:started, phase:completed, phase:blocked |
4 |
| Merge | merge:queued, merge:started, merge:completed, merge:conflicted |
4 |
| Page | page:created, page:updated, page:deleted |
3 |
| Process | process:spawned, process:stopped, process:crashed |
3 |
| Server | server:started, server:stopped |
2 |
| Worktree | worktree:created, worktree:removed, worktree:merged, worktree:conflict |
4 |
| Account | account:credentials_refreshed, account:credentials_expired, account:credentials_validated |
3 |
| Preview | preview:building, preview:ready, preview:stopped, preview:failed |
4 |
| Conversation | conversation:created, conversation:answered |
2 |
| Log | log:entry |
1 |
Key Event Payloads
AgentSpawnedEvent { agentId, name, taskId, worktreeId, provider }
AgentStoppedEvent { agentId, name, taskId, reason }
// reason: 'user_requested'|'task_complete'|'error'|'waiting_for_input'|
// 'context_complete'|'plan_complete'|'detail_complete'|'refine_complete'
AgentWaitingEvent { agentId, name, taskId, sessionId, questions[] }
AgentOutputEvent { agentId, stream, data }
TaskCompletedEvent { taskId, agentId, success, message }
TaskQueuedEvent { taskId, priority, dependsOn[] }
PhaseStartedEvent { phaseId, initiativeId }
MergeConflictedEvent { taskId, agentId, worktreeId, targetBranch, conflictingFiles[] }
AccountCredentialsRefreshedEvent { accountId, expiresAt, previousExpiresAt? }
Task Dispatch
src/dispatch/ — In-memory queue with dependency-ordered dispatch.
Architecture
- Port:
DispatchManagerinterface - Adapter:
DefaultDispatchManager
How Task Dispatch Works
- Queue —
queue(taskId)fetches task dependencies, adds to internal Map - Dispatch —
dispatchNext()finds highest-priority task with all deps complete - Priority order: high > medium > low, then oldest first (FIFO within priority)
- Checkpoint skip — Tasks with type starting with
checkpoint:skip auto-dispatch - Planning skip — Planning-category tasks (research, discuss, plan, detail, refine) skip auto-dispatch — they use the architect flow
- Approval check —
completeTask()checksrequiresApproval(task-level, then initiative-level) - Approval flow — If approval required: status →
pending_approval, emittask:pending_approval
DispatchManager Methods
| Method | Purpose |
|---|---|
queue(taskId) |
Add task to dispatch queue |
dispatchNext() |
Find and dispatch next ready task |
getNextDispatchable() |
Get next task without dispatching |
completeTask(taskId, agentId?) |
Complete with approval check |
approveTask(taskId) |
Approve pending task |
blockTask(taskId, reason) |
Block task with reason |
getQueueState() |
Return queued, ready, blocked tasks |
Phase Dispatch
DefaultPhaseDispatchManager — Same pattern for phases:
- Queue —
queuePhase(phaseId)validates phase is approved, gets dependencies - Dispatch —
dispatchNextPhase()finds phase with all deps complete - Auto-queue tasks — When phase starts, pending execution tasks are queued (planning-category tasks excluded)
- Events —
phase:queued,phase:started,phase:completed,phase:blocked
PhaseDispatchManager Methods
| Method | Purpose |
|---|---|
queuePhase(phaseId) |
Queue approved phase |
dispatchNextPhase() |
Start next ready phase, auto-queue its tasks |
getNextDispatchablePhase() |
Get next phase without dispatching |
completePhase(phaseId) |
Mark phase complete |
blockPhase(phaseId, reason) |
Block phase |
getPhaseQueueState() |
Return queued, ready, blocked phases |