Programs

A webhook handler, a chat bot or a release script reaches a workflow's runs through agentswf/client, in its own process: no awf command per call, no JSON to build, and tsc checks every call against the workflow you import. Client API lists every call, what it returns and what it throws.

import { client } from "agentswf/client";
import review from "./pr-review";

const runs = client();
await runs.emit(review, "pr-7", "pr.pushed", { sha: "abc123", branch: "fix-login" });

tsc refuses an event name the workflow doesn't declare, data its schema refuses, and args its prepare doesn't return.

Where the program runs

The program is a client of the run's files on this machine. A run it starts or wakes runs as every event run does: in a process of its own, in the workflow's project, after the program has moved on or exited. So the program runs where you would type awf:

On the machine and as the user whose runs these are, with the same HOME, and a PATH that holds what Install lists for your harnesses. Not in a serverless function.
With Bun. The client is Bun-only, as awf is.
In a project whose agentswf is installed: that is the engine every run of the project runs on, and the client refuses a project at another version than its own, or with none installed.

The workflow gives file: import.meta.path, by which the client finds its project, and is its file's default export, as awf loads it. A workflow without file doesn't compile as an argument; one its file doesn't export as its default throws a ClientError with code file.

A webhook

The workflow reviews a pull request at each push, posts each review to a thread, and returns when the pull request closes:

export default defineWorkflow({
  meta: {
    name: "pr-review",
    description: "Reviews a pull request at each push, until it closes.",
  },
  file: import.meta.path,
  events: {
    "pr.pushed": PUSHED,
    "pr.closed": CLOSED,
    "thread:reply": ParticipantReply,
  },
  starts: ["pr.pushed"],
  async run(wf) {
    const started = wf.started;
    if (started?.name !== "pr.pushed")
      return wf.stop("started by hand: send pr.pushed");
    const thread = await wf.participants.connect({ key: "thread" });
    const reviewer = await wf.agents.open({
      key: "reviewer",
      runtime: "claude",
    });
    const review = async (prompt: string) => {
      const { outcome } = await reviewer.run<Type.Static<typeof REVIEW>>({
        prompt,
        schema: REVIEW,
      });
      if (outcome.answered) await thread.post(outcome.value.summary);
    };
    // The first wait takes the push that started the run, then each event after it.
    for await (const next of wf.select(
      {
        pushed: wf.signal("pr.pushed"),
        closed: wf.signal("pr.closed"),
        reply: wf.signal("thread:reply"),
      },
      { label: `reviewing ${started.data.branch}` },
    )) {
      if (next.closed) return { merged: next.closed.data.merged };
      if (next.pushed)
        await review(`Review ${next.pushed.data.sha}: what must change?`);
      if (next.reply) await review(`The thread says: ${next.reply.data.text}`);
    }
    return { merged: false };
  },
});

The handler takes the client as a parameter, so its test can pass a fake:

/** A GitHub push or close, as the webhook handler has verified and parsed it. */
export type GithubEvent =
  | { kind: "push"; delivery: string; pr: number; sha: string; branch: string }
  | { kind: "closed"; delivery: string; pr: number; merged: boolean };

export async function onGithub(runs: Client, event: GithubEvent) {
  // The delivery id is the event's: GitHub resends a delivery, and the resend is a duplicate.
  const options = { eventId: eventIdOf(event.delivery), source: "github" };
  const run = `pr-${event.pr}`;
  return event.kind === "push"
    ? runs.emit(
        review,
        run,
        "pr.pushed",
        { sha: event.sha, branch: event.branch },
        options,
      )
    : runs.emit(review, run, "pr.closed", { merged: event.merged }, options);
}

The first push starts the PR's run; each later one wakes it. emit returns what became of the event, as awf emit --json prints it: accepted, duplicate or refused, and what was done to the run: started, nudged, continued, or none with why. ok says whether the run has the event and needs no one: it is false for a refused event, and for a run not woken because its launch failed (launch-failed) or the operator's turn has come (stuck, stopped, cancelled). When the run takes the event is the workflow's: an event its wait doesn't take yet, nothing-due, is ok. A launch that failed is a result, not a throw: the event is kept, and the next event, the same event resent, or awf tick tries again.

eventId names the occurrence, so a resent delivery is a duplicate and never reviewed twice. An event id is lowercase letters, digits, ., _, @ and -, and never starts awf.; a foreign id that doesn't fit is hashed:

/**
 * An event id is lowercase letters, digits, `.`, `_`, `@` and `-`, at most 128, and never starts
 * `awf.`, which is awf's own; a foreign id that doesn't fit is hashed: the same id hashes the
 * same, so a resend is still a duplicate.
 */
export function eventIdOf(foreign: string): string {
  if (
    /^[a-z0-9][a-z0-9._@-]{0,127}$/.test(foreign) &&
    !foreign.startsWith("awf.")
  )
    return foreign;
  return `h-${new Bun.CryptoHasher("sha256").update(foreign).digest("hex").slice(0, 40)}`;
}

source names the integration that sent the event; the event a wait returns carries it.

Answer the webhook first. An emit that starts or continues a run waits until its attempt has started, a Bun start and a workflow load, which can take seconds:

if (import.meta.main) {
  const runs = client();
  Bun.serve({
    port: 8787,
    async fetch(request) {
      const event = await verifiedGithubEvent(request);
      if (!event) return new Response("not ours", { status: 401 });
      // Answered first: an emit that starts a run waits for its attempt, which can take seconds,
      // and GitHub gives a delivery ten.
      void onGithub(runs, event).catch((error) =>
        console.error(`${event.delivery}:`, error),
      );
      return new Response(null, { status: 202 });
    },
  });
}

