Durable execution

Use waits, subtasks, handles, commits, state-machine steps, map/reduce, branches, and ledger views without breaking replay.

Constal records runtime operations at stable journal positions. A resumed invocation replays the recorded result, continues the owned operation, or applies its declared recovery contract.

Wait and delegate

import { all } from "@constal/std";

const approval = ctx.await<{ approved: boolean }>("approval", {
  timeout: 86_400_000,
  onTimeout: { approved: false },
  schema: { type: "object", required: ["approved"], properties: { approved: { type: "boolean" } } },
  maxBytes: 1_024,
});

const checks = [
  ctx.spawn(riskCheck, input, { retries: 2 }),
  ctx.spawn(complianceCheck, input, { retries: 2 }),
];

const [decision, [risk, compliance]] = await Promise.all([approval, all(checks)]);

Handles have stable ids and recorded terminal outcomes. Do not catch and suppress runtime suspension, and do not reuse Ctx or a Handle after the invocation yields. all, select, and race are derived joins from @constal/std, not additional Agent primitives.

Map and reduce registered work

import { agent, validateFoldFn, validatePartitionFn } from "@constal/sdk";

const extract = validatePartitionFn<string, { words: number }, never>({
  id: "word-count", version: "1", effects: "pure", batch: { rows: 100 },
  async run(batch, out) {
    for (const text of batch) out.emit({ words: text.trim().split(/\s+/u).length });
  },
});

const total = validateFoldFn<{ words: number }, { words: number }>({
  id: "sum-words", version: "1", deterministic: true, associative: true,
  async run(group, out) {
    let words = 0;
    for await (const row of group.rows) words += row.words;
    out.emit({ words });
  },
});

export default agent({
  id: "counter", version: "1.0.0", model: "model",
  partitionFns: [extract], foldFns: [total],
  async onMessage(rows: string[], ctx) {
    const mapped = await ctx.map(extract, rows, { partition: { rows: 100 } });
    return ctx.reduce(total, mapped, { scope: "global" });
  },
});

Stage functions must be registered and versioned. Large data moves by content reference; failed partitions do not erase earlier facts.

Durable mode

Set mode: "durable" and implement init, step, and output when progress itself must be explicit state:

import { agent } from "@constal/sdk";

type State = { input: unknown; phase: "review" | "done" | "rejected" };

export default agent<State>({
  id: "review-workflow", version: "1.0.0", model: "model", mode: "durable",
  init: (input) => ({ input, phase: "review" }),
  async step(state, ctx) {
    if (state.phase === "review") {
      const decision = await ctx.await<{ approved: boolean }>("approval", {
        schema: { type: "object", required: ["approved"], additionalProperties: false,
          properties: { approved: { type: "boolean" } } },
      });
      return { state: { ...state, phase: decision.approved ? "done" : "rejected" }, done: true };
    }
    return { state, done: true };
  },
  output: (state) => state,
});

Each step() must return serializable state and a done flag. Operators resolve waits and apply pause, resume, interrupt, cancel, Policy, rebind, truncate, or branch through authenticated Platform API events—not Agent SDK calls. See Run operations and The seven Agent primitives.