/**
 * Pure scheduling decisions for the DB-driven flow-runner — NO database, no IO.
 *
 * This is the decision core of the orchestrator's wave scheduler, lifted out of
 * the IO-heavy `flow-runner.ts` so it can be unit-tested in isolation (the repo
 * pattern: `backends/turn-guard.ts`, `backends/lifecycle-decisions.ts`). It is a
 * faithful port of the Phase-1 in-memory scheduler (`conductor/src/flow.ts:27-41`):
 * a step is ready when every dependency is *settled*, and a ready step is skipped
 * when its `when()` guard returns false against the current context.
 *
 * Vocabulary (DB-adapted from flow.ts's `done`/`skipped` sets):
 *  - done      — a step that completed with output (its result is in ctx.results)
 *  - skipped   — a step whose when() guard returned false at ready-time
 *  - inFlight  — an agent step whose task is dispatched but not yet terminal
 *  - excluded  — a step pre-skipped at launch by its when() guard against the run
 *                input (band/repo gating). It never gets a flow_steps row, so it
 *                is tracked here rather than read back from the DB. A dependency
 *                on an excluded step counts as satisfied (the fan-in step still
 *                runs with whatever the live angles produced — flow.ts:39 / the
 *                fold's `if (out)` guard).
 *
 * "Settled" = done ∪ skipped ∪ excluded. Only settled deps unblock a step;
 * an inFlight dep does NOT (the runner waits for its terminal callback).
 */
import type { Flow, Step, StepContext, TriggerRule } from '../conductor/types.js';

export interface SchedulerState {
  /** step ids that completed successfully (results available) */
  readonly done: ReadonlySet<string>;
  /** step ids skipped at ready-time (when() === false with results in hand) */
  readonly skipped: ReadonlySet<string>;
  /** step ids whose agent task is dispatched but not yet terminal */
  readonly inFlight: ReadonlySet<string>;
  /** step ids pre-skipped at launch (band/when gating) — never given a row */
  readonly excluded: ReadonlySet<string>;
  /** step ids that timed out (terminal — no retries remaining or not retriable) */
  readonly timedOut: ReadonlySet<string>;
  /**
   * Per-batch running sets, populated by buildBatchState from the flow definition
   * and the current inFlight set. Only read by getReadyInBatch; never mutated by
   * decision functions (the caller maintains it across ticks).
   */
  readonly batchState?: Map<string, { running: Set<string>; maxConcurrent: number; joinRule: TriggerRule }>;
  /**
   * Per-switch-step routing results. Populated when a SWITCH step completes.
   * Step ids in any result's `excluded` set are treated as excluded for the
   * remainder of the run — they won't execute and won't block dependents.
   */
  readonly switchResults: ReadonlyMap<string, { chosenCase: string | null; excluded: ReadonlySet<string> }>;
  /** Per-DO_WHILE iteration count; presence in the map indicates an active loop */
  readonly loopIterations: ReadonlyMap<string, number>;
}

/** A dependency is satisfied once it is done, skipped, excluded, or timed out.
 *  Dependencies on a running DO_WHILE step are also satisfied so body steps
 *  execute during an active loop iteration. */
function isSatisfied(state: SchedulerState, id: string): boolean {
  const effectiveExcluded = getEffectiveExcluded(state);
  if (state.done.has(id) || state.skipped.has(id) || effectiveExcluded.has(id) || state.timedOut.has(id)) {
    return true;
  }
  // A dependency on a running DO_WHILE step is satisfied (body runs during the loop).
  if (state.loopIterations.has(id) && state.inFlight.has(id)) return true;
  return false;
}

/**
 * The union of the static `excluded` set and every switch result's excluded
 * step ids. Steps excluded by a SWITCH evaluation act exactly like launch-time
 * excluded steps: they never run and they don't block dependents.
 */
