Planner-Executor-Critic Plus a Shared Workspace: Workspace State, toolBudget, Parallel Fan-Out, a Review Loop

import { Annotation } from '@langchain/langgraph'
 
// D15's version: new writes are appended as-is. Under a parallel fan-out the same id
// leaves both a pending and a done copy
const concatWorkspace = (old, next) => old.concat(next)
 
// Today's replacement: update by id in place, appending only ids never seen
const upsertWorkspace = (old, next) => {
  const merged = old.slice() // a reducer must be pure: copy before changing, never touch the old value
  for (const task of next) {
    const at = merged.findIndex((t) => t.id === task.id)
    if (at === -1) merged.push(task)
    else merged[at] = task
  }
  return merged
}
 
// The merge rule hangs off the field, not the node - so adding a node needs no thought
// about merging with anybody else's writes
export const AgentAnnotation = Annotation.Root({
  workspace: Annotation({ reducer: upsertWorkspace, default: () => [] }),
  reviewRounds: Annotation({ reducer: (_old, next) => next, default: () => 0 }),
})

from typing import Annotated, TypedDict
 
def concat_workspace(old: list[dict], new: list[dict]) -> list[dict]:
    return old + new
 
def upsert_workspace(old: list[dict], new: list[dict]) -> list[dict]:
    # A dict dedupes by key and preserves insertion order, which is exactly upsert
    # semantics with no hand-written lookup
    merged = {task["id"]: task for task in old}
    merged.update({task["id"]: task for task in new})
    return list(merged.values())
 
class AgentState(TypedDict):
    # Annotated's second argument is this channel's merge rule; unannotated means last write wins
    workspace: Annotated[list[dict], upsert_workspace]
    review_rounds: int

// Java has no LangGraph, so this hand-writes "the merge rule hangs off the field"
// idiomatically for JDK 17+. Dependencies: JDK 17+, no third-party library
record Task(String id, String goal, String result, int toolCalls, String status) {}
 
static List<Task> concatWorkspace(List<Task> old, List<Task> next) {
    return Stream.concat(old.stream(), next.stream()).toList();
}
 
// LinkedHashMap preserves insertion order and overwrites the same key, exactly upsert semantics
static List<Task> upsertWorkspace(List<Task> old, List<Task> next) {
    var merged = new LinkedHashMap<String, Task>();
    for (var task : old) merged.put(task.id(), task);
    for (var task : next) merged.put(task.id(), task);
    return List.copyOf(merged.values());
}
 
// A table from field name to merge rule is the thing the framework maintains for you
static final Map<String, BinaryOperator<List<Task>>> REDUCERS =
    Map.of("workspace", Reducers::upsertWorkspace);

struct Task {
    let id: String
    var goal: String
    var result: String?
    var toolCalls: Int
    var status: String
}
 
// Value semantics make a reducer pure by construction: old arrives as a copy, so changing
// it cannot affect the caller - exactly a reducer's hard requirement, which the JS version
// achieves only through a hand-written slice
func concatWorkspace(_ old: [Task], _ next: [Task]) -> [Task] { old + next }
 
func upsertWorkspace(_ old: [Task], _ next: [Task]) -> [Task] {
    var merged = old
    for task in next {
        if let at = merged.firstIndex(where: { $0.id == task.id }) {
            merged[at] = task
        } else {
            merged.append(task)
        }
    }
    return merged
}
 
// The merge rule hangs off the field, not the node
let reducers: [String: ([Task], [Task]) -> [Task]] = ["workspace": upsertWorkspace]

const TOOL_BUDGET = 5 // per subtask: a bounce is billed too, or a graph with a loop has no ceiling
 
export async function runTask(task) {
  const observations = []
  let used = task.toolCalls
  for (const tool of toolsFor(task)) {
    // An exhausted budget does not throw: throwing escalates "half done" into total failure
    if (used >= TOOL_BUDGET) return degrade(task, used, observations, `budget exhausted, stopped at ${tool}`)
    observations.push(await callTool(tool, task))
    used += 1
  }
  return { ...task, toolCalls: used, result: await compose(task, observations), status: 'done' }
}
 
function degrade(task, used, observations, why) {
  const partial = observations.length > 0 ? observations.join('; ') : 'nothing found'
  return { ...task, toolCalls: used, result: `(degraded) ${why}. So far: ${partial}`, status: 'failed' }
}

TOOL_BUDGET = 5  # per subtask: a bounce is billed too
 
async def run_task(task: dict) -> dict:
    observations: list[str] = []
    used = task["toolCalls"]
    for tool in tools_for(task):
        # An exhausted budget does not raise: raising escalates "half done" into total failure
        if used >= TOOL_BUDGET:
            return degrade(task, used, observations, f"budget exhausted, stopped at {tool}")
        observations.append(await call_tool(tool, task))
        used += 1
    result = await compose(task, observations)
    return {**task, "toolCalls": used, "result": result, "status": "done"}
 