What comes of the emit after the answer can only be logged: a launch that failed is retried by the next event or by awf tick, which needs a schedule, but a refused event is lost, as GitHub won't send it again. Verifying that a request came from GitHub is the program's job, with the webhook's secret; awf trusts whatever the program sends.

Relaying what runs say

/**
 * Posts what the runs said to their Slack threads, each once, and returns the cursor to read on
 * from: kept where a restart finds it, or a restarted program posts everything again.
 */
export async function relayPosts(
  runs: Client,
  cursor: string | undefined,
  post: (ref: string, text: string) => Promise<void>,
): Promise<string | undefined> {
  const items = await runs.inbox(
    review,
    cursor === undefined ? {} : { after: cursor },
  );
  for (const item of items)
    if (item.kind === "post") await post(item.ref, item.text);
  return items.at(-1)?.cursor ?? cursor;
}

/** A person's answer in the thread, back to the run: Slack's message ts names it. */
export async function onThreadReply(
  runs: Client,
  pr: number,
  ts: string,
  text: string,
  author: string,
) {
  return runs.emit(
    review,
    `pr-${pr}`,
    "thread:reply",
    { text, author },
    {
      eventId: eventIdOf(ts),
      source: "slack",
    },
  );
}

inbox lists what the workflow's runs posted and asked, oldest first, as awf inbox --workflow --json does. Each item carries a cursor; pass the last one to the next read, and each item is relayed once, or again if a post in its batch throws. Keep the cursor where a restart finds it, or a restarted program relays everything again. onThreadReply sends a person's answer back as the thread's thread:reply, which the workflow declares. A reply goes to a run that is there and never starts one: to an id with no run, emit throws no-run.

Starting a run and waiting for it

start takes the args prepare would return, and makes sure the run exists and has been launched:

/** Starts the notes for a version, and waits, at most half an hour, until nothing will continue it. */
export async function releaseNotes(runs: Client, version: string) {
  const started = await runs.start(
    release,
    { version, dryRun: false },
    { id: `v${version}` },
  );
  if (!started.ok)
    throw new Error(`not launched: ${started.did} ${started.error ?? ""}`);
  return started.run.wait({
    final: true,
    signal: AbortSignal.timeout(30 * 60_000),
  });
}

ok is true when it started the run or found it exists; false when its launch failed, launch-failed, with the attempt's error, or it is stuck. The run records its args, and every attempt runs with them; prepare never runs for it. start of an id already there with the same args is exists; with other args, or of a run an event or awf run began, it throws other-args. Without id, the workflow's id(args) names the run, or a new id does; a workflow with starts and no id(args) is named by its senders, so start without id throws no-id.

A Run is read from the run's files whenever asked. status() says its state and whether it is final: nothing will continue it on its own. wait() resolves at the run's next rest, when it ends, parks or crashes, or at once if it isn't running; wait({ final: true }) at the first rest that is final. Bound a final wait with a signal: an event run that failed is continued only by its next event or awf tick, and with neither, the wait waits on. get(workflow, id) reaches a run another program started.

A plain run that failed is final. awf run {file} --id {id} --continue continues it, from anywhere: a workflow that gives file keeps its runs in its project.

Testing a program

agentswf/testing's fakeClient() is a Client that keeps the calls it is given, runs the real client's checks that need no files, and answers as the real one's rules do (a resent event id is a duplicate, a run whose launch failed is kept), or as the test scripts:

test("a push is pr.pushed to the PR's run, named by the delivery, from github", async () => {
  const runs = fakeClient();
  const push = {
    kind: "push",
    delivery: "9f1c2e40-1d1a-11ef-8a2b-3c7d9e1f2a4b",
    pr: 7,
    sha: "abc123",
    branch: "fix-login",
  } as const;
  expect(await onGithub(runs, push)).toMatchObject({
    outcome: "accepted",
    did: "started",
  });
  expect(
    await onGithub(runs, {
      kind: "closed",
      delivery: "d-2",
      pr: 7,
      merged: true,
    }),
  ).toMatchObject({ did: "nudged" });
  expect(runs.calls).toEqual([
    {
      kind: "emit",
      workflow: "pr-review",
      ref: "pr-review/pr-7",
      name: "pr.pushed",
      data: { sha: "abc123", branch: "fix-login" },
      eventId: push.delivery,
      source: "github",
    },
    {
      kind: "emit",
      workflow: "pr-review",
      ref: "pr-review/pr-7",
      name: "pr.closed",
      data: { merged: true },
      eventId: "d-2",
      source: "github",
    },
  ]);
});

Catches

A run's questions to a person are answered from a shell. A run started from code has no terminal: what its agents ask shows in inbox() and awf inbox, and a person answers with awf reply. Until then the run waits, as long as the question's timeout allows.
A redeployed workflow needs the program restarted. The program checks events against the workflow it imported; the run's attempt loads the file as it is now, and may refuse data the old schema took.
No stop yet. awf has no command that stops a run, and the client none either.
One engine per project. Update the project's agentswf and the program's together; see Install.

Programs ​

Where the program runs ​

A webhook ​

Relaying what runs say ​

Starting a run and waiting for it ​

Testing a program ​

Catches ​

See also ​