diff --git a/.changeset/engine-runtime.md b/.changeset/engine-runtime.md
new file mode 100644
index 0000000..cabed8f
--- /dev/null
+++ b/.changeset/engine-runtime.md
@@ -0,0 +1,13 @@
+---
+"durable-workflows": minor
+---
+
+Engine runtime: `durableWorkflows(options)`. The connector that ties the authoring surface and execution host to a persistence store, exposing the full instance lifecycle — `create`, `get`, `continueWorkflow`, `terminate`, `evict`, `restart`, `dispose`, and `pendingPromises`.
+
+Each lifecycle call resolves to one replay turn: load the instance record + boundary cache from the `WorkflowStore`, resolve the pinned definition version via the store's `getDefinition`, hydrate (and cache, per pinned version) a runner mounted with the plugin shims + any `alias` re-export modules, execute with per-run handlers, then persist the grown cache and new status and return a `RunOutcome`. Storage is minimal (instances + one opaque cache blob each); retry, scheduling and wake-ups stay the caller's job; resume is plain re-execution.
+
+The store is ONE persistent world: instances, caches, and READ-ONLY definition access (`getDefinition(name, version?)` — no version means "active/latest", the pinned version on replay). Writing definitions (upload, versioning, rollback) is deliberately the application's own deploy layer against the same backend; the contract is that a handed-out `(name, version)` stays fetchable and byte-identical while instances pin it.
+
+Plugin handlers receive the structured `DurableHandlerInput` (`{ instanceId, workflow, run, stepId, payload }`) — the `durable-workflows:internal` `operation` shim now forwards the boundary key alongside the args so the engine can recover `stepId` (the kernel never hands a handler its key), and `payload` is the full forwarded argument list. The host now forwards iso4 resource limits (engine defaults ← `options.limits` ← per-definition `limits`).
+
+Ships an in-memory `memoryStore()` reference adapter (with a `deploy` seeding helper playing the deploy layer's role for tests).
diff --git a/AGENTS.md b/AGENTS.md
index 67e39b7..034c210 100644
--- a/AGENTS.md
+++ b/AGENTS.md
@@ -4,7 +4,7 @@
 
 This repo is a **pnpm monorepo** (modeled on the [iso4](https://github.com/schplitt/iso4) repo, releases via [changesets](https://github.com/changesets/changesets)) containing two published packages:
 
-- **`durable-workflows`** — a composable durable workflow library, actively inspired by [Cloudflare Workflows](https://developers.cloudflare.com/workflows/) and built on top of [`iso4`](https://github.com/schplitt/iso4). Workflows run inside a sandbox with a step-based API whose results are persisted, enabling suspend/resume, retries, and event-driven continuation. It consumes `durable-isolates` (`workspace:*`). The **authoring surface** is shipped so far: two preloaded virtual modules — `durable-workflows:workflow` (author-facing `defineWorkflow({ run })` + `step.do(id, fn)`) and `durable-workflows:internal` (the shim-facing wrapper over the kernel, for plugin authors) — plus `durableWorkflowHost`, the execution host that wraps a `durable-isolates` host: `hydrate({ workflow, plugins })` mounts a definition with the core auto-injected, and `runner.execute({ input, cache })` runs one replay turn (input baked into a generated entry as JSON). The higher-level engine runtime (store adapters, the `durableWorkflows()` lifecycle factory in `types.ts`) is not built yet.
+- **`durable-workflows`** — a composable durable workflow library, actively inspired by [Cloudflare Workflows](https://developers.cloudflare.com/workflows/) and built on top of [`iso4`](https://github.com/schplitt/iso4). Workflows run inside a sandbox with a step-based API whose results are persisted, enabling suspend/resume, retries, and event-driven continuation. It consumes `durable-isolates` (`workspace:*`). Three layers ship: the **authoring surface** — two preloaded virtual modules, `durable-workflows:workflow` (author-facing `defineWorkflow({ run })` + `step.do(id, fn)`) and `durable-workflows:internal` (the shim-facing wrapper over the kernel, for plugin authors) — the **execution host** `durableWorkflowHost` (`hydrate({ workflow, plugins, limits })` mounts a definition with the core auto-injected; `runner.execute({ input, cache, handlers, limits })` runs one replay turn, input baked into a generated entry as JSON), and the **engine runtime** `durableWorkflows({ store, plugins, alias, limits, onEvent })` — the full instance lifecycle (`create`/`get`/`continueWorkflow`/`terminate`/`evict`/`restart`/`dispose`/`pendingPromises`) over a `WorkflowStore` adapter. The store is ONE persistent world: instances, boundary-cache blobs, and READ-ONLY definition access (`getDefinition(name, version?)` — instances pin the version they were created on; a handed-out `(name, version)` must stay fetchable and byte-identical). Writing definitions (upload/versioning/rollback) is deliberately app-layer, not engine scope. Plugin handlers get `DurableHandlerInput` (`{ instanceId, workflow, run, stepId, payload }`); the `operation` shim forwards its boundary key as the leading arg so the engine can recover `stepId` (the kernel never passes handlers their key), and `payload` is the full forwarded argument list. `memoryStore()` is the reference store adapter (with a `deploy` seeding helper standing in for the app's deploy layer).
 - **`durable-isolates`** — the replay kernel `durable-workflows` builds on: durably execute one isolate program over a **keyed cache** of boundaries. Implemented as a **memoize-by-key router** — in-sandbox shims form a `key` and call the primitives from `durable-isolates:internal`: `durableCall(key, name, …args)` (host-handler work), `durableLookup`/`durableCommit` (sandbox-side checkpoints) and the nestable `boundary(key, fn)`/`nextKey(name)` sugar over them. The ambient boundary prefix is carried through iso4's `AsyncLocalStorage` (`@iso4/sandbox` ≥ 0.3.0), so nested boundaries key deterministically whether they run sequentially or in parallel (`Promise.all`). The host answers a boundary from the cache by `key` or dispatches the `name` handler. A handler throwing `SuspendIsolate` suspends the run (waiting record + abort); resume is always **re-execution** (the waiting boundary re-dispatches and the handler consults host state — there is no delivery API and no tokens); `handle.suspend()` suspends externally (drains in-flight handler IO into the cache, for server teardown). Determinism is a documented contract, not an enforced check (no divergence detection). The caller owns storage, retry/eviction (cache surgery), and reacting to pending operations.
 
 The project uses ESM modules, Vitest for testing, ESLint for code quality, and tsdown for builds. Node `>=26`.
@@ -17,13 +17,16 @@ The project uses ESM modules, Vitest for testing, ESLint for code quality, and t
 packages/
   durable-workflows/      # Published as `durable-workflows`
     src/
-      index.ts            # Public entry — types + durableWorkflowHost + specifier constants
-      types.ts            # Engine/host public type surface (the not-yet-built engine runtime)
-      host.ts             # durableWorkflowHost: hydrate({ workflow, plugins }) → runner.execute({ input, cache })
+      index.ts            # Public entry — types + durableWorkflows + durableWorkflowHost + memoryStore + specifier constants
+      types.ts            # Engine/store/plugin/outcome public type surface
+      engine.ts           # durableWorkflows(): instance lifecycle over a WorkflowStore, per-version runner cache, handler wrapping
+      memory-store.ts     # memoryStore(): in-memory WorkflowStore + `deploy` seeding helper (tests)
+      host.ts             # durableWorkflowHost: hydrate({ workflow, plugins, limits }) → runner.execute({ input, cache, handlers, limits })
       shim.ts             # Source of the two virtual modules + coreModules bundle (internal)
       internal.ts         # `./internal` export — shim-facing types (operation/boundary) for plugin authors
       workflow.d.ts        # `./workflow` export — ambient `declare module 'durable-workflows:workflow'` (copied to dist)
     tests/authoring-surface.test.ts  # Authoring surface + host, against the real kernel
+    tests/engine.test.ts  # Engine lifecycle end-to-end (memory store + real sandbox)
     __snapshots__/tsnapi/ # tsnapi public-API snapshots (index, internal)
     package.json          # workspace:* dep on durable-isolates
     tsconfig.json         # standalone (no root base)
@@ -50,7 +53,7 @@ pnpm-workspace.yaml       # `packages/*` + dependency catalog
 .changeset/               # changesets config + release docs
 ```
 
-`durable-workflows` public exports: `.` (types + `durableWorkflowHost` + `INTERNAL_SPECIFIER`/`WORKFLOW_SPECIFIER`), `./internal` (shim-facing types), and `./workflow` (author-facing ambient `.d.ts`). The core shim sources and `coreModules` are internal — the host mounts them; callers never do.
+`durable-workflows` public exports: `.` (types + `durableWorkflows` + `durableWorkflowHost` + `memoryStore` + `INTERNAL_SPECIFIER`/`WORKFLOW_SPECIFIER`), `./internal` (shim-facing types), and `./workflow` (author-facing ambient `.d.ts`). The core shim sources and `coreModules` are internal — the host mounts them; callers never do.
 
 Each package owns its own `src/` (public API exported from `src/index.ts`), a standalone `tsconfig.json` (no shared root base), `tsdown.config.ts`, and `vitest.config.ts`. Shared root config is just ESLint and the pnpm catalog.
 
@@ -85,7 +88,7 @@ Root scripts use `pnpm -r --filter="./packages/*"`; `lint`/`lint:fix` run ESLint
 - Each package has its own `vitest.config.ts`; put tests inside that package's `tests/` directory (or alongside source under `src/`) — both are covered by the package `tsconfig.json` `include`, so tests are type-checked against the same config as source
 - Use the `*.test.ts` file naming convention
 - Run `pnpm test:run` from the root for all packages (no watch), or run it inside a single package
-- Both packages have real suites (`durable-isolates/tests/kernel.test.ts`, `durable-workflows/tests/authoring-surface.test.ts`) and neither sets `passWithNoTests`. The `durable-workflows` suite mounts the real shims on a `durable-isolates` runner via `durableWorkflowHost`, so it exercises the whole chain (workflow → `:workflow` → `:internal` → `durable-isolates:internal` → host)
+- Both packages have real suites (`durable-isolates/tests/kernel.test.ts`, `durable-workflows/tests/authoring-surface.test.ts` + `tests/engine.test.ts`) and neither sets `passWithNoTests`. The `durable-workflows` suites mount the real shims on a `durable-isolates` runner (via `durableWorkflowHost` directly, and via the full engine with a `memoryStore`), so they exercise the whole chain (engine → workflow → `:workflow` → `:internal` → `durable-isolates:internal` → host → handlers)
 
 Example test structure:
 
diff --git a/packages/durable-workflows/README.md b/packages/durable-workflows/README.md
index 6d6dbec..b2f79fc 100644
--- a/packages/durable-workflows/README.md
+++ b/packages/durable-workflows/README.md
@@ -4,7 +4,7 @@ Durable workflows for JavaScript, inspired by [Cloudflare Workflows](https://dev
 
 Write a workflow as a plain async function. Each step's result is saved, so a workflow can pause for days, survive restarts and deploys, and continue right where it left off. No state machines, no manual bookkeeping.
 
-> Status: work in progress. The authoring surface and runner are here; the full engine (persistence, scheduling) lands next.
+> Status: work in progress. The authoring surface, the execution host and the engine (instance lifecycle over a pluggable store) are here; first-party capability plugins land next.
 
 ## Features
 
@@ -38,6 +38,43 @@ export default defineWorkflow({
 
 Run one loads the device, asks for approval, and pauses. Days later the approval arrives, the workflow replays through the saved step in milliseconds, applies the fix, and finishes. A deploy in between changes nothing.
 
+## Running workflows: the engine
+
+`durableWorkflows` runs instances of deployed workflow code over a store you provide:
+
+```ts
+import { durableWorkflows, memoryStore } from 'durable-workflows'
+
+const store = memoryStore() // or your own WorkflowStore adapter
+store.deploy('device-fix', 'v1', workflowCode) // in production your own deploy layer writes these rows
+
+const engine = durableWorkflows({
+  store,
+  plugins: {
+    'my:approvals': approvalsPlugin(myBackend), // keys are the import specifiers workflows use
+  },
+})
+
+// your triggers start instances — resolves the active version, pins it, runs the first turn
+app.post('/devices/:id/fix', async (req) => {
+  const outcome = await engine.create('device-fix', { deviceId: req.params.id })
+  // outcome.status: 'completed' | 'waiting' | 'failed'
+})
+
+// and your own wiring (webhook, cron, queue) resumes waiting instances:
+app.post('/approvals/:instanceId/decide', async (req) => {
+  await engine.continueWorkflow(req.params.instanceId)
+})
+```
+
+How the pieces divide:
+
+- **The store** (`WorkflowStore`) is one persistent world: instance records, each instance's saved step history, and read access to deployed workflow code (`getDefinition`). Implement it once against your database.
+- **Definitions are read-only to the engine.** Uploading, versioning and rollback belong to your app — you write code rows into the same backend the store reads. Instances pin the version they started on and always replay exactly that code, so rollbacks never disturb running instances. The one rule: never mutate or delete a version that instances still reference.
+- **Resuming is re-running.** `continueWorkflow(id)` replays the workflow over its saved history; a step that was waiting asks its plugin handler again, and the handler checks your systems (the approval row, the clock, the job status) to answer, keep waiting, or fail. There are no callbacks to register and nothing to inject.
+- **Scheduling is yours.** The engine never wakes anything up — your cron/webhooks/queues decide when to call `continueWorkflow`.
+- **Remediation:** `evict(id, stepId)` deletes a step (and everything after it) from history and replays; `restart(id)` replays from scratch; `terminate(id)` ends an instance.
+
 ## License
 
 MIT
diff --git a/packages/durable-workflows/__snapshots__/tsnapi/index.snapshot.d.ts b/packages/durable-workflows/__snapshots__/tsnapi/index.snapshot.d.ts
index 4fb639b..9c0840f 100644
--- a/packages/durable-workflows/__snapshots__/tsnapi/index.snapshot.d.ts
+++ b/packages/durable-workflows/__snapshots__/tsnapi/index.snapshot.d.ts
@@ -36,7 +36,6 @@ export interface DurableWorkflowsEngine {
 export interface DurableWorkflowsOptions {
   sandbox?: SandboxOptions;
   store: WorkflowStore;
-  resolveDefinition: (_: string, _?: string) => Promise<ResolvedDefinition | null>;
   plugins?: Readonly<Record<string, DurableWorkflowsPlugin>>;
   alias?: Readonly<Record<string, string>>;
   limits?: Partial<ResourceLimits>;
@@ -51,6 +50,9 @@ export interface FailedInstanceRecord extends InstanceRecordBase {
   status: "failed";
   error: SerializedError;
 }
+export interface MemoryWorkflowStore extends WorkflowStore {
+  deploy: (_: string, _: string, _: string, _?: Partial<ResourceLimits>) => void;
+}
 export interface PendingOperation {
   stepId: string;
   operation: string;
@@ -75,10 +77,12 @@ export interface WorkflowExecuteOptions {
   input?: unknown;
   cache: BoundaryCache;
   handlers?: PerExecuteHandlers;
+  limits?: Partial<ResourceLimits>;
 }
 export interface WorkflowHydrateOptions {
   workflow: string;
   plugins?: Readonly<Record<string, ModuleDefinition>>;
+  limits?: Partial<ResourceLimits>;
 }
 export interface WorkflowInstanceHandle {
   readonly id: string;
@@ -96,6 +100,7 @@ export interface WorkflowStore {
   deleteInstance: (_: string) => Promise<void>;
   getCache: (_: string) => Promise<BoundaryCache | null>;
   putCache: (_: string, _: BoundaryCache) => Promise<void>;
+  getDefinition: (_: string, _?: string) => Promise<ResolvedDefinition | null>;
 }
 // #endregion
 
@@ -146,6 +151,8 @@ export type RunOutcome = (RunOutcomeBase & {
 
 // #region Functions
 export declare function durableWorkflowHost(_?: DurableIsolatesOptions): DurableWorkflowHost;
+export declare function durableWorkflows(_: DurableWorkflowsOptions): DurableWorkflowsEngine;
+export declare function memoryStore(): MemoryWorkflowStore;
 // #endregion
 
 // #region Variables
diff --git a/packages/durable-workflows/__snapshots__/tsnapi/index.snapshot.js b/packages/durable-workflows/__snapshots__/tsnapi/index.snapshot.js
index 527cb81..9562988 100644
--- a/packages/durable-workflows/__snapshots__/tsnapi/index.snapshot.js
+++ b/packages/durable-workflows/__snapshots__/tsnapi/index.snapshot.js
@@ -3,6 +3,8 @@
  */
 // #region Functions
 export function durableWorkflowHost(_) {}
+export function durableWorkflows(_) {}
+export function memoryStore() {}
 // #endregion
 
 // #region Variables
diff --git a/packages/durable-workflows/package.json b/packages/durable-workflows/package.json
index 2d49123..828216f 100644
--- a/packages/durable-workflows/package.json
+++ b/packages/durable-workflows/package.json
@@ -56,6 +56,7 @@
   },
   "devDependencies": {
     "@schplitt/eslint-config": "catalog:",
+    "@types/node": "catalog:",
     "eslint": "catalog:",
     "tsdown": "catalog:",
     "tsnapi": "catalog:",
diff --git a/packages/durable-workflows/src/engine.ts b/packages/durable-workflows/src/engine.ts
new file mode 100644
index 0000000..b88c324
--- /dev/null
+++ b/packages/durable-workflows/src/engine.ts
@@ -0,0 +1,361 @@
+/**
+ * The durable-workflows engine runtime — the "connector" that ties the authoring
+ * surface and the execution host to a persistence store and a definition source,
+ * exposing the full instance lifecycle.
+ *
+ * `durableWorkflows(options)` binds ONE {@link durableWorkflowHost} (its own iso4
+ * sandbox, created lazily on the first run) and mounts, once, the caller's plugin
+ * shims plus any `alias` re-export modules. Every lifecycle call resolves down to
+ * a single replay turn:
+ *
+ *   load the instance record + its boundary cache from the store
+ *     → resolve the pinned definition version via the store's `getDefinition`
+ *     → hydrate (or reuse) a runner for that version, mounted with the plugins
+ *     → execute one turn with per-run handlers carrying the instance metadata
+ *     → persist the grown cache and the new instance status
+ *     → return a RunOutcome.
+ *
+ * The engine stores nothing but instances and one opaque cache blob per instance
+ * (see {@link WorkflowStore}); retry, scheduling and wake-ups are the caller's
+ * job. Resume is plain re-execution: a waiting operation re-dispatches and its
+ * handler consults host state.
+ */
+import type { HostHandler, ModuleDefinition } from 'durable-isolates'
+import type { ResourceLimits } from 'durable-isolates/types/iso4'
+import type { WorkflowRunner } from './host'
+import type {
+  DurableHandler,
+  DurableWorkflowsEngine,
+  DurableWorkflowsOptions,
+  InstanceOutcome,
+  InstanceRecord,
+  ResolvedDefinition,
+  RunOutcome,
+  SerializedError,
+  WorkflowInstanceHandle,
+} from './types'
+import { randomUUID } from 'node:crypto'
+import { durableWorkflowHost } from './host'
+
+/**
+ * Engine sandbox defaults (differing from iso4's own) — see
+ * {@link DurableWorkflowsOptions.sandbox}. Replay runs are mostly I/O-idle, so
+ * far more concurrent runs than cores is fine.
+ */
+const DEFAULT_SANDBOX = { maxIsolates: 45 }
+
+/**
+ * Engine per-run limit defaults (differing from iso4's and the kernel's) — see
+ * {@link DurableWorkflowsOptions.limits}. Merged UNDER `options.limits` and a
+ * definition's own `limits`, so explicit settings always win.
+ */
+const DEFAULT_LIMITS: Partial<ResourceLimits> = {
+  maxBridgeCalls: 300,
+  wallTimeMs: 600_000,
+  cpuTimeMs: 30_000,
+}
+
+/**
+ * Lift the kernel's recorded failure (an iso4 `RunError` — `name`/`message`/
+ * `stack`/`fields`, or any thrown value) into the engine's named
+ * {@link SerializedError}. `data` is the error's own fields carried across the
+ * bridge; `class` is a plugin-attached verdict (informational only) if the
+ * thrown error carried a `permanent`/`transient` one.
+ * @param error the kernel's recorded failure value
+ */
+function toSerializedError(error: unknown): SerializedError {
+  if (typeof error !== 'object' || error === null)
+    return { name: 'Error', message: String(error) }
+  const e = error as { name?: unknown, message?: unknown, stack?: unknown, fields?: unknown }
+  const name = typeof e.name === 'string' ? e.name : 'Error'
+  const message = typeof e.message === 'string' ? e.message : String(error)
+  const data = e.fields
+  const rawClass = typeof data === 'object' && data !== null && 'class' in data
+    ? (data as { class?: unknown }).class
+    : undefined
+  const errorClass = rawClass === 'permanent' || rawClass === 'transient' ? rawClass : undefined
+  return {
+    name,
+    message,
+    ...(typeof e.stack === 'string' ? { stack: e.stack } : {}),
+    ...(data === undefined ? {} : { data }),
+    ...(errorClass === undefined ? {} : { class: errorClass }),
+  }
+}
+
+/**
+ * Derive the terminal {@link InstanceOutcome} from a record, or `null` while the
+ * instance is still running/waiting.
+ * @param record the stored instance record
+ */
+function outcomeOf(record: InstanceRecord): InstanceOutcome | null {
+  if (record.status === 'running' || record.status === 'waiting')
+    return null
+  const base = {
+    instanceId: record.instanceId,
+    workflow: record.workflow,
+    version: record.version,
+    runs: record.runs,
+    createdAt: record.createdAt,
+    finishedAt: record.updatedAt,
+  }
+  if (record.status === 'failed')
+    return { ...base, status: 'failed', error: record.error }
+  if (record.status === 'terminated')
+    return { ...base, status: 'terminated' }
+  return { ...base, status: 'completed' }
+}
+
+export function durableWorkflows(options: DurableWorkflowsOptions): DurableWorkflowsEngine {
+  const { store, plugins = {}, alias = {}, limits, onEvent } = options
+
+  const host = durableWorkflowHost({ sandbox: { ...DEFAULT_SANDBOX, ...options.sandbox } })
+
+  // Mount modules built ONCE for the engine's lifetime: each plugin's in-sandbox
+  // shim, plus a tiny re-export module per `alias` remapping a core specifier.
+  const mountedModules: Record<string, ModuleDefinition> = {}
+  for (const [specifier, plugin] of Object.entries(plugins))
+    mountedModules[specifier] = { shim: plugin.shim }
+  for (const [aliasSpecifier, canonical] of Object.entries(alias)) {
+    if (Object.hasOwn(mountedModules, aliasSpecifier))
+      throw new Error(`durable-workflows: alias "${aliasSpecifier}" collides with a mounted plugin`)
+    // A reserved-specifier alias key is rejected by the host on hydrate; the
+    // canonical target is a reserved core module, which is what we re-export.
+    mountedModules[aliasSpecifier] = { shim: `export * from '${canonical}'` }
+  }
+
+  // Flat operation-name → handler map across every plugin. Routing is by name,
+  // so two DISTINCT handlers under one name are ambiguous — reject that; the
+  // same handler mounted under two alias specifiers is fine.
+  const handlerByName = new Map<string, DurableHandler>()
+  for (const plugin of Object.values(plugins)) {
+    for (const [name, handler] of Object.entries(plugin.handlers)) {
+      const existing = handlerByName.get(name)
+      if (existing !== undefined && existing !== handler)
+        throw new Error(`durable-workflows: two plugins register a handler named "${name}"`)
+      handlerByName.set(name, handler)
+    }
+  }
+
+  // Compiled runners, one per pinned definition version, reused across every run
+  // of that version. Keyed by `workflow@version`; the promise is cached so
+  // concurrent first runs share one hydrate, and a failed hydrate is evicted.
+  const runners = new Map<string, Promise<WorkflowRunner>>()
+  function ensureRunner(workflow: string, version: string, def?: ResolvedDefinition): Promise<WorkflowRunner> {
+    const key = `${workflow}@${version}`
+    let runner = runners.get(key)
+    if (runner === undefined) {
+      runner = (async () => {
+        const resolved = def ?? await store.getDefinition(workflow, version)
+        if (resolved === null)
+          throw new Error(`durable-workflows: no definition for "${workflow}" @ "${version}"`)
+        return host.hydrate({
+          workflow: resolved.code,
+          plugins: mountedModules,
+          limits: { ...DEFAULT_LIMITS, ...limits, ...resolved.limits },
+        })
+      })().catch((err: unknown) => {
+        runners.delete(key)
+        throw err
+      })
+      runners.set(key, runner)
+    }
+    return runner
+  }
+
+  const pendingPromises = new Set<Promise<unknown>>()
+  function track<T>(work: Promise<T>): Promise<T> {
+    const tracked: Promise<T> = work.finally(() => {
+      pendingPromises.delete(tracked)
+    })
+    pendingPromises.add(tracked)
+    return tracked
+  }
+
+  // One replay turn against the current record. Resolves the runner, executes
+  // with per-run handlers, persists the grown cache and the new status, emits
+  // events, and returns the outcome. An infrastructure failure (resolve/hydrate/
+  // execute/store) rejects WITHOUT transitioning — nothing is persisted past the
+  // point it threw.
+  async function runTurn(record: InstanceRecord, def?: ResolvedDefinition): Promise<RunOutcome> {
+    const { instanceId, workflow, version } = record
+    const run = record.runs + 1
+
+    const runner = await ensureRunner(workflow, version, def)
+    const cache = (await store.getCache(instanceId)) ?? {}
+
+    const handlers: Record<string, HostHandler> = {}
+    for (const [name, handler] of handlerByName) {
+      handlers[name] = (...forwarded: unknown[]) => {
+        const [stepId, ...payload] = forwarded
+        return handler({ instanceId, workflow, run, stepId: String(stepId), payload })
+      }
+    }
+
+    const result = await runner.execute({ input: record.input, cache, handlers }).result
+    await store.putCache(instanceId, result.cache)
+
+    // Emit a step event for every boundary that settled or changed this run.
+    for (const [stepId, boundary] of Object.entries(result.cache)) {
+      const prev = cache[stepId]
+      if (prev === undefined || prev.status !== boundary.status)
+        onEvent?.({ type: 'step', instanceId, stepId, status: boundary.status, run })
+    }
+
+    const now = new Date().toISOString()
+    const base = { instanceId, workflow, version, input: record.input, runs: run, createdAt: record.createdAt, updatedAt: now }
+
+    let next: InstanceRecord
+    let outcome: RunOutcome
+    if (result.outcome === 'completed') {
+      next = { ...base, status: 'completed' }
+      outcome = { instanceId, run, status: 'completed' }
+    } else if (result.outcome === 'suspended') {
+      next = { ...base, status: 'waiting' }
+      outcome = {
+        instanceId,
+        run,
+        status: 'waiting',
+        pending: result.pending.map((p) => ({ stepId: p.id, operation: p.name, payload: p.payload })),
+      }
+    } else {
+      const error = toSerializedError(result.error)
+      next = { ...base, status: 'failed', error }
+      outcome = { instanceId, run, status: 'failed', error }
+    }
+
+    await store.updateInstance(next)
+    onEvent?.({ type: 'instance', instanceId, status: next.status, run })
+    return outcome
+  }
+
+  return {
+    pendingPromises,
+
+    create: (workflow, input, opts) => track((async () => {
+      const instanceId = opts?.instanceId ?? randomUUID()
+      const existing = await store.getInstance(instanceId)
+      if (existing !== null) {
+        // Idempotent creation: an active instance re-runs (harmless replay,
+        // accurate pending), a terminal one returns its recorded outcome.
+        if (existing.status === 'running' || existing.status === 'waiting')
+          return runTurn(existing)
+        if (existing.status === 'completed')
+          return { instanceId, run: existing.runs, status: 'completed' }
+        if (existing.status === 'failed')
+          return { instanceId, run: existing.runs, status: 'failed', error: existing.error }
+        throw new Error(`durable-workflows: instance "${instanceId}" is terminated`)
+      }
+
+      const def = await store.getDefinition(workflow, opts?.version)
+      if (def === null) {
+        throw new Error(
+          `durable-workflows: no definition for "${workflow}"${opts?.version ? ` @ "${opts.version}"` : ''}`,
+        )
+      }
+      const now = new Date().toISOString()
+      const record: InstanceRecord = {
+        instanceId,
+        workflow,
+        version: def.version,
+        input,
+        runs: 0,
+        status: 'running',
+        createdAt: now,
+        updatedAt: now,
+      }
+      await store.createInstance(record)
+      onEvent?.({ type: 'instance', instanceId, status: 'running', run: 0 })
+      return runTurn(record, def)
+    })()),
+
+    get: async (instanceId): Promise<WorkflowInstanceHandle | null> => {
+      const record = await store.getInstance(instanceId)
+      if (record === null)
+        return null
+      return {
+        id: instanceId,
+        status: async () => {
+          const current = await store.getInstance(instanceId)
+          if (current === null)
+            throw new Error(`durable-workflows: instance "${instanceId}" no longer exists`)
+          return current.status
+        },
+        outcome: async () => {
+          const current = await store.getInstance(instanceId)
+          if (current === null)
+            throw new Error(`durable-workflows: instance "${instanceId}" no longer exists`)
+          return outcomeOf(current)
+        },
+      }
+    },
+
+    continueWorkflow: (instanceId) => track((async () => {
+      const record = await store.getInstance(instanceId)
+      if (record === null)
+        throw new Error(`durable-workflows: no instance "${instanceId}"`)
+      if (record.status === 'terminated')
+        throw new Error(`durable-workflows: instance "${instanceId}" is terminated`)
+      return runTurn(record)
+    })()),
+
+    terminate: (instanceId) => track((async () => {
+      const record = await store.getInstance(instanceId)
+      if (record === null)
+        throw new Error(`durable-workflows: no instance "${instanceId}"`)
+      if (record.status === 'terminated')
+        return
+      const now = new Date().toISOString()
+      await store.updateInstance({
+        instanceId,
+        workflow: record.workflow,
+        version: record.version,
+        input: record.input,
+        runs: record.runs,
+        status: 'terminated',
+        createdAt: record.createdAt,
+        updatedAt: now,
+      })
+      onEvent?.({ type: 'instance', instanceId, status: 'terminated', run: record.runs })
+    })()),
+
+    evict: (instanceId, stepId) => track((async () => {
+      const record = await store.getInstance(instanceId)
+      if (record === null)
+        throw new Error(`durable-workflows: no instance "${instanceId}"`)
+      if (record.status === 'terminated')
+        throw new Error(`durable-workflows: instance "${instanceId}" is terminated`)
+      const cache = (await store.getCache(instanceId)) ?? {}
+      const target = cache[stepId]
+      if (target === undefined)
+        throw new Error(`durable-workflows: no boundary "${stepId}" in instance "${instanceId}"`)
+      // Delete the boundary, everything recorded after it (prefix by seq), AND
+      // its whole subtree (by key-prefix). The subtree matters because a scope's
+      // children commit BEFORE the scope itself, so they carry a lower seq — a
+      // seq-only prune would leave them cached and the replayed body would reuse
+      // them instead of truly re-executing.
+      const pruned = Object.fromEntries(
+        Object.entries(cache).filter(([key, boundary]) =>
+          boundary.seq < target.seq && key !== stepId && !key.startsWith(`${stepId}/`),
+        ),
+      )
+      await store.putCache(instanceId, pruned)
+      return runTurn(record)
+    })()),
+
+    restart: (instanceId) => track((async () => {
+      const record = await store.getInstance(instanceId)
+      if (record === null)
+        throw new Error(`durable-workflows: no instance "${instanceId}"`)
+      if (record.status === 'terminated')
+        throw new Error(`durable-workflows: instance "${instanceId}" is terminated`)
+      await store.putCache(instanceId, {})
+      return runTurn(record)
+    })()),
+
+    dispose: async () => {
+      await Promise.allSettled([...pendingPromises])
+      await host.dispose()
+    },
+  }
+}
diff --git a/packages/durable-workflows/src/host.ts b/packages/durable-workflows/src/host.ts
index a8426f8..59a6eac 100644
--- a/packages/durable-workflows/src/host.ts
+++ b/packages/durable-workflows/src/host.ts
@@ -24,6 +24,7 @@ import type {
   ModuleDefinition,
   PerExecuteHandlers,
 } from 'durable-isolates'
+import type { ResourceLimits } from 'durable-isolates/types/iso4'
 import { durableIsolates } from 'durable-isolates'
 import { coreModules } from './shim'
 
@@ -46,6 +47,12 @@ export interface WorkflowHydrateOptions {
    * automatically; a plugin may not use a reserved specifier.
    */
   plugins?: Readonly<Record<string, ModuleDefinition>>
+  /**
+   * Default iso4 resource limits for every `execute` on this runner;
+   * `WorkflowExecuteOptions.limits` overrides per run. Forwarded to the kernel's
+   * `hydrate`.
+   */
+  limits?: Partial<ResourceLimits>
 }
 
 export interface WorkflowExecuteOptions {
@@ -62,6 +69,10 @@ export interface WorkflowExecuteOptions {
    * Per-run host handlers for the mounted plugins, keyed by operation name.
    */
   handlers?: PerExecuteHandlers
+  /**
+   * iso4 resource limits for this run, overriding the runner's `hydrate` default.
+   */
+  limits?: Partial<ResourceLimits>
 }
 
 export interface WorkflowRunner {
@@ -97,7 +108,7 @@ export interface DurableWorkflowHost {
 export function durableWorkflowHost(options?: DurableIsolatesOptions): DurableWorkflowHost {
   const host = durableIsolates(options)
   return {
-    hydrate: async ({ workflow, plugins = {} }): Promise<WorkflowRunner> => {
+    hydrate: async ({ workflow, plugins = {}, limits }): Promise<WorkflowRunner> => {
       for (const specifier of Object.keys(plugins)) {
         if (Object.hasOwn(coreModules, specifier) || specifier === DEFINITION_SPECIFIER) {
           throw new Error(
@@ -108,11 +119,17 @@ export function durableWorkflowHost(options?: DurableIsolatesOptions): DurableWo
       }
       const runner = await host.hydrate({
         modules: { ...coreModules, [DEFINITION_SPECIFIER]: { shim: workflow }, ...plugins },
+        ...(limits === undefined ? {} : { limits }),
       })
       return {
-        execute: ({ input, cache, handlers }) => {
+        execute: ({ input, cache, handlers, limits: runLimits }) => {
           const code = `import workflow from '${DEFINITION_SPECIFIER}'\nexport default await workflow(${JSON.stringify(input) ?? 'undefined'})`
-          return runner.execute(handlers === undefined ? { code, cache } : { code, cache, handlers })
+          return runner.execute({
+            code,
+            cache,
+            ...(handlers === undefined ? {} : { handlers }),
+            ...(runLimits === undefined ? {} : { limits: runLimits }),
+          })
         },
         dispose: () => runner.dispose(),
       }
diff --git a/packages/durable-workflows/src/index.ts b/packages/durable-workflows/src/index.ts
index 2774ae7..f2309bd 100644
--- a/packages/durable-workflows/src/index.ts
+++ b/packages/durable-workflows/src/index.ts
@@ -10,6 +10,7 @@
  * the shim-facing contract for plugin authors at `durable-workflows/internal`.
  */
 export type * from './types'
+export { durableWorkflows } from './engine'
 export type {
   DurableWorkflowHost,
   WorkflowExecuteOptions,
@@ -17,6 +18,8 @@ export type {
   WorkflowRunner,
 } from './host'
 export { durableWorkflowHost } from './host'
+export type { MemoryWorkflowStore } from './memory-store'
+export { memoryStore } from './memory-store'
 // The canonical specifiers are exported: plugin shims import from them, and
 // `alias` remounts reference them. The core shim SOURCES are not public — the
 // host mounts them for you.
diff --git a/packages/durable-workflows/src/internal.ts b/packages/durable-workflows/src/internal.ts
index 4bdb14d..93dbb5a 100644
--- a/packages/durable-workflows/src/internal.ts
+++ b/packages/durable-workflows/src/internal.ts
@@ -21,7 +21,10 @@
  * A durable OPERATION call — the primitive every waiting/host-backed capability
  * is built on. Forms an auto step id (the boundary key) from `name` via the
  * kernel's ambient `nextKey` (`name#0`, `name#1`, … per scope) and routes to the
- * host handler registered under `name`, forwarding `args`. Resolves with the
+ * host handler registered under `name`. Over the wire it forwards the step id
+ * first, then `args` (the kernel never passes a handler its boundary key, so the
+ * engine recovers `stepId` from this leading id and treats the rest as
+ * `payload`). Resolves with the
  * boundary's value, rejects with a recorded/handler error, and never settles
  * when the handler suspends (the run is aborted). Answered from the cache on
  * replay — the handler runs at most once per boundary until re-dispatch.
diff --git a/packages/durable-workflows/src/memory-store.ts b/packages/durable-workflows/src/memory-store.ts
new file mode 100644
index 0000000..36188af
--- /dev/null
+++ b/packages/durable-workflows/src/memory-store.ts
@@ -0,0 +1,95 @@
+/**
+ * An in-memory {@link WorkflowStore} — the reference adapter, for tests and
+ * single-process throwaway runs. It holds instances, boundary caches and
+ * deployed definitions in plain `Map`s; nothing is persisted, so a process
+ * restart loses everything.
+ *
+ * Records and caches are deep-cloned on the way in and out (`structuredClone`),
+ * mirroring what a real backend does when it serializes across a boundary: a
+ * caller cannot mutate stored state by holding onto a reference it passed in or
+ * got back. A production adapter (SQL, a platform's managed objects) implements
+ * the same methods against durable storage.
+ */
+import type { ResourceLimits } from 'durable-isolates/types/iso4'
+import type { BoundaryCache } from 'durable-isolates'
+import type { InstanceRecord, ResolvedDefinition, WorkflowStore } from './types'
+
+/**
+ * The memory adapter's surface: the engine-facing {@link WorkflowStore} plus
+ * `deploy` — the WRITE side of definitions, which is deliberately NOT part of
+ * the store contract (the engine never writes definitions). In a real
+ * deployment the app's own deploy layer writes into the same backend the store
+ * reads; here `deploy` plays that role for tests.
+ */
+export interface MemoryWorkflowStore extends WorkflowStore {
+  /**
+   * Register a definition version and make it the active one (what
+   * `getDefinition(name)` without a version returns). Versions are immutable:
+   * re-deploying an existing `(name, version)` throws — mirroring the store
+   * contract that a handed-out version never changes.
+   */
+  deploy: (name: string, version: string, code: string, limits?: Partial<ResourceLimits>) => void
+}
+
+/**
+ * Create a fresh in-memory store. Each call is an isolated backend — hand one to
+ * `durableWorkflows({ store })` and seed definitions with `deploy`.
+ */
+export function memoryStore(): MemoryWorkflowStore {
+  const instances = new Map<string, InstanceRecord>()
+  const caches = new Map<string, BoundaryCache>()
+  // name → version → definition; insertion order makes the LAST deploy active.
+  const definitions = new Map<string, Map<string, ResolvedDefinition>>()
+
+  return {
+    createInstance: async (record) => {
+      if (instances.has(record.instanceId))
+        throw new Error(`durable-workflows: instance "${record.instanceId}" already exists`)
+      instances.set(record.instanceId, structuredClone(record))
+    },
+    getInstance: async (instanceId) => {
+      const record = instances.get(instanceId)
+      return record === undefined ? null : structuredClone(record)
+    },
+    updateInstance: async (record) => {
+      if (!instances.has(record.instanceId))
+        throw new Error(`durable-workflows: instance "${record.instanceId}" does not exist`)
+      instances.set(record.instanceId, structuredClone(record))
+    },
+    deleteInstance: async (instanceId) => {
+      instances.delete(instanceId)
+      caches.delete(instanceId)
+    },
+
+    getCache: async (instanceId) => {
+      const cache = caches.get(instanceId)
+      return cache === undefined ? null : structuredClone(cache)
+    },
+    putCache: async (instanceId, cache) => {
+      caches.set(instanceId, structuredClone(cache))
+    },
+
+    getDefinition: async (name, version) => {
+      const versions = definitions.get(name)
+      if (versions === undefined)
+        return null
+      if (version === undefined) {
+        // Active = the most recently deployed version.
+        const latest = [...versions.values()].at(-1)
+        return latest === undefined ? null : structuredClone(latest)
+      }
+      const exact = versions.get(version)
+      return exact === undefined ? null : structuredClone(exact)
+    },
+    deploy: (name, version, code, limits) => {
+      let versions = definitions.get(name)
+      if (versions === undefined) {
+        versions = new Map()
+        definitions.set(name, versions)
+      }
+      if (versions.has(version))
+        throw new Error(`durable-workflows: "${name}" @ "${version}" is already deployed (versions are immutable)`)
+      versions.set(version, { version, code, ...(limits === undefined ? {} : { limits }) })
+    },
+  }
+}
diff --git a/packages/durable-workflows/src/shim.ts b/packages/durable-workflows/src/shim.ts
index 87c7a99..e906e45 100644
--- a/packages/durable-workflows/src/shim.ts
+++ b/packages/durable-workflows/src/shim.ts
@@ -37,9 +37,13 @@ export const WORKFLOW_SPECIFIER = 'durable-workflows:workflow'
  *
  * `operation(name, ...args)` — a durable OPERATION call: forms an auto step id
  * with the kernel's `nextKey` (`name#0`, `name#1`, … within the current scope)
- * and routes to the host handler `name`, forwarding `args`. This is what plugin
- * shims build their capability on (e.g. `export const sleep = ms =>
- * operation('sleep', ms)`).
+ * and routes to the host handler `name`. It forwards the step id FIRST, then the
+ * caller's `args` — the kernel hands a handler only what the shim forwards and
+ * never the boundary key, so carrying it in the args is how the engine recovers
+ * `stepId` when it builds the handler's `{ instanceId, workflow, run, stepId,
+ * payload }` input (it strips this leading id and passes the rest as `payload`).
+ * This is what plugin shims build their capability on (e.g. `export const sleep =
+ * ms => operation('sleep', ms)`).
  *
  * `boundary(id, fn)` — the nestable BOUNDARY: `fn` runs in-sandbox, its return
  * value is committed, and `id` is prepended as the ambient prefix so inner
@@ -51,7 +55,8 @@ export const internalShim: string = /* js */ `
 import { durableCall, boundary as kernelBoundary, nextKey } from 'durable-isolates:internal'
 
 export function operation(name, ...args) {
-  return durableCall(nextKey(String(name)), String(name), ...args)
+  const key = nextKey(String(name))
+  return durableCall(key, String(name), key, ...args)
 }
 
 export function boundary(id, fn) {
diff --git a/packages/durable-workflows/src/types.ts b/packages/durable-workflows/src/types.ts
index 8818853..321dd52 100644
--- a/packages/durable-workflows/src/types.ts
+++ b/packages/durable-workflows/src/types.ts
@@ -3,7 +3,7 @@
  *
  * Entry point shape:
  *
- *   const engine = durableWorkflows({ store, resolveDefinition, plugins: [...] })
+ *   const engine = durableWorkflows({ store, plugins: { ... } })
  *
  * Plugins never augment the engine surface — it is fixed. What a plugin
  * extends is the WORKFLOW's world: its shim defines a virtual module inside
@@ -39,20 +39,12 @@ export interface DurableWorkflowsOptions {
    */
   sandbox?: SandboxOptions
   /**
-   * The only mandatory adapter: where instances and their boundary cache live.
-   * Values are stored exactly as they cross the iso4 bridge (V8-serializable
-   * data) — there is no codec layer.
+   * The only mandatory adapter: ONE persistent world — instances, their
+   * boundary caches, and read access to workflow definitions. Values are
+   * stored exactly as they cross the iso4 bridge (V8-serializable data) —
+   * there is no codec layer.
    */
   store: WorkflowStore
-  /**
-   * Where workflow definitions come from — the engine deliberately has NO
-   * registry of its own. Called with a concrete version when replaying an
-   * instance (instances pin the version they started on), and without one
-   * when creating a new instance — the resolver decides what "latest" means
-   * and returns the concrete version to pin. Definitions live wherever the
-   * application wants: disk, database, git, an upload endpoint.
-   */
-  resolveDefinition: (name: string, version?: string) => Promise<ResolvedDefinition | null>
   /**
    * Capabilities workflow code can import. Each plugin is a pair of an
    * in-sandbox shim and a host-side implementation.
@@ -139,7 +131,7 @@ export interface DurableWorkflowsEngine {
   readonly pendingPromises: ReadonlySet<Promise<unknown>>
   /**
    * Start a new instance and run its first turn. Resolves the definition via
-   * `resolveDefinition` (without a version unless `opts.version` says
+   * the store's `getDefinition` (without a version unless `opts.version` says
    * otherwise), pins the returned concrete version to the instance for all
    * future replays, and returns the run's outcome.
    */
@@ -161,9 +153,10 @@ export interface DurableWorkflowsEngine {
   continueWorkflow: (instanceId: string) => Promise<RunOutcome>
   terminate: (instanceId: string) => Promise<void>
   /**
-   * Prefix invalidation: deletes the boundary and every boundary recorded after
-   * it (by `seq`), then replays. Rare manual remediation — not part of normal
-   * operation.
+   * Prefix invalidation: deletes the boundary, every boundary recorded after it
+   * (by `seq`), and its whole subtree (by key-prefix — a scope's children commit
+   * before the scope itself, so a seq-only prune would leave them cached), then
+   * replays. Rare manual remediation — not part of normal operation.
    */
   evict: (instanceId: string, stepId: string) => Promise<RunOutcome>
   /**
@@ -237,7 +230,7 @@ export interface CreateOptions {
    */
   instanceId?: string
   /**
-   * Pin a specific definition version instead of the resolver's "latest".
+   * Pin a specific definition version instead of the store's "latest"/active.
    */
   version?: string
 }
@@ -475,6 +468,11 @@ export interface TerminatedInstanceRecord extends InstanceRecordBase {
  * in-memory by the engine on the loaded cache before it re-executes, so the
  * store only ever reads and writes the whole blob. A new backend should be an
  * afternoon.
+ *
+ * Definitions are the store's READ-ONLY third concern: the engine never
+ * writes them. Writing (upload, versioning, rollback, deletion) is the
+ * application's own deploy layer operating on the same backend — deliberately
+ * out of engine scope.
  */
 export interface WorkflowStore {
   createInstance: (record: InstanceRecord) => Promise<void>
@@ -488,6 +486,23 @@ export interface WorkflowStore {
    */
   getCache: (instanceId: string) => Promise<BoundaryCache | null>
   putCache: (instanceId: string, cache: BoundaryCache) => Promise<void>
+
+  /**
+   * Where workflow definitions come from — the engine only ever READS them.
+   * Called without a `version` when creating a new instance (the adapter
+   * decides what "latest"/active means and returns the concrete version to
+   * pin), and with the pinned version on every replay of an existing
+   * instance.
+   *
+   * CONTRACT (the price of storing a reference instead of the bytes): a
+   * `(name, version)` pair is immutable and must stay fetchable —
+   * byte-identical, forever — while any instance pins it. Think docker image
+   * digest: the deploy layer never mutates a handed-out version and refuses
+   * to delete one that instances still reference; otherwise those instances
+   * strand or (worse) replay divergent code. The engine cannot enforce this —
+   * it lives in the adapter/deploy layer.
+   */
+  getDefinition: (name: string, version?: string) => Promise<ResolvedDefinition | null>
 }
 
 // ─────────────────────────────────────────────────────────────────────────────
diff --git a/packages/durable-workflows/tests/engine.test.ts b/packages/durable-workflows/tests/engine.test.ts
new file mode 100644
index 0000000..7d42628
--- /dev/null
+++ b/packages/durable-workflows/tests/engine.test.ts
@@ -0,0 +1,334 @@
+import { afterEach, describe, expect, test } from 'vitest'
+import { SuspendIsolate } from 'durable-isolates'
+import type {
+  DurableHandlerInput,
+  DurableWorkflowsEngine,
+  DurableWorkflowsPlugin,
+} from '../src'
+import { durableWorkflows, INTERNAL_SPECIFIER, memoryStore, WORKFLOW_SPECIFIER } from '../src'
+
+// The engine is tested end-to-end: a real memory store (definitions seeded via
+// its `deploy` helper — the test's stand-in for the app's deploy layer), and
+// plugin shims built on `durable-workflows:internal` mounted on a REAL sandbox.
+// Every assertion drives the whole chain
+// (engine → :workflow → :internal → durable-isolates:internal → host → handler).
+
+const APPROVALS_SPECIFIER = 'test:approvals'
+const APPROVALS_SHIM = /* js */ `
+  import { operation } from '${INTERNAL_SPECIFIER}'
+  export const approve = (subject) => operation('approve', subject)
+  export const tick = () => operation('tick')
+`
+
+// A test harness bundling engine + the store + captured handler inputs, so each
+// test can compose exactly what it needs and dispose cleanly. The store's
+// `getDefinition` is instrumented to record every call, so tests can assert
+// version pinning.
+function harness(pluginOverrides?: Partial<DurableWorkflowsPlugin>) {
+  const store = memoryStore()
+  const resolveCalls: Array<{ name: string, version?: string }> = []
+  const rawGetDefinition = store.getDefinition
+  store.getDefinition = async (name, version) => {
+    resolveCalls.push(version === undefined ? { name } : { name, version })
+    return rawGetDefinition(name, version)
+  }
+  const seen: DurableHandlerInput[] = []
+  let approvalAnswer: string | undefined
+
+  const approvals: DurableWorkflowsPlugin = {
+    id: 'approvals',
+    shim: APPROVALS_SHIM,
+    handlers: {
+      approve: (input) => {
+        seen.push(input)
+        if (approvalAnswer === undefined)
+          throw new SuspendIsolate({ subject: input.payload })
+        return approvalAnswer
+      },
+      tick: (input) => {
+        seen.push(input)
+        return 'tock'
+      },
+    },
+    ...pluginOverrides,
+  }
+
+  const engines: DurableWorkflowsEngine[] = []
+  const make = (extra?: Partial<Parameters<typeof durableWorkflows>[0]>) => {
+    const engine = durableWorkflows({
+      store,
+      plugins: { [APPROVALS_SPECIFIER]: approvals },
+      sandbox: { maxIsolates: 4 },
+      ...extra,
+    })
+    engines.push(engine)
+    return engine
+  }
+
+  return {
+    store,
+    resolveCalls,
+    seen,
+    make,
+    setAnswer: (v: string | undefined) => {
+      approvalAnswer = v
+    },
+    dispose: () => Promise.all(engines.map((e) => e.dispose())),
+  }
+}
+
+let active: { dispose: () => Promise<unknown> } | undefined
+afterEach(async () => {
+  await active?.dispose()
+  active = undefined
+})
+
+describe('create', () => {
+  test('runs the first turn, pins the resolved version, completes', async () => {
+    const h = harness()
+    active = h
+    h.store.deploy('greet', 'v1', `import { defineWorkflow } from '${WORKFLOW_SPECIFIER}'
+      export default defineWorkflow({ async run({ input }) { return input.name } })`)
+    const engine = h.make()
+
+    const outcome = await engine.create('greet', { name: 'ada' })
+    expect(outcome.status).toBe('completed')
+    expect(outcome.run).toBe(1)
+
+    // resolver was called WITHOUT a version (create resolves "latest").
+    expect(h.resolveCalls).toEqual([{ name: 'greet' }])
+
+    const handle = await engine.get(outcome.instanceId)
+    expect(await handle?.status()).toBe('completed')
+    const finished = await handle?.outcome()
+    expect(finished).toMatchObject({ status: 'completed', workflow: 'greet', version: 'v1', runs: 1 })
+  }, 20_000)
+
+  test('idempotent for a repeated instanceId (no second start)', async () => {
+    const h = harness()
+    active = h
+    h.store.deploy('greet', 'v1', `import { defineWorkflow } from '${WORKFLOW_SPECIFIER}'
+      export default defineWorkflow({ async run() { return 'ok' } })`)
+    const engine = h.make()
+
+    const a = await engine.create('greet', undefined, { instanceId: 'fixed' })
+    const b = await engine.create('greet', undefined, { instanceId: 'fixed' })
+    expect(a.status).toBe('completed')
+    expect(b).toEqual({ instanceId: 'fixed', run: 1, status: 'completed' })
+    // Only the first create resolved a definition.
+    expect(h.resolveCalls).toEqual([{ name: 'greet' }])
+  }, 20_000)
+})
+
+describe('handler input', () => {
+  test('handler receives { instanceId, workflow, run, stepId, payload } with payload = full arg list', async () => {
+    const h = harness()
+    active = h
+    h.setAnswer('yes')
+    h.store.deploy('flow', 'v1', `import { defineWorkflow, step } from '${WORKFLOW_SPECIFIER}'
+      import { approve } from '${APPROVALS_SPECIFIER}'
+      export default defineWorkflow({
+        async run() { return await step.do('gate', () => approve('ship it')) }
+      })`)
+    const engine = h.make()
+
+    const outcome = await engine.create('flow', undefined, { instanceId: 'inst-1' })
+    expect(outcome.status).toBe('completed')
+
+    expect(h.seen).toHaveLength(1)
+    expect(h.seen[0]).toEqual({
+      instanceId: 'inst-1',
+      workflow: 'flow',
+      run: 1,
+      stepId: 'gate/approve#0',
+      payload: ['ship it'], // the whole argument list, never blended with metadata
+    })
+  }, 20_000)
+})
+
+describe('suspend + continue', () => {
+  test('waiting outcome surfaces pending; continueWorkflow resumes on re-dispatch', async () => {
+    const h = harness()
+    active = h
+    h.store.deploy('flow', 'v1', `import { defineWorkflow, step } from '${WORKFLOW_SPECIFIER}'
+      import { approve } from '${APPROVALS_SPECIFIER}'
+      export default defineWorkflow({
+        async run() { return await step.do('gate', () => approve('need sign-off')) }
+      })`)
+    const engine = h.make()
+
+    const first = await engine.create('flow', undefined, { instanceId: 'inst-2' })
+    expect(first.status).toBe('waiting')
+    if (first.status !== 'waiting')
+      return
+    expect(first.pending).toEqual([
+      { stepId: 'gate/approve#0', operation: 'approve', payload: { subject: ['need sign-off'] } },
+    ])
+    const waitingHandle = await engine.get('inst-2')
+    expect(await waitingHandle?.status()).toBe('waiting')
+    expect(await waitingHandle?.outcome()).toBeNull()
+
+    // Host state changes, then the caller's own wiring re-runs.
+    h.setAnswer('approved')
+    const second = await engine.continueWorkflow('inst-2')
+    expect(second).toMatchObject({ status: 'completed', run: 2 })
+  }, 20_000)
+
+  test('a fresh engine sharing the store resumes against the PINNED version', async () => {
+    const h = harness()
+    active = h
+    h.store.deploy('flow', 'v7', `import { defineWorkflow, step } from '${WORKFLOW_SPECIFIER}'
+      import { approve } from '${APPROVALS_SPECIFIER}'
+      export default defineWorkflow({
+        async run() { return await step.do('gate', () => approve('x')) }
+      })`)
+    const first = h.make()
+    const created = await first.create('flow', undefined, { instanceId: 'inst-3' })
+    expect(created.status).toBe('waiting')
+
+    // A second engine (simulating a restarted process) shares the store; its
+    // runner cache is empty, so continue must re-resolve — with the pinned v7.
+    h.setAnswer('ok')
+    const second = h.make()
+    const resumed = await second.continueWorkflow('inst-3')
+    expect(resumed).toMatchObject({ status: 'completed', run: 2 })
+    expect(h.resolveCalls).toContainEqual({ name: 'flow', version: 'v7' })
+  }, 20_000)
+})
+
+describe('eviction & restart', () => {
+  test('evict deletes a boundary and everything after it, then replays', async () => {
+    const h = harness()
+    active = h
+    h.store.deploy('flow', 'v1', `import { defineWorkflow, step } from '${WORKFLOW_SPECIFIER}'
+      import { tick } from '${APPROVALS_SPECIFIER}'
+      export default defineWorkflow({
+        async run() { return await step.do('beat', () => tick()) }
+      })`)
+    const engine = h.make()
+
+    await engine.create('flow', undefined, { instanceId: 'inst-4' })
+    expect(h.seen).toHaveLength(1) // the operation dispatched once
+
+    const outcome = await engine.evict('inst-4', 'beat')
+    expect(outcome.status).toBe('completed')
+    expect(h.seen).toHaveLength(2) // the evicted step re-executed
+  }, 20_000)
+
+  test('restart clears the whole cache and re-runs from scratch', async () => {
+    const h = harness()
+    active = h
+    h.store.deploy('flow', 'v1', `import { defineWorkflow, step } from '${WORKFLOW_SPECIFIER}'
+      import { tick } from '${APPROVALS_SPECIFIER}'
+      export default defineWorkflow({ async run() { return await step.do('beat', () => tick()) } })`)
+    const engine = h.make()
+
+    await engine.create('flow', undefined, { instanceId: 'inst-5' })
+    const restarted = await engine.restart('inst-5')
+    expect(restarted).toMatchObject({ status: 'completed', run: 2 })
+    expect(h.seen).toHaveLength(2)
+  }, 20_000)
+})
+
+describe('terminate', () => {
+  test('marks the instance terminated and blocks continuation', async () => {
+    const h = harness()
+    active = h
+    h.store.deploy('flow', 'v1', `import { defineWorkflow, step } from '${WORKFLOW_SPECIFIER}'
+      import { approve } from '${APPROVALS_SPECIFIER}'
+      export default defineWorkflow({ async run() { return await step.do('gate', () => approve('x')) } })`)
+    const engine = h.make()
+
+    await engine.create('flow', undefined, { instanceId: 'inst-6' })
+    await engine.terminate('inst-6')
+
+    expect(await (await engine.get('inst-6'))?.status()).toBe('terminated')
+    expect((await (await engine.get('inst-6'))?.outcome())?.status).toBe('terminated')
+    await expect(engine.continueWorkflow('inst-6')).rejects.toThrow(/terminated/)
+  }, 20_000)
+})
+
+describe('failure', () => {
+  test('an uncaught throw surfaces as a failed outcome + failed instance', async () => {
+    const h = harness()
+    active = h
+    h.store.deploy('boom', 'v1', `import { defineWorkflow, step } from '${WORKFLOW_SPECIFIER}'
+      export default defineWorkflow({
+        async run() {
+          await step.do('explode', () => { const e = new Error('kaboom'); e.reason = 'test'; throw e })
+        }
+      })`)
+    const engine = h.make()
+
+    const outcome = await engine.create('boom', undefined, { instanceId: 'inst-7' })
+    expect(outcome.status).toBe('failed')
+    if (outcome.status !== 'failed')
+      return
+    expect(outcome.error.name).toBe('Error')
+    expect(outcome.error.message).toBe('kaboom')
+    expect(outcome.error.data).toMatchObject({ reason: 'test' })
+  }, 20_000)
+})
+
+describe('alias', () => {
+  test('a whitelabeled specifier re-exports the core workflow module', async () => {
+    const h = harness()
+    active = h
+    h.store.deploy('flow', 'v1', `import { defineWorkflow, step } from 'my:workflow'
+      import { tick } from '${APPROVALS_SPECIFIER}'
+      export default defineWorkflow({ async run() { return await step.do('beat', () => tick()) } })`)
+    const engine = h.make({ alias: { 'my:workflow': WORKFLOW_SPECIFIER } })
+
+    const outcome = await engine.create('flow', undefined, { instanceId: 'inst-8' })
+    expect(outcome.status).toBe('completed')
+  }, 20_000)
+})
+
+describe('observability & lifecycle', () => {
+  test('onEvent emits instance + step events; dispose drains pendingPromises', async () => {
+    const events: unknown[] = []
+    const h = harness()
+    active = h
+    h.setAnswer('done')
+    h.store.deploy('flow', 'v1', `import { defineWorkflow, step } from '${WORKFLOW_SPECIFIER}'
+      import { approve } from '${APPROVALS_SPECIFIER}'
+      export default defineWorkflow({ async run() { return await step.do('gate', () => approve('go')) } })`)
+    const engine = h.make({ onEvent: (e) => events.push(e) })
+
+    const p = engine.create('flow', undefined, { instanceId: 'inst-9' })
+    expect(engine.pendingPromises.size).toBe(1)
+    await p
+
+    expect(events).toContainEqual({ type: 'instance', instanceId: 'inst-9', status: 'running', run: 0 })
+    expect(events).toContainEqual({ type: 'instance', instanceId: 'inst-9', status: 'completed', run: 1 })
+    expect(events).toContainEqual({ type: 'step', instanceId: 'inst-9', stepId: 'gate', status: 'completed', run: 1 })
+    expect(engine.pendingPromises.size).toBe(0)
+  }, 20_000)
+})
+
+describe('definitions in the store', () => {
+  test('new instances get the last deployed version; deployed versions are immutable', async () => {
+    const h = harness()
+    active = h
+    h.store.deploy('flow', 'v1', `import { defineWorkflow } from '${WORKFLOW_SPECIFIER}'
+      export default defineWorkflow({ async run() { return 1 } })`)
+    h.store.deploy('flow', 'v2', `import { defineWorkflow } from '${WORKFLOW_SPECIFIER}'
+      export default defineWorkflow({ async run() { return 2 } })`)
+    const engine = h.make()
+
+    const outcome = await engine.create('flow', undefined, { instanceId: 'inst-10' })
+    expect(outcome.status).toBe('completed')
+    expect((await (await engine.get('inst-10'))?.outcome())?.version).toBe('v2')
+
+    expect(() => h.store.deploy('flow', 'v2', 'export default null')).toThrow(/immutable/)
+  }, 20_000)
+})
+
+describe('unknown definition', () => {
+  test('create rejects when the store has no definition', async () => {
+    const h = harness()
+    active = h
+    const engine = h.make()
+    await expect(engine.create('missing')).rejects.toThrow(/no definition/)
+  }, 20_000)
+})
diff --git a/packages/durable-workflows/tsconfig.json b/packages/durable-workflows/tsconfig.json
index 832a80c..84d6c87 100644
--- a/packages/durable-workflows/tsconfig.json
+++ b/packages/durable-workflows/tsconfig.json
@@ -13,6 +13,7 @@
     "moduleResolution": "bundler",
     "noUncheckedSideEffectImports": true,
     "resolveJsonModule": true,
+    "types": ["node"],
     "declaration": true,
     "noEmit": true,
     "erasableSyntaxOnly": true,
diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml
index c8ab259..86a8000 100644
--- a/pnpm-lock.yaml
+++ b/pnpm-lock.yaml
@@ -100,6 +100,9 @@ importers:
       '@schplitt/eslint-config':
         specifier: 'catalog:'
         version: 1.5.1(@angular/cli@21.0.4(@types/node@24.13.2))(@vue/compiler-sfc@3.5.26)(jiti@2.6.1)(ts-declaration-location@1.0.7(typescript@6.0.3))(typescript-eslint@8.50.1(eslint@10.5.0(jiti@2.6.1))(typescript@6.0.3))(typescript@6.0.3)
+      '@types/node':
+        specifier: 'catalog:'
+        version: 24.13.2
       eslint:
         specifier: 'catalog:'
         version: 10.5.0(jiti@2.6.1)