step

@alexkroman1/aai/step — the surface a step body is written against.

The reader is a workflows/*.ts module in an agent project. Nothing here is durable on its own: a call becomes a journaled step only when the body puts it inside ctx.step(name, fn), and outside one it runs inline on every replay with no journal and no retry. So the loop is: workflow on the root DECLARES the run and types its input, a workflows/*.ts module holds the body, this subpath is what that body is written against, and useWorkflowRun in @alexkroman1/aai-ui renders it.

What is here is one reader’s whole vocabulary, in the order a pipeline needs it:

Bounded fan-out — mapConcurrent, a WINDOW over a cursor so a slow item costs only itself. Replay-safe at any width; its own doc carries the rule that makes it so.
Environment — stepEnv / requireStepEnv. A step body has no ToolContext, so this is how it reads agent({ requiredEnv }).
HTTP — stepFetch (HTTP/1.1-pinned; fetch speaks h2 and a fan-out on one connection turns a rate limit into an unreadable stream reset) and multipartBody.
Narration — stepReport / stepEmit, what a page’s progress stream renders.
Being woken — stepWebhookUrl, the public callback URL a step hands a third party so a delivery resolves the body’s ctx.waitFor instead of the run polling for an answer. The tool-side spelling of the same URL is ctx.workflows.publicWebhookUrl, which a body and its steps cannot reach.
The model — stepGenerate (one fetch to the LLM gateway on the agent’s own key, because the AI SDK would be megabytes in a ~7 KB artifact) and stepGenerateJson / stripJsonFence. For a whole tool LOOP rather than one prompt, stepDelegate — a published slot for the same reason stepGenerate is a raw fetch: the host owns ToolLoopAgent, and this barrel’s whole budget is that nothing here drags it into the agent bundle.
Audio, both directions — stepWriteUpload / stepReadUpload / stepUploadInfo, stepSpeak and encodeWav out, and stepTranscribeUpload / stepTranscribeSubmit / stepTranscribePoll for the async job API or stepTranscribeSync for the one-request one, back in. transcribeUploadDurably is the async trio and the poll as ONE body-side call over the body’s ctx — the only name here that takes one.
Phone calls, outbound — stepPlaceCall dials through Twilio and streams the answered call to an agent’s WS /phone, with <Parameter>s the answering session reads as call.parameters; stepCallStatus follows it to one of the statuses that are over (isCallOver).
The device and the owner — stepSayOnClient says a sentence on a device (sayFailureOnClient is a run’s failure, as workflow({ onFailure })), and stepTextOwner texts the owner by the text_me builtin’s rule.
Retry classification — isTransientStatus / retryAfter, for a body deciding whether a failure is worth another round, and stepInfo, which says which ATTEMPT this is and whether it is the last. That is what lets a step degrade rather than fail — a smaller model on the final try beats a failed run — and it is the DevKit’s getStepMetadata() with the two differences its own module doc gives.

The zod-free budget still applies here and is now a property of BOTH subpaths rather than the reason one of them exists: a workflows/*.ts module is bundled separately, so the root barrel’s graph would ride into the step bundle. That is also why stepSpeak carries the SLOT and the WAV framing rather than a synthesizer, the same split stepFetch makes with its undici dispatcher.

Two neighbours that are deliberately elsewhere. The failure a body THROWS (toStepError / throwStepError / throwFatalStepError, and the FatalError / RetryableError they resolve to) is on @alexkroman1/aai/step-errors, so that importing a classifier is an opt-in rather than something every /step reader pays for. And the durable wait is ctx.sleep on the WorkflowContext the engine hands the body — this SDK owns what is INSIDE a step and never the steps.

Functions

blockAlign()

blockAlign(format): number

Bytes one sample frame occupies — every channel of one instant.

Parameters

format

Pick<WavFormat, "channels" | "bitsPerSample">

Returns

number

Example

import { blockAlign } from "@alexkroman1/aai/step";

blockAlign({ channels: 1, bitsPerSample: 16 }); // 2
blockAlign({ channels: 2, bitsPerSample: 16 }); // 4

bytesPerSecond()

bytesPerSecond(format): number

Bytes of audio per second of wall clock — the constant that converts a byte offset into a timestamp, and a duration into a request size.

Parameters

format

Pick<WavFormat, "channels" | "bitsPerSample" | "sampleRate">

Returns

number

Example

import { bytesPerSecond } from "@alexkroman1/aai/step";

bytesPerSecond({ sampleRate: 16_000, channels: 1, bitsPerSample: 16 }); // 32000

encodeWav()

encodeWav(samples, format): Uint8Array<ArrayBuffer>

Wrap raw linear-PCM samples in a WAV container.

Parameters

samples

Uint8Array<ArrayBufferLike> | readonly Uint8Array<ArrayBufferLike>[]

The PCM bytes, little-endian and channel-interleaved. A LIST is joined in order, which is what a synthesizer’s or a capture’s frames arrive as.

format

PcmFormat

How to read them. See PcmFormat for the two defaults.

Returns

Uint8Array<ArrayBuffer>

A complete .wav file: WAV_HEADER_BYTES of header followed by samples unchanged.

Throws

for a format no header can describe — a non-integer or non-positive rate or channel count, or a bit depth that is not a positive multiple of 8.

isCallOver()

isCallOver(status): boolean

Whether a call in status is over — one of CALL_OVER_STATUSES.

Takes a string rather than a PlacedCallStatus so a loop can start from a status it made up ("queued") or read back from its own store without a cast; anything unknown is not over.

Parameters

status

string

Returns

boolean

isTransientStatus()

isTransientStatus(status): boolean

Will another attempt plausibly answer differently?

408 counts because it is the far side saying “too slow”, not “no”; 429 and every 5xx are the ordinary transient pair. Everything else — a 400, a 401, a 404 — answers the same way on the fourth attempt, and retrying it spends the step’s whole budget to arrive at the same failure several seconds later.

Parameters

status

number

Returns

boolean

mapConcurrent()

mapConcurrent<T, R>(items, width, run): Promise<R[]>

Map items through run, at most width at a time, in a replay-safe order.

Results come back in ITEM order however the individual calls settle, so it substitutes directly for Promise.all(items.map(run)) where a bound is needed.

A rejection stops the window taking new items and then propagates once the calls already in flight have SETTLED — not the instant it happens. That order is what a workflow body wants: every sibling that finished is journaled, so a resume replays it for free and re-issues only what is missing, where throwing immediately would discard siblings that were mid-call and have the resume pay for them a second time. Catching per item to salvage a partial result is a decision only the caller can make — do it inside run.

Type Parameters

T

T

R

R

Parameters

items

readonly T[]

What to map. An empty list runs nothing and resolves [].

width

number

Most calls in flight at once. Rounded down, and floored at 1. A width of zero would otherwise start no slot at all — a hang, not an error, and a hang inside a workflow body is a run that never completes. A non-finite width is worse and needs the same floor for a different reason: Math.min(NaN, n) is NaN, so Array.from({ length: NaN }) is empty and the map silently does NOTHING, which reads as an empty input.

run

(item, index) => R | Promise<R>

Called once per item, with the item and its index in items. Inside a workflow body this is where a ctx.step call goes, and it must be the only one, issued synchronously — a callback that awaits before its step call, or issues two in a row, interleaves with its siblings by completion order and a resume hands the Nth journal entry to a different call. A body needing two steps per item runs them as two fan-outs. The module doc above carries why, and why the window itself needs no barrier.

Returns

Promise<R[]>

Example

import type { WorkflowContext } from "@alexkroman1/aai";
import { mapConcurrent, stepGenerate } from "@alexkroman1/aai/step";

// In a workflow body: one step per segment, four in flight. The callback
// issues its `ctx.step` immediately, under ONE literal name.
export async function cleanFlow(input: { segments: string[] }, ctx: WorkflowContext) {
  const cleaned = await mapConcurrent(input.segments, 4, (text) =>
    ctx.step("postProcess", () => stepGenerate(`Clean up this transcript: ${text}`)),
  );
  return { cleaned };
}

mapSettled()

mapSettled<T, R>(items, width, run): Promise<Settled<T, R>[]>

Map items through run, at most width at a time, settling each item rather than racing to the first rejection.

Results come back in ITEM order however the calls settle. A run that throws (or rejects) for one item produces { ok: false, error } for that item and nothing else changes: the window keeps taking items, and every sibling’s result is kept.

Type Parameters

T

T

R

R

Parameters

items

readonly T[]

What to map. An empty list runs nothing and resolves [].

width

number

Most calls in flight at once — mapConcurrent’s bound, with one addition: Infinity means every item at once, the shape Promise.allSettled(items.map(run)) had. mapConcurrent alone floors a non-finite width to 1, which would turn an author’s “all at once” into “one at a time” without a word; here the one non-finite value that has an obvious meaning gets it. Any other width is passed through unchanged, floors included.

run

(item, index) => R | Promise<R>

Called once per item, with the item and its index. Inside a workflow body this is where the step call goes, and the rule mapConcurrent’s doc states — one step per item, issued synchronously — applies unchanged.

Returns

Promise<Settled<T, R>[]>

Example

import { mapSettled, partitionSettled } from "@alexkroman1/aai/step";

async function score(name: string): Promise<number> {
  if (name === "") throw new Error("blank");
  return name.length;
}

const settled = await mapSettled(["ann", "", "bo"], 2, score);
// [{ item: "ann", ok: true, value: 3 }, { item: "", ok: false, error: "blank" }, …]
const { ok, failed } = partitionSettled(settled);
if (ok.length === 0) throw new Error(`Every name failed: ${failed[0]?.error}`);

multipartBody()

multipartBody(…parts): MultipartBody

Encode multipart/form-data as BYTES.

The reason this exists rather than new FormData(): a FormData is a branded object, and handing one to a fetch from a different undici than your realm’s global silently sends the string [object FormData] — see StepFetchInit. Bytes cannot be got wrong that way, and a step’s multipart body is always one or two known parts rather than a form somebody filled in.

The boundary is generated per call and is not derived from the content, so a body containing the boundary token is astronomically unlikely rather than impossible; endpoints behave the same way.

A part’s name and filename are ESCAPED, because they are the one thing here that is routinely not the author’s own string. An upload’s name is “the filename the uploader gave” (stepUploadInfo), so it reaches a step from a browser form and lands in a header this function writes — and a ", a CR or an LF in it closed the quoted string and appended headers of the caller’s choosing to the request. The escaping is the HTML form-encoding algorithm’s, which is what the FormData this replaces would have applied: " becomes %22, CR %0D, LF %0A.

Parameters

parts

…readonly MultipartPart[]

Returns

MultipartBody

offsetToMs()

offsetToMs(format, offset): number

Where a byte offset in the FILE falls in the recording, in milliseconds.

Takes a file offset rather than a sample offset, because that is what a ranged read deals in — dataStart is subtracted here so no caller has to remember to.

Parameters

format

WavFormat

offset

number

Returns

number

parseWav()

parseWav(head, totalBytes): WavFormat

Read a WAV header out of the first bytes of a recording.

Parameters

head

Uint8Array

The start of the file. Must reach past the data chunk’s own header: this walks the chunk list, and a file with a large LIST or bext chunk ahead of its samples pushes that well past 44 bytes. 64 KiB is a comfortable probe; the failure for too little is an error naming the shortfall, never a wrong answer.

totalBytes

number

The size of the whole file, from Content-Range or Content-Length. Bounds dataEnd, because the header cannot be trusted for it — see this module’s doc.

Returns

WavFormat

Throws

for anything that is not linear-PCM WAV, for a format nothing can be cut on (a zero rate, zero bytes per frame, or a bit depth that is not a whole number of bytes), and for a header longer than head.

Example

import { blockAlign, bytesPerSecond, parseWav } from "@alexkroman1/aai/step";

declare const head: Uint8Array;
const format = parseWav(head, 1_000_000);
const seconds = (format.dataEnd - format.dataStart) / bytesPerSecond(format);
const frame = blockAlign(format);

partitionSettled()

partitionSettled<T, R>(settled): object

Split what mapSettled answered into the successes and the failures, each still beside its item and each list typed as its own arm.

The typing is the point: a .filter((one) => !one.ok) over the union keeps the union, so reading failed[0].error afterwards needs a re-narrowing (failed[0]?.ok === false ? failed[0].error : …) that three templates each wrote slightly differently. Both lists keep ITEM order.

Type Parameters

T

T

R

R

Parameters

settled

readonly Settled<T, R>[]

Returns

object

failed

failed: object[]

ok

ok: object[]

pcmDurationMs()

pcmDurationMs(byteLength, format): number

How long a run of PCM samples lasts, in milliseconds.

Beside the encoder because it divides by the same derived blockAlign the header states, and a duration computed from a different one is how a progress bar and a file disagree. Rounded, since a caller reporting milliseconds has no use for the fraction.

It takes a LENGTH where its neighbours take bytes, so a Uint8Array handed to it is refused rather than divided: bytes / blockAlign is NaN, and a NaN duration is journaled by a step, rendered into a progress bar and reported to a caller without anything on the way saying which call produced it. That is the one misuse this signature invites — encodeWav, stepSpeak and stepReadUpload all deal in the bytes themselves. wavHeader is the other function here taking a count, and shares this check for that reason.

Parameters

byteLength

number

format

PcmFormat

Returns

number

Throws

for a format no header can describe — the same check encodeWav makes, so the two cannot disagree about what is legal — or for a byteLength that is not a length.

requireStepEnv()

requireStepEnv(name): string

stepEnv, failing by name when the key is not set.

The failure a step wants for a credential: an absent key is not transient, so it should say which key and how to set it rather than surface three layers down as an HTTP 401 the engine then retries.

It throws a plain Error rather than a FatalError on purpose — that class is /step-errors’ and this module must stay importable from a tool body and a spec, neither of which has a workflow around it. A step that wants the retries skipped wraps the call:

import { requireStepEnv } from "@alexkroman1/aai/step";
import { FatalError } from "@alexkroman1/aai/step-errors";
import { errorMessage } from "@alexkroman1/aai/utils";

export function apiKey(): string {
  try {
    return requireStepEnv("ASSEMBLYAI_API_KEY");
  } catch (err) {
    throw new FatalError(errorMessage(err));
  }
}

Parameters

name

string

Returns

string

retryAfter()

retryAfter(from): Date | undefined

When the far side asked to be called back, as a Date.

Reads Retry-After in both spellings RFC 9110 allows — delta-seconds (Retry-After: 30) and an HTTP date (Retry-After: Wed, 21 Oct 2026 07:28:00 GMT) — and answers undefined for a header that is absent, unparsable, or in the past. undefined is what a caller wants there: it means “you decide”, which is the engine’s own default delay, rather than a date that would retry instantly or never.

Parameters

from

{ headers: Headers; } | Headers

A Response, or its headers. Both spellings are accepted because a caller holding only the headers should not have to fake a response to ask.

Returns

Date | undefined

sayFailureOnClient()

sayFailureOnClient<I>(options): object

A workflow({ onFailure }) handler that says a failed run’s reason on the device its input names: delivery id ${runId}:failed, data.failed: true, the step’s budget DEFAULT_CLIENT_DELIVERY_ATTEMPTS. A no-op when clientId resolves undefined or "".

Write the input type as the type argument (sayFailureOnClient<Input>): tsc cannot infer it through workflow({ … }), whose own inference of the schema is still open when this call is checked, so the callbacks would see unknown.

Also accepted as deepResearchWorkflow({ onFailure }) (@alexkroman1/aai/experimental), where it is handed to the engine the same way.

Type Parameters

I

I

Parameters

options

SayFailureOnClientOptions<I>

Returns

object

maxAttempts

maxAttempts: number

run

run: WorkflowFailureHook<I>

Example

import { workflow } from "@alexkroman1/aai";
import { sayFailureOnClient } from "@alexkroman1/aai/step";
import { z } from "zod";

declare function research(topic: string): Promise<string>;

type DigestInput = { clientId?: string | undefined; topic: string };

export const digest = workflow({
  input: z.object({ clientId: z.string().optional(), topic: z.string() }),
  run: async ({ topic }) => ({ summary: await research(topic) }),
  onFailure: sayFailureOnClient<DigestInput>({
    clientId: (input) => input.clientId,
    event: "research",
    text: (_err, input, reason) => `Sorry, the research on ${input.topic} didn't finish. ${reason}`,
    data: (input) => ({ topic: input.topic }),
  }),
});

stepCallStatus()

stepCallStatus(options): Promise<PlacedCallStatus>

Where a placed call is now, as the carrier reports it, normalized to PlacedCallStatus.

Parameters

options

CallStatusOptions

Returns

Promise<PlacedCallStatus>

Throws

On every failure, including a status this SDK does not know (non-retryable, naming it).

stepClientTranscript()

stepClientTranscript(clientId, options?): Promise<ClientTranscript>

Read what clientId’s sessions said — the durable log behind ?client=.

import type { WorkflowContext } from "@alexkroman1/aai";
import { stepClientTranscript } from "@alexkroman1/aai/step";

export async function memorize(
  input: { clientId: string; after?: { sessionId: string; index: number } },
  ctx: WorkflowContext,
) {
  // Everything said since the last digest this run's caller recorded.
  const { sessions } = await ctx.step("read", () =>
    stepClientTranscript(input.clientId, { afterEventIndex: input.after }),
  );
  return { lines: sessions.reduce((n, s) => n + s.messages.length, 0) };
}

Parameters

clientId

string

options?

StepClientTranscriptOptions

Returns

Promise<ClientTranscript>

stepDelegate()

stepDelegate(subagent, options): Promise<DelegateResult>

Hand a bounded task to a SUBAGENT from inside a step.

The step-side ctx.delegate: its own instructions, model, tools and context window, and what comes back is DelegateResult — the final message plus what the run cost, never the tool results that stayed inside it.

Rejects when nothing has published a runner (see the module doc), and for everything ctx.delegate rejects on: no LLM named, an unknown builtin, a cancelled parent.

Parameters

stepEmit()

stepEmit<T>(namespace, chunk): Promise<void>

Write one structured chunk into a NAMED stream of this run.

The other half of stepReport, and the split is what each is FOR: stepReport writes a sentence for a person, into the run’s default stream and the server log. This writes a VALUE for a program — a partial result, as the step produces it — into a stream a reader asks for by name.

That is what makes a long run’s output streamable rather than only its narration. A run’s snapshot carries a status and, once terminal, an output, so a fan-out that has transcribed forty of sixty segments has forty results and no way to hand any of them over. Emitting each one as it lands means a page renders the answer growing instead of a spinner:

import { stepEmit } from "@alexkroman1/aai/step";

export async function transcribeSegment(index: number) {
  const text = await transcribe(index);
  await stepEmit("transcript", { index, text });
  return { index, text };
}

declare function transcribe(index: number): Promise<string>; // your own step

// The reader, which the SDK already had: one stream per namespace.
import { useWorkflowProgress } from "@alexkroman1/aai-ui";

export function Transcript({ runId }: { runId: string }) {
  const { progress } = useWorkflowProgress<{ index: number; text: string }>(runId, {
    namespace: "transcript",
  });
  return (
    <ol>
      {progress.map((chunk) => (
        <li key={chunk.index}>{chunk.text}</li>
      ))}
    </ol>
  );
}

The namespace is REQUIRED, and that is the point of the argument. The default stream is stepReport()’s, carrying lines a page renders verbatim — an object written into it comes back as [object Object] in the middle of the progress log, which is a trap rather than a decision. A named stream is also how a reader gets ONE kind of chunk per subscription, so useWorkflowProgress<T> can be typed at all.

Call it from a STEP, never from the workflow body, for the reason stepReport says: a body replays from the top on every resume, so a chunk written there is re-emitted on each one.

Chunks are RETAINED with the run, so a reader that arrives late or reloads gets the whole stream from the beginning rather than only what arrives next.

Failures are swallowed, exactly as stepReport’s are: a run must not fail because a reader could not be told about a result the run itself has.

Type Parameters

T

T

Parameters

namespace

string

Which of the run’s streams this belongs in. A short, stable name — a reader subscribes by it.

chunk

T

The value, which must survive the run’s own serialization.

Returns

Promise<void>

stepEnv()

stepEnv(name): string | undefined

Read one key of the agent’s env from inside a step.

Parameters

name

string

The env key, as declared in .env or set with aai secret put. Listing it in agent({ requiredEnv }) is what makes a deploy check it is there.

Returns

string | undefined

The value, or undefined when the agent env does not declare it.

Example

import { stepEnv } from "@alexkroman1/aai/step";

export async function fetchReport(id: string): Promise<string> {
  const base = stepEnv("REPORT_BASE_URL") ?? "https://reports.example.com";
  return await (await fetch(`${base}/${id}`)).text();
}

stepEnvContext()

stepEnvContext(): EnvContext

The whole agent env, plus the running step’s cancel signal, as an EnvContext — what a helper shared with tools is handed inside a step.

env is the same record stepEnv reads key by key, so the parity rule is the same: exactly what .env and aai secret put declare once a host has published it, process.env in a process where nothing has (a spec, a script). It is a frozen SNAPSHOT: a copy of process.env in the fallback case, so a helper cannot write back through it.

signal is present inside a step of a running workflow, and aborts when the run is cancelled; outside one (a spec calling the step directly) it is absent.

Returns

EnvContext

Example

import { type EnvContext, stepEnvContext } from "@alexkroman1/aai/step";

// Shared with tools, which pass their own `ctx`.
async function readProfile(ctx: EnvContext, id: string): Promise<unknown> {
  const res = await fetch(`${ctx.env.PROFILE_API ?? ""}/profiles/${id}`, {
    ...(ctx.signal ? { signal: ctx.signal } : {}),
  });
  return await res.json();
}

export async function loadProfile(id: string): Promise<unknown> {
  return await readProfile(stepEnvContext(), id);
}

stepFetch()

stepFetch(url, init?): Promise<Response>

Make one HTTP request from inside a step.

Prefer this to fetch in any step, and especially in a fan-out: it pins HTTP/1.1 (so a concurrent batch gets a socket each rather than N streams on one connection), reuses connections across a fan-out’s calls, and reports a connection failure with its whole cause chain instead of a bare TypeError: fetch failed.

Parameters

url

string

init?

StepFetchInit

Returns

Promise<Response>

Remarks

globalThis.fetch speaks HTTP/2 now, and a fan-out is the worst case for that. undici 8 — the copy backing it from Node 26 — defaults allowH2 to true, so every concurrent request from one process is multiplexed onto ONE TCP connection sharing one flow-control window. Measured against AssemblyAI’s sync transcription endpoint, 8 concurrent 17.66 MB uploads, same bytes and key, one minute apart:

transport	landed	p50	throughput
`globalThis.fetch` (h2)	14/16	8094ms	20.8 MB/s
HTTP/1.1	16/16	3719ms	29.9 MB/s
HTTP/1.1, keep-alive pool	16/16	3037ms	38.6 MB/s

The two lost requests matter more than the 2.7x. On HTTP/2 a capacity limit arrives as a stream reset, and a stream error carries no HTTP status — so neither isTransientStatus nor retryAfter can see it, every sibling in a bounded fan-out retries in lockstep into the same reset, and the run dies on TypeError: fetch failed with its real cause two cause hops down. Over HTTP/1.1 the identical limit arrives as a 503 or 429 carrying retry-after, which those helpers already read. Verified end to end: the same 65-segment run that failed on fetch completes on HTTP/1.1 at every concurrency up to 48, and at 64 pays 20 retried 503s instead of dying.

Throws

when the request never got an answer — a reset connection, a DNS failure, a timeout. Distinct from a response with a bad status, which is returned like any other: only the caller knows whether a 404 is fatal.

From a step, prefer orFail(stepFetch) (orFail is on @alexkroman1/aai/step-errors). The engine’s retry policy is decided by WHICH error a step throws, and raw every failure looks alike to it — a bad API key is retried until the attempts run out. It also turns a non-2xx into a throw, which stepFetch deliberately does not.

stepGenerate()

stepGenerate(prompt, options?): Promise<string>

Ask the AssemblyAI LLM Gateway one question and return its reply.

From a step, prefer orFail(stepGenerate) (orFail is on @alexkroman1/aai/step-errors). It is this call plus throwStepError, and the engine decides its retry policy from WHICH error a step throws: raw, a terminal failure burns every remaining attempt and a rate limit backs off for one second while the delay the far side named sits unread. Reach for the raw call where the failure is not simply a failure — a 404 that means “already deleted”.

Parameters

prompt

string

The user message.

options?

StepGenerateOptions

Returns

Promise<string>

The reply, trimmed. Never empty — a 200 carrying no content is a StepGenerateError with retryable: true, because it is a real and transient thing a gateway does and a step returning "" would file a blank report and report success.

Example

import { stepGenerate, StepGenerateError } from "@alexkroman1/aai/step";
import { FatalError } from "@alexkroman1/aai/step-errors";

export async function summarize(text: string): Promise<string> {
  try {
    return await stepGenerate(text, { system: "Summarize in two sentences." });
  } catch (err) {
    if (err instanceof StepGenerateError && !err.retryable) throw new FatalError(err.message);
    throw err;
  }
}

Throws

On EVERY failure of this call, which is the point of the class: a non-2xx, an empty completion, a reply that is not JSON, a request that never got an answer (a reset, a DNS failure, this call’s own deadline), and a missing API key. Only the last is retryable: false — three more attempts find the same gap.

stepGenerateJson()

stepGenerateJson<S>(prompt, options): Promise<InferSchemaOutput<S>>

Ask the model for JSON and return it validated.

The reply is unfenced, parsed, and checked against schema; the validated value is what comes back, typed as the schema’s output.

From a step, prefer orFail(stepGenerateJson) (orFail is on @alexkroman1/aai/step-errors). It is this call plus throwStepError, and the engine decides its retry policy from WHICH error a step throws: raw, a terminal failure burns every remaining attempt and a rate limit backs off for one second while the delay the far side named sits unread. Reach for the raw call where the failure is not simply a failure — a 404 that means “already deleted”.

Type Parameters

S

S extends StandardSchemaV1<unknown, unknown>

Parameters

prompt

string

The user message. The SHAPE is added for you: the schema is rendered as JSON Schema and appended to system (see shapeInstruction), so a caller writes only the wording that gets a model to comply and never restates the fields.

options

StepGenerateJsonOptions<S>

Returns

Promise<InferSchemaOutput<S>>

The validated reply.

Throws

A plain error — retryable by the engine’s default, which is the point — when the reply is not JSON, is not an object, or does not satisfy schema. All three are things a model may get right next time.

Throws

On any gateway failure, exactly as stepGenerate does. Classify it with toStepError from @alexkroman1/aai/step-errors.

Example

import { stepGenerateJson } from "@alexkroman1/aai/step";
import { z } from "zod";

const Digest = z.object({ headline: z.string(), points: z.array(z.string()) });

export async function summarize(article: string): Promise<{ headline: string }> {
  return await stepGenerateJson(article, {
    schema: Digest,
    system: 'Reply with JSON only: {"headline": string, "points": string[]}.',
  });
}

stepInfo()

stepInfo(): StepInfo | undefined

Which step this code is running inside, or undefined when it is not in one.

undefined is ORDINARY and is not an error: a workflow BODY is not a step, a tool is not a step, and a spec calling an exported step directly has no run at all. A body that branches on the attempt should read the undefined case as “not retrying”, which is what a spec wants and what a first attempt would have said anyway.

Read it once, at the top. The value is a snapshot of the attempt in flight, so calling it again after an await inside the same step answers the same thing — but a helper that reads it per call is asking a question whose answer cannot change and reads as though it could.

Returns

StepInfo | undefined

stepNotifyClient()

stepNotifyClient(clientId, notice, options?): Promise<void>

Push notice to the client connected to this agent’s WS /inbox as clientId, and resolve once it acks.

Throws ClientUnreachableError (retryable) when the client is not connected, is busy, or does not ack in time — so call it from a step and let the step’s retries redeliver:

import type { WorkflowContext } from "@alexkroman1/aai";
import { stepNotifyClient, stepSpeak } from "@alexkroman1/aai/step";

export async function remindFlow(
  input: { clientId: string; text: string; dueAt: number },
  ctx: WorkflowContext,
) {
  await ctx.sleep("due", new Date(input.dueAt));
  const { runId } = ctx;
  await ctx.step(
    "deliver",
    async () => {
      const spoken = await stepSpeak(`Reminder: ${input.text}`, { sampleRate: 16_000 });
      await stepNotifyClient(input.clientId, { id: runId, event: "reminder", audio: spoken.pcm });
    },
    { maxAttempts: 120 }, // an hour of 30 s retries
  );
}

Parameters

clientId

string

notice

ClientNotice

options?

StepNotifyClientOptions

Returns

Promise<void>

stepPlaceCall()

stepPlaceCall(options): Promise<PlacedCall>

Dial to from from and, once answered, stream the call to the agent at agentUrl — this agent, unless another is named. Resolves the carrier’s call id as soon as the carrier ACCEPTS the request — the phone may not have rung yet; stepCallStatus follows it.

Parameters

options

PlaceCallOptions

Returns

Promise<PlacedCall>

Example

Dial, then follow the call until it is over

import type { WorkflowContext } from "@alexkroman1/aai";
import { isCallOver, requireStepEnv, stepCallStatus, stepPlaceCall } from "@alexkroman1/aai/step";

export async function callFlow(input: { to: string; callRef: string }, ctx: WorkflowContext) {
  const { callId } = await ctx.step(
    "dial",
    () =>
      stepPlaceCall({
        carrier: "twilio",
        to: input.to,
        from: requireStepEnv("TWILIO_FROM_NUMBER"),
        parameters: { call: input.callRef },
      }),
    { maxAttempts: 2 },
  );
  let status = "queued";
  for (let i = 0; i < 60 && !isCallOver(status); i++) {
    await ctx.sleep("poll", new Date((await ctx.now()) + 10_000));
    status = await ctx.step("status", () => stepCallStatus({ carrier: "twilio", callId }));
  }
  return { callId, status };
}

Throws

On every failure — see the module doc.

stepPollUntil()

stepPollUntil<T>(check, options): Promise<PollResult<T>>

Check check() every everyMs until done(value) holds or maxMs has passed, inside a step, and resolve the last value.

Type Parameters

T

T

Parameters

check

() => T | Promise<T>

options

StepPollUntilOptions<T>

Returns

Promise<PollResult<T>>

Example

import { stepPollUntil } from "@alexkroman1/aai/step";

declare function jobStatus(id: string): Promise<"PENDING" | "SUCCEEDED" | "FAILED">;

export async function waitForExtraction(id: string): Promise<string> {
  const polled = await stepPollUntil(() => jobStatus(id), {
    everyMs: 3_000,
    maxMs: 60_000,
    done: (status) => status !== "PENDING",
  });
  if (polled.value === "FAILED") throw new Error(`extraction ${id} failed`);
  return polled.value;
}

stepReadUpload()

stepReadUpload(id, options?): Promise<UploadSlice>

Read a window of an uploaded file.

Omitting both bounds reads the whole file, which is the right call only when the file is small: everything else names the window it needs, so a fan-out over a large file moves each byte once.

Bounds are CLAMPED rather than rejected — a plan computed from a file’s own header can legitimately end one byte past it, and the returned start/end say what was actually read. That is also exactly what makes a STREAMED upload readable: the clamp is to what has ARRIVED, so a window that runs past the bytes stored so far comes back short rather than failing, and end is how a caller learns which it got.

Parameters

id

string

options?

ReadUploadOptions

Returns

Promise<UploadSlice>

Example

Write in one step, read a window back in another — an id crosses the journal, bytes never do.

import { stepReadUpload, stepWriteUpload } from "@alexkroman1/aai/step";

export async function store(bytes: Uint8Array): Promise<string> {
  const { id } = await stepWriteUpload(bytes, { name: "summary.wav" });
  return id;
}

export async function firstSecond(uploadId: string): Promise<Uint8Array> {
  const { bytes } = await stepReadUpload(uploadId, { start: 44, end: 44 + 32_000 });
  return bytes;
}

stepReport()

stepReport(line): Promise<void>

Write one progress line for the run this step belongs to.

The line reaches two readers: the run’s own output stream, which GET /workflows/runs/:id/stream serves and useWorkflowProgress in @alexkroman1/aai-ui renders, and the server log, so an operator watching a deploy can see which step is running without a page open.

Call it from a STEP, never from the workflow body. A body replays from the top on every resume, so a line written there is re-emitted on each one.

Failures are swallowed: narration must never fail a run. It resolves either way, so awaiting it is safe and is what keeps the ordering of a step’s own lines.

Parameters

line

string

One line of progress, as a reader should see it. Prefer a sentence naming what is happening and to what ("Transcribing 0:00–0:58.") over a machine token — the page renders these verbatim.

Returns

Promise<void>

stepRequireCompleteUpload()

stepRequireCompleteUpload(id): Promise<UploadInfo>

One upload’s metadata, refused unless every byte is in.

The read for a step that consumes a file END TO END — an upload to a provider, a copy to local disk, a length a segment plan is computed from. A step that works on a WINDOW wants stepUploadInfo and clamping, which is what stepReadUpload already does.

Parameters

id

string

Returns

Promise<UploadInfo>

Example

import { stepRequireCompleteUpload } from "@alexkroman1/aai/step";

export async function wholeFileSize(uploadId: string): Promise<number> {
  const stored = await stepRequireCompleteUpload(uploadId);
  return stored.size;
}

Throws

when the upload is still arriving — see this module’s doc for why that is a refusal rather than a wait.

Throws

when the id names no upload, exactly as stepUploadInfo does.

From a step, prefer orFail(stepRequireCompleteUpload) (orFail is on @alexkroman1/aai/step-errors). The refusal carries retryable: false, and orFail is what turns that into the FatalError the engine stops on; the raw call’s throw is retried, against an upload no attempt can finish.

stepSayOnClient()

stepSayOnClient(clientId, options): Promise<string>

Say text on the device connected to WS /inbox as clientId: synthesize it, push the audio with data.said = text, and resolve with the text once the device acks.

Throws ClientUnreachableError (retryable) when the device is offline, busy or slow to ack — call it from ONE step and let the step’s retries redeliver; DEFAULT_CLIENT_DELIVERY_ATTEMPTS as that step’s maxAttempts rides out an hour. A retry re-uses the audio it already synthesized.

Parameters

clientId

string

The device’s ?client= id.

options

StepSayOnClientOptions

The notice (id, event, text, data) and how to speak it.

Returns

Promise<string>

The text that was said.

Example

import type { WorkflowContext } from "@alexkroman1/aai";
import { DEFAULT_CLIENT_DELIVERY_ATTEMPTS, stepSayOnClient } from "@alexkroman1/aai/step";

export async function remindFlow(
  input: { clientId: string; text: string; dueAt: number },
  ctx: WorkflowContext,
) {
  await ctx.sleep("due", new Date(input.dueAt));
  const { runId } = ctx;
  await ctx.step(
    "deliver",
    () =>
      stepSayOnClient(input.clientId, {
        id: runId,
        event: "reminder",
        text: `Reminder: ${input.text}`,
        data: { text: input.text },
      }),
    { maxAttempts: DEFAULT_CLIENT_DELIVERY_ATTEMPTS },
  );
}

stepSpeak()

stepSpeak(text, options?): Promise<SpokenAudio>

Speak text, and answer with the whole utterance as a WAV.

Parameters

text

string

What to say. Refused when it is blank: a synthesizer answers an empty request with an empty file, which is a zero-length audio element on somebody’s page rather than an error, and no retry finds the missing words.

options?

SpeakOptions

Returns

Promise<SpokenAudio>

Throws

when no synthesizer is published — the message names both causes (a process serving no agent, and a spec calling the step directly).

Throws

when the credential named by apiKeyEnv is not in the agent’s env, which requireStepEnv reports by name. (stubSpeech’s fake needs none unless told to — see StubSpeechOptions.requireApiKey.)

Example

Speak and STORE in one step, and return the id. A step is journaled by what it returns, so an id is replayed on a resume and bytes are not — splitting this in two would carry the audio across the queue on every resume.

import { stepSpeak, stepWriteUpload } from "@alexkroman1/aai/step";

export async function narrate(summary: string): Promise<string> {
  const spoken = await stepSpeak(summary, { voice: "aai-jane" });
  const stored = await stepWriteUpload(spoken.audio, {
    name: "summary.wav",
    type: "audio/wav",
  });
  return stored.id;
}

stepTextOwner()

stepTextOwner(text, options?): Promise<StepTextOwnerResult>

Text the owner text, choosing the number exactly as the text_me builtin does, from the step env (TEXTBELT_KEY, SMS_TO_PHONE, SMS_ALLOWED_PHONES, TEXTBELT_LINKS).

No recipient (no SMS_TO_PHONE, no listed claim): { sent: false }.
A refusal that will refuse again (a bad number, no credit, a link on a key not allowed links), or a text that was only links on such a key: { sent: false, why }.
A transient failure (a 5xx, a 429, no answer): thrown, classified for the step’s retry as throwStepError classifies it.
A recipient but no TEXTBELT_KEY: thrown FATAL, naming the key — the owner asked for texts and a retry cannot set it.

Parameters

text

string

options?

StepTextOwnerOptions

Returns

Promise<StepTextOwnerResult>

Example

import type { WorkflowContext } from "@alexkroman1/aai";
import { stepTextOwner } from "@alexkroman1/aai/step";

export async function reportFlow(input: { phone?: string; report: string }, ctx: WorkflowContext) {
  const texted = await ctx.step(
    "text",
    () => stepTextOwner(input.report, { phone: input.phone }),
    { maxAttempts: 3 },
  );
  return { texted: texted.sent };
}

stepTranscribePoll()

stepTranscribePoll(transcriptId, options?): Promise<TranscribeProgress>

Ask once whether a job has finished, and read it when it has.

Parameters

transcriptId

string

options?

TranscribeRequestOptions

Returns

Promise<TranscribeProgress>

Remarks

Polling READS, so there is no separate read. Both templates this replaced polled GET /v2/transcript/:id for a status and then fetched the identical URL again for the text — the completed poll had the transcript in its hand and threw it away. This answers with it, so a finished job costs one round trip rather than two and the value journaled by the last poll IS the transcript.

The provider’s vocabulary stays inside: branch on done, never on a status string, so a new status the service invents cannot read as “not finished yet” forever.

Throws

, NOT retryable, when the provider failed the job or transcribed no words at all. A recording of silence succeeds and answers with an empty string, which is the failure this flow is most likely to meet and the one that reads least like a failure: everything downstream would otherwise be handed no words and asked to work anyway.

From a step, prefer orFail(stepTranscribePoll) (orFail is on @alexkroman1/aai/step-errors). The engine’s retry policy is decided by WHICH error a step throws, and raw every failure looks alike to it — a bad API key is retried until the attempts run out.

stepTranscribeSubmit()

stepTranscribeSubmit(audioUrl, options?): Promise<{ id: string; }>

Create the transcription job, and answer with the id that outlives this run.

Parameters

audioUrl

string

What to transcribe. stepTranscribeUpload’s answer, or any URL the service can reach — a recording already sitting in a bucket never needs to pass through this process at all.

options?

TranscribeSubmitOptions

Returns

Promise<{ id: string; }>

Throws

on a refusal, or when the API creates no id.

Remarks

This trio is for a recording of arbitrary length; stepTranscribeSync is for one that fits in a single request. That endpoint answers with the words in the response and pays for it with a hard 120-second, 40 MB ceiling. Under the ceiling it is one round trip against these three steps plus a polling loop; over it, the job API is the only thing that works. Choosing between them is the one decision this subpath forces, and it is decided by what the audio IS rather than by anything either function can see.

Example

The whole job, as three steps and a durable wait. The submit is journaled, so a resumed run polls the same job rather than paying for a second one.

import {
  stepTranscribePoll,
  stepTranscribeSubmit,
  stepTranscribeUpload,
} from "@alexkroman1/aai/step";

export async function startJob(uploadId: string): Promise<string> {
  const { audioUrl } = await stepTranscribeUpload(uploadId);
  const { id } = await stepTranscribeSubmit(audioUrl);
  return id;
}

export async function checkJob(id: string): Promise<string | undefined> {
  const progress = await stepTranscribePoll(id);
  // Branch on `done`, never on a provider status string.
  return progress.done ? progress.transcript.text : undefined;
}

From a step, prefer orFail(stepTranscribeSubmit) (orFail is on @alexkroman1/aai/step-errors). The engine’s retry policy is decided by WHICH error a step throws, and raw every failure looks alike to it — a bad API key is retried until the attempts run out.

stepTranscribeSync()

stepTranscribeSync(bytes, options?): Promise<{ text: string; }>

Transcribe one complete audio file.

From a step, prefer orFail(stepTranscribeSync) (orFail is on @alexkroman1/aai/step-errors). It is this call plus throwStepError, and the engine decides its retry policy from WHICH error a step throws: raw, a terminal failure burns every remaining attempt and a rate limit backs off for one second while the delay the far side named sits unread. Reach for the raw call where the failure is not simply a failure — a 404 that means “already deleted”.

Parameters

bytes

Uint8Array<ArrayBufferLike> | readonly Uint8Array<ArrayBufferLike>[]

A whole file, header included. The endpoint decodes each request independently, so a headerless tail is bytes it will refuse. A LIST is still a whole file — the same bytes in the same order, just not contiguous in memory — which is what lets a caller cutting a WAV hand over [wavHeader(format, window.byteLength), window] rather than joining the two into a buffer this function would then copy into the body a second time.

options?

TranscribeSyncOptions

Returns

Promise<{ text: string; }>

The text, trimmed. An EMPTY string is a legitimate answer here and is not refused, unlike the async API’s: a caller fanning out over segments routinely gets silent ones, and a throw would fail the whole recording over a pause in it. A caller transcribing exactly one clip should check for it.

Throws

on a refusal, carrying the verdict toStepError reads — which matters most here, because a fan-out hits a rate limit all at once and retryAfter is what makes the batch drain instead of colliding a second later.

Remarks

The ceiling is the whole decision: 120 seconds and 40 MB per request, both enforced by the service. Under it, this is one round trip against stepTranscribeUpload/stepTranscribeSubmit/stepTranscribePoll’s three steps and a polling loop — no job id, nothing to journal between phases, no wait to make durable. Over it, the audio has to be CUT into segments and fanned out, which is a subject of its own: where to cut so a word is not split, how to re-attach a header to each piece, how wide to run the fan-out, and how to stitch the results back together. So reach for this when one request is enough and for the async trio when it is not; the cut is not a decision this function can make, because it depends on what the audio IS.

Whole files only. A caller cutting a WAV re-attaches a header to every window — encodeWav is the 44 bytes, or wavHeader and the window as two chunks — and a caller handed complete files (parts of a multi-file upload, stepSpeak’s output) passes them through untouched.

Example

One clip, one request. Compare stepTranscribeSubmit for a recording that cannot fit in one.

import { stepReadUpload, stepTranscribeSync } from "@alexkroman1/aai/step";

export async function transcribeClip(uploadId: string): Promise<string> {
  const clip = await stepReadUpload(uploadId);
  const { text } = await stepTranscribeSync(clip.bytes);
  return text;
}

stepTranscribeUpload()

stepTranscribeUpload(uploadId, options?): Promise<{ audioUrl: string; size: number; source: string; }>

Send a stored upload to the provider, and answer with the URL it gave.

The recording STREAMS out of the app’s own store: stepReadUpload hands back bytes and a two-hour recording is not a value this process can hold, so the body is an async iterable of windows — which stepFetch accepts precisely for this. Nothing is buffered beyond one window, and one window of READ-AHEAD keeps the store and the socket busy at the same time.

The upload has to be FINISHED, and that is checked rather than assumed. UploadInfo.size is the contiguous readable prefix, so reading it off a still-arriving recording used to upload only what had landed and transcribe a truncated file — a plausible wrong answer with no error anywhere. stepRequireCompleteUpload refuses instead, BEFORE the expensive leg; its module doc carries why that is a refusal rather than a wait.

Parameters

uploadId

string

An upload in the agent’s own store, as stepWriteUpload or a page’s api.upload(file) produced. Must be complete.

options?

TranscribeRequestOptions

Returns

Promise<{ audioUrl: string; size: number; source: string; }>

The provider’s URL, plus the upload’s label as source and its size. Both ride the JOURNALED result on purpose: a body that reports “Transcribing call.m4a (12 MB)” after this step, or names the recording in its output, reads them off the value it already has rather than paying a second stepUploadInfo round trip — which on a deployed guest is a platform call, and inside a step is one more thing to classify.

Throws

when the upload is still arriving.

Throws

on a refusal, carrying the verdict toStepError reads. Give this step extra retries: it is the one call here worth another attempt, and the only one whose cost is the file.

From a step, prefer orFail(stepTranscribeUpload) (orFail is on @alexkroman1/aai/step-errors). The engine’s retry policy is decided by WHICH error a step throws, and raw every failure looks alike to it — a bad API key is retried until the attempts run out.

stepUploadInfo()

stepUploadInfo(id): Promise<UploadInfo>

Read one upload’s metadata: its name (and the label to show a person), what has ARRIVED, and whether that is all of it.

The poll a body waiting on a streamed upload runs.

complete is the field to branch on, never size. A size that stopped growing means “nothing arrived recently”, which is what a slow link and a dead client both look like; only complete says the file is all there. A body that treated a stalled size as the end would return a transcript of most of a recording and report success.

Parameters

id

string

Returns

Promise<UploadInfo>

Throws

when the id names no upload — a step that reaches for one and finds nothing has been handed a stale or invented id, which no retry fixes, so the engine stops on the first attempt. Note a streamed upload EXISTS from its first byte, so this answers for one that is still arriving.

stepWebhookUrl()

Call Signature

stepWebhookUrl(token): string

The public URL a third party POSTs to in order to resolve ctx.waitFor(token) for the run this step belongs to.

The same URL ctx.workflows.publicWebhookUrl(token) mints for a tool, reached from a step — which is what lets a workflowApp() with no tools hand a provider a callback instead of polling it. Hand it out in the step that submits the work, so the far side is told about the waitpoint before the body parks on it.

One waitFor park per token per run, with a poll as the backstop. A token can be claimed at most ONCE per run — a second claim under a different occurrence key THROWS, and the token is only released when the run goes terminal (onRunSettled in aai-runtime/workflow-journal-memory.ts, whose comment records a template bitten by exactly this: a derived token served one run, the second claim conflicted, the conflict is not a suspend, and the saga compensated a transcript away). So this belongs in a submit-then-park shape, never a waitFor inside a loop — and because a delivery can be lost, missed or never sent, the reconciling backstop is a waitFor(token, { timeoutMs }) whose undefined sends the body to poll the provider once. The callback is what makes the common case fast; the poll is what makes it correct.

Under aai dev this throws, and even where a local origin IS configured it is not reachable from the internet. A public URL is a property of a deployment: the platform bakes it into the guest’s exec env, a self-hosted server passes publicUrl. A laptop has none, so a real third-party callback cannot be exercised locally — drive that path against a deployed agent, or point a tunnel at the dev server’s BACKEND port (the Vite port a developer opens does not proxy /.well-known/) and set PUBLIC_URL to the tunnel. A spec drives it by publishing a minter of its own.

Parameters

token

string

The waitpoint’s token, exactly as the body passes it to ctx.waitFor. Derived from the run’s input, never random.

Returns

string

The absolute URL, encoded for the route’s single token segment.

Example

Submit the work and hand the callback over in the same step, so the far side is told about the waitpoint before the body parks on it.

import { stepReport, stepFetch, stepWebhookUrl } from "@alexkroman1/aai/step";

// The token is DERIVED from the run's own input, so the step handing the URL
// out and the body parking on it agree — the rule `ctx.waitFor` states.
export const renderToken = (id: string) => `render:${id}`;

export async function submitRender(id: string): Promise<void> {
  await stepReport(`Submitting render ${id}.`);
  await stepFetch("https://renders.example.com/jobs", {
    method: "POST",
    headers: { "content-type": "application/json" },
    body: JSON.stringify({ id, callbackUrl: stepWebhookUrl(renderToken(id)) }),
  });
}

Throws

when this process cannot mint one — the message names the configuration and says what aai dev can and cannot do.

Throws

when token is empty: that composes to the route’s own prefix, which the parser refuses, so the failure would otherwise arrive at the far end as a 404 on a URL nobody can re-issue. Refused under optional too — a missing URL is a deployment’s state, an empty token is a bug.

Call Signature

stepWebhookUrl(token, options): string | undefined

The URL, or undefined when this process cannot mint one — for a step that has a callback-free path and wants to take it rather than fail.

Parameters

token

string

options

optional

true

Returns

string | undefined

Example

import { stepFetch, stepWebhookUrl } from "@alexkroman1/aai/step";

export async function submitRender(id: string): Promise<void> {
  // Under `aai dev` there is no public URL; the body polls instead, so the
  // submit goes out without a callback rather than not at all.
  const callbackUrl = stepWebhookUrl(`render:${id}`, { optional: true });
  await stepFetch("https://renders.example.com/jobs", {
    method: "POST",
    headers: { "content-type": "application/json" },
    body: JSON.stringify({ id, ...(callbackUrl === undefined ? {} : { callbackUrl }) }),
  });
}

Call Signature

stepWebhookUrl(token, options?): string | undefined

stepWebhookUrl with the options bag assembled elsewhere.

Parameters

token

string

options?

optional?

boolean

Returns

string | undefined

stepWriteUpload()

stepWriteUpload(bytes, options?): Promise<UploadInfo>

Store a file a step PRODUCED, and answer with the record naming it.

The other direction of stepReadUpload, and the half a workflow app needs the moment its output is not text. A run’s output is journaled and read back as JSON, so audio, an image or a PDF cannot travel in one — the same rule that keeps a recording’s bytes out of a run’s INPUT, arriving at the other end of the run. So the bytes go to the store and the output carries the id, which a page turns back into a file with api.download(id).

import { stepSpeak, stepWriteUpload } from "@alexkroman1/aai/step";

export async function narrate(summary: string) {
  const spoken = await stepSpeak(summary);
  const stored = await stepWriteUpload(spoken.audio, { name: "summary.wav", type: "audio/wav" });
  return { audio: stored.id, durationMs: spoken.durationMs };
}

Write it in the step that MAKES it, and return the id. A step is journaled by what it returns, so an id is replayed and bytes are not: a resumed run re-reads the same file rather than re-synthesizing it. The corollary is that a RETRIED step writes a second upload and abandons the first, which is the cost of the store having no way to know two calls meant one file — worth knowing, and cheap next to the alternative of a step that cannot retry at all.

Parameters

bytes

Uint8Array<ArrayBufferLike> | readonly Uint8Array<ArrayBufferLike>[] | AsyncIterable<Uint8Array<ArrayBufferLike>, any, any>

The file. A LIST is stored in order and an async iterable is streamed, so a step producing something large — a long recording, a concatenation of many utterances — never has to hold it whole.

options?

WriteUploadOptions

What to declare about it. Both fields are stored verbatim and neither is inferred; see WriteUploadOptions.

Returns

Promise<UploadInfo>

Throws

when the process published no store, or published a READ-ONLY one — two different sentences, because the remedies differ and the call site cannot tell them apart. Also when the deployment has nowhere durable to put bytes at all, which the store reports by naming the variable that is missing.

stripJsonFence()

stripJsonFence(reply): string

Unwrap a ```json fence, which models add however firmly they are told not to.

Refusing one would cost a whole retry for a reply that was otherwise correct. Text that carries no fence is returned trimmed and otherwise untouched.

Parameters

reply

string

Returns

string

transcribeUploadDurably()

transcribeUploadDurably(ctx, uploadId, options?): Promise<DurableTranscript>

Transcribe a stored upload through the async API, durably: upload it, create the job, and poll until the transcript is in, each phase a journaled step of ctx.

Call it from the BODY, handing it the body’s own ctx — it journals through that context exactly as the hand-written three-step loop did (step-transcribe.ts’s module doc shows that loop), so there is nothing to wrap it in. A resumed run replays the upload and the submit from the journal and polls the same job.

Parameters

ctx

TranscribeDurablyHost

The workflow body’s context, or anything with its step and poll (the /testing recorder qualifies).

uploadId

string

An upload in the agent’s own store. Must be complete; one still arriving is refused BEFORE the upload leg, as stepTranscribeUpload refuses it, and that refusal is fatal.

options?

TranscribeDurablyOptions

See TranscribeDurablyOptions.

Returns

Promise<DurableTranscript>

Throws

/ {RetryableError} out of each step, by throwStepError’s verdict — the engine reads them; a caller sees the run fail or retry.

Throws

from the body, retryable: false, when the budget is spent with the job still running. The message carries the transcript id and where to read it.

Example

import { stepReport, transcribeUploadDurably } from "@alexkroman1/aai/step";
import type { WorkflowContext } from "@alexkroman1/aai/workflow-api";

export async function transcribe(input: { recording: string }, ctx: WorkflowContext) {
  const transcript = await transcribeUploadDurably(ctx, input.recording, {
    params: { speaker_labels: true },
    maxMs: 2 * 3_600_000,
  });
  await ctx.step("report", () =>
    stepReport(`Transcribed ${transcript.source}: ${transcript.text.length} characters.`),
  );
  return transcript;
}

wavHeader()

wavHeader(format, byteLength): Uint8Array<ArrayBuffer>

The WAV header for a payload of byteLength bytes, and nothing else.

For a caller handing the header and the samples to something that takes a LIST — multipartBody, a stream — where encodeWav’s joined buffer would be a second full copy of the audio held at the same time. Two chunks are the same bytes in the same order on the wire.

Parameters

format

PcmFormat

How to read the samples the header will sit in front of. See PcmFormat for the two defaults.

byteLength

number

How many bytes of PCM follow. The header states TWO lengths and both derive from this one, so a value that does not match what is actually sent is a file a decoder reads past the end of or stops short inside.

Returns

Uint8Array<ArrayBuffer>

Exactly WAV_HEADER_BYTES bytes.

Throws

for a format no header can describe — the same check encodeWav makes, so the two cannot disagree about what is legal — or for a byteLength that is not a length.

Classes

ClientUnreachableError

The client did not take the notice. Retryable: throw it (or let it propagate) and the step runs again after retryAfter.

Extends

RetryableError

Constructors

Constructor

new ClientUnreachableError(clientId, reason, retryAfterMs): ClientUnreachableError

Parameters

clientId

string

reason

ClientUnreachableReason

retryAfterMs

number

Returns

ClientUnreachableError

Overrides

RetryableError.constructor

Methods

is()

static is(value): value is RetryableError

Is value a RetryableError, including one from another copy of this module?

Parameters

value

unknown

Returns

value is RetryableError

Inherited from

RetryableError.is

Properties

clientId

readonly clientId: string

reason

readonly reason: ClientUnreachableReason

retryAfter

readonly retryAfter: Date

When the next attempt may run. Always a Date — a number passed to the constructor is resolved against the clock AT CONSTRUCTION, which is the moment the caller meant.

Inherited from

RetryableError.retryAfter

retryAfterNamed?

readonly optional retryAfterNamed?: boolean

Whether the thrower NAMED the delay, or retryAfter is the DEFAULT_RETRY_DELAY_MS fill.

The field StepOptions.maxWaits reads: only a delay the far side named (a Retry-After on a 429, a rate-limited gateway’s own “come back at”) is a WAIT out of that budget. The default is a second ahead of the clock at construction and so would read as “still in the future” on every catch, which made every unclassified transient failure a free retry under maxWaits. A clock test cannot tell the two apart; this can.

Optional in the TYPE, because of the two-copies case the brand exists for (the module doc): an instance from an older copy of this class carries no such field, and a reader treats its absence as “not named” — the direction that spends an attempt. Every instance this constructor makes sets it.

Inherited from

RetryableError.retryAfterNamed

PlaceCallError

The carrier refused, or never answered. message is a sentence a person can act on and never quotes a credential; branch on retryable, which throwStepError also reads.

Extends

Error

Constructors

Constructor

new PlaceCallError(message, init): PlaceCallError

Parameters

message

string

init

carrier

string

cause?

unknown

code?

number

retryable

boolean

retryAfter?

Date

status?

number

Returns

PlaceCallError

Overrides

Error.constructor

Properties

carrier

readonly carrier: string

The carrier asked.

code

readonly code: number | undefined

The carrier’s own error code — Twilio’s 21219, and so on — when it gave one.

name

readonly name: "PlaceCallError" = "PlaceCallError"

Overrides

Error.name

retryable

readonly retryable: boolean

Whether another attempt could plausibly succeed: a 429, a 5xx, or no answer at all.

retryAfter

readonly retryAfter: Date | undefined

When the carrier named a Retry-After, the moment it asked for.

status

readonly status: number | undefined

The HTTP status, or undefined when no request was answered (or none was made).

StepGenerateError

A model call that failed, with the one thing a step has to decide from.

retryable is the whole point. The engine retries a step that throws and a caller has to choose between letting it (a rate limit, a 5xx) and refusing (a bad key, a rejected request) — and getting that backwards is either five pointless attempts against a 401 or one attempt against a blip.

It is a BOOLEAN on the error rather than a FatalError thrown for you, because FatalError lives on @alexkroman1/aai/step-errors and reaching for it is the caller’s opt-in: whether a terminal failure should burn a step’s remaining attempts is not this module’s call. The mapping is one line at the call site:

import { StepGenerateError, stepGenerate } from "@alexkroman1/aai/step";
import { FatalError } from "@alexkroman1/aai/step-errors";

export async function summarize(prompt: string, system: string) {
  try {
    return await stepGenerate(prompt, { system });
  } catch (err) {
    if (err instanceof StepGenerateError && !err.retryable) throw new FatalError(err.message);
    throw err;
  }
}

Extends

Error

Constructors

Constructor

new StepGenerateError(message, options): StepGenerateError

Parameters

message

string

options

cause?

unknown

retryable

boolean

retryAfter?

Date

status?

number

Returns

StepGenerateError

Overrides

Error.constructor

Properties

retryable

readonly retryable: boolean

Will another attempt plausibly answer differently?

retryAfter

readonly retryAfter: Date | undefined

When the gateway asked to be called back, from its own Retry-After.

Present on a rate limit that named a delay, and what a caller should hand to RetryableError — the engine’s default delay is a guess, and this is the number the far side chose.

status

readonly status: number | undefined

The gateway’s status, when there was a response at all.

StepTransportError

A request that never got an answer.

Its own class because the DISTINCTION is what a retry policy turns on: a response with a status can be classified (isTransientStatus, retryAfter), and this cannot — so a caller’s choice is between retrying a connection failure and giving up on one, with nothing to read. Retrying is almost always right, which is why StepTransportError is what the SDK raises rather than making every step write the catch.

The message carries the whole cause chain, because the top of one never says anything useful: TypeError: fetch failed and socket hang up are wrappers, and the code that identifies the failure — ECONNRESET, UND_ERR_SOCKET, ETIMEDOUT, ERR_HTTP2_STREAM_ERROR — is a hop or two below.

Extends

Error

Constructors

Constructor

new StepTransportError(url, options): StepTransportError

Parameters

url

string

options

cause

unknown

Returns

StepTransportError

Overrides

Error.constructor

Properties

codes

readonly codes: readonly string[]

Every code in the chain, outermost first — what a caller would branch on.

TranscribeError

A failure from either endpoint, carrying what the caller needs to classify it.

The SDK does not decide fatal-vs-retryable, for the same reason stepSpeak does not: a helper that guessed would be guessing for every caller. What it can do is carry the evidence, which is what retryable and retryAfter are — read by toStepError on @alexkroman1/aai/step-errors, exactly as StepGenerateError’s are. So a step body says .catch(throwStepError) and gets the three-way call for free.

Extends

Error

Constructors

Constructor

new TranscribeError(message, init): TranscribeError

Parameters

message

string

init

retryable

boolean

retryAfter?

Date

status?

number

Returns

TranscribeError

Overrides

Error.constructor

Properties

retryable

readonly retryable: boolean

Whether asking again could plausibly answer differently.

retryAfter

readonly retryAfter: Date | undefined

How long the service asked us to wait, when it said.

status

readonly status: number | undefined

HTTP status the endpoint answered, when it answered one.

UnsupportedRecordingError

A recording parseWav will not read.

Its own class rather than a RangeError because every caller has to decide one thing about it — this is TERMINAL, no retry helps, and the recording needs converting — and a caller distinguishing that from a transient read failure needs to be able to name it.

Extends

Error

Constructors

Constructor

new UnsupportedRecordingError(message): UnsupportedRecordingError

Parameters

message

string

Returns

UnsupportedRecordingError

Overrides

Error.constructor

UploadIncompleteError

An upload that is still arriving, where the whole file was needed.

retryable: false, and that is the interesting field: toStepError (@alexkroman1/aai/step-errors) recognises a carried verdict STRUCTURALLY, so a step ending .catch(throwStepError) turns this into a FatalError and the run stops on the spot with this sentence. Which is right — the default retry cadence is ~0 ms, so three attempts would spend the whole budget of the most expensive step in the flow inside a millisecond and still find the upload unfinished. What has to change is the run’s ORDER, and no number of attempts changes that.

Extends

Error

Constructors

Constructor

new UploadIncompleteError(message, stored): UploadIncompleteError

Parameters

message

string

stored

number

Returns

UploadIncompleteError

Overrides

Error.constructor

Properties

retryable

readonly retryable: false = false

Whether asking again could plausibly answer differently. It cannot.

stored

readonly stored: number

Bytes readable when the check ran — the PREFIX, never a total.

Type Aliases

CallStatusOptions

CallStatusOptions = object

Input

What stepCallStatus takes.

Properties

callId

callId: string

PlacedCall.callId.

carrier

carrier: "twilio"

The carrier the call was placed through.

credentials?

optional credentials?: PlaceCallCredentials

Explicit credentials; the step env otherwise.

signal?

optional signal?: AbortSignal

ClientNotice

ClientNotice = object

Input

One thing to tell a client.

Properties

audio?

optional audio?: Uint8Array

Bytes sent after the header as binary frames — audio the client plays, most often (stepSpeak’s pcm, at a rate the client expects).

data?

optional data?: unknown

Anything JSON-serializable the client reads alongside event.

event

event: string

What the client should do with it, e.g. "reminder".

id

id: string

Identifies the DELIVERY, so a client can drop a repeat — the run id, when a run sends one notice. At most 128 characters.

ClientTranscript

ClientTranscript = object

Sealed

What stepClientTranscript returns.

Properties

sessions

sessions: ClientTranscriptSession[]

Oldest session first.

ClientTranscriptMessage

ClientTranscriptMessage = object

Sealed

One committed line of a client’s conversation.

Properties

at

at: number

Epoch ms the line was recorded.

role

role: "user" | "assistant" | string & object

text

text: string

ClientTranscriptSession

ClientTranscriptSession = object

Sealed

One of a client’s sessions, as stepClientTranscript reads it.

Properties

lastEventIndex

lastEventIndex: number

The index of the last event read for this session (-1 when none) — the index half of the next call’s afterEventIndex.

messages

messages: ClientTranscriptMessage[]

sessionId

sessionId: string

startedAt

startedAt: number

Epoch ms the session was first bound to the client.

tools

tools: ClientTranscriptTool[]

ClientTranscriptTool

ClientTranscriptTool = object

Sealed

One settled tool call in a client’s conversation.

Properties

args

args: Readonly<Record<string, unknown>> | undefined

The arguments it was called with, when the call is still in the log.

at

at: number

Epoch ms the result was recorded.

name

name: string | undefined

The tool’s name, or undefined when the log’s front no longer holds the call.

result

result: string

What it returned, serialized and capped as the client’s own frame carries it.

ClientUnreachableReason

ClientUnreachableReason = "offline" | "busy" | "no-ack" | "disconnected" | string & object

Why a notice was not taken.

DurableTranscript

DurableTranscript = Transcript & object

Sealed

What transcribeUploadDurably resolves with: the transcript, plus what the recording was CALLED.

Type Declaration

source

source: string

The upload’s label — its filename, or its id when the uploader gave none. Read off the journaled upload step, so a resumed run names the same file without a second look at the store.

EnvContext

EnvContext = object

Sealed

The part of a context a helper needs to reach an outside service: the agent’s env, and a signal to stop on.

What a helper shared between a TOOL and a STEP should take. A ToolContext and a RouteContext both satisfy it as they are, and stepEnvContext builds one inside a step — so a supabase.ts or an apps.ts written against this runs unchanged in all three, instead of every app hand-building { env: { KEY: stepEnv("KEY"), … } } per workflow and forgetting a key the helper later grew.

Properties

env

env: Readonly<Partial<Record<string, string>>>

The agent’s env — ctx.env in a tool, the published agent env in a step.

signal?

optional signal?: AbortSignal

Aborted when the caller goes away — the tool call, the request, or (in a step) the run being cancelled. Absent where nothing can cancel.

MultipartBody

MultipartBody = object

Sealed

A ready-to-send multipart body, as multipartBody returns it.

Properties

body

body: Uint8Array

The whole encoded body.

headers

headers: object

The Content-Type naming the boundary — spread into headers.

Content-Type

Content-Type: string

MultipartPart

MultipartPart = object

Input

One file part, as multipartBody takes it.

Properties

bytes

bytes: Uint8Array | readonly Uint8Array[]

The bytes, as one buffer or as the chunks they already are.

A LIST is concatenated into the body in order and is indistinguishable on the wire from the same content passed as one buffer. It exists so a caller holding a payload in pieces — a wavHeader in front of the samples it describes, a file read window by window — need not JOIN them first only for this function to copy the join into the body: for a multi-megabyte part that intermediate buffer is a second full copy of the payload, held at the same time as the first. The body itself is still one buffer, which is what buys Content-Length and a retryable request; this removes the copy in front of it, not the one it is.

filename?

optional filename?: string

Filename to declare. Omitted makes this an ordinary field rather than a file.

name

name: string

The form field name the endpoint reads.

type?

optional type?: string

Content type of the part. Defaults to application/octet-stream for a file.

PcmFormat

PcmFormat = object

Input

How to read the samples handed to encodeWav.

Only sampleRate is required, because the other two have an answer that is right for every synthesizer and every phone call this SDK talks to, and a caller repeating channels: 1, bitsPerSample: 16 at every site is a caller who will one day repeat it beside samples that are neither.

Properties

bitsPerSample?

optional bitsPerSample?: number

Bits per sample. Defaults to 16, and must be a multiple of 8.

DESCRIBES the bytes rather than requesting a conversion: nothing here transcodes, so a wrong value writes a header that misreads samples it never touched.

channels?

optional channels?: number

Interleaved channel count. Defaults to 1 — mono, which speech is.

sampleRate

sampleRate: number

Samples per second, e.g. 24_000. Must be a positive integer.

PlaceCallCredentials

PlaceCallCredentials = object

Input

A Twilio account’s credentials. Pass them to name them yourself; omitted, they are read from the step env as TWILIO_ACCOUNT_SID_ENV and TWILIO_AUTH_TOKEN_ENV. Never put them in a run’s INPUT, which is journaled: read them inside the step.

Properties

accountSid

readonly accountSid: string

The account SID, AC….

authToken

readonly authToken: string

The auth token. A credential: it never appears in an error.

PlaceCallOptions

PlaceCallOptions = object

Input

What stepPlaceCall takes.

Properties

agentUrl?

optional agentUrl?: string

The PUBLIC base URL of the agent that answers — https://… or wss://…, with any path prefix it is served under. Twilio fetches the webhook <agentUrl>/phone?carrier=twilio once the call is answered, and that agent must declare telephony for Twilio.

Defaults to THIS agent’s public URL, which the host publishes: the deployed agent’s own URL on the managed platform, the tunnel under aai dev --tunnel, PUBLIC_URL for a self-hosted server. Pass it only to have a DIFFERENT agent answer.

brief?

optional brief?: CallBrief

The call’s BRIEF — the app’s own call id and the task prompt — signed with the secret in briefSecretEnv and carried on the call as the parameters call, sig and p0…pN (signCallBrief), so the answering session reads it back with readCallBrief(call, secret) and refuses a stream that was not signed by this app. parameters may not use those names. The prompt is at most MAX_CALL_BRIEF_PROMPT_CHARS.

briefSecretEnv?

optional briefSecretEnv?: string

The step-env variable holding the secret brief is signed with. Default CALL_BRIEF_SECRET_ENV ("CALLS_API_KEY"), which an empty string also means; the answering agent reads the same variable from its env. Unset with a brief is a non-retryable refusal.

carrier

carrier: "twilio"

The carrier that dials. "twilio" only, for now — see the module doc.

credentials?

optional credentials?: PlaceCallCredentials

Explicit credentials; the step env otherwise.

from

from: string

The caller id: a number on the Twilio account, E.164.

parameters?

optional parameters?: Readonly<Record<string, string>>

Custom parameters for the answering session, each a TwiML <Parameter>, read there as call.parameters. At most 32, each name plus value under 500 characters (Twilio’s limit). Put an unguessable id here and check it in sessionContext.

ringTimeoutS?

optional ringTimeoutS?: number

How long it rings before no-answer, in seconds. Default DEFAULT_CALL_RING_TIMEOUT_S.

signal?

optional signal?: AbortSignal

timeLimitS?

optional timeLimitS?: number

Hard cap on the connected call, in seconds. Default DEFAULT_CALL_TIME_LIMIT_S.

to

to: string

The number to call, E.164 (+15555550123). Personal data.

PlacedCall

PlacedCall = object

Sealed

What stepPlaceCall resolves: the carrier’s id for the call (Twilio’s CA… SID).

Properties

callId

callId: string

PlacedCallStatus

PlacedCallStatus = "queued" | "ringing" | "in-progress" | "completed" | "busy" | "no-answer" | "failed" | "canceled" | string & object

Where a placed call is. The last five are OVER — nothing moves off them: completed (answered and hung up), busy, no-answer, failed (never connected: a bad number, a carrier refusal), canceled (hung up by the API before it was answered). Twilio’s initiated reads as queued.

ReadUploadOptions

ReadUploadOptions = object

Input

Options for stepReadUpload.

Both bounds admit undefined explicitly rather than being merely optional: a step computes them from a plan, and under exactOptionalPropertyTypes a { start: maybe } would otherwise be a compile error at every call site that has one — which is most of them, since “no start” and “start 0” mean the same thing here.

Properties

end?

optional end?: number

One past the last byte to read. Defaults to the end of the file.

start?

optional start?: number

First byte to read. Defaults to 0.

SayFailureOnClientOptions

SayFailureOnClientOptions<I> = Pick<StepSayOnClientOptions, "event" | "sampleRate" | "voice" | "language" | "ackTimeoutMs" | "retryAfterMs"> & object

Input

Options for sayFailureOnClient.

Type Declaration

clientId

(input) => string | undefined

The device to say it on, from the run’s input — undefined (or "") says nothing, for a run started without one (a page, not a speaker).

data?

optional data?: Record<string, unknown> | ((input) => Record<string, unknown>)

What the device reads beside the text. failed: true is set on top of it, and said on top of that (by stepSayOnClient).

maxAttempts?

optional maxAttempts?: number

The hook step’s attempt budget. Defaults to DEFAULT_CLIENT_DELIVERY_ATTEMPTS, an hour’s outage.

text

(error, input, reason) => string

What to say. reason is spokenErrorReason(error) — the first sentence, credentials redacted, URLs out — which is what to put in a sentence a person hears; error is there for a branch on its type.

Type Parameters

I

I

The workflow’s parsed input.

Settled

Settled<T, R> = { item: T; ok: true; value: R; } | { error: string; item: T; ok: false; }

What one item of a mapSettled came back as: its value, or the reason it failed, beside the item itself.

error is a STRING (via errorMessage) rather than the thrown value, because the reason exists to be spoken or stored — a ToolFailure sentence, a slot’s unscored list — and a template that wants the raw cause has mapConcurrent and its own catch.

Type Parameters

T

T

R

R

SpeakOptions

SpeakOptions = object

Input

What stepSpeak accepts.

Properties

apiKeyEnv?

optional apiKeyEnv?: string

Env var holding the credential, replacing ASSEMBLYAI_API_KEY.

Names a VARIABLE, not a key — the same contract every provider descriptor keeps, so nothing here can end up in a journaled step argument.

language?

optional language?: string

Spoken language as an ISO 639-1 code ("en", "fr", "de", "es", "it", "pt"). Omitted by default so the service infers it from the voice — set it only alongside a voice that speaks it.

sampleRate?

optional sampleRate?: number

Samples per second to synthesize at. Defaults to STEP_SPEAK_SAMPLE_RATE.

Worth lowering only when the audio is going somewhere that resamples it anyway (a phone line is 8 kHz), since the file scales linearly with it.

signal?

optional signal?: AbortSignal

Abort the synthesis. Combined with STEP_SPEAK_TIMEOUT_MS rather than replacing it, so a caller passing one still cannot hang forever.

voice?

optional voice?: string

Voice id, e.g. "aai-jane", "aai-michael", "aai-vera". Defaults to ASSEMBLYAI_TTS_DEFAULT_VOICE; the library voices are ASSEMBLYAI_TTS_VOICES on @alexkroman1/aai/tts, each speaking every supported language.

SpokenAudio

SpokenAudio = object

Sealed

What stepSpeak resolves with.

Properties

audio

audio: Uint8Array<ArrayBuffer>

A complete .wav file — the container a browser, a bucket and a transcription API all accept.

durationMs

durationMs: number

How long it lasts. Derived from the byte count, not claimed by the service.

pcm

pcm: Uint8Array

The same samples WITHOUT the header, for a caller joining several utterances and framing them once.

sampleRate

sampleRate: number

Samples per second the audio was synthesized at.

voice

voice: string

The voice that actually spoke, with the default filled in.

StepClientTranscriptOptions

StepClientTranscriptOptions = object

Input

Options for stepClientTranscript.

Properties

afterEventIndex?

optional afterEventIndex?: object

A cursor: only what came AFTER event index of session sessionId — the rest of that session, and every session that started after it. An unknown session id reads from the beginning, which is the safe way to be wrong for a summarizer that keys its own writes.

index

index: number

sessionId

sessionId: string

since?

optional since?: number

Epoch ms: only sessions active, and events recorded, at or after it.

StepFetchInit

StepFetchInit = object

Input

What stepFetch accepts.

Deliberately NARROWER than RequestInit, and the narrowing is the API’s main safety property. A FormData, Blob, File, Headers or Request handed to a fetch that is not the one your realm’s global came from is brand-checked against that other undici’s classes, matches no branch, and is silently stringified — Content-Type: text/plain with the 17-byte body [object FormData], answered 415 by a server that was told nothing. body therefore takes BYTES or a string, and multipartBody is how a file becomes bytes.

Properties

body?

optional body?: Uint8Array | string | AsyncIterable<Uint8Array>

The request body: bytes, a string, or an async iterable of chunks.

The iterable form is what lets a step send a file it must not hold in memory — a stored upload read window by window, which is the only way a step can hand a multi-gigabyte recording to another service. It requires duplex: "half", which the published fetch adds; the caller passes only the iterable.

Note a streaming body cannot be RETRIED by the transport, because an iterable is consumed once. That is a property of streaming rather than of this option, and it is why a step sending one should be the step the engine retries — a fresh attempt re-reads the upload from the start.

headers?

optional headers?: Record<string, string>

Plain record, not a Headers — see the type’s own doc for why.

method?

optional method?: string

signal?

optional signal?: AbortSignal

StepGenerateJsonOptions

StepGenerateJsonOptions<S> = StepGenerateOptions & object

Input

Options for stepGenerateJson: StepGenerateOptions plus the shape.

Type Declaration

schema

schema: S

The shape the reply must satisfy — any Standard Schema, zod being the documented default. Its OUTPUT type is what this call returns, so a schema that coerces (dropping the elements a model got wrong, say) is a supported and often better answer than one that rejects the whole reply.

Type Parameters

S

S extends StandardSchemaV1

StepGenerateOptions

StepGenerateOptions = object

Input

Options for stepGenerate.

Properties

apiKeyEnv?

optional apiKeyEnv?: string

Env key holding the AssemblyAI API key. Defaults to ASSEMBLYAI_API_KEY — the same name every AssemblyAI stage reads.

gatewayUrl?

optional gatewayUrl?: string

Gateway base URL, including the version path — e.g. https://llm-gateway.eu.assemblyai.com/v1 for EU residency.

maxTokens?

optional maxTokens?: number

Cap on the reply, forwarded only when set.

model?

optional model?: string

Gateway model id. Defaults to ASSEMBLYAI_LLM_DEFAULT_MODEL, the same one an agent’s own pipeline resolves, so a workflow and its agent do not silently run on different models.

responseSchema?

optional responseSchema?: Record<string, unknown>

A JSON Schema the reply must match, sent as the gateway’s response_format (json_schema) so the model is CONSTRAINED to it rather than asked. The reply still arrives as a string: stepGenerateJson is the call that parses and validates it, and passes this for you. The gateway supports it on GPT-4.1/5.x, Gemini and Claude models, not on gpt-oss.

system?

optional system?: string

The system instruction. Omitted entirely when unset, rather than sent empty — an empty system message is a message the model still reads.

temperature?

optional temperature?: number

Sampling temperature, forwarded only when set. Left to the model’s own default otherwise, which is what an unset knob should mean.

timeoutMs?

optional timeoutMs?: number

Request deadline in milliseconds. Defaults to 60s.

StepInfo

StepInfo = object

Sealed

Which step is running, and which attempt of it.

Properties

attempt

readonly attempt: number

Which try this is, 1-based.

The WALK’s count, not the journal’s charge. Two overlapping deliveries of one run each start at 1, because maxAttempts means how many times to try and how many workers happen to be trying is not that number — see “An attempt is a LEASE, and it EXPIRES” in packages/aai-runtime/JOURNAL-CLAUDE.md.

isLastAttempt

readonly isLastAttempt: boolean

Is this the last try, so that a throw fails the step for good?

Provided rather than left as attempt === maxAttempts because the subtraction is where the mistake is: a body that hard-codes the ceiling degrades early on every run when the call site’s maxAttempts changes, and nothing reports it.

A FatalError still ends the step wherever it is thrown, so this being false is not a promise that another attempt will happen.

key

readonly key: string

name#occurrence — the journal key, which is what makes a loop’s rounds distinct.

maxAttempts

readonly maxAttempts: number

The ceiling this step was given — StepOptions.maxAttempts, or its default.

name

readonly name: string

The step’s own name, as ctx.step was given it.

signal?

readonly optional signal?: AbortSignal

Aborted when this delivery stops — the run was cancelled, or the delivery was abandoned. Pass it to whatever the step waits on; stepFetch already reads it. Absent from a reader that has none (an eval’s).

StepNotifyClientOptions

StepNotifyClientOptions = object

Input

Options for stepNotifyClient.

Properties

ackTimeoutMs?

optional ackTimeoutMs?: number

How long to wait for the ack once the notice is sent. It covers the client TAKING it — reading the audio, for one that acks after queueing it — not the whole playback. Default DEFAULT_CLIENT_ACK_TIMEOUT_MS.

retryAfterMs?

optional retryAfterMs?: number

When the next attempt runs after the client was unreachable. Default DEFAULT_CLIENT_RETRY_MS.

signal?

optional signal?: AbortSignal

StepPollUntilOptions

StepPollUntilOptions<T> = object

Input

Options for stepPollUntil.

Type Parameters

T

T

Properties

done

done: (value) => boolean

Is this check’s value the answer?

Parameters

value

T

Returns

boolean

everyMs

everyMs: number

Milliseconds to wait between two checks. At least 1.

maxMs

maxMs: number

Wall-clock budget in milliseconds, measured from the first check. No wait starts that would end past it, so the poll returns done: false at or before maxMs (plus the last check’s own time).

signal?

optional signal?: AbortSignal

Stop waiting when this aborts — the abort’s reason is thrown. Defaults to the running step’s own signal, so a cancelled run stops polling.

StepSayOnClientOptions

StepSayOnClientOptions = object

Input

Options for stepSayOnClient.

Properties

ackTimeoutMs?

optional ackTimeoutMs?: number

How long to wait for the device’s ack — see stepNotifyClient.

data?

optional data?: Record<string, unknown>

Anything JSON-serializable the device reads alongside event. said is set to text on top of it.

event

event: string

What the device should do with it, e.g. "reminder".

id

id: string

Identifies the DELIVERY, so the device drops a repeat — the run id for a run’s one announcement, ${runId}:failed for its failure. 1 to 128 characters.

language?

optional language?: string

Spoken language, as stepSpeak takes it.

retryAfterMs?

optional retryAfterMs?: number

When the step’s next attempt runs after the device was unreachable.

sampleRate?

optional sampleRate?: number

Samples per second to synthesize at. Defaults to the agent’s clientInbox.sampleRate, then to stepSpeak’s own default.

signal?

optional signal?: AbortSignal

Abort the synthesis and the delivery.

text

text: string

What to say. Also sent as data.said.

voice?

optional voice?: string

Voice id, as stepSpeak takes it.

StepTextOwnerOptions

StepTextOwnerOptions = object

Input

Options for stepTextOwner.

Properties

links?

optional links?: "keep" | "strip"

"strip" leaves every link out, for a key Textbelt has not yet allowed links. Defaults to the agent env’s TEXTBELT_LINKS, as text_me reads it.

phone?

optional phone?: string

The number the client CLAIMED (sessionClientPhone(ctx), carried in the run’s input). Used only when the owner listed it (SMS_TO_PHONE or SMS_ALLOWED_PHONES); otherwise the text goes to SMS_TO_PHONE.

StepTextOwnerResult

StepTextOwnerResult = { sent: true; to: string; } | { sent: false; why?: string; }

What stepTextOwner resolves: sent, to whom — or not, and when there is one, a reason a person can hear. why is absent when texting is simply not set up (no SMS_TO_PHONE).

to is personal data, and a step’s result is JOURNALED: return what the body needs from it rather than the whole result when the run’s output is shown to anyone but the owner.

TranscribeDurablyHost

TranscribeDurablyHost = object

Input

The two context methods the composition is made of, with the step NAMES unconstrained — the PollHost shape, for the same reason: WorkflowContext refuses a non-literal name at compile time, and the names here are this module’s own literals. Method syntax, so a WorkflowContext is assignable.

Methods

poll()

poll<T>(name, check, options): Promise<PollResult<T>>

Type Parameters

T

T

Parameters

name

string

check

() => T | Promise<T>

options

PollOptions<T>

Returns

Promise<PollResult<T>>

step()

step<T>(name, fn, options?): Promise<T>

Type Parameters

T

T

Parameters

name

string

fn

() => T | Promise<T>

options?

StepOptions

Returns

Promise<T>

TranscribeDurablyOptions

TranscribeDurablyOptions = object

Input

What transcribeUploadDurably accepts.

Properties

apiKeyEnv?

optional apiKeyEnv?: string

Env var holding the credential, replacing ASSEMBLYAI_API_KEY. Passed to every call.

everyMs?

optional everyMs?: number

Durable wait between polls, in milliseconds. Defaults to TRANSCRIBE_POLL_EVERY_MS.

maxAttempts?

optional maxAttempts?: number

The UPLOAD step’s attempt budget. The submit and poll steps keep the engine’s default: they are JSON round trips, and the upload is the one call here whose cost is the file.

maxMs?

optional maxMs?: number

The polling budget, in milliseconds. Defaults to TRANSCRIBE_POLL_MAX_MS.

Spending it is a refusal naming the job, not a silent undefined — see the module doc.

maxWaits?

optional maxWaits?: number

Every step’s StepOptions.maxWaits: how many dated Retry-Afters the provider may ask for — a 429 on the submit, a 503 on a poll — before one spends an attempt. Passed to the upload, the submit and each poll check, because a rate limit met by a fan-out lands on whichever leg is in flight.

models?

optional models?: readonly string[]

Models to ask for. stepTranscribeSubmit’s own option, passed through.

params?

optional params?: Record<string, unknown>

Extra fields for the create-job body, VERBATIM — speaker_labels, language_code, and so on. stepTranscribeSubmit’s own option, passed through.

TranscribeProgress

TranscribeProgress = { done: false; status: "queued" | "processing" | string & object; } | { done: true; status: "completed" | string & object; transcript: Transcript; }

Where a submitted job has got to.

A discriminated union rather than { done, transcript? }, so a caller that checks done has the transcript without a second narrowing and one that forgets cannot read undefined text.

TranscribeRequestOptions

TranscribeRequestOptions = object

Input

Options every call in this family accepts.

Properties

apiKeyEnv?

optional apiKeyEnv?: string

Env var holding the credential, replacing ASSEMBLYAI_API_KEY.

Names a VARIABLE, not a key — the same contract every provider descriptor keeps, so nothing here can end up in a journaled step argument.

signal?

optional signal?: AbortSignal

Abort the request. Combined with the deadline rather than replacing it, so a caller passing one still cannot hang forever.

timeoutMs?

optional timeoutMs?: number

Deadline for this one request. Defaults to TRANSCRIBE_TIMEOUT_MS.

TranscribeSubmitOptions

TranscribeSubmitOptions = TranscribeRequestOptions & object

Input

What stepTranscribeSubmit accepts.

Type Declaration

models?

optional models?: readonly string[]

Models to ask for. Defaults to TRANSCRIBE_MODELS.

params?

optional params?: Record<string, unknown>

Extra fields merged into the create-job body, VERBATIM.

The async API has a large surface this deliberately does not mirror — speaker_labels, language_code, redact_pii, auto_chapters — and mirroring it would mean a wrapper that goes stale against the service every time it grows a feature. Nothing here interprets these; they are the caller’s request, sent as given.

audio_url and speech_models are set from this function’s own arguments and cannot be overridden here, so the two ways to say the same thing cannot disagree.

TranscribeSyncOptions

TranscribeSyncOptions = TranscribeRequestOptions & object

Input

What stepTranscribeSync accepts.

Type Declaration

filename?

optional filename?: string

Filename to declare in the multipart part. Defaults to "audio.wav".

The endpoint reads the CONTENT, so this is for the service’s logs and for a failure that names something a reader recognises — a segment’s timestamp, a part’s index — rather than for routing.

label?

optional label?: string

How this piece is named in a failure message. Defaults to filename.

The CALLER’s vocabulary, because a fan-out’s log is read by someone holding segment numbers rather than filenames.

model?

optional model?: string

Model to route to. Defaults to TRANSCRIBE_SYNC_MODEL.

Singular and a header rather than a body field — this endpoint’s shape, not the async API’s, and the two are deliberately not unified here because unifying them would mean inventing a name for something the service has two different names for.

type?

optional type?: string

Content type of the part. Defaults to "audio/wav".

Worth setting for anything that is not linear PCM in a WAV container; the endpoint accepts the common encodings and decodes on what it is told.

Transcript

Transcript = object

Sealed

A finished transcript, as stepTranscribePoll answers with one.

Properties

durationMs

durationMs: number

How long the recording runs, from the PROVIDER’s own measurement rather than from a byte count — it decoded the file and this did not.

id

id: string

The job id, so a caller can quote it in a log or fetch it again later.

text

text: string

What was said, trimmed. Never empty — see stepTranscribePoll.

UploadInfo

UploadInfo = UploadRecord & object

Sealed

What a stored upload is, minus its bytes — as a STEP reads it.

UploadRecord with one derived field on top, filled by every reader on @alexkroman1/aai/step (stepUploadInfo, stepReadUpload, stepRequireCompleteUpload, stepWriteUpload).

Type Declaration

label

label: string

What to CALL this upload in a sentence: its name, or its id when the uploader named none.

Derived here rather than at each call site because every step that narrates (“Transcribing call.m4a”) wrote stored.name || uploadId itself, and two templates had already diverged on the fallback ("the recording" against the id). It is the string a progress line and a run’s output should show a person; name stays the raw declaration for a caller that needs to know the uploader gave none.

UploadRange

UploadRange = object

Sealed

A half-open window of an upload’s bytes, [start, end).

Properties

end

end: number

start

start: number

UploadRecord

UploadRecord = object

Sealed

What a STORE answers about an upload: the record as stored, minus its bytes.

UploadInfo is this plus UploadInfo.label, which the reader module derives; a store never writes it. See labelUpload (/host-internal) for why the two shapes are split.

Properties

complete

complete: boolean

Whether every byte is in.

The one field a body waiting on a streamed upload may branch on. false means more may arrive; a size that has stopped growing means only that nothing arrived recently, which a slow link and a dead client both produce. An ordinary upload is true from the moment it exists, because it does not exist until it is finished.

id

id: string

The handle a run input carries.

Minted by the store for an ordinary upload, and CHOSEN BY THE CALLER for a streamed one — see UPLOAD_TOKEN_RE, which is what a chosen id has to satisfy.

name

name: string

Filename the uploader gave, or "" when it named none.

ranges?

optional ranges?: readonly UploadRange[]

Which windows have LANDED, for an unfinished upload that arrived as parts.

Absent for every other upload, and that absence is the honest answer rather than an omission: a whole-file write has no windows (its bytes are one contiguous prefix, which UploadInfo.size already states), and a finished parts upload is covered end to end by construction.

A READER may act on it, and stepReadUpload already does. The store maps a window onto the objects covering it and never consults the prefix, so the hazard is a read STRADDLING a hole, and clamping to the containing run protects exactly that while making a landed window readable. Without it a parts upload publishes nothing a run can use until its first window lands, which under a fan-out is the end of the upload; readableEnd carries the measurement.

The other reader is the UPLOADER: a client re-sending a parts upload can skip the windows that are already stored instead of sending the file again, which is the difference between resuming a recording and starting it over.

Sorted, non-overlapping, and half-open like every other range here.

A LIST rather than a single offset, and that is the whole of what it buys. The obvious cheaper shape is one number — “everything up to here has landed” — which is what a sequential append protocol reports (tus’s Upload-Offset, where a PATCH at any other offset is a 409). A single cursor cannot represent a GAP at all, so under it an upload whose second part was lost has to re-send everything after the first, and a fan-out that lands parts out of order has nothing to report until they happen to join up. UploadInfo.size already IS that number. This is the strictly larger fact.

Absent also means “cannot say”, not “nothing landed” — the store may decline to report windows, and an agent too old to have this field says nothing either. A reader’s answer to an absent list is therefore to assume nothing about what is stored, which for an uploader means sending the file.

size

size: number

Bytes STORED so far.

The whole file for a finished upload, and a growing number for one still arriving — which is why stepReadUpload clamps to it rather than to anything the uploader declared. Never trust it as a total: see UploadInfo.complete.

type

type: string

MIME type the uploader declared, or "". Never sniffed from the bytes.

UploadSlice

UploadSlice = object

Sealed

One window of an upload, as stepReadUpload resolves it.

Properties

bytes

bytes: Uint8Array

The requested bytes, clamped to the file.

end

end: number

One PAST the last byte offset returned, after clamping.

info

info: UploadInfo

The upload this came from — size is the WHOLE file, not this window.

start

start: number

First byte offset returned, after clamping.

WavFormat

WavFormat = object

Sealed

A WAV’s fmt chunk plus where its samples actually live.

Properties

bitsPerSample

bitsPerSample: number

Bits per sample.

channels

channels: number

Interleaved channel count.

dataEnd

dataEnd: number

Byte offset one past the last readable sample: the declared data length and the file’s real length, whichever is smaller.

dataStart

dataStart: number

Byte offset of the first sample — the END of the data chunk’s own header, NOT WAV_HEADER_BYTES. See this module’s doc for what assuming 44 costs.

sampleRate

sampleRate: number

Samples per second, as the file declares it.

WriteUploadOptions

WriteUploadOptions = object

Input

Options for stepWriteUpload.

Properties

name?

optional name?: string

Filename to store, e.g. "summary.wav".

Worth passing even though nothing reads it: it is what UploadInfo.name answers, so it is the name a page puts on a download link and the string a person sees instead of an opaque id.

type?

optional type?: string

MIME type to store, e.g. "audio/wav".

This one IS read — the byte route serves it as Content-Type, so a browser given an upload with none downloads a file it will not play inline. There is no sniffing anywhere in the store, by design.

Variables

CALL_OVER_STATUSES

const CALL_OVER_STATUSES: ReadonlySet<PlacedCallStatus>

The PlacedCallStatus values nothing moves off — completed, busy, no-answer, failed, canceled. What a loop following a call with stepCallStatus stops on.

Published because every body that dialled wrote its own copy of the list, and a copy missing canceled polls a call the API already hung up until its budget runs out.

DEFAULT_CLIENT_DELIVERY_ATTEMPTS

const DEFAULT_CLIENT_DELIVERY_ATTEMPTS: number

A step maxAttempts that rides out an hour-long device outage: 120 attempts at stepNotifyClient’s 30-second retry.

STEP_SPEAK_SAMPLE_RATE

const STEP_SPEAK_SAMPLE_RATE: number

Sample rate stepSpeak asks for when a caller names none.

TRANSCRIBE_API

const TRANSCRIBE_API: string

The async API’s base.

TRANSCRIBE_DURABLY_STEPS

const TRANSCRIBE_DURABLY_STEPS: object

The three journal names transcribeUploadDurably writes under.

Exported so a spec driving a body over the /testing recorder can assert the journal shape by these rather than by copying the strings — and so a aai workflow listing reads the same way as the code that produced it.

Type Declaration

poll

readonly poll: "transcribePoll"

submit

readonly submit: "transcribeSubmit"

upload

readonly upload: "transcribeUpload"

TRANSCRIBE_MODELS

const TRANSCRIBE_MODELS: readonly string[]

The models a job asks for when a caller names none.

speech_models, PLURAL and an array. The singular speech_model is deprecated on the async API and answers 400 for any current model name — which is exactly the fault that produced the retry measurement in the module doc. Omitting the field entirely is legal and routes to the service’s default; naming it is what stops a default change silently moving a workflow’s output.

TRANSCRIBE_POLL_EVERY_MS

const TRANSCRIBE_POLL_EVERY_MS: number

How often transcribeUploadDurably polls when the caller names no interval: every 10 seconds.

The async API finishes a recording in a fraction of its length, so a tighter interval buys seconds on a job that takes minutes and costs a journal entry per round; a looser one leaves a short recording waiting on the sleep rather than the provider.

TRANSCRIBE_POLL_MAX_MS

const TRANSCRIBE_POLL_MAX_MS: number

The polling budget when the caller names none: one hour.

Generous on purpose — a two-hour call transcribes in well under it — and sized so the failure it bounds is a job the provider never finishes rather than a long one. A COUNT of sleeps rather than a clock (PollOptions.maxMs says why), so floor(maxMs / everyMs) rounds of the default is 360.

TRANSCRIBE_SYNC_MODEL

const TRANSCRIBE_SYNC_MODEL: string

Required on every sync request; the endpoint routes on it.

TWILIO_ACCOUNT_SID_ENV

const TWILIO_ACCOUNT_SID_ENV: "TWILIO_ACCOUNT_SID" = "TWILIO_ACCOUNT_SID"

The step env key stepPlaceCall reads the Twilio account SID from.

TWILIO_AUTH_TOKEN_ENV

const TWILIO_AUTH_TOKEN_ENV: "TWILIO_AUTH_TOKEN" = "TWILIO_AUTH_TOKEN"

The step env key stepPlaceCall reads the Twilio auth token from.

WAV_HEADER_BYTES

const WAV_HEADER_BYTES: number

Bytes of WAV header encodeWav writes — RIFF, fmt , and data.

References

PollResult

Re-exports PollResult