feat(conductor): Wave 2 — parallel batch execution + SWITCH branching step
- Parallel batch execution: batch field on Step, batchConfig on Flow, batch-aware readySteps with maxConcurrent gating, getReadyInBatch helper - SWITCH branching step: new 'switch' StepKind with cases/programmed conditions, resolveSwitch() pure function, switch-excluded steps tracked in SchedulerState, non-selected branches excluded from execution
This commit is contained in:
@@ -1,16 +1,20 @@
|
||||
import { describe, it, expect } from 'vitest';
|
||||
import type { Flow, Step, StepContext } from '../../conductor/types.js';
|
||||
import {
|
||||
buildBatchState,
|
||||
getReadyInBatch,
|
||||
manifestSteps,
|
||||
readySteps,
|
||||
partitionReady,
|
||||
readySteps,
|
||||
isRunComplete,
|
||||
isStuck,
|
||||
reconcileResumeStep,
|
||||
reconcileRun,
|
||||
resolveSwitch,
|
||||
shouldFailOnMissingAgent,
|
||||
type SchedulerState,
|
||||
} from '../flow-runner-decisions.js';
|
||||
import type { StepContext } from '../../conductor/types.js';
|
||||
|
||||
/**
|
||||
* The DB-driven flow-runner replaces the Phase-1 in-memory wave scheduler
|
||||
@@ -53,6 +57,7 @@ const emptyState = (over: Partial<SchedulerState> = {}): SchedulerState => ({
|
||||
inFlight: new Set(),
|
||||
excluded: new Set(),
|
||||
timedOut: new Set(),
|
||||
switchResults: new Map(),
|
||||
...over,
|
||||
});
|
||||
|
||||
@@ -238,6 +243,442 @@ describe('isRunComplete / isStuck', () => {
|
||||
});
|
||||
});
|
||||
|
||||
// ─── SWITCH branching (v2.9) ─────────────────────────────────────────────────
|
||||
|
||||
describe('resolveSwitch', () => {
|
||||
const baseCtx: StepContext = { input: { question: 'q', band: 'small' }, results: {} };
|
||||
|
||||
it('selects the first matching case and excludes other branches', () => {
|
||||
const step: Step = {
|
||||
id: 'router',
|
||||
kind: 'switch',
|
||||
run: () => '',
|
||||
cases: [
|
||||
{ label: 'a', condition: () => false, stepIds: ['a1', 'a2'] },
|
||||
{ label: 'b', condition: () => true, stepIds: ['b1', 'b2'] },
|
||||
{ label: 'c', condition: () => true, stepIds: ['c1', 'c2'] },
|
||||
],
|
||||
};
|
||||
const result = resolveSwitch(step, baseCtx);
|
||||
expect(result.chosenCase).toBe('b');
|
||||
expect(result.excluded).toEqual(['a1', 'a2', 'c1', 'c2']);
|
||||
});
|
||||
|
||||
it('falls back to defaultBranch when no case matches', () => {
|
||||
const step: Step = {
|
||||
id: 'router',
|
||||
kind: 'switch',
|
||||
run: () => '',
|
||||
cases: [
|
||||
{ label: 'x', condition: () => false, stepIds: ['x1'] },
|
||||
{ label: 'y', condition: () => false, stepIds: ['y1'] },
|
||||
],
|
||||
defaultBranch: ['z1', 'z2'],
|
||||
};
|
||||
const result = resolveSwitch(step, baseCtx);
|
||||
expect(result.chosenCase).toBeNull();
|
||||
// Only case branch steps are excluded; default steps are not.
|
||||
expect(result.excluded).toEqual(['x1', 'y1']);
|
||||
});
|
||||
|
||||
it('excludes all branch steps when no case matches and no default', () => {
|
||||
const step: Step = {
|
||||
id: 'router',
|
||||
kind: 'switch',
|
||||
run: () => '',
|
||||
cases: [
|
||||
{ label: 'p', condition: () => false, stepIds: ['p1'] },
|
||||
{ label: 'q', condition: () => false, stepIds: ['q1', 'q2'] },
|
||||
],
|
||||
};
|
||||
const result = resolveSwitch(step, baseCtx);
|
||||
expect(result.chosenCase).toBeNull();
|
||||
expect(result.excluded).toEqual(['p1', 'q1', 'q2']);
|
||||
});
|
||||
|
||||
it('excludes defaultBranch when a case matched', () => {
|
||||
const step: Step = {
|
||||
id: 'router',
|
||||
kind: 'switch',
|
||||
run: () => '',
|
||||
cases: [
|
||||
{ label: 'hit', condition: () => true, stepIds: ['h1'] },
|
||||
{ label: 'miss', condition: () => false, stepIds: ['m1'] },
|
||||
],
|
||||
defaultBranch: ['d1'],
|
||||
};
|
||||
const result = resolveSwitch(step, baseCtx);
|
||||
expect(result.chosenCase).toBe('hit');
|
||||
expect(result.excluded).toEqual(['m1', 'd1']);
|
||||
});
|
||||
|
||||
it('returns empty excluded for a degenerate switch with no cases and no default', () => {
|
||||
const step: Step = {
|
||||
id: 'noop',
|
||||
kind: 'switch',
|
||||
run: () => '',
|
||||
};
|
||||
const result = resolveSwitch(step, baseCtx);
|
||||
expect(result.chosenCase).toBeNull();
|
||||
expect(result.excluded).toEqual([]);
|
||||
});
|
||||
|
||||
it('uses ctx.results in condition evaluation', () => {
|
||||
const step: Step = {
|
||||
id: 'router',
|
||||
kind: 'switch',
|
||||
run: () => '',
|
||||
cases: [
|
||||
{ label: 'has', condition: (ctx) => ctx.results['prev'] === 'yes', stepIds: ['yes-branch'] },
|
||||
{ label: 'no', condition: () => true, stepIds: ['no-branch'] },
|
||||
],
|
||||
};
|
||||
const ctxWithResult: StepContext = { input: { question: 'q', band: 'small' }, results: { prev: 'yes' } };
|
||||
const result = resolveSwitch(step, ctxWithResult);
|
||||
expect(result.chosenCase).toBe('has');
|
||||
expect(result.excluded).toEqual(['no-branch']);
|
||||
});
|
||||
});
|
||||
|
||||
describe('readySteps with switch-excluded steps', () => {
|
||||
// Flow: switch router → branch-a/branch-b → fold
|
||||
function switchFlow(): Flow {
|
||||
const steps: Step[] = [
|
||||
{
|
||||
id: 'switch', kind: 'switch', run: () => '',
|
||||
cases: [
|
||||
{ label: 'a', condition: () => true, stepIds: ['branch-a'] },
|
||||
{ label: 'b', condition: () => false, stepIds: ['branch-b'] },
|
||||
],
|
||||
},
|
||||
{ id: 'branch-a', kind: 'agent', agent: 'x', deps: ['switch'], run: () => 'p' },
|
||||
{ id: 'branch-b', kind: 'agent', agent: 'y', deps: ['switch'], run: () => 'q' },
|
||||
{ id: 'fold', kind: 'code', deps: ['branch-a', 'branch-b'], run: () => 'r' },
|
||||
];
|
||||
return { name: 'switch-demo', description: '', steps, render: () => '' };
|
||||
}
|
||||
|
||||
it('excludes non-selected branch steps and treats them as satisfied deps', () => {
|
||||
const flow = switchFlow();
|
||||
// switch completed, branch-b excluded by switch (branch-a selected)
|
||||
const switchResult = new Map<string, { chosenCase: string | null; excluded: Set<string> }>([
|
||||
['switch', { chosenCase: 'a', excluded: new Set(['branch-b']) }],
|
||||
]);
|
||||
const state: SchedulerState = {
|
||||
done: new Set(['switch']),
|
||||
skipped: new Set(),
|
||||
inFlight: new Set(),
|
||||
excluded: new Set(),
|
||||
timedOut: new Set(),
|
||||
switchResults: switchResult,
|
||||
};
|
||||
const ready = readySteps(flow, state).map((s) => s.id);
|
||||
// branch-a is ready (dep switch is done), branch-b is excluded
|
||||
expect(ready).toContain('branch-a');
|
||||
expect(ready).not.toContain('branch-b');
|
||||
});
|
||||
|
||||
it('fold unblocks once selected branch completes (excluded branch satisfied)', () => {
|
||||
const flow = switchFlow();
|
||||
const switchResult = new Map<string, { chosenCase: string | null; excluded: Set<string> }>([
|
||||
['switch', { chosenCase: 'a', excluded: new Set(['branch-b']) }],
|
||||
]);
|
||||
const state: SchedulerState = {
|
||||
done: new Set(['switch', 'branch-a']),
|
||||
skipped: new Set(),
|
||||
inFlight: new Set(),
|
||||
excluded: new Set(),
|
||||
timedOut: new Set(),
|
||||
switchResults: switchResult,
|
||||
};
|
||||
const ready = readySteps(flow, state).map((s) => s.id);
|
||||
// fold's deps: branch-a done, branch-b excluded (via switch) → satisfied
|
||||
expect(ready).toContain('fold');
|
||||
});
|
||||
|
||||
it('fold stays blocked until selected branch completes, even with excluded dep', () => {
|
||||
const flow = switchFlow();
|
||||
const switchResult = new Map<string, { chosenCase: string | null; excluded: Set<string> }>([
|
||||
['switch', { chosenCase: 'a', excluded: new Set(['branch-b']) }],
|
||||
]);
|
||||
const state: SchedulerState = {
|
||||
done: new Set(['switch']),
|
||||
skipped: new Set(),
|
||||
inFlight: new Set(['branch-a']),
|
||||
excluded: new Set(),
|
||||
timedOut: new Set(),
|
||||
switchResults: switchResult,
|
||||
};
|
||||
const ready = readySteps(flow, state).map((s) => s.id);
|
||||
// branch-a in flight, branch-b excluded — only branch-a offered
|
||||
expect(ready).not.toContain('fold');
|
||||
});
|
||||
|
||||
it('isRunComplete returns true when switch-excluded steps are the only unsettled', () => {
|
||||
const flow = switchFlow();
|
||||
// All non-excluded steps done; branch-b is excluded via switch
|
||||
const switchResult = new Map<string, { chosenCase: string | null; excluded: Set<string> }>([
|
||||
['switch', { chosenCase: 'a', excluded: new Set(['branch-b']) }],
|
||||
]);
|
||||
const state: SchedulerState = {
|
||||
done: new Set(['switch', 'branch-a', 'fold']),
|
||||
skipped: new Set(),
|
||||
inFlight: new Set(),
|
||||
excluded: new Set(),
|
||||
timedOut: new Set(),
|
||||
switchResults: switchResult,
|
||||
};
|
||||
expect(isRunComplete(flow, state)).toBe(true);
|
||||
expect(isStuck(flow, state)).toBe(false);
|
||||
});
|
||||
|
||||
it('combines static excluded with switch-excluded', () => {
|
||||
const flow = switchFlow();
|
||||
// band gating excludes branch-b at launch, AND switch also excludes it
|
||||
const switchResult = new Map<string, { chosenCase: string | null; excluded: Set<string> }>([
|
||||
['switch', { chosenCase: 'a', excluded: new Set(['branch-b']) }],
|
||||
]);
|
||||
const state: SchedulerState = {
|
||||
done: new Set(['switch', 'branch-a']),
|
||||
skipped: new Set(),
|
||||
inFlight: new Set(),
|
||||
excluded: new Set(['branch-b']),
|
||||
timedOut: new Set(),
|
||||
switchResults: switchResult,
|
||||
};
|
||||
// branch-b excluded both ways; fold sees branch-a done, branch-b excluded
|
||||
const ready = readySteps(flow, state).map((s) => s.id);
|
||||
expect(ready).toContain('fold');
|
||||
});
|
||||
});
|
||||
|
||||
// ─── Batch parallelism (v2.8.22) ─────────────────────────────────────────────
|
||||
|
||||
describe('buildBatchState', () => {
|
||||
it('returns empty map when flow has no batchConfig', () => {
|
||||
const flow: Flow = {
|
||||
name: 'no-batch',
|
||||
description: '',
|
||||
steps: [
|
||||
{ id: 'a', kind: 'agent', agent: 'x', run: () => 'p' },
|
||||
{ id: 'b', kind: 'code', deps: ['a'], run: () => 'r' },
|
||||
],
|
||||
render: () => '',
|
||||
};
|
||||
const bs = buildBatchState(flow, new Set());
|
||||
expect(bs.size).toBe(0);
|
||||
});
|
||||
|
||||
it('maps each batch group to its running set and config', () => {
|
||||
const flow: Flow = {
|
||||
name: 'batched',
|
||||
description: '',
|
||||
steps: [
|
||||
{ id: 'a1', kind: 'agent', agent: 'x', batch: 'review', run: () => 'p' },
|
||||
{ id: 'a2', kind: 'agent', agent: 'y', batch: 'review', run: () => 'q' },
|
||||
{ id: 'b1', kind: 'agent', agent: 'z', batch: 'check', run: () => 'r' },
|
||||
{ id: 'fold', kind: 'code', deps: ['a1', 'a2', 'b1'], run: () => 's' },
|
||||
],
|
||||
render: () => '',
|
||||
batchConfig: { maxConcurrent: 2 },
|
||||
};
|
||||
// a1 is in flight → review batch has 1 running, check has 0.
|
||||
const bs = buildBatchState(flow, new Set(['a1']));
|
||||
expect(bs.size).toBe(2);
|
||||
|
||||
const review = bs.get('review');
|
||||
expect(review).toBeDefined();
|
||||
expect([...review!.running]).toEqual(['a1']);
|
||||
expect(review!.maxConcurrent).toBe(2);
|
||||
expect(review!.joinRule).toBe('all_success');
|
||||
|
||||
const check = bs.get('check');
|
||||
expect(check).toBeDefined();
|
||||
expect(check!.running.size).toBe(0);
|
||||
expect(check!.maxConcurrent).toBe(2);
|
||||
});
|
||||
|
||||
it('uses joinRule from batchConfig when provided', () => {
|
||||
const flow: Flow = {
|
||||
name: 'join',
|
||||
description: '',
|
||||
steps: [
|
||||
{ id: 'x', kind: 'agent', agent: 'a', batch: 'g1', run: () => 'p' },
|
||||
],
|
||||
render: () => '',
|
||||
batchConfig: { maxConcurrent: 1, joinRule: 'one_success' },
|
||||
};
|
||||
const bs = buildBatchState(flow, new Set());
|
||||
expect(bs.get('g1')!.joinRule).toBe('one_success');
|
||||
});
|
||||
|
||||
it('ignores steps without a batch field', () => {
|
||||
const flow: Flow = {
|
||||
name: 'mixed',
|
||||
description: '',
|
||||
steps: [
|
||||
{ id: 'a', kind: 'agent', agent: 'x', run: () => 'p' },
|
||||
{ id: 'b', kind: 'agent', agent: 'y', batch: 'g1', run: () => 'q' },
|
||||
],
|
||||
render: () => '',
|
||||
batchConfig: { maxConcurrent: 3 },
|
||||
};
|
||||
const bs = buildBatchState(flow, new Set(['a', 'b']));
|
||||
// a is inFlight but has no batch — it does not create an entry
|
||||
expect(bs.size).toBe(1);
|
||||
expect(bs.has('g1')).toBe(true);
|
||||
expect(bs.get('g1')!.running.has('b')).toBe(true);
|
||||
// a is not in any batch entry
|
||||
for (const entry of bs.values()) {
|
||||
expect(entry.running.has('a')).toBe(false);
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
describe('getReadyInBatch', () => {
|
||||
function makeBatchState(
|
||||
overrides?: Map<string, { running: Set<string>; maxConcurrent: number; joinRule: TriggerRule }>,
|
||||
): Map<string, { running: Set<string>; maxConcurrent: number; joinRule: TriggerRule }> {
|
||||
return overrides ?? new Map();
|
||||
}
|
||||
|
||||
it('passes all steps through when batchState is empty', () => {
|
||||
const steps: Step[] = [
|
||||
{ id: 'a', kind: 'agent', agent: 'x', run: () => 'p' },
|
||||
{ id: 'b', kind: 'agent', agent: 'y', batch: 'g1', run: () => 'q' },
|
||||
];
|
||||
const state: SchedulerState = {
|
||||
done: new Set(),
|
||||
skipped: new Set(),
|
||||
inFlight: new Set(),
|
||||
excluded: new Set(),
|
||||
timedOut: new Set(),
|
||||
switchResults: new Map(),
|
||||
batchState: makeBatchState(),
|
||||
};
|
||||
const result = getReadyInBatch(steps, state, {} as Flow);
|
||||
expect(result.map((s) => s.id)).toEqual(['a', 'b']);
|
||||
});
|
||||
|
||||
it('passes non-batched steps through regardless of batch capacity', () => {
|
||||
const batchState = new Map();
|
||||
batchState.set('g1', { running: new Set(['a']), maxConcurrent: 1, joinRule: 'all_success' });
|
||||
const steps: Step[] = [
|
||||
{ id: 'nobatch', kind: 'agent', agent: 'z', run: () => 'r' },
|
||||
{ id: 'batched', kind: 'agent', agent: 'x', batch: 'g1', run: () => 'p' },
|
||||
];
|
||||
const state: SchedulerState = {
|
||||
done: new Set(),
|
||||
skipped: new Set(),
|
||||
inFlight: new Set(['a']),
|
||||
excluded: new Set(),
|
||||
timedOut: new Set(),
|
||||
switchResults: new Map(),
|
||||
batchState,
|
||||
};
|
||||
const result = getReadyInBatch(steps, state, {} as Flow);
|
||||
// nobatch passes, batched is at maxConcurrent=1 with a already running → blocked
|
||||
expect(result.map((s) => s.id)).toEqual(['nobatch']);
|
||||
});
|
||||
|
||||
it('allows batch steps up to maxConcurrent', () => {
|
||||
const batchState = new Map();
|
||||
batchState.set('g1', { running: new Set(), maxConcurrent: 2, joinRule: 'all_success' });
|
||||
const steps: Step[] = [
|
||||
{ id: 's1', kind: 'agent', agent: 'x', batch: 'g1', run: () => 'p' },
|
||||
{ id: 's2', kind: 'agent', agent: 'y', batch: 'g1', run: () => 'q' },
|
||||
{ id: 's3', kind: 'agent', agent: 'z', batch: 'g1', run: () => 'r' },
|
||||
];
|
||||
const state: SchedulerState = {
|
||||
done: new Set(),
|
||||
skipped: new Set(),
|
||||
inFlight: new Set(),
|
||||
excluded: new Set(),
|
||||
timedOut: new Set(),
|
||||
switchResults: new Map(),
|
||||
batchState,
|
||||
};
|
||||
// All 0 running, maxConcurrent=2 → all 3 pass through (readySteps would return them,
|
||||
// but the flow-runner dispatches them one-by-one in the agent dispatch loop; getReadyInBatch
|
||||
// is called each tick to allow up to maxConcurrent. Since batch is empty on this tick,
|
||||
// all are allowed — the runner's dispatch loop will put 2 in flight, then next tick blocks.)
|
||||
const result = getReadyInBatch(steps, state, {} as Flow);
|
||||
expect(result.map((s) => s.id)).toEqual(['s1', 's2', 's3']);
|
||||
});
|
||||
|
||||
it('blocks batch steps when at capacity', () => {
|
||||
const batchState = new Map();
|
||||
batchState.set('g1', { running: new Set(['a', 'b']), maxConcurrent: 2, joinRule: 'all_success' });
|
||||
const steps: Step[] = [
|
||||
{ id: 'c', kind: 'agent', agent: 'x', batch: 'g1', run: () => 'p' },
|
||||
{ id: 'd', kind: 'agent', agent: 'y', batch: 'g1', run: () => 'q' },
|
||||
];
|
||||
const state: SchedulerState = {
|
||||
done: new Set(),
|
||||
skipped: new Set(),
|
||||
inFlight: new Set(['a', 'b']),
|
||||
excluded: new Set(),
|
||||
timedOut: new Set(),
|
||||
switchResults: new Map(),
|
||||
batchState,
|
||||
};
|
||||
// Both batches at capacity → everything filtered out
|
||||
expect(getReadyInBatch(steps, state, {} as Flow)).toEqual([]);
|
||||
});
|
||||
|
||||
it('handles multiple independent batch groups', () => {
|
||||
const batchState = new Map();
|
||||
batchState.set('g1', { running: new Set(['a']), maxConcurrent: 1, joinRule: 'all_success' });
|
||||
batchState.set('g2', { running: new Set(), maxConcurrent: 5, joinRule: 'all_success' });
|
||||
const steps: Step[] = [
|
||||
{ id: 'b', kind: 'agent', agent: 'x', batch: 'g1', run: () => 'p' }, // g1 at capacity → blocked
|
||||
{ id: 'c', kind: 'agent', agent: 'y', batch: 'g2', run: () => 'q' }, // g2 has room → passes
|
||||
{ id: 'd', kind: 'agent', agent: 'z', batch: 'g2', run: () => 'r' }, // g2 has room → passes
|
||||
];
|
||||
const state: SchedulerState = {
|
||||
done: new Set(),
|
||||
skipped: new Set(),
|
||||
inFlight: new Set(['a']),
|
||||
excluded: new Set(),
|
||||
timedOut: new Set(),
|
||||
switchResults: new Map(),
|
||||
batchState,
|
||||
};
|
||||
expect(getReadyInBatch(steps, state, {} as Flow).map((s) => s.id)).toEqual(['c', 'd']);
|
||||
});
|
||||
|
||||
it('lets a step pass when its batch group is known but has no running steps yet', () => {
|
||||
const batchState = new Map();
|
||||
batchState.set('g1', { running: new Set(), maxConcurrent: 2, joinRule: 'all_success' });
|
||||
const steps: Step[] = [
|
||||
{ id: 'first', kind: 'agent', agent: 'x', batch: 'g1', run: () => 'p' },
|
||||
];
|
||||
const state: SchedulerState = {
|
||||
done: new Set(),
|
||||
skipped: new Set(),
|
||||
inFlight: new Set(),
|
||||
excluded: new Set(),
|
||||
timedOut: new Set(),
|
||||
switchResults: new Map(),
|
||||
batchState,
|
||||
};
|
||||
expect(getReadyInBatch(steps, state, {} as Flow).map((s) => s.id)).toEqual(['first']);
|
||||
});
|
||||
|
||||
it('handles empty step list gracefully', () => {
|
||||
const state: SchedulerState = {
|
||||
done: new Set(),
|
||||
skipped: new Set(),
|
||||
inFlight: new Set(),
|
||||
excluded: new Set(),
|
||||
timedOut: new Set(),
|
||||
switchResults: new Map(),
|
||||
batchState: makeBatchState(),
|
||||
};
|
||||
expect(getReadyInBatch([], state, {} as Flow)).toEqual([]);
|
||||
});
|
||||
});
|
||||
|
||||
// ─── Resume reconciliation (D-9) ─────────────────────────────────────────────
|
||||
|
||||
describe('reconcileResumeStep', () => {
|
||||
|
||||
Reference in New Issue
Block a user