def degrade(task: dict, used: int, observations: list[str], why: str) -> dict:
    partial = "; ".join(observations) or "nothing found"
    return {**task, "toolCalls": used, "result": f"(degraded) {why}. So far: {partial}", "status": "failed"}

// Dependencies: JDK 17+, no third-party library. A record is immutable, so every step
// returns a fresh Task, the same discipline as "a node must not mutate state" - here
// guaranteed directly by the type system
static final int TOOL_BUDGET = 5;   // per subtask: a bounce is billed too
 
static Task runTask(Task task) throws Exception {
    var observations = new ArrayList<String>();
    int used = task.toolCalls();
    for (var tool : toolsFor(task)) {
        // Return a degraded result rather than throwing - an exception fails the whole thing
        if (used >= TOOL_BUDGET) return degrade(task, used, observations, "budget exhausted, stopped at " + tool);
        observations.add(callTool(tool, task));
        used += 1;
    }
    return new Task(task.id(), task.goal(), compose(task, observations), used, "done");
}
 
static Task degrade(Task task, int used, List<String> observations, String why) {
    var partial = observations.isEmpty() ? "nothing found" : String.join("; ", observations);
    return new Task(task.id(), task.goal(), "(degraded) " + why + ". So far: " + partial, used, "failed");
}

let toolBudget = 5  // per subtask: a bounce is billed too
 
func runTask(_ task: Task) async throws -> Task {
    var observations: [String] = []
    var updated = task          // a value type, so this changes our own copy
    for tool in toolsFor(task) {
        // Take the degradation exit rather than throwing - a throw propagates to the
        // caller and the whole thing fails
        guard updated.toolCalls < toolBudget else {
            return degrade(updated, observations, "budget exhausted, stopped at \(tool)")
        }
        observations.append(try await callTool(tool, task))
        updated.toolCalls += 1
    }
    updated.result = try await compose(task, observations)
    updated.status = "done"
    return updated
}
 
func degrade(_ task: Task, _ observations: [String], _ why: String) -> Task {
    var degraded = task
    let partial = observations.isEmpty ? "nothing found" : observations.joined(separator: "; ")
    degraded.result = "(degraded) \(why). So far: \(partial)"
    degraded.status = "failed"
    return degraded
}

const CONCURRENCY = 2 // the ceiling comes from your quota, not your machine
 
// Promise.all fires all 20 requests at once, so run a fixed number of workers pulling
// from a queue instead
export async function runAll(tasks) {
  const queue = [...tasks]
  const results = []
  const workers = Array.from({ length: CONCURRENCY }, async () => {
    for (let task = queue.shift(); task; task = queue.shift()) {
      try {
        results.push(await runTask(task))
      } catch (error) {
        // The failure is confined to this one item: the rest deliver as usual
        results.push(degrade(task, task.toolCalls, [], String(error)))
      }
    }
  })
  await Promise.all(workers)
  return results
}

import asyncio
 
CONCURRENCY = 2  # the ceiling comes from your quota, not your machine
 
async def run_all(tasks: list[dict]) -> list[dict]:
    # A Semaphore is Python's idiom for limiting concurrency: gather still submits all,
    # and only 2 run at once
    gate = asyncio.Semaphore(CONCURRENCY)
 
    async def guarded(task: dict) -> dict:
        async with gate:
            try:
                return await run_task(task)
            except Exception as error:  # the failure is confined to this one item
                return degrade(task, task["toolCalls"], [], str(error))
 
    return await asyncio.gather(*(guarded(task) for task in tasks))

// Dependencies: JDK 17+. A fixed-size pool is itself the concurrency gate, with no
// hand-written semaphore
static final int CONCURRENCY = 2;
 
static List<Task> runAll(List<Task> tasks) {
    var pool = Executors.newFixedThreadPool(CONCURRENCY);
    try {
        var futures = tasks.stream()
            .map(task -> CompletableFuture
                .supplyAsync(() -> callUnchecked(task), pool)
                // exceptionally confines the failure to this subtask rather than the batch
                .exceptionally(error -> degrade(task, task.toolCalls(), List.of(), error.getMessage())))
            .toList();
        // allOf is only for waiting; values still come from each future, so the failed
        // one has a degraded result too
        CompletableFuture.allOf(futures.toArray(CompletableFuture[]::new)).join();
        return futures.stream().map(CompletableFuture::join).toList();
    } finally {
        pool.shutdown();
    }
}

let concurrency = 2  // the ceiling comes from your quota, not your machine
 
// A TaskGroup queues every child task by default, so fill one window first and add one
// each time one finishes
func runAll(_ tasks: [Task]) async -> [Task] {
    await withTaskGroup(of: Task.self) { group in
        var index = 0
        var results: [Task] = []
 
        func submit(_ task: Task) {
            group.addTask {
                do { return try await runTask(task) }
                // Turn the error into a degraded result inside the child task, so it
                // never escapes the group
                catch { return degrade(task, [], "\(error)") }
            }
        }
 
        while index < min(concurrency, tasks.count) {
            submit(tasks[index])
            index += 1
        }
        for await finished in group {
            results.append(finished)
            if index < tasks.count {
                submit(tasks[index])
                index += 1
            }
        }
        return results
    }
}

