Jaal vs enterprise orchestrators

const { invoke } = proxyActivities<typeof activities>({ startToCloseTimeout: '30s' });

export async function wf(input: In[]): Promise<Out[]> {
  const output = await Promise.all(input.map((x) => invoke(x)));
  return output;
}

const promises = [];
for (const x of input) {
  promises.push(ctx.serviceClient(invokeService).invoke(x));
}
const output = await RestatePromise.all(promises);

export const wf = inngest.createFunction(
  { id: "wf", triggers: { event: "app/wf.run" } },
  async ({ event, step }) => {
    const input: In[] = event.data.input;
    const output = await Promise.all(
      input.map((x, i) => step.run(`invoke-${i}`, () => invoke(x)))
    );
    return output;
  }
);

// in the coordinator agent's method; Worker is a separate agent type with an invoke method
async fanOut({ input }) {
  const promises = input.map((x, i) => Worker.client.get({ id: i }).invoke({ data: x }));
  return await Promise.all(promises);
}

export function* wf(ctx: Context, input: In[]) {
  const futures = [];
  for (const x of input) futures.push(yield* ctx.beginRun(invoke, x));
  const output = [];
  for (const f of futures) output.push(yield* f);
  return output;
}

const invokeStep = DBOS.registerStep(invoke);

const wf = DBOS.registerWorkflow(async (input: In[]) => {
  const settled = await Promise.allSettled(input.map((x) => invokeStep(x)));
  return settled.map((r) => { if (r.status === "rejected") throw r.reason; return r.value; });
});

export const wf = hatchet.task({
  name: 'wf',
  fn: async (input: { input: In[] }, ctx) => {
    const promises = [];
    for (const x of input.input) promises.push(invoke.run(x));
    const output = await Promise.all(promises);
    return { output };
  },
});

export const wf = task({
  id: "wf",
  run: async (input: In[]) => {
    const results = await invokeTask.batchTriggerAndWait(input.map((x) => ({ payload: x })));
    return results.runs.map((r) => { if (!r.ok) throw r.error; return r.output; });
  },
});

{
  "name": "invoke_fork", "taskReferenceName": "invoke_fork_ref",
  "type": "FORK_JOIN_DYNAMIC",
  "inputParameters": {
    "forkTaskName": "invoke",
    "forkTaskType": "SIMPLE",
    "forkTaskInputs": "${workflow.input.input}"
  }
},
{ "name": "invoke_join", "taskReferenceName": "invoke_join_ref", "type": "JOIN" }

func Wf(ctx workflow.Context, input []In) ([]Out, error) {
	futures := make([]workflow.Future, len(input))
	for i, x := range input {
		futures[i] = workflow.ExecuteActivity(ctx, Invoke, x)
	}
	output := make([]Out, len(input))
	for i, f := range futures {
		if err := f.Get(ctx, &output[i]); err != nil { return nil, err }
	}
	return output, nil
}

const wf: TWorkflow = async function* (ctx: WorkflowContext, input: In[]): any {
  const tasks: Task<any>[] = [];
  for (const x of input) {
    tasks.push(ctx.callActivity(invoke, x));
  }
  const output: Out[] = yield ctx.whenAll(tasks);
  return output;
};

"Invoke All": {
  "Type": "Map",
  "ItemsPath": "$.input",
  "ResultPath": "$.output",
  "ItemProcessor": {
    "ProcessorConfig": { "Mode": "INLINE" },
    "StartAt": "Invoke",
    "States": { "Invoke": {
      "Type": "Task", "Resource": "arn:aws:states:::lambda:invoke",
      "Parameters": { "FunctionName": "<invoke-arn>", "Payload.$": "$" },
      "OutputPath": "$.Payload", "End": true
    } }
  },
  "End": true
}

export const handler = withDurableExecution(
  async (event: { input: In[] }, context: DurableContext): Promise<Out[]> => {
    const result: BatchResult<Out> = await context.map(
      "invoke-all",
      event.input,
      async (ctx, x, i) => ctx.step(`invoke-${i}`, async () => invoke(x)),
    );
    return result.getResults();
  },
);

const wf: OrchestrationHandler = function* (context: OrchestrationContext) {
  const input: In[] = context.df.getInput();
  const tasks = input.map((x) => context.df.callActivity("invoke", x));
  const output: Out[] = yield context.df.Task.all(tasks);
  return output;
};
df.app.orchestration("wf", wf);

