diff --git a/apps/server/execution/orchestrator.test.ts b/apps/server/execution/orchestrator.test.ts new file mode 100644 index 0000000..57efcd5 --- /dev/null +++ b/apps/server/execution/orchestrator.test.ts @@ -0,0 +1,306 @@ +/** + * ExecutionOrchestrator Tests + * + * Tests phase completion transitions, especially when initiative has no branch. + */ + +import { describe, it, expect, vi, beforeEach } from 'vitest'; +import { ExecutionOrchestrator } from './orchestrator.js'; +import type { BranchManager } from '../git/branch-manager.js'; +import type { PhaseRepository } from '../db/repositories/phase-repository.js'; +import type { TaskRepository } from '../db/repositories/task-repository.js'; +import type { InitiativeRepository } from '../db/repositories/initiative-repository.js'; +import type { ProjectRepository } from '../db/repositories/project-repository.js'; +import type { DispatchManager, PhaseDispatchManager } from '../dispatch/types.js'; +import type { ConflictResolutionService } from '../coordination/conflict-resolution-service.js'; +import type { EventBus, TaskCompletedEvent, DomainEvent } from '../events/types.js'; + +function createMockEventBus(): EventBus & { handlers: Map; emitted: DomainEvent[] } { + const handlers = new Map(); + const emitted: DomainEvent[] = []; + return { + handlers, + emitted, + emit: vi.fn((event: DomainEvent) => { + emitted.push(event); + const fns = handlers.get(event.type) ?? []; + for (const fn of fns) fn(event); + }), + on: vi.fn((type: string, handler: Function) => { + const fns = handlers.get(type) ?? []; + fns.push(handler); + handlers.set(type, fns); + }), + off: vi.fn(), + once: vi.fn(), + }; +} + +function createMocks() { + const branchManager: BranchManager = { + ensureBranch: vi.fn(), + mergeBranch: vi.fn().mockResolvedValue({ success: true, message: 'merged' }), + diffBranches: vi.fn().mockResolvedValue(''), + deleteBranch: vi.fn(), + branchExists: vi.fn().mockResolvedValue(true), + remoteBranchExists: vi.fn().mockResolvedValue(false), + listCommits: vi.fn().mockResolvedValue([]), + diffCommit: vi.fn().mockResolvedValue(''), + getMergeBase: vi.fn().mockResolvedValue('abc123'), + pushBranch: vi.fn(), + }; + + const phaseRepository = { + findById: vi.fn(), + findByInitiativeId: vi.fn().mockResolvedValue([]), + update: vi.fn().mockImplementation(async (id: string, data: any) => ({ id, ...data })), + create: vi.fn(), + } as unknown as PhaseRepository; + + const taskRepository = { + findById: vi.fn(), + findByPhaseId: vi.fn().mockResolvedValue([]), + findByInitiativeId: vi.fn().mockResolvedValue([]), + } as unknown as TaskRepository; + + const initiativeRepository = { + findById: vi.fn(), + findByStatus: vi.fn().mockResolvedValue([]), + update: vi.fn(), + } as unknown as InitiativeRepository; + + const projectRepository = { + findProjectsByInitiativeId: vi.fn().mockResolvedValue([]), + } as unknown as ProjectRepository; + + const phaseDispatchManager: PhaseDispatchManager = { + queuePhase: vi.fn(), + getNextDispatchablePhase: vi.fn().mockResolvedValue(null), + dispatchNextPhase: vi.fn().mockResolvedValue({ success: false, phaseId: '', reason: 'none' }), + completePhase: vi.fn(), + blockPhase: vi.fn(), + getPhaseQueueState: vi.fn().mockResolvedValue({ queued: [], ready: [], blocked: [] }), + }; + + const dispatchManager = { + queue: vi.fn(), + getNextDispatchable: vi.fn().mockResolvedValue(null), + dispatchNext: vi.fn().mockResolvedValue({ success: false, taskId: '' }), + completeTask: vi.fn(), + blockTask: vi.fn(), + retryBlockedTask: vi.fn(), + getQueueState: vi.fn().mockResolvedValue({ queued: [], ready: [], blocked: [] }), + } as unknown as DispatchManager; + + const conflictResolutionService: ConflictResolutionService = { + handleConflict: vi.fn(), + }; + + const eventBus = createMockEventBus(); + + return { + branchManager, + phaseRepository, + taskRepository, + initiativeRepository, + projectRepository, + phaseDispatchManager, + dispatchManager, + conflictResolutionService, + eventBus, + }; +} + +function createOrchestrator(mocks: ReturnType) { + const orchestrator = new ExecutionOrchestrator( + mocks.branchManager, + mocks.phaseRepository, + mocks.taskRepository, + mocks.initiativeRepository, + mocks.projectRepository, + mocks.phaseDispatchManager, + mocks.dispatchManager, + mocks.conflictResolutionService, + mocks.eventBus, + '/tmp/test-workspace', + ); + orchestrator.start(); + return orchestrator; +} + +function emitTaskCompleted(eventBus: ReturnType, taskId: string) { + const event: TaskCompletedEvent = { + type: 'task:completed', + timestamp: new Date(), + payload: { taskId, agentId: 'agent-1', success: true, message: 'done' }, + }; + eventBus.emit(event); +} + +describe('ExecutionOrchestrator', () => { + let mocks: ReturnType; + + beforeEach(() => { + mocks = createMocks(); + }); + + describe('phase completion when initiative has no branch', () => { + it('should transition phase to pending_review in review mode even without a branch', async () => { + const task = { + id: 'task-1', + phaseId: 'phase-1', + initiativeId: 'init-1', + category: 'execute', + status: 'completed', + }; + const phase = { id: 'phase-1', initiativeId: 'init-1', name: 'Phase 1', status: 'in_progress' }; + const initiative = { id: 'init-1', branch: null, executionMode: 'review_per_phase' }; + + vi.mocked(mocks.taskRepository.findById).mockResolvedValue(task as any); + vi.mocked(mocks.phaseRepository.findById).mockResolvedValue(phase as any); + vi.mocked(mocks.initiativeRepository.findById).mockResolvedValue(initiative as any); + vi.mocked(mocks.taskRepository.findByPhaseId).mockResolvedValue([task] as any); + + createOrchestrator(mocks); + + emitTaskCompleted(mocks.eventBus, 'task-1'); + + // Allow async handler to complete + await vi.waitFor(() => { + expect(mocks.phaseRepository.update).toHaveBeenCalledWith('phase-1', { status: 'pending_review' }); + }); + }); + + it('should complete phase in yolo mode even without a branch', async () => { + const task = { + id: 'task-1', + phaseId: 'phase-1', + initiativeId: 'init-1', + category: 'execute', + status: 'completed', + }; + const phase = { id: 'phase-1', initiativeId: 'init-1', name: 'Phase 1', status: 'in_progress' }; + const initiative = { id: 'init-1', branch: null, executionMode: 'yolo' }; + + vi.mocked(mocks.taskRepository.findById).mockResolvedValue(task as any); + vi.mocked(mocks.phaseRepository.findById).mockResolvedValue(phase as any); + vi.mocked(mocks.initiativeRepository.findById).mockResolvedValue(initiative as any); + vi.mocked(mocks.initiativeRepository.findByStatus).mockResolvedValue([]); + vi.mocked(mocks.taskRepository.findByPhaseId).mockResolvedValue([task] as any); + vi.mocked(mocks.phaseRepository.findByInitiativeId).mockResolvedValue([phase] as any); + + createOrchestrator(mocks); + + emitTaskCompleted(mocks.eventBus, 'task-1'); + + await vi.waitFor(() => { + expect(mocks.phaseDispatchManager.completePhase).toHaveBeenCalledWith('phase-1'); + }); + + // Should NOT have attempted any branch merges + expect(mocks.branchManager.mergeBranch).not.toHaveBeenCalled(); + }); + }); + + describe('phase completion when merge fails', () => { + it('should still check phase completion even if task merge throws', async () => { + const task = { + id: 'task-1', + phaseId: 'phase-1', + initiativeId: 'init-1', + category: 'execute', + status: 'completed', + }; + const phase = { id: 'phase-1', initiativeId: 'init-1', name: 'Phase 1', status: 'in_progress' }; + const initiative = { id: 'init-1', branch: 'cw/test', executionMode: 'review_per_phase' }; + const project = { id: 'proj-1', name: 'test', url: 'https://example.com', defaultBranch: 'main' }; + + vi.mocked(mocks.taskRepository.findById).mockResolvedValue(task as any); + vi.mocked(mocks.phaseRepository.findById).mockResolvedValue(phase as any); + vi.mocked(mocks.initiativeRepository.findById).mockResolvedValue(initiative as any); + vi.mocked(mocks.taskRepository.findByPhaseId).mockResolvedValue([task] as any); + vi.mocked(mocks.projectRepository.findProjectsByInitiativeId).mockResolvedValue([project] as any); + // Merge fails + vi.mocked(mocks.branchManager.mergeBranch).mockResolvedValue({ + success: false, + message: 'conflict', + conflicts: ['file.ts'], + }); + + createOrchestrator(mocks); + + emitTaskCompleted(mocks.eventBus, 'task-1'); + + // Phase should still transition despite merge failure + await vi.waitFor(() => { + expect(mocks.phaseRepository.update).toHaveBeenCalledWith('phase-1', { status: 'pending_review' }); + }); + }); + }); + + describe('phase completion with branch (normal flow)', () => { + it('should merge task branch and transition phase when all tasks done', async () => { + const task = { + id: 'task-1', + phaseId: 'phase-1', + initiativeId: 'init-1', + category: 'execute', + status: 'completed', + }; + const phase = { id: 'phase-1', initiativeId: 'init-1', name: 'Phase 1', status: 'in_progress' }; + const initiative = { id: 'init-1', branch: 'cw/test', executionMode: 'review_per_phase' }; + const project = { id: 'proj-1', name: 'test', url: 'https://example.com', defaultBranch: 'main' }; + + vi.mocked(mocks.taskRepository.findById).mockResolvedValue(task as any); + vi.mocked(mocks.phaseRepository.findById).mockResolvedValue(phase as any); + vi.mocked(mocks.initiativeRepository.findById).mockResolvedValue(initiative as any); + vi.mocked(mocks.taskRepository.findByPhaseId).mockResolvedValue([task] as any); + vi.mocked(mocks.projectRepository.findProjectsByInitiativeId).mockResolvedValue([project] as any); + vi.mocked(mocks.branchManager.branchExists).mockResolvedValue(true); + vi.mocked(mocks.branchManager.mergeBranch).mockResolvedValue({ success: true, message: 'ok' }); + + createOrchestrator(mocks); + + emitTaskCompleted(mocks.eventBus, 'task-1'); + + await vi.waitFor(() => { + expect(mocks.phaseRepository.update).toHaveBeenCalledWith('phase-1', { status: 'pending_review' }); + }); + }); + + it('should not transition phase when some tasks are still pending', async () => { + const task1 = { + id: 'task-1', + phaseId: 'phase-1', + initiativeId: 'init-1', + category: 'execute', + status: 'completed', + }; + const task2 = { + id: 'task-2', + phaseId: 'phase-1', + initiativeId: 'init-1', + category: 'execute', + status: 'pending', + }; + const phase = { id: 'phase-1', initiativeId: 'init-1', name: 'Phase 1', status: 'in_progress' }; + const initiative = { id: 'init-1', branch: 'cw/test', executionMode: 'review_per_phase' }; + + vi.mocked(mocks.taskRepository.findById).mockResolvedValue(task1 as any); + vi.mocked(mocks.phaseRepository.findById).mockResolvedValue(phase as any); + vi.mocked(mocks.initiativeRepository.findById).mockResolvedValue(initiative as any); + vi.mocked(mocks.taskRepository.findByPhaseId).mockResolvedValue([task1, task2] as any); + vi.mocked(mocks.projectRepository.findProjectsByInitiativeId).mockResolvedValue([]); + + createOrchestrator(mocks); + + emitTaskCompleted(mocks.eventBus, 'task-1'); + + // Give the async handler time to run + await new Promise((r) => setTimeout(r, 50)); + + expect(mocks.phaseRepository.update).not.toHaveBeenCalled(); + expect(mocks.phaseDispatchManager.completePhase).not.toHaveBeenCalled(); + }); + }); +}); diff --git a/apps/server/execution/orchestrator.ts b/apps/server/execution/orchestrator.ts index 973cc59..2883f18 100644 --- a/apps/server/execution/orchestrator.ts +++ b/apps/server/execution/orchestrator.ts @@ -145,27 +145,29 @@ export class ExecutionOrchestrator { if (!task?.phaseId || !task.initiativeId) return; const initiative = await this.initiativeRepository.findById(task.initiativeId); - if (!initiative?.branch) return; - const phase = await this.phaseRepository.findById(task.phaseId); if (!phase) return; - // Skip merge for review/merge tasks — they already work on the phase branch directly - if (task.category !== 'merge' && task.category !== 'review') { - const initBranch = initiative.branch; - const phBranch = phaseBranchName(initBranch, phase.name); - const tBranch = taskBranchName(initBranch, task.id); + // Merge task branch into phase branch (only when branches exist) + if (initiative?.branch && task.category !== 'merge' && task.category !== 'review') { + try { + const initBranch = initiative.branch; + const phBranch = phaseBranchName(initBranch, phase.name); + const tBranch = taskBranchName(initBranch, task.id); - // Serialize merges per phase - const lock = this.phaseMergeLocks.get(task.phaseId) ?? Promise.resolve(); - const mergeOp = lock.then(async () => { - await this.mergeTaskIntoPhase(taskId, task.phaseId!, tBranch, phBranch); - }); - this.phaseMergeLocks.set(task.phaseId, mergeOp.catch(() => {})); - await mergeOp; + // Serialize merges per phase + const lock = this.phaseMergeLocks.get(task.phaseId) ?? Promise.resolve(); + const mergeOp = lock.then(async () => { + await this.mergeTaskIntoPhase(taskId, task.phaseId!, tBranch, phBranch); + }); + this.phaseMergeLocks.set(task.phaseId, mergeOp.catch(() => {})); + await mergeOp; + } catch (err) { + log.error({ taskId, err: err instanceof Error ? err.message : String(err) }, 'task merge failed, still checking phase completion'); + } } - // Check if all phase tasks are done + // Check if all phase tasks are done — always, regardless of branch/merge status const phaseTasks = await this.taskRepository.findByPhaseId(task.phaseId); const allDone = phaseTasks.every((t) => t.status === 'completed'); if (allDone) { @@ -233,10 +235,13 @@ export class ExecutionOrchestrator { if (!phase) return; const initiative = await this.initiativeRepository.findById(phase.initiativeId); - if (!initiative?.branch) return; + if (!initiative) return; if (initiative.executionMode === 'yolo') { - await this.mergePhaseIntoInitiative(phaseId); + // Merge phase branch into initiative branch (only when branches exist) + if (initiative.branch) { + await this.mergePhaseIntoInitiative(phaseId); + } await this.phaseDispatchManager.completePhase(phaseId); // Re-queue approved phases (self-healing: survives server restarts that wipe in-memory queue) diff --git a/apps/server/trpc/routers/task.ts b/apps/server/trpc/routers/task.ts index 1f21074..44f0e95 100644 --- a/apps/server/trpc/routers/task.ts +++ b/apps/server/trpc/routers/task.ts @@ -50,6 +50,14 @@ export function taskProcedures(publicProcedure: ProcedureBuilder) { message: `Task '${input.id}' not found`, }); } + + // Route through dispatchManager when completing — emits task:completed + // event so the orchestrator can check phase completion and merge branches + if (input.status === 'completed' && ctx.dispatchManager) { + await ctx.dispatchManager.completeTask(input.id); + return (await taskRepository.findById(input.id))!; + } + return taskRepository.update(input.id, { status: input.status }); }),