const MAX_REVIEW_ROUNDS = 2 // at most 2 bounces, 3 executions with the first
 
export async function criticNode(state) {
  const tasks = latestById(state.workspace)
  const reviews = await reviewTasks(tasks.filter((t) => t.status === 'done'))
  const rejected = reviews.filter((r) => !r.ok)
  if (rejected.length === 0) return {} // all passed: write nothing, the edge sends it to assembly
 
  const byId = new Map(tasks.map((t) => [t.id, t]))
  if (state.reviewRounds >= MAX_REVIEW_ROUNDS) {
    // Ceiling reached: stop bouncing and return what exists. Bouncing forever means the
    // user never gets a reply
    return {
      degraded: true,
      workspace: rejected.map((r) => ({ ...byId.get(r.id), status: 'failed' })),
    }
  }
  // A bounce writes the reason into the goal: given only "not good enough", the executor
  // resubmits the same thing
  return {
    reviewRounds: state.reviewRounds + 1,
    workspace: rejected.map((r) => ({
      ...byId.get(r.id),
      status: 'pending',
      goal: `${baseGoal(byId.get(r.id).goal)} | review note: ${r.reason}`,
    })),
  }
}

MAX_REVIEW_ROUNDS = 2  # at most 2 bounces, 3 executions with the first
 
async def critic_node(state: dict) -> dict:
    tasks = latest_by_id(state["workspace"])
    reviews = await review_tasks([t for t in tasks if t["status"] == "done"])
    rejected = [r for r in reviews if not r["ok"]]
    if not rejected:
        return {}  # all passed: write nothing, the edge sends it to assembly
 
    by_id = {t["id"]: t for t in tasks}
    if state["review_rounds"] >= MAX_REVIEW_ROUNDS:
        # Ceiling reached: stop bouncing and return what exists
        return {
            "degraded": True,
            "workspace": [{**by_id[r["id"]], "status": "failed"} for r in rejected],
        }
    return {
        "review_rounds": state["review_rounds"] + 1,
        # A bounce writes the reason into the goal so the executor knows what to fix
        "workspace": [
            {
                **by_id[r["id"]],
                "status": "pending",
                "goal": f'{base_goal(by_id[r["id"]]["goal"])} | review note: {r["reason"]}',
            }
            for r in rejected
        ],
    }

// Dependencies: JDK 17+. The returned Map is the delta of fields changed this step
static final int MAX_REVIEW_ROUNDS = 2;
 
static Map<String, Object> criticNode(AgentState state) {
    var tasks = latestById(state.workspace());
    var rejected = reviewTasks(tasks.stream().filter(t -> t.status().equals("done")).toList())
        .stream().filter(r -> !r.ok()).toList();
    if (rejected.isEmpty()) return Map.of();   // all passed: write no fields
 
    var byId = tasks.stream().collect(Collectors.toMap(Task::id, t -> t));
    if (state.reviewRounds() >= MAX_REVIEW_ROUNDS) {
        // Ceiling reached: return what exists rather than bouncing again
        var failed = rejected.stream().map(r -> withStatus(byId.get(r.id()), "failed")).toList();
        return Map.of("degraded", true, "workspace", failed);
    }
    // A bounce writes the reason into the goal so the executor knows what to fix
    var redo = rejected.stream()
        .map(r -> withFeedback(byId.get(r.id()), r.reason()))
        .toList();
    return Map.of("reviewRounds", state.reviewRounds() + 1, "workspace", redo);
}

let maxReviewRounds = 2  // at most 2 bounces, 3 executions with the first
 
// The delta uses Optional to say "this field did not change", more explicit than the
// JS version's absent key
struct Delta {
    var workspace: [Task]?
    var reviewRounds: Int?
    var degraded: Bool?
}
 
func criticNode(_ state: AgentState) async throws -> Delta {
    let tasks = latestById(state.workspace)
    let rejected = try await reviewTasks(tasks.filter { $0.status == "done" }).filter { !$0.ok }
    guard !rejected.isEmpty else { return Delta() }  // all passed: change no field
 
    let byId = Dictionary(uniqueKeysWithValues: tasks.map { ($0.id, $0) })
    guard state.reviewRounds < maxReviewRounds else {
        // Ceiling reached: return what exists rather than bouncing again
        let failed = rejected.compactMap { byId[$0.id] }.map { task -> Task in
            var copy = task
            copy.status = "failed"
            return copy
        }
        return Delta(workspace: failed, degraded: true)
    }
    let redo = rejected.compactMap { review -> Task? in
        guard var task = byId[review.id] else { return nil }
        task.status = "pending"
        // A bounce writes the reason into the goal so the executor knows what to fix
        task.goal = "\(baseGoal(task.goal)) | review note: \(review.reason)"
        return task
    }
    return Delta(workspace: redo, reviewRounds: state.reviewRounds + 1)
}