export class Wf extends WorkflowEntrypoint<Env, { input: In[] }> {
  async run(event: WorkflowEvent<{ input: In[] }>, step: WorkflowStep) {
    const output = await Promise.all(
      event.payload.input.map((x, i) => step.do(`invoke ${i}`, async () => invoke(x)))
    );
    return output;
  }
}

async function invoke(x: In): Promise<Out> {
  "use step";
  // ...
}

export async function wf(input: In[]) {
  "use workflow";
  const output = await Promise.all(input.map((x) => invoke(x)));
  return output;
}

@task
def invoke(x): ...

@flow
def wf(input: list):
    futures = invoke.map(input)
    output = futures.result()
    return output

@task
def invoke(x): ...

@task
def collect(output):
    return list(output)

collect(invoke.expand(x=input))

const invoke = task("invoke", async (x: In) => {
  // ...
});

const wf = entrypoint({ checkpointer, name: "wf" }, async (input: In[]) => {
  return await Promise.all(input.map(invoke));
});

@ray.remote
def invoke(x): ...

futures = [invoke.remote(x) for x in input]
output = ray.get(futures)

import { proxyActivities } from '@temporalio/workflow';
import type * as activities from './activities';

const { fraudScore, creditCheck, authorize } =
  proxyActivities<typeof activities>({ startToCloseTimeout: '1 minute' });

export async function authorizePayment(order: Order): Promise<Auth> {
  const [risk, limit] = await Promise.all([fraudScore(order), creditCheck(order.customer)]);
  return await authorize(order, risk, limit);
}

import * as restate from "@restatedev/restate-sdk";
import { RestatePromise } from "@restatedev/restate-sdk";

export const payments = restate.service({
  name: "payments",
  handlers: {
    authorize: async (ctx: restate.Context, order: Order) => {
      const riskP = ctx.run("fraud score", () => fraud.score(order));
      const limitP = ctx.run("credit check", () => credit.check(order.customer));
      const [risk, limit] = await RestatePromise.all([riskP, limitP]);
      return ctx.run("authorize", () => paymentsApi.authorize(order, risk, limit));
    },
  },
});

import { Inngest } from "inngest";
const inngest = new Inngest({ id: "payments" });

export const authorizePayment = inngest.createFunction(
  { id: "authorize-payment", triggers: { event: "order/placed" } },
  async ({ event, step }) => {
    const order = event.data.order;
    const [risk, limit] = await Promise.all([
      step.run("fraud-score", () => fraud.score(order)),
      step.run("credit-check", () => credit.check(order.customer)),
    ]);
    return step.run("authorize", () => payments.authorize(order, risk, limit));
  }
);

// inside the PaymentAgent method implementation (defineAgent/.implement setup omitted)
import { Fraud } from './fraud-agent.js';
import { Credit } from './credit-agent.js';
import { Payments } from './payments-agent.js';

const fraud = Fraud.client.get({ orderId: order.id });
const credit = Credit.client.get({ customerId: order.customer });
const [risk, limit] = await Promise.all([
  fraud.score({ order }),
  credit.check({ customer: order.customer }),
]);
return await Payments.client.get({ orderId: order.id }).authorize({ order, risk, limit });

import { Context, Resonate } from "@resonatehq/sdk";
const resonate = new Resonate();

resonate.register("authorizePayment", function* (ctx: Context, order: Order) {
  const riskP = yield* ctx.beginRun(fraudScore, order);
  const limitP = yield* ctx.beginRun(creditCheck, order.customer);
  const risk = yield* riskP;
  const limit = yield* limitP;
  return yield* ctx.run(authorize, order, risk, limit);
});
// fraudScore/creditCheck/authorize are (ctx: Context, ...args) => result functions

import { DBOS } from "@dbos-inc/dbos-sdk";

async function authorizePaymentFn(order: Order) {
  const [r, l] = await Promise.allSettled([
    DBOS.runStep(() => fraud.score(order), { name: "fraudScore" }),
    DBOS.runStep(() => credit.check(order.customer), { name: "creditCheck" }),
  ]);
  if (r.status === "rejected") throw r.reason;
  if (l.status === "rejected") throw l.reason;
  return DBOS.runStep(() => payments.authorize(order, r.value, l.value), { name: "authorize" });
}
export const authorizePayment = DBOS.registerWorkflow(authorizePaymentFn);

import { hatchet } from "./hatchet-client";

export const payment = hatchet.workflow<{ order: Order }>({ name: "authorize-payment" });

const fraud = payment.task({ name: "fraud-score",
  fn: async (input) => ({ risk: await fraudSvc.score(input.order) }) });
