fix: Phase completion check runs regardless of branch/merge status
handleTaskCompleted and handlePhaseAllTasksDone both bailed early when initiative had no branch, silently skipping phase status transitions. Also, merge failures would skip the phase completion check entirely. - Decouple phase completion check from branch existence - Wrap merge in try/catch so phase check runs even if merge fails - Route updateTaskStatus through dispatchManager.completeTask when completing, so the task:completed event fires for orchestration
This commit is contained in:
Lukas May
306
apps/server/execution/orchestrator.test.ts
Normal file
306
apps/server/execution/orchestrator.test.ts
Normal file
@@ -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<string, Function[]>; emitted: DomainEvent[] } {
|
||||
const handlers = new Map<string, Function[]>();
|
||||
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<typeof createMocks>) {
|
||||
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<typeof createMockEventBus>, 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<typeof createMocks>;
|
||||
|
||||
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();
|
||||
});
|
||||
});
|
||||
});
|
||||
@@ -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)
|
||||
|
||||
@@ -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 });
|
||||
}),
|
||||
|
||||
|
||||
Reference in New Issue
Block a user