function getEffectiveExcluded(state: SchedulerState): ReadonlySet<string> {
  // Fast path: no switch results → static excluded only.
  if (state.switchResults.size === 0) return state.excluded;
  const combined = new Set(state.excluded);
  for (const result of state.switchResults.values()) {
    for (const id of result.excluded) {
      combined.add(id);
    }
  }
  return combined;
}

/**
 * The steps that will appear in this run — every step NOT pre-skipped by its
 * `when()` guard evaluated against the launch context (input only; results are
 * empty at launch). The complement (steps filtered out here) is the `excluded`
 * set. Mirrors the Spine factory's band gating (`spine.ts:71`). Pure.
 */
export function manifestSteps(flow: Flow, launchCtx: StepContext): Step[] {
  return flow.steps.filter((s) => !s.when || s.when(launchCtx));
}

/**
 * Steps whose dependencies are all satisfied and that are not themselves
 * settled, excluded, or in flight — i.e. ready to evaluate/dispatch this tick.
 * Faithful to `conductor/flow.ts:27-36`. Pure.
 */
export function readySteps(flow: Flow, state: SchedulerState): Step[] {
  const effectiveExcluded = getEffectiveExcluded(state);
  return flow.steps.filter(
    (s) =>
      !state.done.has(s.id) &&
      !state.skipped.has(s.id) &&
      !state.inFlight.has(s.id) &&
      !effectiveExcluded.has(s.id) &&
      ((s.deps ?? []).length === 0 || evaluateTriggerRule(s.deps ?? [], state.done, state.skipped, effectiveExcluded, s.trigger_rule)),
  );
}

/**
 * Partition a ready set into the steps to skip (when() === false at ready-time)
 * vs the steps to run, evaluating each guard against the current context
 * (input + completed results). Faithful to `conductor/flow.ts:39`. Pure.
 */
export function partitionReady(
  ready: readonly Step[],
  ctx: StepContext,
): { toRun: Step[]; toSkip: Step[] } {
  const toRun: Step[] = [];
  const toSkip: Step[] = [];
  for (const s of ready) {
    if (s.when && !s.when(ctx)) toSkip.push(s);
    else toRun.push(s);
  }
  return { toRun, toSkip };
}

/** True when every step is settled (done ∪ skipped ∪ excluded). Pure. */
export function isRunComplete(flow: Flow, state: SchedulerState): boolean {
  return flow.steps.every((s) => isSatisfied(state, s.id));
}

/**
 * True when the run can make no further progress yet cannot complete: no ready
 * step, nothing in flight to eventually unblock one, and not complete. Signals a
 * dependency cycle or an unsatisfiable dep (flow.ts:33 throws on this). Pure.
 */
export function isStuck(flow: Flow, state: SchedulerState): boolean {
  return (
    !isRunComplete(flow, state) &&
    state.inFlight.size === 0 &&
    readySteps(flow, state).length === 0
  );
}

/**
 * Build the batchState Map from the flow definition and the current inFlight set.
 * Only steps with a `batch` field are tracked. Empty map when `flow.batchConfig`
 * is absent or no steps belong to a batch. Pure — no IO.
 */
export function buildBatchState(
  flow: Flow,
  inFlight: ReadonlySet<string>,
): Map<string, { running: Set<string>; maxConcurrent: number; joinRule: TriggerRule }> {
  const result = new Map<string, { running: Set<string>; maxConcurrent: number; joinRule: TriggerRule }>();
  if (!flow.batchConfig) return result;

  // Collect every unique batch group referenced by the flow's steps.
  const groups = new Set<string>();
  for (const s of flow.steps) {
    if (s.batch) groups.add(s.batch);
  }

  const { maxConcurrent, joinRule } = flow.batchConfig;
  for (const batch of groups) {
    const running = new Set<string>(
      flow.steps.filter((s) => s.batch === batch && inFlight.has(s.id)).map((s) => s.id),
    );
    result.set(batch, { running, maxConcurrent, joinRule: joinRule ?? 'all_success' });
  }
  return result;
}

