Run a graph and reuse its result

import { pathToFileURL } from "node:url";
import { InMemoryDurableTaskStore } from "@agentium/core";
import { ExecutionGraphRuntime, type GraphToolBinding } from "@agentium/harness";
import { z } from "zod";

export async function runCalculation() {
  let dispatches = 0;
  const valueInput = z.object({ value: z.number() }).strict();
  const sumInput = z.object({ left: z.number(), right: z.number() }).strict();
  const reference = z.object({
    $ref: z.object({ node: z.string().min(1), path: z.array(z.string()).length(0) }).strict(),
  }).strict();
  const value: GraphToolBinding = {
    id: "value", version: "1", kind: "tool",
    input: valueInput, output: z.number(), referenceOutputs: [""],
    recovery: "pure", // The host attests that no external mutation occurs.
    tool: {
      name: "value", description: "Read a fixture value", parameters: valueInput,
      execute: async (args) => { dispatches++; return JSON.stringify(valueInput.parse(args).value); },
    },
  };
  const sum: GraphToolBinding = {
    id: "sum", version: "1", kind: "tool",
    input: sumInput, output: z.number(), referenceInputs: ["left", "right"],
    validateTemplate: (template) => {
      z.object({ left: reference, right: reference }).strict().parse(template);
    },
    recovery: "pure",
    tool: {
      name: "sum", description: "Add two fixture values", parameters: sumInput,
      execute: async (args) => {
        dispatches++;
        const input = sumInput.parse(args);
        return JSON.stringify(input.left + input.right);
      },
    },
  };
  const graph = new ExecutionGraphRuntime({
    store: new InMemoryDurableTaskStore(), // Local fixture; cannot survive process loss.
    catalog: [value, sum],
    program: { id: "calculation", version: "1", plan: {
      schemaVersion: 1, revision: 1,
      nodes: [
        { id: "left", binding: "value", version: "1", args: { value: 2 } },
        { id: "right", binding: "value", version: "1", args: { value: 3 } },
        { id: "total", binding: "sum", version: "1", args: {
          left: { $ref: { node: "left", path: [] } },
          right: { $ref: { node: "right", path: [] } },
        } },
      ],
    } },
    limits: { maxConcurrentWork: 2, maxExpandedNodes: 3,
      maxWorkAttempts: 3, maxPlanRevisions: 1 },
    harness: { grants: { toolIds: ["value", "sum"], modelRoles: [] },
      budgets: { maxModelCalls: 0, maxToolCalls: 3, maxTokens: 0 } },
  });
  const identity = { tenantId: "demo", actorId: "reader", sessionId: "calculation",
    runId: "calculation-run", rootRunId: "calculation-run" };
  const key = { tenantId: identity.tenantId, taskId: "calculation-task" };
  await graph.create({ id: key.taskId, input: {}, identity, deadline: Date.now() + 60_000 });
  const first = await graph.run(key, "worker-1");
  const second = await graph.run(key, "worker-2");
  const output = z.object({ left: z.literal(2), right: z.literal(3), total: z.literal(5) }).parse(second.output);
  console.log("status:", first.task.state, "output:", JSON.stringify(output));
  console.log("dispatches after retained result:", dispatches);
  console.log("persistent store:", graph.capabilities.durable);
  return { first, second, dispatches, output, durable: graph.capabilities.durable };
}

if (process.argv[1] && import.meta.url === pathToFileURL(process.argv[1]).href) {
  await runCalculation();
}