Chapter 17

Schedule a Ready Set, Not a Queue of Conversations

A DAG scheduler runs only nodes with completed dependencies, limits concurrency, and makes the critical path visible. On each iteration it computes the ready set - pending work items whose dependencies are all completed - and takes no more than the concurrency limit from it. After a batch finishes, new candidates appear. This is simpler and more reliable than asking a supervisor to "track who finished" in a long transcript.

TypeScript

import type { TaskStatus, WorkItem } from "./types.js";

export function readyTasks(
  tasks: readonly WorkItem[],
  status: Readonly<Record<string, TaskStatus>>
): readonly WorkItem[] {
  return tasks.filter((task) =>
    status[task.id] === "pending" &&
    task.dependsOn.every((dependency) => status[dependency] === "completed")
  );
}

export function createInitialStatus(tasks: readonly WorkItem[]): Record<string, TaskStatus> {
  return Object.fromEntries(tasks.map((task) => [task.id, "pending" as const]));
}

export function hasUnfinishedTasks(status: Readonly<Record<string, TaskStatus>>): boolean {
  return Object.values(status).some((value) => value === "pending" || value === "running");
}

The batch itself runs with bounded parallelism, and results are checked against the declared write set before writing.

TypeScript

    const store = new ArtifactStore(handlers.map((handler) => handler.contract));
    let failureReason: string | undefined;

    eventLog.append({ type: "run.started", runId });

    while (hasUnfinishedTasks(status) && !failureReason) {
      const batch = readyTasks(this.#tasks, status).slice(0, this.#maxConcurrency);
      if (batch.length === 0) {
        failureReason = "No runnable task remains";
        break;
      }

      for (const task of batch) {
        status[task.id] = "running";
      }

      const outcomes = await Promise.all(
        batch.map(async (task) => {
          const agentId = routeTask(task);
          const handler = this.#handlers.get(agentId);
          if (!handler) {
            return { task, reason: `No handler for ${agentId}` };
          }
          assertTaskAccepted(handler.contract, task);

          for (let attempt = 1; attempt <= task.maxAttempts; attempt += 1) {
            eventLog.append({ type: "task.started", taskId: task.id, agentId, attempt });
            const openSpan = trace.start(task.id, agentId, attempt);
            try {
              const result = await handler.run({
                runId,
                task,
                inputArtifacts: store.list(),
                attempt
              });
              budget.consume(result.usage);

              for (const draft of result.artifacts) {
                if (!task.writeSet.includes(draft.key)) {
                  throw new Error(`Task ${task.id} produced undeclared artifact ${draft.key}`);
                }
                const artifact = store.write(agentId, draft);

To assess a graph's effective parallelism and see the critical path, use the lab.

Interactive lab 3

Assess effective parallelism

Number of branchesConcurrency limitShared-context shareWrite-overlap risk

Theoretically at once: 3. Effective estimate after coordination penalties: 1.

Both limits are needed. A concurrency limit bounds simultaneously active workers, while a total worker limit bounds the whole run. Otherwise a finished worker can spawn a new fan-out and bypass the concurrency bound across several waves. And count wall-clock correctly: for a parallel batch, latency is the slowest branch plus fan-out/fan-in overhead, while cost is the sum of consumption of all branches.

The schedule exists. But some nodes will fail, and the system is defined by how it classifies a failure before a retry.