Step-Through Execution

type StepEvent =
  | { event: "run_started";    data: { executionId: string; totalSteps: number; steps: PlanStep[] } }
  | { event: "block_started";  data: { stepId: string } }
  | { event: "block_completed"; data: { stepId: string; output: unknown; durationMs: number } }
  | { event: "step_paused";    data: { executionId: string; completedStepIndex: number; nextStepIndex: number } }
  | { event: "step_progress";  data: { executionId: string; completedStepIndex: number; nextStepIndex: number } }
  | { event: "run_completed";  data: { runId: string; status: "completed" | "failed"; error?: string } };
 
interface PlanStep {
  index: number;
  blocks: { stepId: string; blockName: string; processorType: string }[];
  isParallel: boolean;
  isLoop: boolean;
}
 
interface SessionState {
  executionId: string | null;
  cursor: number;
  accumulatedOutputs: Record<string, unknown>;
  plan: PlanStep[] | null;
}
 
async function* stepOnce(
  flowPath: string,
  state: SessionState,
  body: Record<string, unknown>,
): AsyncGenerator<StepEvent> {
  const res = await fetch(
    `${process.env.NOUKAI_API_URL}/api/v1/seq/${flowPath}/step`,
    {
      method: "POST",
      headers: {
        Authorization: `Bearer ${process.env.NOUKAI_API_KEY}`,
        "Content-Type": "application/json",
        Accept: "text/event-stream",
      },
      body: JSON.stringify({
        executionId: state.executionId,
        stepIndex: state.cursor,
        accumulatedOutputs: state.accumulatedOutputs,
        ...body,
      }),
    },
  );
 
  if (!res.ok) {
    const detail = await res.json().catch(() => ({}));
    throw new Error(`step ${res.status}: ${JSON.stringify(detail)}`);
  }
 
  const reader = res.body!.getReader();
  const decoder = new TextDecoder();
  let buffer = "";
 
  while (true) {
    const { value, done } = await reader.read();
    if (done) break;
    buffer += decoder.decode(value, { stream: true });
 
    let sep: number;
    while ((sep = buffer.indexOf("\n\n")) !== -1) {
      const raw = buffer.slice(0, sep);
      buffer = buffer.slice(sep + 2);
 
      let event = "message";
      let data = "";
      for (const line of raw.split("\n")) {
        if (line.startsWith("event:")) event = line.slice(6).trim();
        else if (line.startsWith("data:")) data += line.slice(5).trim();
      }
      if (data) yield { event, data: JSON.parse(data) } as StepEvent;
    }
  }
}
 
function applyEvent(state: SessionState, ev: StepEvent): SessionState {
  switch (ev.event) {
    case "run_started":
      return { ...state, executionId: ev.data.executionId, plan: ev.data.steps };
    case "block_completed":
      return {
        ...state,
        accumulatedOutputs: { ...state.accumulatedOutputs, [ev.data.stepId]: ev.data.output },
      };
    case "step_paused":
    case "step_progress":
      return { ...state, cursor: ev.data.nextStepIndex };
    default:
      return state;
  }
}
 
// Example: run the flow step-by-step, logging each step's output.
async function runStepByStep(flowPath: string, message: string) {
  let state: SessionState = {
    executionId: null,
    cursor: 0,
    accumulatedOutputs: {},
    plan: null,
  };
 
  // First call carries `message`; later calls don't.
  let body: Record<string, unknown> = { message };
 
  while (true) {
    let done = false;
    for await (const ev of stepOnce(flowPath, state, body)) {
      state = applyEvent(state, ev);
      if (ev.event === "block_completed") {
        console.log(`step ${ev.data.stepId} →`, ev.data.output);
      }
      if (ev.event === "run_completed") {
        if (ev.data.status === "failed") throw new Error(ev.data.error);
        done = true;
      }
    }
    if (done) break;
    body = {}; // subsequent calls send only cursor + outputs
  }
 
  return state.accumulatedOutputs;
}