/**
 * Gate a ready step list by batch parallelism limits. Steps without a `batch`
 * field always pass through. Steps belonging to a batch are only included if
 * that batch's currently-running count is below its `maxConcurrent` cap.
 *
 * This is ADDITIVE to the existing wave scheduler: pure dep-based readiness
 * is computed first (readySteps), then this function applies the batch ceiling.
 * Steps excluded here remain pending and will be picked up on the next tick
 * when a running batch step completes.
 */
export function getReadyInBatch(ready: readonly Step[], state: SchedulerState, _flow: Flow): Step[] {
  const batchState = state.batchState;
  if (!batchState || batchState.size === 0) return [...ready];
  return ready.filter((s) => {
    if (!s.batch) return true;
    const bs = batchState.get(s.batch);
    if (!bs) return true;
    return bs.running.size < bs.maxConcurrent;
  });
}

/**
 * Per-step action for `initResume`. Pure — no IO; callers supply DB rows.
 *
 * - 'keep':           step is already settled, or has a live pending task that the
 *                     dispatcher's startup poll will run automatically.
 * - 're-dispatch':    step is running but its task died (non-terminal non-pending
 *                     state, or absent) — re-insert a fresh task with the stored
 *                     prompt.
 * - 'mark-done':      task completed before the terminal callback ran; write output
 *                     and advance.
 * - 'mark-failed':    task failed; propagate so advance() fails the run.
 * - 'mark-cancelled': task was cancelled before the callback ran; propagate so
 *                     advance() cancels the run.
 */
/**
 * True when the step definition allows retries on timeout.
 * Pure — no IO.
 */
export function isRetriable(step: { maxRetries?: number }): boolean {
  return (step.maxRetries ?? 0) > 0;
}

/**
 * True when the step has retries remaining.
 * Pure — no IO.
 */
export function shouldRetry(maxRetries: number | undefined | null, retryCount: number): boolean {
  return retryCount < (maxRetries ?? 0);
}

export type ResumeAction =
  | 'keep'
  | 're-dispatch'
  | 'mark-done'
  | 'mark-failed'
  | 'mark-cancelled'
  | 'retry';

/**
 * Decide what to do with ONE flow step during startup resume (D-9). Pure.
 *
 * @param status     - flow_steps.status
 * @param taskId     - flow_steps.task_id (null for code steps or unstarted agent steps)
 * @param taskState  - tasks.state for taskId, or null if the task row is absent
 * @param retryCount - flow_steps.retry_count (default 0)
 * @param maxRetries - flow_steps.max_retries (null = no retry)
 */
export function reconcileResumeStep(
  status: string,
  taskId: string | null,
  taskState: string | null,
  retryCount?: number,
  maxRetries?: number | null,
): ResumeAction {
  if (status === 'timed_out') {
    if (shouldRetry(maxRetries, retryCount ?? 0)) return 'retry';
    return 'mark-failed';
  }
  if (status !== 'running') return 'keep';
  if (!taskId || taskState === null) return 're-dispatch'; // task gone or never created
  switch (taskState) {
    case 'completed': return 'mark-done';
    case 'failed':    return 'mark-failed';
    case 'cancelled': return 'mark-cancelled';
    case 'pending':   return 'keep'; // dispatcher startup poll will run it normally
    default:          return 're-dispatch'; // 'running' or 'blocked' — PTY is dead
  }
}

export interface StepResumeDecision {
  stepId: string;
  action: ResumeAction;
}

/**
 * Returns true when a task whose named agent is unavailable must FAIL HARD
 * rather than fall through to native inference. Orchestrator steps (qwen+plan)
 * are the only case: the PTY `--approval-mode plan` path is the only route
 * that enforces the read-only invariant, so if qwen is missing the step must
 * fail instead of silently gaining write capability. Pure.
 */
