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

apps/coder/src/services/flow-runner.ts

@@ -40,11 +40,14 @@ import { getFlow } from '../conductor/flows/index.js';
 import { loadPersona } from '../conductor/persona-loader.js';
 import type { Band, DispatchFn, Flow, FlowInput, Step, StepContext } from '../conductor/types.js';
 import {
+  buildBatchState,
+  getReadyInBatch,
   isRunComplete,
   manifestSteps,
   partitionReady,
   readySteps,
   reconcileRun,
+  resolveSwitch,
   type SchedulerState,
   type StepResumeDecision,
 } from './flow-runner-decisions.js';
@@ -95,7 +98,7 @@ interface Deps {
 
 interface FlowStepRow {
   step_id: string;
-  kind: 'agent' | 'code';
+  kind: 'agent' | 'code' | 'switch';
   agent: string | null;
   status: string;
   chat_id: string | null;
@@ -280,6 +283,8 @@ export function createFlowRunner(deps: Deps): FlowRunner {
     const skipped = new Set<string>();
     const inFlight = new Set<string>();
     const timedOut = new Set<string>();
+    /** Per-switch routing results — maps switch step id → resolved branch details */
+    const switchExcluded = new Map<string, { chosenCase: string | null; excluded: Set<string> }>();
     const results: Record<string, string> = {};
     for (const r of rows) {
       switch (r.status) {
@@ -311,6 +316,8 @@ export function createFlowRunner(deps: Deps): FlowRunner {
     // ─── Timeout detection ───────────────────────────────────────────────────────
     // Check running steps. If a step has been 'running' longer than
     // FLOW_STEP_TIMEOUT_MS, mark it timed_out or re-dispatch if retriable.
+    // Build a context here so the timeout retry path can re-dispatch the step.
+    const timeoutCtx = buildCtx(input, results, model, dispatch);
     const timeoutMs = config.FLOW_STEP_TIMEOUT_MS;
     const nowDate = new Date();
     let detectedTimedOut = false;
@@ -341,7 +348,7 @@ export function createFlowRunner(deps: Deps): FlowRunner {
           SET retry_count = ${retryCount + 1}, updated_at = clock_timestamp()
           WHERE run_id = ${runId} AND step_id = ${r.step_id} AND status = 'running'
         `;
-        await dispatchAgentStep(runId, run.project_id, model, step, ctx);
+        await dispatchAgentStep(runId, run.project_id, model, step, timeoutCtx);
         inFlight.add(r.step_id);
         log.warn({ runId, stepId: r.step_id, retry: retryCount + 1, maxRetries },
           'flow-runner: step timed out, retrying');
@@ -369,14 +376,16 @@ export function createFlowRunner(deps: Deps): FlowRunner {
     // Drain ready skips + code steps (synchronous), re-evaluating after each batch,
     // then dispatch the full ready agent wave and wait for their terminal callbacks.
     for (;;) {
-      const state: SchedulerState = { done, skipped, inFlight, excluded, timedOut };
+      // Build per-batch state from the current inFlight set for batch parallelism gating.
+      const batchState = buildBatchState(flow, inFlight);
+      const state: SchedulerState = { done, skipped, inFlight, excluded, timedOut, batchState, switchResults: switchExcluded };
 
       if (isRunComplete(flow, state)) {
         await finishRun(runId, flow, input, results, model, dispatch);
         return;
       }
 
-      const ready = readySteps(flow, state);
+      const ready = getReadyInBatch(readySteps(flow, state), state, flow);
       if (ready.length === 0) {
         if (inFlight.size > 0) return; // agents in flight will re-enter via the hook
         await failRun(runId, flow, input, model, 'unsatisfiable dependencies / cycle');
@@ -395,6 +404,31 @@ export function createFlowRunner(deps: Deps): FlowRunner {
         continue; // re-evaluate — a skip can settle a fan-in step's deps
       }
 
+      // SWITCH steps run synchronously — evaluate conditions, update the excluded
+      // set in SchedulerState, and mark themselves complete. Non-selected branch
+      // step ids are excluded from ever running.
+      const switchReady = toRun.filter((s) => s.kind === 'switch');
+      if (switchReady.length > 0) {
+        for (const s of switchReady) {
+          let result: { chosenCase: string | null; excluded: string[] };
+          try {
+            result = resolveSwitch(s, buildCtx(input, results, model, dispatch));
+          } catch (err) {
+            await failRun(runId, flow, input, model, `switch step '${s.id}' threw: ${errMsg(err)}`, s.id);
+            return;
+          }
+          switchExcluded.set(s.id, {
+            chosenCase: result.chosenCase,
+            excluded: new Set(result.excluded),
+          });
+          const outputText = result.chosenCase ? `branch:${result.chosenCase}` : '';
+          await markStep(runId, s.id, 'completed', outputText);
+          results[s.id] = outputText;
+          done.add(s.id);
+        }
+        continue; // re-evaluate — excluded steps may unblock dependents
+      }
+
       const codeReady = toRun.filter((s) => s.kind === 'code');
       if (codeReady.length > 0) {
         for (const s of codeReady) {