const credit = payment.task({ name: "credit-check",
  fn: async (input) => ({ limit: await creditSvc.check(input.order.customer) }) });

payment.task({ name: "authorize", parents: [fraud, credit],
  fn: async (input, ctx) => {
    const { risk } = await ctx.parentOutput(fraud);
    const { limit } = await ctx.parentOutput(credit);
    return { auth: await paymentsSvc.authorize(input.order, risk, limit) };
  } });

import { task, batch } from "@trigger.dev/sdk";
export const fraudScore = task({ id: "fraud-score", run: async (order: Order) => fraud.score(order) });
export const creditCheck = task({ id: "credit-check", run: async (customer: string) => credit.check(customer) });

export const authorizePayment = task({
  id: "authorize-payment",
  run: async (order: Order) => {
    const { runs: [r, l] } = await batch.triggerByTaskAndWait([
      { task: fraudScore, payload: order },
      { task: creditCheck, payload: order.customer },
    ]);
    if (!r.ok || !l.ok) throw new Error("check failed");
    return payments.authorize(order, r.output, l.output);
  },
});

{ "name": "authorize_payment", "version": 1, "schemaVersion": 2, "tasks": [
  { "name": "fork", "taskReferenceName": "fork_ref", "type": "FORK_JOIN", "forkTasks": [
    [ { "name": "fraud_score", "taskReferenceName": "fraud_ref", "type": "SIMPLE",
        "inputParameters": { "order": "${workflow.input.order}" } } ],
    [ { "name": "credit_check", "taskReferenceName": "credit_ref", "type": "SIMPLE",
        "inputParameters": { "customer": "${workflow.input.order.customer}" } } ] ] },
  { "name": "join", "taskReferenceName": "join_ref", "type": "JOIN",
    "joinOn": ["fraud_ref", "credit_ref"] },
  { "name": "authorize", "taskReferenceName": "authorize_ref", "type": "SIMPLE",
    "inputParameters": { "order": "${workflow.input.order}",
      "risk": "${fraud_ref.output.risk}", "limit": "${credit_ref.output.limit}" } } ] }

func AuthorizePayment(ctx workflow.Context, order Order) (Auth, error) {
	ao := workflow.ActivityOptions{ScheduleToStartTimeout: time.Minute, StartToCloseTimeout: time.Minute}
	ctx = workflow.WithActivityOptions(ctx, ao)
	riskF := workflow.ExecuteActivity(ctx, FraudScore, order)
	limitF := workflow.ExecuteActivity(ctx, CreditCheck, order.Customer)
	var risk Risk
	var limit Limit
	if err := riskF.Get(ctx, &risk); err != nil { return Auth{}, err }
	if err := limitF.Get(ctx, &limit); err != nil { return Auth{}, err }
	var auth Auth
	err := workflow.ExecuteActivity(ctx, Authorize, order, risk, limit).Get(ctx, &auth)
	return auth, err
}

import { WorkflowContext, WorkflowActivityContext, TWorkflow } from "@dapr/dapr";

const fraudScore = async (_: WorkflowActivityContext, order: Order) => fraud.score(order);
const creditCheck = async (_: WorkflowActivityContext, customer: string) => credit.check(customer);
const authorize = async (_: WorkflowActivityContext, a: AuthInput) => payments.authorize(a.order, a.risk, a.limit);

const authorizePayment: TWorkflow = async function* (ctx: WorkflowContext, order: Order): any {
  const tasks = [ctx.callActivity(fraudScore, order), ctx.callActivity(creditCheck, order.customer)];
  const [risk, limit] = yield ctx.whenAll(tasks);
  return yield ctx.callActivity(authorize, { order, risk, limit });
};
// plus: new WorkflowRuntime(...).registerWorkflow(authorizePayment).registerActivity(...) x3

{ "StartAt": "Checks", "States": {
  "Checks": { "Type": "Parallel", "ResultPath": "$.checks", "Next": "Authorize",
    "Branches": [
      { "StartAt": "FraudScore", "States": { "FraudScore": { "Type": "Task",
          "Resource": "arn:aws:lambda:REGION:ACCOUNT:function:FraudScore", "End": true } } },
      { "StartAt": "CreditCheck", "States": { "CreditCheck": { "Type": "Task", "InputPath": "$.order.customer",
          "Resource": "arn:aws:lambda:REGION:ACCOUNT:function:CreditCheck", "End": true } } } ] },
  "Authorize": { "Type": "Task", "End": true,
    "Resource": "arn:aws:lambda:REGION:ACCOUNT:function:Authorize",
    "Parameters": { "order.$": "$.order", "risk.$": "$.checks[0]", "limit.$": "$.checks[1]" } } } }