export function shouldFailOnMissingAgent(agent: string, modeId: string | null): boolean {
  return agent === 'qwen' && modeId === 'plan';
}

/**
 * Evaluate a SWITCH step: iterate cases in declaration order and return the
 * label of the first matching case plus every step id that belongs to a
 * non-selected branch. When no case matches, the defaultBranch (if present)
 * is the effective choice. If there is no default, all branch steps are
 * excluded and the switch returns `chosenCase: null`.
 *
 * Pure — no IO. The caller adds the returned `excluded` ids to the scheduler
 * state's switchResults so downstream decision functions see them as excluded.
 */
export function resolveSwitch(
  step: Step,
  ctx: StepContext,
): { chosenCase: string | null; excluded: string[] } {
  const cases = step.cases;
  if (!cases || cases.length === 0) {
    // Degenerate switch — nothing to evaluate.
    return { chosenCase: null, excluded: [] };
  }

  // Evaluate conditions in order.
  for (const c of cases) {
    if (c.condition(ctx)) {
      // This case matches — exclude all OTHER branches.
      const excluded: string[] = [];
      for (const other of cases) {
        if (other.label !== c.label) {
          excluded.push(...other.stepIds);
        }
      }
      // The default branch is also excluded when a case matched.
      if (step.defaultBranch) excluded.push(...step.defaultBranch);
      return { chosenCase: c.label, excluded };
    }
  }

  // No case matched — use default branch if present.
  if (step.defaultBranch) {
    // Default is the chosen branch: exclude all explicit case branches.
    const excluded: string[] = [];
    for (const c of cases) {
      excluded.push(...c.stepIds);
    }
    return { chosenCase: null, excluded };
  }

  // No case matched and no default — exclude everything.
  const excluded: string[] = [];
  for (const c of cases) {
    excluded.push(...c.stepIds);
  }
  return { chosenCase: null, excluded };
}

/**
 * Evaluate a trigger rule against dependency results.
 * - all_success: every dep must be done (not skipped/failed)
 * - one_success: at least one dep must be done
 * - all_done: every dep must be settled regardless of outcome
 */
export function evaluateTriggerRule(
  deps: string[],
  done: ReadonlySet<string>,
  skipped: ReadonlySet<string>,
  excluded: ReadonlySet<string>,
  rule?: TriggerRule,
): boolean {
  if (deps.length === 0) return true;
  const satisfied = new Set([...done, ...skipped, ...excluded]);

  switch (rule ?? 'all_success') {
    case 'all_success':
      return deps.every((d) => done.has(d) || skipped.has(d) || excluded.has(d));
    case 'one_success':
      return deps.some((d) => done.has(d));
    case 'all_done':
      return deps.every((d) => satisfied.has(d));
  }
}

/**
 * Reconcile every step of an in-flight run for startup resume. Returns one
 * decision per step. Pure — no IO.
 */
export function reconcileRun(
  steps: ReadonlyArray<{ stepId: string; taskId: string | null; status: string; retryCount?: number; maxRetries?: number | null }>,
  taskStates: ReadonlyMap<string, string>,
): StepResumeDecision[] {
  return steps.map((step) => ({
    stepId: step.stepId,
    action: reconcileResumeStep(
      step.status,
      step.taskId,
      step.taskId ? (taskStates.get(step.taskId) ?? null) : null,
      step.retryCount,
      step.maxRetries,
    ),
  }));
}

/**
 * True when a DO_WHILE loop should stop: the condition returned false or the
 * iteration cap was reached. Pure — no IO.
 *
 * @param step       - the DO_WHILE step definition
 * @param ctx        - current step context (input + accumulated results)
 * @param iterations - number of completed iterations so far
 */
export function isLoopTerminated(step: Step, ctx: StepContext, iterations: number): boolean {
  if (iterations >= (step.loopMaxIterations ?? 100)) return true;
  if (step.loopCondition) return !step.loopCondition(ctx);
  return false;
}
