# Reference Kassette is published as four packages. `@usekassette/core` provides the journal API (`start`, `resume`, `fork`, `Run.record`) and storage backends. `@usekassette/kassette` provides the workflow wrapper (`kassette()`, `Context`, `ctx.step`, `ctx.suspend`, `ctx.parallel`, `ctx.sleep`). `@usekassette/s3` provides an S3-compatible `ObjectStoreClient` for `RemoteStorage`. `@usekassette/cli` provides the `kassette` command for inspecting and forking journals. CLI usage and flags are documented in [Operations](operations.md#the-cli). ## Journal file layout Each `runId` has one append-only journal. The journal is newline-delimited JSON: one JSON entry per line. `LocalStorage` writes `{dir}/{runId}.jsonl`. `RemoteStorage` writes `{runId}/journal.jsonl`, or `{prefix}/{runId}/journal.jsonl` when a prefix is set. Values stored in the journal must be JSON-compatible. `metadata`, step `result`, and resume `value` are written with `JSON.stringify` and read with `JSON.parse`. Cycles and `BigInt` throw. Dates become strings. `undefined` object fields are omitted. ## Entry envelope ```typescript interface BaseEntry { readonly session: number; readonly timestamp: string; // ISO 8601 } type Entry = StartEntry | StepEntry | SuspendEntry | ResumeEntry | CompleteEntry | ErrorEntry | CancelEntry; type JournalEntry = Entry & { readonly offset: number }; ``` `session` is the session number of the writer that appended the entry. It increases across the run. `timestamp` is when the entry was appended. `offset` is the position assigned by storage; it starts at 0 and increases by 1 for each entry. Entries returned by `readAll` include `offset`. Entries passed to `append` do not, because storage assigns the offset. ## Entry types ### start Marks the beginning of a session. ```typescript interface StartEntry extends BaseEntry { readonly type: 'start'; readonly version?: string; readonly source?: { readonly runId: string; readonly fromOffset: number }; readonly metadata?: unknown; } ``` `version` is written only when the caller supplies one. Version checks compare the caller's version to the first `start` entry that has a version. `source` is written on the session that continues a forked run, not on the initial `start` entry that creates the fork journal. `metadata` is written on the first `start` entry of a new run. For a fork, inherited metadata is written on that initial `start` entry. ### step Records the result of one `record()` call. ```typescript interface StepEntry extends BaseEntry { readonly type: 'step'; readonly stepId: string; readonly name: string; readonly result: unknown; } ``` `stepId` is the unique stored identifier (`llm`, `llm#2`, `llm#3`). `name` is the caller-supplied name before repeat numbering. Step names must not contain `#`; that character is reserved as the repeat separator. Both fields are stored so replay can check that the current call is asking for the same step result that was recorded earlier. A mismatch throws `ReplayMismatchError`. See [Concurrency](concurrency.md) for how parallel branch names are prefixed. ### suspend Records that the session is waiting for an external event. ```typescript interface SuspendEntry extends BaseEntry { readonly type: 'suspend'; readonly reason: string; readonly waitingFor: string; readonly timeout?: string; } ``` `waitingFor` is the event name. `reason` is a human-readable note (defaults to `Waiting for event: ${eventName}`). `timeout` is an ISO 8601 deadline. Kassette does not wake itself up when the deadline passes. The next `start()` or `resume()` checks the deadline; if it has expired, Kassette writes `cancel` and throws `CancelledError`. ### resume Carries an event payload into the run. ```typescript interface ResumeEntry extends BaseEntry { readonly type: 'resume'; readonly eventName: string; readonly value: unknown; } ``` Written by `resume()` after it opens a new session and storage has rejected stale writers. If a matching `resume` entry already exists, Kassette does not write a duplicate. The value already in the journal wins (Journal Authority). On replay, the matching `waitForEvent` call returns the stored `value` instead of suspending. ### complete Terminal. The workflow returned successfully. ```typescript interface CompleteEntry extends BaseEntry { readonly type: 'complete'; } ``` After this entry is written, `start()` and `resume()` on the same run throw `TerminalRunError`. Forking _from_ a complete source run is allowed because the fork gets its own journal. Use a new target `runId` for forks. ### error Terminal. The workflow threw an error that was not `SuspendError`. ```typescript interface ErrorEntry extends BaseEntry { readonly type: 'error'; readonly name?: string; readonly message: string; readonly stack?: string; } ``` Written by `Run.fail(error)`. After this entry is written, `start()` and `resume()` on the same run throw `TerminalRunError`. Kassette stores the thrown error as `name`, `message`, and `stack`. There is no machine-readable error code. ### cancel Terminal. The run was cancelled before completing. ```typescript interface CancelEntry extends BaseEntry { readonly type: 'cancel'; readonly reason?: string; } ``` Today, only suspend-timeout auto-cancel writes this entry. When `start()` or `resume()` sees an expired `suspend.timeout`, it writes a fresh `start` and a `cancel` with `reason: 'suspend_timeout_expired'`, then throws `CancelledError`. There is no public cancel API. ## @usekassette/core ### Storage interface ```typescript interface Storage { append(runId: string, entry: Entry): Promise; readAll(runId: string): Promise; list(): Promise; } ``` The `Storage` interface has three methods. `append` must write an entry atomically: the entry is fully written or not written at all. It assigns an increasing offset and rejects writes from older sessions with `FencedError`. `readAll` returns every entry in append order. `list` returns every `runId` present in storage. Core includes two storage backends. Third-party code can implement this interface for any backend that can append safely and reject stale writers. ```typescript new LocalStorage(dir: string) new RemoteStorage(client: ObjectStoreClient, options?: { prefix?: string }) ``` `LocalStorage` writes one JSONL file per run under `dir`. It uses a lockfile held for the whole session to reject stale writers. `RemoteStorage` writes one object per run at `{runId}/journal.jsonl` or `{prefix}/{runId}/journal.jsonl`. It rejects stale writers with conditional object writes: `If-Match` for updates and create-if-absent for first writes. It retries `PreconditionFailedError` 5 times. `@usekassette/s3` provides the S3-compatible adapter. ### Run interface ```typescript interface Run { readonly runId: string; readonly metadata: unknown; record(name: string, fn: () => Promise, options?: RecordOptions): Promise; waitForEvent(eventName: string, options?: WaitForEventOptions): Promise; complete(): Promise; fail(error: Error): Promise; } ``` Returned by `start`, `resume`, and `fork`. A `Run` represents the active session. `metadata` is the value passed to the first `start()` for the run and replayed on every later session. #### record ```typescript record(name: string, fn: () => Promise, options?: RecordOptions): Promise ``` On first execution, `fn` runs and the result is written as a `step` entry. On replay, Kassette returns the journaled result and does not run `fn`. `name` is the caller-supplied identifier and must not contain `#`. The stored `stepId` is `name` for the first call, then `name#2`, `name#3`, etc. for repeats. Throws `ReplayMismatchError` if the stored step name does not match the current call's name. #### waitForEvent ```typescript waitForEvent(eventName: string, options?: WaitForEventOptions): Promise ``` If a `resume` entry for `eventName` already exists, returns its value. Otherwise writes a `suspend` entry, marks the session suspended, and throws `SuspendError`. Do not swallow `SuspendError`; let it escape so the process can stop or the wrapper can return `suspended`. Re-using an event name within the same run throws `UsageError`. Use unique names, such as `approval:${itemId}`, for repeated waits. #### complete / fail ```typescript complete(): Promise fail(error: Error): Promise ``` Write a terminal entry and close the session. Later `record` or `waitForEvent` calls throw `SessionClosedError`. Calling `complete` or `fail` after the session has suspended throws `SuspendedError`. `kassette()` calls these for you. Direct callers of `start`, `resume`, or `fork` must call one of them explicitly. #### RecordOptions ```typescript interface RecordOptions { onReplay?: (result: T) => void; } ``` `onReplay` runs synchronously when replay returns a stored result. Streaming integrations use it to send stored events to the consumer when the step is replayed instead of recomputed. See `examples/vercel-ai-sdk` for the streaming pattern. #### WaitForEventOptions ```typescript interface WaitForEventOptions { timeout?: string; reason?: string; } ``` `timeout` is an ISO 8601 absolute deadline written into the suspend entry. `reason` is a human-readable note. Both are optional. Kassette does not wake itself up when the deadline passes. The next `start()` or `resume()` enforces the deadline. ### Lifecycle functions #### start ```typescript function start(storage: Storage, runId: string, options?: StartRunOptions): Promise; interface StartRunOptions extends RunOptions { metadata?: unknown; } interface RunOptions { version?: string | undefined; } ``` Opens a new session for `runId`. On a new run, writes the first `start` entry and stores `metadata`. On recovery, replays the existing journal, checks `version` against the first `start` entry that has a version, and checks `metadata` only when the caller provides it. If metadata is omitted, Kassette uses the value already in the journal. Throws `TerminalRunError` if the run is terminal, `EventPendingError` if the run is suspended (use `resume`), and `MetadataMismatchError` or `VersionMismatchError` on validation failures. #### resume ```typescript function resume(storage: Storage, runId: string, eventName: string, value: unknown, options?: RunOptions): Promise; ``` Delivers an event to a suspended run and opens the next session. Writes a new `start` entry, then a `resume` entry carrying `value`. If a `resume` entry for `eventName` already exists, the value in the journal wins, the new `value` is ignored, and no duplicate `resume` entry is written. This makes retrying the same resume call safe after a crash. Throws `UsageError` if the run is not suspended, or if it is suspended on a different event name. #### fork ```typescript function fork( storage: Storage, runId: string, source: { runId: string } & ({ fromOffset: number } | { fromStepId: string }), options?: RunOptions, ): Promise; ``` Branches a new run from the beginning of an existing run. The cut point is `fromOffset`, or the offset of the first step entry matching `fromStepId`. Entries with source offsets lower than the cut point are copied. The cut point itself is not copied. Only `step` and `resume` entries are copied; `start`, `suspend`, and terminal entries are not copied. The target first gets an initial `start` entry, with inherited metadata when present. A later session `start` records `source: { runId, fromOffset }`. `version` comes from the new caller. The copy is not atomic. A crash during the copy leaves a partial target journal with no terminal entry. Reopening that target run treats it as crashed work and replays or re-executes from the first missing step. Use a new target `runId`. Throws `UsageError` if `fromStepId` does not match any step. #### Validation order `start` and `resume` share this order: 1. Read the target journal with `storage.readAll(runId)`. 2. If the run is terminal (`complete`, `error`, or `cancel`), throw `TerminalRunError`. 3. Rebuild replay state and check `version`. If the caller provided a version and it differs from the first journaled version, throw `VersionMismatchError`. 4. If the run is suspended with an expired `timeout`, write a fresh `start` plus `cancel`, then throw `CancelledError`. 5. `start`: if the run is still suspended, throw `EventPendingError`; otherwise check provided metadata against journaled metadata and write a fresh `start`. 6. `resume`: require either a matching unresolved suspend or an existing matching `resume` entry from an earlier retry. Write a fresh `start`; write `resume` only if it was not already journaled. `fork` follows a different path. It reads the source journal, resolves the cut point, writes an initial `start` to the target, copies eligible `step` and `resume` entries, then reads the target and opens a session with the normal terminal and version checks. It does not auto-cancel expired suspends in the source run. ### Errors All errors extend `KassetteError`. One `instanceof KassetteError` catches every kassette throw. | Error | When | Notable fields | | -------------------------------------- | -------------------------------------------------------------------------------------------- | -------------------------------------- | | `UsageError` | Caller misuse (bad input, wrong call order, misconfiguration). | | | `TerminalRunError` (`UsageError`) | `start`/`resume` on a terminal run. | `terminalState` | | `MetadataMismatchError` (`UsageError`) | Recovery `start` with provided metadata that differs from journaled metadata. | `storedMetadata`, `providedMetadata` | | `EventPendingError` (`UsageError`) | `start` on a suspended run with an unresolved event. Use `resume`. | `waitingFor` | | `SuspendError` | Control-flow error thrown by `waitForEvent` when the workflow suspends. | `eventName` | | `SuspendedError` | Session method called after suspension, usually because a `catch` swallowed `SuspendError`. | | | `SessionClosedError` | Session method called after `complete()` or `fail()`. | | | `VersionMismatchError` | `options.version` disagrees with the first journaled version. | `storedVersion`, `currentVersion` | | `CancelledError` | Suspend-timeout auto-cancel. | `reason` | | `ReplayMismatchError` | Replay found a stored step name that does not match the current call. | `stepId`, `expectedName`, `actualName` | | `FencedError` | Storage rejected an append from an older session. | `rejectedSession`, `activeSession` | | `WriteContentionError` | Backend cannot get exclusive write access (lockfile held, conditional-write retries failed). | | | `PreconditionFailedError` | `ObjectStoreClient` conditional put failed; `RemoteStorage` catches this during retries. | | | `JournalCorruptionError` | Journal data cannot be parsed or fails structural validation. | `line` | | `InternalError` | Library invariant violated. File a bug. | | Every instance carries `runId` (when known) on the base class. ### Helpers #### createRunId ```typescript function createRunId(): string; ``` Returns a fresh UUID via `crypto.randomUUID()`. Callers can create run IDs another way if they prefer. Every API takes `runId` as a string. #### isSuspendError ```typescript function isSuspendError(err: unknown): err is SuspendError; ``` Cross-package-safe `SuspendError` check. Prefer this over `instanceof` when bundled code may contain more than one copy of `@usekassette/core`. #### isTerminal ```typescript function isTerminal(entry: Entry): entry is CompleteEntry | ErrorEntry | CancelEntry; ``` Type guard. Returns `true` for `complete`, `error`, or `cancel` entries. Useful when walking a journal looking for the closing entry. #### getMetadata ```typescript function getMetadata(entries: readonly JournalEntry[]): T | undefined; ``` Pulls the caller-supplied metadata from the first `start` entry of a journal. Returns `undefined` if the journal is empty or no metadata was passed at start time. #### runStatus / RunStatus ```typescript function runStatus(entries: readonly JournalEntry[]): RunStatus; type RunStatus = | { status: 'completed' } | { status: 'failed'; message: string; name?: string; stack?: string } | { status: 'cancelled'; reason?: string } | { status: 'suspended'; waitingFor: string; timeout?: string } | { status: 'unsettled' }; ``` Stateless helper over `JournalEntry[]`. It is the shared answer to "what is this run doing right now?" `unsettled` covers a live session in another process, a crashed-and-abandoned session, and an empty journal. The journal alone cannot distinguish those cases, so live-process detection needs a separate signal. #### ObjectStoreClient / GetObjectResult / PreconditionFailedError ```typescript interface ObjectStoreClient { getObject(key: string): Promise; putObject(key: string, content: string, etag: string | undefined): Promise; listPrefixes(prefix: string): Promise; } interface GetObjectResult { readonly content: string; readonly etag: string; } class PreconditionFailedError extends KassetteError {} function isPreconditionFailedError(err: unknown): err is PreconditionFailedError; ``` The extension point for `RemoteStorage`. Implementations wrap a specific object-store SDK and translate its "write only if the object version matches" feature into conditional writes. `putObject` with `etag: string` updates only if the current object still has that ETag. `putObject` with `etag: undefined` creates the object only if it does not already exist. Throw `PreconditionFailedError` on HTTP 412. `listPrefixes(prefix)` must return run IDs with the parent prefix removed and no trailing slash. `isPreconditionFailedError` is the cross-package-safe check used by `RemoteStorage`. `@usekassette/s3` provides the S3 adapter. ## @usekassette/s3 ### S3ObjectStoreClient ```typescript class S3ObjectStoreClient implements ObjectStoreClient { constructor(options: S3ObjectStoreClientOptions); } interface S3ObjectStoreClientOptions { readonly bucket: string; readonly client?: S3Client; // pre-configured @aws-sdk/client-s3 instance readonly clientConfig?: S3ClientConfig; // ignored if `client` is provided } ``` Implements `ObjectStoreClient` using the AWS SDK v3 (`@aws-sdk/client-s3`). Pass it to `new RemoteStorage(client, { prefix })` to journal runs into an S3-compatible bucket. It works with AWS S3, Cloudflare R2, DigitalOcean Spaces, GCS interop mode, Tigris, and MinIO. The backend must provide read-after-write consistency and conditional writes (`If-Match` / `If-None-Match`). Eventually consistent or CDN-fronted stores can let stale writers overwrite newer sessions. Maps the three `ObjectStoreClient` operations to `GetObjectCommand`, `PutObjectCommand`, and `ListObjectsV2Command`. `putObject` sends `If-Match: ` for updates and `If-None-Match: *` for first writes. HTTP 412 from either path is detected via `$metadata.httpStatusCode`, with an `error.name === 'PreconditionFailed'` fallback for non-AWS providers, and re-thrown as `PreconditionFailedError` for `RemoteStorage`'s conditional-write retry loop. `getObject` returns `null` on `NoSuchKey`. `@aws-sdk/client-s3` is a peer dependency. ## @usekassette/kassette ### kassette() ```typescript function kassette = Record>( fn: (ctx: Context, input: TInput) => Promise, options: KassetteOptions, ): Kassette; ``` Wraps a workflow function and returns an object with `start`, `resume`, and `fork`. The wrapper opens the session, runs `fn`, calls `complete()` on success, calls `fail()` on non-suspend errors, and returns `suspended` when it sees `SuspendError`. `TEvents` maps event names to payload types for `ctx.suspend` and `Kassette.resume` type checking. #### KassetteOptions ```typescript interface KassetteOptions { storage: Storage; version?: string | undefined; onFinish?: (result: RunResult) => void | Promise; onError?: (info: { runId: string; error: Error }) => void | Promise; } ``` `storage` is the journal backend. `version` is recorded when supplied and checked on later sessions. `onFinish` runs for settled wrapper results (`success`, `suspended`, or `failed`) from `start`, `resume`, or `fork`. `onError` runs only for `failed`. Both hooks are for logging or metrics. Hook errors are caught and logged, not passed back to the caller. Core lifecycle errors thrown before the workflow function runs (`TerminalRunError`, `VersionMismatchError`, `CancelledError`, etc.) are passed to the caller as thrown errors. They are not converted into `RunResult`, and hooks are not called. ### Context `Context` is the API available inside the workflow function. Each method forwards to the core run or a small helper. #### ctx.runId ```typescript ctx.runId: string ``` The current run's identifier. Stable across sessions of the same run. #### ctx.input ```typescript ctx.input: unknown ``` The input passed to the original `start()`. On replay, returns the value stored as metadata on the first `start` entry. The type is `unknown` because the journal stores it without TypeScript type information. The same value also arrives as the typed second argument to the workflow function, which is usually what you want. #### ctx.step ```typescript ctx.step(name: string, fn: () => Promise, opts?: StepOptions): Promise ``` Thin wrapper over `Run.record`. On first execution, `fn` runs and the result is written to the journal. On replay, Kassette returns the journaled result and does not run `fn`. Names containing `#` throw `UsageError`. `StepOptions.retry` wraps `fn` in a retry loop inside one `record()` call. All attempts run in memory before any journal write. A successful return is written to the journal exactly once. If every attempt fails, the last error throws out of `ctx.step` and no step entry is written. The workflow can catch it like any other exception. #### ctx.suspend ```typescript ctx.suspend(eventName: K, opts?: WaitForEventOptions): Promise ``` Calls `Run.waitForEvent` with payload typing from `TEvents`. If a `resume` entry for `eventName` already exists, returns the journaled payload. Otherwise writes a `suspend` entry and throws `SuspendError`. Let `SuspendError` escape. The wrapper turns it into `RunResult` with `status: 'suspended'`. #### ctx.sleep ```typescript ctx.sleep(ms: number): Promise ``` Crash-aware in-process delay. Writes an absolute wake deadline as a step (`delay:${ms}ms`) and waits the remaining time. On replay after a crash, it waits only the time that was still left. Use this only for short waits. If the wait could outlive the process, use `ctx.suspend` with a timeout and an external timer. #### ctx.parallel ```typescript ctx.parallel) => Promise>>( branches: T, ): Promise<{ [K in keyof T]: Awaited> }> ``` Runs named branches concurrently with `Promise.allSettled`. Each branch receives a `Context` whose `step` calls are prefixed with the branch key (`${key}:${name}`), so step IDs from different branches do not collide. If any branch throws `SuspendError`, the whole call throws `SuspendError` and the parallel block suspends. If any branch throws a non-suspend error and no branch suspended, the first error throws. Event names are not prefixed because they are part of the external resume API. #### StepOptions ```typescript interface StepOptions { retry?: RetryConfig; onReplay?: (result: T) => void; } ``` `retry` activates the in-memory retry loop. `onReplay` mirrors `RecordOptions.onReplay` for streaming integrations. #### RetryConfig ```typescript interface RetryConfig { maxAttempts: number; delay?: number; // initial delay in ms (default 1000) backoffRate?: number; // multiplier per attempt (default 1) maxDelay?: number; // cap on delay (default Infinity) } ``` `maxAttempts` is the maximum number of tries. Wait time between attempts starts at `delay`, multiplies by `backoffRate` each round, and is capped at `maxDelay`. All retries run in-process. No journal entry is written until success. ### Kassette object `kassette()` returns an object with three methods. Each opens a session through the core lifecycle and runs the workflow function to completion, suspension, or failure. #### workflow.start ```typescript start( input: Record extends TInput ? TInput | undefined : TInput, options?: { runId?: string }, ): Promise> ``` Begins a run. Uses `options.runId` if provided; otherwise creates one with `createRunId`. `input` is stored as metadata on the first `start` entry. TypeScript allows `undefined` input when `TInput` permits `Record`. At runtime, Kassette journals whatever value is passed. Returns once the wrapper has a settled result. Calling `start` again with the same `runId` after a crash recovers and replays. #### workflow.resume ```typescript resume( runId: string, event: { [K in keyof TEvents & string]: { eventName: K; value: TEvents[K] } }[keyof TEvents & string], ): Promise> ``` Delivers an event to a suspended run. The wrapper opens a new session through core `resume`, replays to the suspend point, and continues. Retrying after a crash is safe: repeat calls with a different `value` for the same event ignore the new value and use the journaled value. #### workflow.fork ```typescript fork( source: { runId: string } & ({ fromOffset: number } | { fromStepId: string }), options?: { runId?: string }, ): Promise> ``` Branches a new run from the beginning of `source.runId` up to the cut point. The new `runId` is `options.runId` or a fresh ID. Replays the copied entries and continues live from the cut point. The source run is unchanged. ### RunResult ```typescript type RunResult = | { status: 'success'; result: TOutput; runId: string } | { status: 'failed'; error: Error; runId: string } | { status: 'suspended'; event: string; runId: string }; ``` Returned by `start`, `resume`, and `fork`. `success` carries the workflow's return value. `failed` carries the thrown error, which has already been written to the journal. `suspended` carries the event name the workflow is waiting on. The next call should be `workflow.resume(runId, { eventName, value })`. Switch on `status` to handle each case. ### Re-exports from core `@usekassette/kassette` re-exports the symbols workflow authors usually need, so most workflows do not need to import `@usekassette/core` directly. - `LocalStorage`, `RemoteStorage`, `createRunId`, `runStatus` - `PreconditionFailedError`, `isPreconditionFailedError` - `KassetteError`, `UsageError`, `SuspendError`, `isSuspendError`, `TerminalRunError`, `EventPendingError`, `MetadataMismatchError`, `ReplayMismatchError`, `FencedError`, `WriteContentionError`, `VersionMismatchError`, `CancelledError`, `JournalCorruptionError` - Types: `Storage`, `WaitForEventOptions`, `RunStatus`, `ObjectStoreClient`, `GetObjectResult`, and all journal entry types For lower-level work, import from `@usekassette/core`: `start`, `resume`, `fork`, the `Run` interface, `isTerminal`, `getMetadata`, `InternalError`, `SessionClosedError`, `SuspendedError`, and custom storage internals.