import { withDurableExecution, DurableContext } from "@aws/durable-execution-sdk-js";

export const handler = withDurableExecution(async (event: { order: Order }, context: DurableContext) => {
  const { order } = event;
  const checks = await context.parallel("checks", [
    async (ctx) => ctx.step("fraud-score", async () => fraud.score(order)),
    async (ctx) => ctx.step("credit-check", async () => credit.check(order.customer)),
  ]);
  checks.throwIfError();
  const [risk, limit] = checks.getResults() as [Risk, Limit];
  return context.step("authorize", async () => payments.authorize(order, risk, limit));
});

import * as df from "durable-functions";

df.app.orchestration("authorizePayment", function* (context) {
  const order: Order = context.df.getInput();
  const [risk, limit] = yield context.df.Task.all([
    context.df.callActivity("fraudScore", order),
    context.df.callActivity("creditCheck", order.customer),
  ]);
  return yield context.df.callActivity("authorize", { order, risk, limit });
});
df.app.activity("fraudScore", { handler: (order: Order) => fraud.score(order) });
df.app.activity("creditCheck", { handler: (customer: string) => credit.check(customer) });
df.app.activity("authorize", { handler: (a: AuthInput) => payments.authorize(a.order, a.risk, a.limit) });

import { WorkflowEntrypoint, WorkflowEvent, WorkflowStep } from "cloudflare:workers";

export class AuthorizePayment extends WorkflowEntrypoint<Env, { order: Order }> {
  async run(event: WorkflowEvent<{ order: Order }>, step: WorkflowStep) {
    const { order } = event.payload;
    const [risk, limit] = await Promise.all([
      step.do("fraud score", async () => fraud.score(order)),
      step.do("credit check", async () => credit.check(order.customer)),
    ]);
    return step.do("authorize", async () => payments.authorize(order, risk, limit));
  }
}

async function fraudScore(order: Order) { "use step"; return fraud.score(order); }
async function creditCheck(customer: string) { "use step"; return credit.check(customer); }
async function authorize(order: Order, risk: Risk, limit: Limit) {
  "use step"; return payments.authorize(order, risk, limit);
}

export async function authorizePayment(order: Order) {
  "use workflow";
  const [risk, limit] = await Promise.all([fraudScore(order), creditCheck(order.customer)]);
  return authorize(order, risk, limit);
}

from prefect import flow, task

@task
def fraud_score(order): return fraud.score(order)
@task
def credit_check(customer): return credit.check(customer)
@task
def authorize(order, risk, limit): return payments.authorize(order, risk, limit)

@flow
def authorize_payment(order):
    risk = fraud_score.submit(order)
    limit = credit_check.submit(order["customer"])
    return authorize.submit(order, risk, limit).result()

from airflow.sdk import dag, task

@dag(schedule=None, params={"order": {}})
def authorize_payment():
    @task
    def fraud_score(params=None): return fraud.score(params["order"])
    @task
    def credit_check(params=None): return credit.check(params["order"]["customer"])
    @task
    def authorize(risk, limit, params=None): return payments.authorize(params["order"], risk, limit)

    authorize(fraud_score(), credit_check())

authorize_payment()

import { StateGraph, Annotation, START, END } from "@langchain/langgraph";

const State = Annotation.Root({
  order: Annotation<Order>, risk: Annotation<Risk>, limit: Annotation<Limit>, auth: Annotation<Auth>,
});

export const graph = new StateGraph(State)
  .addNode("fraud", async (s) => ({ risk: await fraud.score(s.order) }))
  .addNode("credit", async (s) => ({ limit: await credit.check(s.order.customer) }))
  .addNode("authorize", async (s) => ({ auth: await payments.authorize(s.order, s.risk, s.limit) }))
  .addEdge(START, "fraud").addEdge(START, "credit")
  .addEdge("fraud", "authorize").addEdge("credit", "authorize")
  .addEdge("authorize", END)
  .compile();

import ray

@ray.remote
def fraud_score(order): return fraud.score(order)
@ray.remote
def credit_check(customer): return credit.check(customer)
@ray.remote
def authorize(order, risk, limit): return payments.authorize(order, risk, limit)

risk_ref = fraud_score.remote(order)
limit_ref = credit_check.remote(order["customer"])
auth = ray.get(authorize.remote(order, risk_ref, limit_ref))