// A WASM-backed {@link EngineClient} for in-process, deterministic e2e tests // (ADR 0059 test kit, S1 — issue #157). It backs the exact same `EngineClient` // seam the live `@nanobpm/nano-sdk` engine adapter (in `@nanobpm/urban`) // implements, so the shared contract suite in `./contract.ts` can be run against // both — the seam that would have caught the cancelled-instance state-mapping bug. // // The engine (`@nanobpm/engine-wasm`, class `TestEngine`) is synchronous and // pull-based (`activateJobs` + `completeJob`), with a virtual clock. This adapter // gives the runtime the *push* worker semantics it expects by draining every // registered worker to quiescence after each mutating call — so a registered // worker runs autonomously exactly as it would against a live engine, but // deterministically and with no wall-clock waits. import { AsyncLocalStorage } from "node:async_hooks"; import type { TestEngine } from "@nanobpm/engine-wasm/readmodel"; // The engine's derived read-model DTO types — the single source of truth for the // shapes its REST read channel returns (`searchUserTasks` / `searchProcessInstances` / // `getFormByKey`). Generated from the Camunda-parity OpenAPI in engine-wasm (#881), so // this adapter reads those results through the *derived* types instead of hand-mirroring // their fields — No Drift Surfaces (AGENTS.md). Type-only, so nothing is pulled at runtime. import type { FormResult, ProcessInstanceSearchQueryResult, UserTaskSearchQueryResult, VariableSearchQueryResult, } from "@nanobpm/engine-wasm/readmodel-types"; import { applyAmbientLineage, assertDeployedWaitStateType, type AgentHistoryFilter, type AgentHistoryRecord, type AgentInstanceFilter, type AgentInstanceSummary, type EngineClient, type ElementInstanceSummary, type ElementInstanceFilter, type ElementInstanceWaitState, type ElementInstanceWaitStateFilter, type EngineJob, isBpmnError, type IncidentFilter, type IncidentSummary, type JobFilter, type JobHandler, type JobSummary, type UserTaskState, type UserTaskFilter, type VariableFilter, type VariableSummary, type WorkerSubscription, } from "@nanobpm/urban/runtime"; import { applyOutcome, MockWorkerBuilder } from "./worker-mock.ts"; // `ProcessInstanceState` and the wasm→state projection // `wasmStateToProcessInstanceState` are the canonical read-model mapping owned by // `@nanobpm/engine-testkit` (issue Magikcraft/nano-bpm#894); import and re-export // them so this adapter and the lifted assertion DSL share ONE definition // (No Drift Surfaces, AGENTS.md) instead of the two byte-identical copies they had // before. The state mapping is therefore sourced from `@nanobpm/engine-testkit` // (this adapter already imports `@nanobpm/urban/runtime` too). `ProcessInstanceSnapshot` // stays declared here as a local structural mirror of urban's shape — not re-exported // from engine-testkit — so a scaffolded app can still pin the *current* urban release // without engine-testkit dictating that DTO; `isRecord` is a generic JSON guard. import { type ProcessInstanceState, wasmStateToProcessInstanceState, } from "@nanobpm/engine-testkit"; export { type ProcessInstanceState, wasmStateToProcessInstanceState }; /** A single process instance's lifecycle snapshot, as returned by * {@link EngineClient.searchProcessInstances}. Structurally identical to urban's. */ export interface ProcessInstanceSnapshot { readonly processInstanceKey: string; readonly state: ProcessInstanceState; readonly processDefinitionKey?: string; readonly parentProcessInstanceKey?: string; readonly rootProcessInstanceKey?: string; } /** Narrow an untyped JSON value to a plain object. Used to bridge the wasm * engine's JSON-string API into the typed `EngineClient` contract. */ function isRecord(v: unknown): v is Record { return typeof v === "object" && v !== null && !Array.isArray(v); } /** Defensively narrow an untyped read-model search response to its object rows. The DTO * annotation on a `JSON.parse`d body is a shape *claim*, not a runtime guarantee, so guard * the body, its `items` array, and each row: a malformed/changed engine response (a non-object * body, a non-array `items`, or `null`/non-object rows) yields `[]`/drops the bad row instead * of throwing downstream. Keeps the caller's DTO row type — the single source of truth both * `searchUserTasks` and `searchProcessInstances` extract through, so they cannot drift. */ export function searchRows(body: { items: T[] }): T[] { if (!isRecord(body) || !Array.isArray(body.items)) return []; return body.items.filter(isRecord); } /** Whether a deploy resource is an executable engine model (BPMN or DMN) the WASM engine can parse. * BPMN/DMN are XML (`text/xml`); forms (`application/json`) and other assets are not engine models. * Falls back to the file extension when a contentType is absent. */ function isEngineModel(r: { name?: string; contentType?: string }): boolean { const ct = (r.contentType ?? "").toLowerCase(); if (ct.includes("xml")) return true; if (ct.length > 0) return false; const name = (r.name ?? "").toLowerCase(); return name.endsWith(".bpmn") || name.endsWith(".dmn"); } /** Return a copy of `source` containing only the requested keys that are * actually present — the client-side analogue of an engine `fetchVariables` * projection. */ function pick( source: Record, keys: readonly string[], ): Record { const out: Record = {}; for (const k of keys) { if (Object.hasOwn(source, k)) out[k] = source[k]; } return out; } // A minimal ambient `Deno` declaration lets this file's runtime branch compile // under Node's tsc (mirrors the pattern in the urban runtime Deno adapter); the // `typeof Deno` guard picks the right file-read path at runtime on either host. declare const Deno: { readFile(url: URL): Promise } | undefined; /** A lock horizon for `activateJobs`. Jobs are completed synchronously inside * {@link WasmEngineClient.drain}, so the lock never actually expires in a run; * the value only has to be comfortably positive. */ const JOB_LOCK_MS = 60_000; /** Default fan-out per `activateJobs` call when a worker sets no `maxParallelJobs`. */ const DEFAULT_MAX_JOBS = 32; /** A hard cap on drain iterations, so a worker that endlessly re-creates work * (a modelling bug) surfaces as a thrown error instead of hanging the test. */ const MAX_DRAIN_ITERATIONS = 100_000; /** A real-timer macrotask yield: runs after every currently-queued microtask, so awaiting it * lets a just-dispatched worker handler's async chain drain to the point it either completes * or parks on a virtual-clock `app.wait` timer. Uses the real `setTimeout` captured up front so * it is immune to a handler (or test) swapping `globalThis.setTimeout`. */ const realSetTimeout: typeof setTimeout = globalThis.setTimeout; const flushMacrotask = (): Promise => new Promise((resolve) => realSetTimeout(resolve, 0)); /** A registered worker's dispatch parameters. */ interface RegisteredWorker { readonly handler: JobHandler; readonly workerName: string; readonly maxParallelJobs: number; /** When set, a job surfaces only this subset of variables (mirrors the live * SDK adapter's `fetchVariables`, which the engine applies server-side). */ readonly fetchVariables?: readonly string[]; } /** Per-handler async-context payload (see {@link WasmEngineClient.close}/`#track`/`#handlerContext`): * carries the handler's OWN tracked promise so a nested self-close() can exclude itself from the * teardown settlement wait. Single source of truth for the store shape — used by the * `AsyncLocalStorage`, `#track`, and `#runJob`. */ interface HandlerStore { own?: Promise; } /** A synthetic activation descriptor for a mock-only type (a mocked `taskType` with no real * `registerWorker`). It only supplies the `activateJobs` parameters so a mock-only type's jobs * can be pulled; its `handler` is never called (the mock either resolves the job or, on no * clause match, `#runJob` leaves it because `hasRealWorker` is false). */ function mockOnlyWorker(jobType: string): RegisteredWorker { return { handler: () => { throw new Error(`mock-only worker for "${jobType}" has no handler — this should be unreachable`); }, workerName: `urban-testkit:mock:${jobType}`, maxParallelJobs: DEFAULT_MAX_JOBS, }; } let bootPromise: Promise | undefined; /** Boot the wasm module once per process, single-flight. The first caller starts * initialization; concurrent callers await the *same* promise, so `initSync` * runs exactly once even when parallel tests each construct an engine (a plain * boolean flag would let two callers race past the guard and double-init). * * `@nanobpm/engine-wasm/readmodel` is imported dynamically so merely importing the * testkit (e.g. the contract runner) does not eagerly load the wasm engine; it is * pulled in only when an engine is actually constructed. We import the **read-model** * subpath (not the lean default): it is the lean engine plus the gateway's SQLite * read model compiled to wasm, so the REST read methods (`getFormByKey`, * `searchUserTasks`, `searchProcessInstances`, …) are served by the *real* read * channel instead of a hand-maintained JS twin (epic Magikcraft/nano-bpm#796). * Loads the `.wasm` bytes via `import.meta.resolve` so it works from a published * package or a workspace checkout, on both Node and Deno, with no bundler * import-attribute support. */ function bootEngineWasm(): Promise { if (!bootPromise) { bootPromise = (async () => { const mod = await import("@nanobpm/engine-wasm/readmodel"); const url = new URL( import.meta.resolve( "@nanobpm/engine-wasm/readmodel/nanobpmn_engine_bg.wasm", ), ); const bytes = typeof Deno !== "undefined" ? await Deno.readFile(url) : new Uint8Array(await (await import("node:fs/promises")).readFile(url)); mod.initSync({ module: bytes }); return mod; })().catch((err) => { // A transient failure (e.g. an fs read error) must not poison every // later create(): clear the cached rejection so the next call retries. bootPromise = undefined; throw err; }); } return bootPromise; } /** * An {@link EngineClient} backed by the in-process `@nanobpm/engine-wasm` * engine. Construct via {@link createWasmEngineClient}. Beyond the interface it * exposes {@link advanceTime}, {@link snapshot}, and {@link now} for the settle * loop and assertions (S2). */ export class WasmEngineClient implements EngineClient { readonly #engine: TestEngine; readonly #workers = new Map(); /** Job-worker mocks keyed by jobType (epic #296, S1). A mocked type is resolved by its * {@link MockWorkerBuilder} at dispatch (see {@link mockWorker}) instead of running the * real handler; an un-mocked type — or a mock whose clauses don't match a given job — * runs real code. Empty and untouched unless a test calls {@link mockWorker}, so mocking * is strictly opt-in and zero-cost when unused. */ readonly #workerMocks = new Map(); /** Optional observer notified with a job's type each time a job is dispatched to a * worker handler. Additive, default-absent seam used by the S4 coverage gate to know * which worker/job types were actually exercised; a no-op for every other caller. The * second argument reports whether the dispatch was satisfied by a mock (epic #296, S4), * so a mocked-but-exercised type is recorded as covered AND flagged as mocked. */ #onJob: ((jobType: string, mocked: boolean) => void) | undefined; /** Worker handlers that have been dispatched but not yet resolved — a real push worker runs * autonomously, so a handler doing time-bounded work (`app.wait` on the virtual clock) must * NOT block {@link drain}: it is dispatched fire-and-forget and tracked here, then driven to * completion as virtual time advances (`advanceTime` fires its waits). {@link drain} quiesces * these each iteration — awaiting the ones that finish (or park on a *future* wait) — so a * quick handler's effects stay visible after `settle()` exactly as before, while a parked one * is simply left in-flight instead of deadlocking the drain on a clock that only moves later. */ readonly #inflight = new Set>(); /** Per-handler async context carrying the handler's OWN tracked promise (see {@link #track}). A * worker handler can itself call {@link close} (workers reach the engine through `AppApi`); its * tracked promise is then in {@link #inflight} while it awaits close(), yet can only settle AFTER * close() returns. {@link close} reads this store to record the caller as a close-awaiter (see * {@link #closeAwaiters}), so {@link #settleInflight} excludes it from its settlement wait and a * handler-initiated close() drains its PEERS without dead-awaiting itself (close() waiting on the * handler that is waiting on close()). Scoped by {@link #runJob} to the actual `worker.handler` * invocation ONLY — never the dispatch-time `#onJob`/mock callbacks that run first — so a dispatch * observer re-entering close() is an EXTERNAL close, not the parked handler. Empty for an external * close() (no handler on the stack), so that path is unchanged. */ readonly #handlerContext = new AsyncLocalStorage(); /** Tracked promises of handlers that are currently parked inside a {@link close} call — the * initiator AND any peer that re-entered the memoized run. Such a promise sits in {@link #inflight} * but can only settle AFTER close() returns, so {@link #settleInflight} excludes EVERY member of * this set from its wait; otherwise close() dead-awaits a handler that is itself awaiting close(). * A single `#handlerContext.getStore()?.own` exclusion is not enough: `#settleInflight()` runs in * the INITIATOR's async context, so it would only exclude the initiator and still hang on a second * handler that re-entered `close()` (returns the memoized `#closeRun`) while parked. Members are * removed when their handler settles (see {@link #track}). Empty for a purely external close(). */ readonly #closeAwaiters = new Set>(); /** One-shot wake for a {@link #settleInflight} pass that is mid-await. `#settleInflight()` waits on * `Promise.allSettled(peers)`, but a peer that re-enters {@link close} AFTER that wait began joins * {@link #closeAwaiters} and will never settle (it awaits this very run) — so the pass must stop * waiting on it and re-evaluate. {@link close} resolves this wake whenever it adds a new awaiter, * racing it against the peer wait so the pass re-filters the moment a peer parks on close(). */ #closeAwaiterWake: (() => void) | undefined; /** Job keys whose handler is currently in-flight (dispatched, not yet settled). Advancing virtual * time past a job's activation lock makes the engine re-offer a still-running job; this set lets * {@link drain} skip spawning a duplicate handler for one the in-flight instance already owns. */ readonly #inflightJobKeys = new Set(); /** The first error a tracked in-flight handler surfaced from its engine-completion call, held so * {@link drain} can rethrow it at the next quiesce point (preserving fail-loud), then cleared. */ #inflightError: unknown; /** Monotonic count of tracked handlers that have *settled* (completed or failed and left * {@link #inflight}) — never incremented by a handler still parked on a future virtual wait. * {@link drain} samples it around {@link #quiesce} to detect a handler that finished mid-quiesce * and may have enqueued fresh engine work, so it loops for another activation pass instead of * returning early on `!activatedAny` and leaving that work undrained until a later `settle`. */ #completions = 0; /** Shutdown lifecycle for {@link close}. `#closing` flips true the instant close() begins and * gates {@link drain} so a concurrently-suspended drain cannot activate jobs on an engine that is * being (or has been) freed. `#shutdown` is aborted at the same point and surfaced via * {@link shutdownSignal}: a handler parked on the app clock's `wait()` (see the testkit scheduler * wiring) sits on a virtual timer that no `advanceTime` fires during teardown, so aborting lets it * unwind — its throw maps to `failJob` on the still-live engine — and {@link #settleInflight} * actually settles instead of hanging. Real-time work is unaffected and still genuinely awaited * before `free()` (issue #446). */ #closing = false; readonly #shutdown = new AbortController(); /** Memoized shutdown run (see {@link close}). `close()` is idempotent: the FIRST call starts the * one-and-only teardown and every later/concurrent call awaits this same promise, so the WASM * handle is `free()`d exactly once — two `close()` calls can never double-free it. A run that * fails to settle *before* `free()` (engine left allocated, not freed) clears this so the caller * can cancel/retry; once `free()` has run the memo is kept even if the run then rejects (a late * in-flight error surfaced by {@link #throwInflightError}), so a retry can never re-`free()`. */ #closeRun: Promise | undefined; /** Flips true the instant {@link #doClose} reaches `#engine.free()`. Guards the `#closeRun` memo: * a rejection *after* the handle is freed must stay memoized (clearing it would let a later * `close()` re-enter `#doClose` and `free()` the same handle twice — the use-after-free #446 * exists to prevent), whereas a pre-free rejection may clear it for a legitimate retry. */ #freed = false; /** The live engine handle, or a thrown error once {@link close} has {@link TestEngine.free freed} * it. EVERY public {@link EngineClient} operation reaches the engine through this accessor so a * call made *after* close() — most importantly by a worker handler that self-initiated teardown * (`await app.close()`) and then keeps using `AppApi.engine` (e.g. `publishMessage`, * `createInstance`) — faults with a clear, catchable error instead of driving the freed WASM * handle (the opaque "null pointer passed to rust" issue #446 use-after-free). The `#freed` guard * in `#runJob` only covers a resumed handler's final `completeJob`/`failJob`; this closes the same * hole categorically across the whole EngineClient surface. Internal drain/`#settleInflight`/ * `#runJob` paths keep touching `#engine` directly on purpose: they run *before* `free()` (and * carry their own `#freed` guards), so they must be able to complete the last in-flight handlers. */ get #liveEngine(): TestEngine { if (this.#freed) { throw new Error( "WasmEngineClient: engine used after close() — the engine has been freed and can no longer service requests", ); } return this.#engine; } /** A shutdown {@link AbortSignal} that aborts when {@link close} begins. The test kit threads it * into the scheduler backing `app.wait()`, so a worker handler parked on the virtual clock is * cancelled at teardown rather than wedging `close()` on a timer that will never fire (the * virtual-timer sibling of the issue #446 real-time use-after-free). */ get shutdownSignal(): AbortSignal { return this.#shutdown.signal; } private constructor(engine: TestEngine) { this.#engine = engine; } /** Boot the wasm module (idempotent) and return a fresh, empty engine. */ static async create(): Promise { const mod = await bootEngineWasm(); return new WasmEngineClient(new mod.TestEngine()); } async deployResources( resources: { name: string; content: string; contentType: string }[], ): Promise<{ deployed: number }> { // The runtime's `deployModels` sends every deployable here — BPMN + DMN (`text/xml`), // `.form` (`application/json`), and, under ADR 0062 deploy-by-convention, any other file // swept from `resources/`. Only executable models can run under the WASM engine: BPMN/DMN // are parsed by the engine. This adapter does *not* forward `.form` resources into the engine // (only `isEngineModel` resources are deployed above), so a deployed `.form` is accepted and // counted but is not resolvable here — deploying it populates no read model. The `getForm` // read path delegates to the engine's real read model (`getFormByKey`), not a JS shadow store, // but it can only return a form once the `.form` *write* path (Magikcraft/nano-bpm#815) lands // *and* this adapter is updated to forward `.form` content; that does not happen automatically. // Any *other* generic resource likewise has no read surface here. Every non-executable resource // is inert to the BPMN parser here. // // The authored XML is deployed UNMODIFIED: a `` reaches the engine as-is and // executes natively — the engine instantiates the called process as a real child instance and // the parent parks until it completes (`@nanobpm/engine-wasm` ≥ 0.6.0). Consequence for // consumers: the called process must be deployed to this same engine, or the engine raises a // recoverable `CalledElementError` incident on the call activity — production-faithful behaviour. for (const r of resources) { if (isEngineModel(r)) this.#liveEngine.deploy(r.content); } // Match `SdkEngineClient.deployResources`: the deployment accepts every resource, so the // `deployed` count is the total — a form (or any non-executable asset) still counts as // deployed even though the WASM engine doesn't execute it. return { deployed: resources.length }; } async createInstance(input: { processDefinitionId: string; variables?: Record; awaitCompletion?: boolean; }): Promise<{ processInstanceKey: string; variables?: Record }> { const snap = this.#parseObj( this.#liveEngine.createInstance( input.processDefinitionId, // Auto-thread the `_urban.lineage` envelope via the same shared step the live // SdkEngineClient uses (No Drift Surfaces), so lineage is observable in-harness (issue #254). JSON.stringify(applyAmbientLineage(input.variables)), ), ); const processInstanceKey = requireCreated(snap.created); // Registered workers run autonomously against a live engine; mirror that by // draining to quiescence so a job whose worker is registered is served now. await this.drain(); if (input.awaitCompletion) { return { processInstanceKey, variables: this.#instanceVariables(processInstanceKey), }; } return { processInstanceKey }; } async cancelInstance(input: { processInstanceKey: string }): Promise { this.#liveEngine.cancelInstance(input.processInstanceKey); await this.drain(); } async publishMessage(input: { name: string; correlationKey?: string; variables?: Record; }): Promise { this.#liveEngine.correlateMessage( input.name, input.correlationKey ?? "", JSON.stringify(applyAmbientLineage(input.variables)), ); await this.drain(); } async searchUserTasks(filter?: UserTaskFilter & { state?: UserTaskState; }): Promise<{ userTaskKey: string; elementId?: string; variables?: Record; formKey?: string; externalFormReference?: string; processInstanceKey?: string; parentProcessInstanceKey?: string; rootProcessInstanceKey?: string; }[]> { // Delegate to the engine's real REST read channel (`POST /user-tasks/search`) instead of // scraping the primary-state snapshot. The read model honours the `{ state? }` filter // (e.g. `"CREATED"`) itself, so pass it through rather than re-implementing state matching. // Parse the result through the engine's DERIVED `UserTaskSearchQueryResult` DTO // (`@nanobpm/engine-wasm/readmodel-types`) — the single source of truth for the row shape — // so `items` is a typed `UserTaskResult[]` rather than a hand-scraped `Record` bag. const body: UserTaskSearchQueryResult = JSON.parse( this.#liveEngine.searchUserTasks( JSON.stringify(filter?.state ? { state: filter.state } : {}), ), ); // `body`/`items` come straight from an untyped `JSON.parse` boundary, so the DTO annotation // is a shape *claim*, not a runtime guarantee. `searchRows` guards the body and drops // non-object rows so a malformed/changed engine response can't throw downstream. // Normalize the process-instance / parent / root selectors through the shared presence rule // (`presentKey`) so a blank/whitespace-only selector is *dropped* (matches everything) rather // than compared literally (which would silently filter out every row), and each row key is // compared under the same rule. This mirrors `SdkEngineClient`, which has these honoured // server-side after an identical `presentEngineKey` normalization (No Drift Surfaces). const wantProcessInstanceKey = presentKey(filter?.processInstanceKey); const wantRootProcessInstanceKey = presentKey(filter?.rootProcessInstanceKey); const wantParentProcessInstanceKey = presentKey(filter?.parentProcessInstanceKey); return searchRows(body) // The read model does not yet honour the non-lifecycle selectors // (`processInstanceKey`/`assignee`/`candidateGroup` — the write/index side is // Magikcraft/nano-bpm#815's follow-up), so apply them client-side here. This mirrors the // *effective* behaviour of the gateway-backed `SdkEngineClient`, which gets them honoured // server-side; the results are identical, only the filtering site differs. .filter((t) => wantProcessInstanceKey === undefined || presentKey(t.processInstanceKey) === wantProcessInstanceKey ) .filter((t) => filter?.assignee === undefined || t.assignee === filter.assignee) .filter((t) => filter?.candidateGroup === undefined || (Array.isArray(t.candidateGroups) && t.candidateGroups.includes(filter.candidateGroup)) ) // Parent/root selectors mirror the live `SdkEngineClient`, which has them honoured // server-side; apply them client-side here against the engine's row linkage. The user-task // read model carries `rootProcessInstanceKey` (and the owning `processInstanceKey`) but not a // `parentProcessInstanceKey` today, so a parent selector reads the row's parent through the // same optional accessor — absent → matches nothing — exactly as the live gateway would // against a read model that omits it (No Drift Surfaces). .filter((t) => wantRootProcessInstanceKey === undefined || presentKey(t.rootProcessInstanceKey) === wantRootProcessInstanceKey ) .filter((t) => wantParentProcessInstanceKey === undefined || presentKey(Reflect.get(t, "parentProcessInstanceKey")) === wantParentProcessInstanceKey ) .flatMap((t) => { // A keyless row cannot be acted on — drop it (parity with `SdkEngineClient`, which logs // and skips such a row). `presentKey` also normalises a numeric key to a trimmed string. const userTaskKey = presentKey(t.userTaskKey); if (userTaskKey === undefined) return []; // The read model already resolves a `` linkage to the // latest deployed form's key server-side, so `formKey`/`externalFormReference` arrive // resolved on the row — no client-side id→key map (the deleted `#formKeyById` shadow). // Presence is type-aware (mirrors the shared form contract's `pickFormLinkage`): a // `formKey` counts only when a string/number, an `externalFormReference` only when a // string, so a non-string value can never coerce into a garbage `"[object Object]"` id. const formKey = presentKey(t.formKey); const externalFormReference = presentString(t.externalFormReference); // Parent/root linkage the engine surfaces on the user-task row, mapped through the same // presence rule as the live adapter so a native-child task correlates back to its // parent/root subject through the typed seam. `parentProcessInstanceKey` is read via the // optional accessor (the read model omits it today) so it flows through automatically if // the engine starts carrying it — until then it is honestly absent. const processInstanceKey = presentKey(t.processInstanceKey); const parentProcessInstanceKey = presentKey(Reflect.get(t, "parentProcessInstanceKey")); const rootProcessInstanceKey = presentKey(t.rootProcessInstanceKey); const variablesRaw = Reflect.get(t, "variables"); const variables = isRecord(variablesRaw) ? variablesRaw : undefined; return [{ userTaskKey, elementId: typeof t.elementId === "string" ? t.elementId : undefined, // Surface the row's variables when the engine carries them, mirroring the live // `SdkEngineClient` (which maps `variables` through the same `isRecord` guard) so // tests that rely on user-task variables see identical shape from both adapters // (No Drift Surfaces). The engine's `UserTaskResult` read model does not declare // `variables` today, so — exactly like `parentProcessInstanceKey` above — it is read // through the optional accessor and flows through automatically if the engine starts // carrying it; until then it is honestly absent. ...(variables ? { variables } : {}), ...(formKey ? { formKey } : {}), ...(externalFormReference ? { externalFormReference } : {}), ...(processInstanceKey ? { processInstanceKey } : {}), ...(parentProcessInstanceKey ? { parentProcessInstanceKey } : {}), ...(rootProcessInstanceKey ? { rootProcessInstanceKey } : {}), }]; }); } /** The open (answerable) user tasks — `searchUserTasks` pinned to `state: "CREATED"`. * Mirrors `SdkEngineClient.openUserTasks`: the single safe accessor for reconcile/ * affordance paths, derived from `searchUserTasks` so the two cannot drift. */ openUserTasks(filter?: UserTaskFilter): Promise<{ userTaskKey: string; elementId?: string; variables?: Record; formKey?: string; externalFormReference?: string; processInstanceKey?: string; parentProcessInstanceKey?: string; rootProcessInstanceKey?: string; }[]> { return this.searchUserTasks({ ...filter, state: "CREATED" }); } /** Fetch a deployed form's form-js schema from the engine's real read model * (`GET /forms/{formKey}` via `getFormByKey`). Structurally matches urban's * `EngineClient.getForm`; returns `null` when no such form exists in the read model. */ async getForm(input: { formKey?: string; formId?: string }): Promise< { formKey?: string; formId?: string; version?: number; schema: Record } | null > { // Mirror `SdkEngineClient.getForm`'s identifier normalization exactly (a behavioral // drift surface guarded by a test): an empty/whitespace-only identifier is *absent*, so a // blank `formKey` falls through to whatever `formId` is present. The engine addresses a form // by a single deploy key, so pass whichever identifier is present straight through to // `getFormByKey` (no local id→key map — the read model owns that resolution now). That // fallback identifier need not be a usable key: a malformed key — e.g. an authored `formId` // handed through as the fallback — makes `getFormByKey` *throw*; like the REST gateway's 404, // that is treated below as "no such form" (null), not propagated. const key = present(input.formKey) ?? present(input.formId); if (key == null) return null; // Resolve the live engine BEFORE the try below: a used-after-close fault must surface as a loud // error, not be swallowed by the "malformed key → no such form (null)" catch that wraps the // `getFormByKey` call itself. const engine = this.#liveEngine; // The engine addresses a form by a numeric deploy key and *throws* on a malformed key (e.g. an // authored id passed through as the fallback). Mirror `SdkEngineClient.getForm`, which treats a // failed fetch as "no such form" and returns null rather than propagating. Parse through the // engine's DERIVED `FormResult` DTO (`@nanobpm/engine-wasm/readmodel-types`) — its `schema` // (JSON string), `formKey`, `formId`, and `version` fields are the single source of truth. let body: FormResult | null; try { body = JSON.parse(engine.getFormByKey(key)); } catch { return null; } // Build the typed result through the shared `parseForm` boundary guard (single source of // truth): `getFormByKey` returns JSON `null` for an unknown key, and the `FormResult` DTO // annotation is only a shape *claim*, so `parseForm` guards the body (excluding arrays) and a // missing/invalid schema, treating either as "no such form" (`null`). return parseForm(body); } async completeUserTask( userTaskKey: string, variables?: Record, ): Promise { this.#liveEngine.completeUserTask(userTaskKey, JSON.stringify(variables ?? {})); await this.drain(); } async searchProcessInstances(filter?: { processInstanceKeys?: string[]; state?: ProcessInstanceState; parentProcessInstanceKey?: string; rootProcessInstanceKey?: string; }): Promise { // Normalize the wanted keys through the shared presence rule (`presentKey`) so a padded key // (`" 123 "`) matches the normalized row keys below, and treat an empty/all-blank list as // *absent* (no key filter) rather than a set that matches nothing — mirroring // `SdkEngineClient`, whose `$in` filter is only sent when it has ≥1 present key. No Drift Surfaces. const wantedKeys = filter?.processInstanceKeys ?.map((k) => presentKey(k)) .filter((k): k is string => k !== undefined); const wanted = wantedKeys && wantedKeys.length > 0 ? new Set(wantedKeys) : undefined; // Delegate to the engine's real REST read channel (`POST /process-instances/search`) // instead of scraping the primary-state snapshot. The read model does not yet honour // filter/sort/page fields server-side (it returns every instance — the write/index side is // Magikcraft/nano-bpm#815's follow-up), so apply the key/state selectors client-side. This // mirrors the *effective* behaviour of the gateway-backed `SdkEngineClient`, which gets them // honoured server-side; only the filtering site differs. Parse through the engine's DERIVED // `ProcessInstanceSearchQueryResult` DTO (`@nanobpm/engine-wasm/readmodel-types`) so each // row is a typed `ProcessInstanceResult` rather than a hand-scraped `Record` bag. const body: ProcessInstanceSearchQueryResult = JSON.parse( this.#liveEngine.searchProcessInstances("{}"), ); const out: ProcessInstanceSnapshot[] = []; // Normalize the parent/root selectors through the shared presence rule (`presentKey`) so a // blank/whitespace-only selector is *dropped* (matches everything) rather than compared // literally against normalized row keys — which would treat a padded/blank selector as present // and silently filter out every row, diverging from `SdkEngineClient` (which normalizes each // selector via `presentEngineKey` before sending it server-side). No Drift Surfaces. const wantParentProcessInstanceKey = presentKey(filter?.parentProcessInstanceKey); const wantRootProcessInstanceKey = presentKey(filter?.rootProcessInstanceKey); // Same untyped-JSON defence as `searchUserTasks`: `searchRows` guards the body and drops // non-object rows so a malformed/changed engine response can't throw while reading // `inst.processInstanceKey`. for (const inst of searchRows(body)) { const key = presentKey(inst.processInstanceKey); // Skip keyless items so a missing/null key can't leak in as "" (matches // the live SDK adapter, which drops instances with no key). if (key === undefined) continue; if (wanted && !wanted.has(key)) continue; const state = wasmStateToProcessInstanceState(inst.state); if (state === undefined) continue; if (filter?.state !== undefined && state !== filter.state) continue; // Parent/root selectors mirror the live `SdkEngineClient` (honoured server-side there); // apply them client-side against the engine's row linkage. The process-instance read model // carries both keys, so a native child/descendant is selectable by its parent or root. const parentProcessInstanceKey = presentKey(inst.parentProcessInstanceKey); const rootProcessInstanceKey = presentKey(inst.rootProcessInstanceKey); if ( wantParentProcessInstanceKey !== undefined && parentProcessInstanceKey !== wantParentProcessInstanceKey ) { continue; } if ( wantRootProcessInstanceKey !== undefined && rootProcessInstanceKey !== wantRootProcessInstanceKey ) { continue; } const processDefinitionKey = presentKey(inst.processDefinitionKey); out.push({ processInstanceKey: key, state, ...(processDefinitionKey ? { processDefinitionKey } : {}), ...(parentProcessInstanceKey ? { parentProcessInstanceKey } : {}), ...(rootProcessInstanceKey ? { rootProcessInstanceKey } : {}), }); } return out; } /** * Search element instances ("tokens") — derived from the primary-state snapshot, which the * WASM engine keeps in lieu of a `/v2/element-instances/search` read channel. Each running * instance records its *active* element instances (`instances[].activeElements[]`), so this * surfaces the ACTIVE set — "the furthest element reached", including active non-user-task * elements a user-task search cannot see. A completed/terminated element instance is not * retained with a key in the snapshot, so a caller filtering `state: "COMPLETED"` / * `"TERMINATED"` gets an empty result here (unlike the gateway-backed `SdkEngineClient`, * which the engine serves those from server-side). `elementType` is likewise absent — the * snapshot does not carry the BPMN type — matching `ElementInstanceSummary`'s optional field. */ async searchElementInstances( filter?: ElementInstanceFilter, ): Promise { return deriveElementInstances(this.#snapshot(), filter); } /** * Search element-instance *wait states* — the parks the deployed engine's read model * serves, derived from the snapshot's `jobs` (JOB), `messageSubscriptions` (MESSAGE) and * `userTasks` (USER_TASK). A JOB/MESSAGE park row carries its `elementId` + owning process * instance but not the element-instance key, so it is joined to the same instance's * `activeElements` to resolve the key `searchElementInstances` reports; a USER_TASK row * instead carries its `elementInstanceKey` directly — either way the two stay consistent. * * **Deployed floor (`JOB | MESSAGE | USER_TASK`).** This synthesizes only the park kinds the * deployed nanobpmn gateway implements ({@link DEPLOYED_WAIT_STATE_TYPES}), and rejects a * `waitStateType` filter outside it with {@link UnsupportedWaitStateTypeError} — exactly as * the gateway-backed `SdkEngineClient` does (the live gateway answers HTTP 422). A * `USER_TASK` park is synthesized here (Magikcraft/nano-bpm#1042 shipped it in the gateway's * read model) and can also be read via `searchUserTasks`; `TIMER`/`SIGNAL`/`CONDITION` are * still *not* synthesized (the 8.10 follow-on), so the emulation cannot report a park a real * engine would not — a query that passes here passes against a live engine too. Ref: * Magikcraft/nano-bpm#1042. */ async searchElementInstanceWaitStates( filter?: ElementInstanceWaitStateFilter, ): Promise { return deriveWaitStates(this.#snapshot(), filter); } /** Fetch a single element instance by key — the snapshot-derived active set (see * {@link searchElementInstances}) narrowed to the one whose key matches, or `null` when no * such active element instance exists (mirrors `SdkEngineClient.getElementInstance`, which * returns `null` on a 404). A blank key can address nothing → `null`. */ async getElementInstance( elementInstanceKey: string, ): Promise { if (typeof elementInstanceKey !== "string" || elementInstanceKey.trim() === "") { return null; } const key = elementInstanceKey.trim(); const match = deriveElementInstances(this.#snapshot(), undefined) .find((e) => e.elementInstanceKey === key); return match ?? null; } /** Incidents the WASM engine currently records. The engine retains an incident in its snapshot * only while it is *open* (a resolved incident drops out), so every reported incident is * `ACTIVE`; a request for any other lifecycle state therefore yields nothing. Each snapshot * incident carries `key`/`instanceKey`/`elementId`/`kind`/`reason` but no job key, so `jobKey` * is left absent (matching `IncidentSummary`'s optional field); the element-instance key is * resolved by joining `(instanceKey, elementId)` to the owning instance's `activeElements` — * the same index `searchElementInstanceWaitStates` uses, so the two cannot drift. The * `processInstanceKey`/`state` selectors are applied client-side, mirroring the effective * behaviour of the gateway-backed `SdkEngineClient`. */ async searchIncidents(filter?: IncidentFilter): Promise { // Only open (ACTIVE) incidents are retained in the snapshot; a request for any other // lifecycle state cannot be satisfied here, so it yields nothing rather than a wrong row. if (filter?.state !== undefined && filter.state !== "ACTIVE") return []; const snapshot = this.#snapshot(); const keyIndex = activeElementKeyIndex(snapshot); // Normalize the request selector the same way the response mapper normalizes keys: a // whitespace-only/padded `processInstanceKey` must not silently filter out every row (a blank // selector is treated as absent, matching the presence rule used elsewhere in this file and the // `SdkEngineClient` adapter — No Drift Surfaces). const wantInstanceKey = presentKey(filter?.processInstanceKey); const out: IncidentSummary[] = []; for (const inc of records(snapshot.incidents)) { const incidentKey = presentKey(inc.key); const processInstanceKey = presentKey(inc.instanceKey); if (incidentKey === undefined || processInstanceKey === undefined) continue; if ( wantInstanceKey !== undefined && processInstanceKey !== wantInstanceKey ) { continue; } const elementId = presentString(inc.elementId); const elementInstanceKey = elementId !== undefined ? keyIndex.get(`${processInstanceKey}\u0000${elementId}`) : undefined; const errorType = presentString(inc.kind); const errorMessage = presentString(inc.reason); out.push({ incidentKey, processInstanceKey, ...(elementId ? { elementId } : {}), ...(elementInstanceKey ? { elementInstanceKey } : {}), ...(errorType ? { errorType } : {}), ...(errorMessage ? { errorMessage } : {}), state: "ACTIVE", }); } return out; } /** Search process variables — the read counterpart to {@link setVariables}. Delegates to the * engine's real REST read channel (`POST /v2/variables/search`) and parses through the derived * `VariableSearchQueryResult` DTO. The read model does not yet honour filter/sort/page fields * server-side (it returns every variable), so the `processInstanceKey`/`scopeKey`/`name` * selectors are applied client-side — mirroring the *effective* behaviour of the gateway-backed * `SdkEngineClient`, which gets them honoured server-side; only the filtering site differs. Each * row carries its serialized JSON `value` and the engine's `isTruncated` flag. */ async searchVariables(filter?: VariableFilter): Promise { const body: VariableSearchQueryResult = JSON.parse( this.#liveEngine.searchVariables("{}"), ); const wantInstanceKey = presentKey(filter?.processInstanceKey); const wantScopeKey = presentKey(filter?.scopeKey); const wantName = present(filter?.name); const out: VariableSummary[] = []; // Same untyped-JSON defence as `searchProcessInstances`: `searchRows` guards the body and // drops non-object rows so a malformed/changed engine response can't throw while mapping. for (const row of searchRows(body)) { const variableKey = presentKey(row.variableKey); const processInstanceKey = presentKey(row.processInstanceKey); const name = presentString(row.name); if (variableKey === undefined || processInstanceKey === undefined || name === undefined) { continue; } // A process-level variable is scoped to the process instance itself; fall back to the // process-instance key when the engine omits `scopeKey`, matching `SdkEngineClient`. const scopeKey = presentKey(row.scopeKey) ?? processInstanceKey; if (wantInstanceKey !== undefined && processInstanceKey !== wantInstanceKey) continue; if (wantScopeKey !== undefined && scopeKey !== wantScopeKey) continue; if (wantName !== undefined && name !== wantName) continue; // The engine serializes `value` as a JSON string; keep a string as-is, otherwise JSON-encode // so the field is always a string the caller can `JSON.parse` (parity with `SdkEngineClient`). const value = typeof row.value === "string" ? row.value : JSON.stringify(row.value ?? null); out.push({ variableKey, name, value, scopeKey, processInstanceKey, isTruncated: row.isTruncated === true, }); } return out; } /** Search jobs — the "is it actually stuck" read (a `CREATED` job WITH a `worker` set is leased, * one with NONE set is queued) and the on-tool source of a `jobKey` for {@link updateJobRetries}. * Derived from the snapshot's `jobs` (the WASM engine has no `/jobs/search` read channel): each * snapshot job carries `key`/`jobType`/`instanceKey`/`elementId`/`retries`/`state`. The pull-based * WASM engine locks a job to a worker only transiently inside {@link drain}, so a parked job has * no `worker`/`deadline` — both are emitted only when the snapshot carries them (matching * `JobSummary`'s optional fields). The `processInstanceKey`/`state`/`type`/`elementId`/`worker` * selectors are applied client-side, mirroring the effective behaviour of the gateway-backed * `SdkEngineClient`. */ async searchJobs(filter?: JobFilter): Promise { const wantInstanceKey = presentKey(filter?.processInstanceKey); const wantState = present(filter?.state); const wantType = present(filter?.type); const wantElementId = present(filter?.elementId); const wantWorker = present(filter?.worker); const out: JobSummary[] = []; for (const job of records(this.#snapshot().jobs)) { const jobKey = presentKey(job.key); const type = presentString(job.jobType); const processInstanceKey = presentKey(job.instanceKey); if (jobKey === undefined || type === undefined || processInstanceKey === undefined) continue; // The snapshot spells job state in PascalCase (e.g. `"Created"`); normalize to the REST/v2 // enum spelling the gateway-backed `SdkEngineClient` reports so the two adapters agree. const state = normalizeJobState(job.state); if (state === "") continue; const worker = presentString(job.worker); const elementId = presentString(job.elementId); const deadline = presentString(job.deadline); const retries = typeof job.retries === "number" && Number.isFinite(job.retries) ? job.retries : undefined; if (wantInstanceKey !== undefined && processInstanceKey !== wantInstanceKey) continue; if (wantState !== undefined && state !== wantState) continue; if (wantType !== undefined && type !== wantType) continue; if (wantElementId !== undefined && elementId !== wantElementId) continue; // An unset `worker` cannot match a `worker` selector — a queued job is excluded from a // "leased by worker X" search, matching the live engine's server-side filter. if (wantWorker !== undefined && worker !== wantWorker) continue; out.push({ jobKey, type, state, processInstanceKey, ...(worker ? { worker } : {}), ...(retries !== undefined ? { retries } : {}), ...(elementId ? { elementId } : {}), ...(deadline ? { deadline } : {}), }); } return out; } /** Fetch the deployed BPMN XML of a process definition by key — the deployed routing model (with * its FEEL gateway conditions), the source of truth for WHY an instance routed where it did. The * WASM engine has no `/v2/process-definitions/{key}/xml` read channel, but its event log retains a * `ProcessDeployed` event carrying `process_definition_key` + the full `process.xml`, so this * scans the log for the matching key — engine truth, no side-channel capture. Returns `null` for * a blank/unknown key (mirroring `SdkEngineClient.getProcessDefinitionXml`, which returns `null` * on a 404). */ async getProcessDefinitionXml(processDefinitionKey: string): Promise { const key = presentKey(processDefinitionKey); if (key === undefined) return null; // The most recent deploy of a key wins if a key ever recurred; scan newest-first. const events = this.#parseArray(this.#liveEngine.events()); for (let i = events.length - 1; i >= 0; i--) { const ev = events[i]; if (ev.type !== "ProcessDeployed") continue; if (presentKey(ev.process_definition_key) !== key) continue; const process = isRecord(ev.process) ? ev.process : undefined; const xml = presentString(process?.xml); if (xml !== undefined) return xml; } return null; } /** Engine-native agent instances the WASM double records. The bundled μ-nano.wasm read model has * no AgentInstance/AgentHistory channel (it exposes none of the `/agent-instances/*` endpoints * the `SdkEngineClient` reads), so this emulates the floor a consumer needs to boot against the * testkit without a `TypeError`: no agent instance is ever recorded, so the search is empty. * This keeps the `EngineClient` surface complete (issue #341 conformance) and lets the historical * transcript consumer (nanobpm/nano-workforce#747) exercise its read path here; the *behavioural* * agent-history parity is validated against a live engine, which this in-process double omits. * The read-as-absence shape (empty list / `null`) matches `SdkEngineClient`, so the two adapters * agree on the "no such instance" answer. */ async searchAgentInstances(_filter?: AgentInstanceFilter): Promise { return []; } /** One agent instance's conversation history — see {@link searchAgentInstances} for why the WASM * double records none. A blank key addresses nothing (parity with `SdkEngineClient`, which * short-circuits it); every key yields the empty list here. */ async searchAgentInstanceHistory( _agentInstanceKey: string, _filter?: AgentHistoryFilter, ): Promise { return []; } /** Fetch a single agent instance by key — always `null` here (the WASM double records none; see * {@link searchAgentInstances}), matching `SdkEngineClient`'s read-as-absence on a 404 and its * blank-key short-circuit. */ async getAgentInstance(_agentInstanceKey: string): Promise { return null; } /** Resolve an open incident by key, returning the parked token to progress (a job incident * returns its job to the activatable pool — which must have retries left, see * {@link updateJobRetries}). Drains afterward so a re-activated job is served, mirroring a live * push engine (parity with `SdkEngineClient`, whose engine serves it server-side). */ async resolveIncident(input: { incidentKey: string }): Promise { this.#liveEngine.resolveIncident(requireKey(input.incidentKey, "incidentKey")); await this.drain(); } /** Set a failed job's remaining retries — the "retry a failed job" operation. Bumping retries * back above zero makes the job activatable again so a paired {@link resolveIncident} can * return it to the pool. Drains afterward for the same push-parity reason as the other * mutating calls. */ async updateJobRetries(input: { jobKey: string; retries: number }): Promise { const jobKey = requireKey(input.jobKey, "jobKey"); // `retries` reaches the engine as the job's remaining retry count, and `number` admits `NaN`, // fractional, and negative values — none of which is a valid retry count. Reject them up front // so a tool-driven call fails with a clear message instead of an opaque engine error, matching // `SdkEngineClient.updateJobRetries`. if (!Number.isInteger(input.retries) || input.retries < 0) { throw new Error("retries must be a non-negative integer"); } this.#liveEngine.updateRetries(jobKey, input.retries); await this.drain(); } /** Set (merge) variables into a scope (`scopeKey` is a process-instance or element-instance * key). `local` merges strictly into that local scope; the default (`false`) propagates to the * outermost scope — the same semantics the live SDK adapter's `createElementInstanceVariables` * honours. Drains afterward so a variable that unblocks a waiting worker takes effect. */ async setVariables(input: { scopeKey: string; variables: Record; local?: boolean; }): Promise { this.#liveEngine.setVariables( requireKey(input.scopeKey, "scopeKey"), JSON.stringify(input.variables), input.local ?? false, ); await this.drain(); } async registerWorker( jobType: string, handler: JobHandler, options?: { workerName?: string; maxParallelJobs?: number; fetchVariables?: string[] }, ): Promise { // Gate registration on the same live-engine boundary every other public op uses (No Drift // Surfaces): after close()/free() a fresh worker would be inserted into `#workers` (already // cleared by `#doClose`) and — because `drain()` bails on `#closing` — the call would still // resolve with a live-looking subscription that retains the handler on a freed engine. Touch // `#liveEngine` for its freed-throw side effect so that misuse faults with the categorical // "used after close()" error instead of leaking a subscription over a dead handle. void this.#liveEngine; this.#workers.set(jobType, { handler, workerName: options?.workerName ?? `urban-testkit:${jobType}`, maxParallelJobs: options?.maxParallelJobs ?? DEFAULT_MAX_JOBS, fetchVariables: options?.fetchVariables, }); // A freshly-registered worker picks up any jobs already waiting for its type. await this.drain(); const workers = this.#workers; return { jobType, async unsubscribe() { workers.delete(jobType); }, }; } async close(): Promise { // A worker handler can call close() (it reaches the engine via `AppApi`). Record its OWN tracked // promise as a close-awaiter so `#settleInflight()` excludes it from the wait: that promise cannot // settle until close() returns, so waiting on it would deadlock. Register here for EVERY handler // that enters close() — the initiator AND any peer that re-enters the memoized run — not just the // one whose async context `#settleInflight()` happens to run in, so the two-handler case (a peer // parks on `#closeRun` while the initiator's `#settleInflight()` waits on that peer) cannot wedge. // Waking a mid-await settle pass (below) is what lets it drop a peer that parks on close() AFTER // the pass already started awaiting it. const ownAwaiter = this.#handlerContext.getStore()?.own; if (ownAwaiter !== undefined && !this.#closeAwaiters.has(ownAwaiter)) { this.#closeAwaiters.add(ownAwaiter); this.#closeAwaiterWake?.(); } // Idempotent teardown: memoize the single shutdown run so a second close() (concurrent, or after // the first resolves) awaits the SAME promise instead of re-running `#settleInflight()` and a // second `#engine.free()` on the same WASM handle (a double-free). A run that fails *before* // `free()` deliberately leaves the engine allocated (see `#doClose`); clear the memo in that case // so the caller can cancel/retry. But once `free()` has run (`#freed`), keep even a rejected run // memoized: `#doClose` can still reject *after* `free()` when `#throwInflightError()` surfaces a // late completion error, and re-running it would `free()` the already-freed handle a second time. // // Re-entrancy: `#doClose()` synchronously aborts `#shutdown`, and `shutdownSignal` is PUBLIC, so an // abort listener can call `close()` again DURING that dispatch. A naive `this.#closeRun ??= // this.#doClose()` evaluates `#doClose()` (which aborts, dispatching the listener) BEFORE the `??=` // stores its promise, so the re-entrant call still sees `#closeRun === undefined`, starts a SECOND // `#doClose()`, and both reach `free()` — the double-free the memo exists to prevent. Publish the // memo (a deferred) BEFORE invoking `#doClose()` so the re-entrant call observes the in-progress run // and awaits it. `#closing`/`#shutdown.abort()` stay synchronous inside `#doClose()`, unchanged. if (this.#closeRun === undefined) { let settle!: () => void; let fail!: (err: unknown) => void; this.#closeRun = new Promise((resolve, reject) => { settle = resolve; fail = reject; }); this.#doClose().then(settle, (err: unknown) => { if (!this.#freed) this.#closeRun = undefined; fail(err); }); } return this.#closeRun; } async #doClose(): Promise { // Enter the shutdown state up front (idempotent). `#closing` gates a concurrently-suspended // `drain()` from activating jobs on an engine that is about to be freed, and aborting `#shutdown` // cancels handlers parked on the app clock's `wait()` — a virtual timer no `advanceTime` will // fire during teardown. Aborting BEFORE `#settleInflight()` is what makes that settle finite: the // parked handler's `app.wait()` rejects, the handler unwinds, and its throw is mapped to `failJob` // on the still-live engine (this all runs before `free()`), so `#inflight` drains to empty. A // handler parked on *real-time* async work ignores the signal and is still fully awaited (#446). this.#closing = true; if (!this.#shutdown.signal.aborted) { this.#shutdown.abort(new Error("WasmEngineClient closing")); } // Fire-and-forget dispatch (see #track/drain) means a drain can return with a handler still // in-flight. Await every in-flight handler to FULL settlement before `free()` — NOT just a // macrotask fixpoint. A handler parked on *real-time* async work (the reproduction was a // worker spawning a subprocess) holds #inflight.size steady across macrotasks, so the old // `#quiesce()` fixpoint declared quiescence and returned with the handler still pending; // `free()` then released the engine underneath it, and when the handler resumed it called // `completeJob` on a released wasm handle — the opaque "null pointer passed to rust" // use-after-free (issue #446). Awaiting the actual handler promises closes that race // categorically. // // CRITICAL: `free()` must be reached ONLY once `#inflight` is empty. If `#settleInflight()` // exhausts its iteration guard it throws with handlers still pending — we deliberately do NOT // free in that case (no `finally`): a handler that later resumes would call `completeJob` on a // released engine, reproducing the very issue #446 use-after-free the drain exists to prevent. // Leaving the engine allocated is a leak, but a leak is strictly safer than a use-after-free, // and it keeps the engine available for the caller to cancel/retry while the failure surfaces. await this.#settleInflight(); // Reached only once `#inflight` is empty — safe to release the engine. this.#workers.clear(); this.#workerMocks.clear(); this.#engine.free(); this.#freed = true; // Settlement succeeded; surface (once, fail-loud) the first engine-completion error a tracked // handler captured, so a late worker failure is not silently swallowed at teardown. this.#throwInflightError(); } // --- Extras beyond EngineClient (used by the settle loop + assertions) --- /** * Register (or fetch the existing) job-worker mock for `taskType` (epic #296, S1). * A mocked type is resolved by its {@link MockWorkerBuilder} at the dispatch seam * ({@link drain} → `#runJob`) — completed / failed / errored by the mock's matching * clause **instead of** the app's real handler — while un-mocked types, and jobs a * mock's `when(...)` clauses don't match, still run real code. The builder shadows * the real handler for the app's lifetime; call `.reset()` on it (or * {@link clearWorkerMock}) to remove the mock and restore real behaviour. * * Idempotent per type: repeated calls return the SAME builder, so conditional * clauses accumulate in registration order across calls. Purely opt-in — no mock * bookkeeping happens on any dispatch until this is first called for a type. */ mockWorker(taskType: string): MockWorkerBuilder { let builder = this.#workerMocks.get(taskType); if (builder === undefined) { builder = new MockWorkerBuilder(() => { this.#workerMocks.delete(taskType); }); this.#workerMocks.set(taskType, builder); } return builder; } /** * Remove any job-worker mock for `taskType`, restoring its real handler. No-op if * unmocked. Delegates to the builder's {@link MockWorkerBuilder.reset} so a test that * still holds the builder reference sees its clauses and pending predicate cleared too * — not just the registry entry dropped. */ clearWorkerMock(taskType: string): void { this.#workerMocks.get(taskType)?.reset(); } /** Advance the virtual clock by `ms`, firing due timers, then drain workers so * any jobs the timers created are served. */ async advanceTime(ms: number): Promise { this.#liveEngine.advanceTime(ms); await this.drain(); } /** The current virtual clock (ms). */ get now(): number { return this.#liveEngine.now; } /** * Register an observer notified with the `jobType` each time a job is dispatched to a * worker handler (before the handler runs, so a failing handler still counts as * "exercised"). The second argument reports whether that dispatch was satisfied by a * job-worker mock (epic #296, S4) — the coverage gate records the type as exercised * either way, and additionally flags it as mocked so a mocked worker stays an honest, * visible entry rather than a silently-hidden gap. Returns an unsubscribe. A single * observer is held — a later call replaces the earlier one — which is all the coverage * gate needs; pass `undefined` (or call the returned unsubscribe) to clear it. The * observer is a passive spectator: a throw from it is swallowed and never affects job * completion, the drain, or incidents. */ observeJobs(onJob: ((jobType: string, mocked: boolean) => void) | undefined): () => void { this.#onJob = onJob; return () => { if (this.#onJob === onJob) this.#onJob = undefined; }; } /** The raw engine snapshot (parsed). */ snapshot(): Record { return this.#snapshot(); } /** * Serve every registered worker's — and every mock-only type's — activatable jobs to * quiescence: repeatedly activate + run + acknowledge until a full pass activates * nothing. A handler that returns a value completes the job with it; a * {@link isBpmnError} throw raises a BPMN error; any other throw fails the job * (decrementing retries, raising an incident at zero) — parity with the live adapter. * * A **mock-only** type (one with a {@link mockWorker} mock but no real * `registerWorker`) is also drained, so a test can mock a worker the app never * registered (e.g. an unimplemented external integration). Its jobs are activated and * resolved by the mock; a job the mock's clauses don't match is left locked (no real * handler exists to run it), which keeps the drain a fixpoint rather than spinning. */ async drain(): Promise { for (let i = 0; i < MAX_DRAIN_ITERATIONS; i++) { // Shutdown lifecycle guard: once close() has begun, a drain that starts — or a suspended drain // that resumes at this loop top — must NOT run another activation pass. close() may have already // freed (or be about to free) the engine once #settleInflight drains #inflight, so a post-close // activateJobs would drive the freed handle (the #settleInflight-vs-drain race the #446 fix's // reviewer flagged). Any handler THIS drain already dispatched is in #inflight and is still // awaited by close()'s #settleInflight before free(), so bailing here strands nothing. if (this.#closing) return; let activatedAny = false; for (const jobType of this.#dispatchableJobTypes()) { // Batch-level shutdown guard (companion to the loop-top `#closing` guard above, which only // stops the NEXT pass). A handler dispatched EARLIER in THIS synchronous pass can self-initiate // teardown — `await app.close()` runs close()'s synchronous head (setting `#closing`, aborting // `#shutdown`, and snapshotting `#inflight` in `#settleInflight()`) before it yields — so by the // time this loop advances to the next jobType/job, `#closing` is already true. Without bailing // here, this pass would `activateJobs` and `#track` MORE handlers onto an engine whose // `#settleInflight()` already declared quiescence (it saw only the excluded self-closer), and // `free()` would then run with those later handlers still in-flight — the very issue #446 // use-after-free the settle barrier exists to prevent. Everything already dispatched this pass // is in `#inflight` and awaited by close() before `free()`, so bailing strands nothing. if (this.#closing) return; const realWorker = this.#workers.get(jobType); // A mock-only type has no registered worker; use a synthetic activation // descriptor so its jobs can be pulled. Its handler is never invoked. const worker = realWorker ?? mockOnlyWorker(jobType); const jobs = this.#parseArray( this.#engine.activateJobs( jobType, worker.maxParallelJobs, JOB_LOCK_MS, worker.workerName, ), ); for (const raw of jobs) { // Same batch-level guard, now per-job: a handler dispatched a moment ago in THIS same // `jobs` array can have synchronously self-closed, flipping `#closing` mid-batch. Stop // dispatching the remaining jobs so none is `#track`ed onto an engine being freed. if (this.#closing) return; // Advancing virtual time can expire a still-running job's activation lock, so the engine // re-offers a job whose handler is already in-flight (parked on a future `app.wait`). Don't // spawn a duplicate handler for it, and don't count it as progress — the in-flight instance // owns its single completion (a second `completeJob` would hit "not in a state that can be // acted on"). const jobKey = str(raw.key); if (jobKey !== "" && this.#inflightJobKeys.has(jobKey)) continue; activatedAny = true; // Fire-and-forget: a push worker runs autonomously, so dispatching must not block the // drain on the handler. A quick handler is still awaited to completion by the `#quiesce` // below (so its effects are visible after `settle`); a handler parked on a *future* // virtual-clock `app.wait` is left in-flight for `advanceTime` to drive, instead of // deadlocking the drain on a clock that only moves later. this.#track(jobKey, (store) => this.#runJob(worker, realWorker !== undefined, raw, store)); } } // Let every handler dispatched this iteration run until it completes (its `completeJob` may // enqueue fresh engine work the next iteration picks up) or parks on a future wait. const completionsBefore = this.#completions; await this.#quiesce(); // A handler that *settled* during `#quiesce` (as opposed to merely parking on a future wait) // may have enqueued fresh engine work via its `completeJob`. Even when this pass activated // nothing, that newly enqueued work must get an activation pass, so only reach the fixpoint — // and return — once a pass both activates nothing new AND settles no in-flight handler. const settledAny = this.#completions !== completionsBefore; if (!activatedAny && !settledAny) return; } throw new Error( `drain did not quiesce after ${MAX_DRAIN_ITERATIONS} iterations (worker re-creating work?)`, ); } /** Track a fire-and-forget worker handler promise (and its job key) until it settles. `#runJob` * maps a handler throw onto the engine's completion surface internally, so a rejection here can * only come from the engine completion call itself; capture it (deduped) so {@link drain} can * rethrow it loudly rather than let it escape as an unhandled rejection. */ #track(jobKey: string, run: (store: HandlerStore) => Promise): void { if (jobKey !== "") this.#inflightJobKeys.add(jobKey); // Register the tracked promise in `#inflight` and publish it as the handler's OWN promise BEFORE // `run()` executes. A handler that calls close() synchronously — before its first await — must // find its tracked promise already in `#inflight` and already published as `store.own`; otherwise // close() -> `#settleInflight()` sees no peer for it, frees the engine, and `#track` then adds a // still-pending promise to a freed engine while the handler resumes on the dead handle (the issue // #446 use-after-free). `tracked` is derived from the handler promise (which does not exist until // `run()` starts), so use a DEFERRED: add it to `#inflight`/`store.own` up front, run the handler // inside its async context, and resolve the deferred when the handler settles. let settleTracked!: () => void; const tracked = new Promise((resolve) => { settleTracked = resolve; }); // `store.own` identifies THIS handler's tracked promise so a nested close() the handler makes can // exclude itself from `#settleInflight()`. `#runJob` scopes `#handlerContext` to the actual // `worker.handler` invocation ONLY (see there): dispatch-time callbacks it runs first — the // `observeJobs` coverage observer and mock predicates — must NOT run under this store, or an // observer re-entering close() would read this still-parked handler's promise as its OWN and // exclude it, freeing the engine under a live handler (#446). So `#track` passes `store` to `run` // by value instead of entering the context here; only the handler call enters it. const store: HandlerStore = { own: tracked }; this.#inflight.add(tracked); void run(store) .catch((err: unknown) => { if (this.#inflightError === undefined) this.#inflightError = err; }) .finally(() => { this.#completions++; this.#inflight.delete(tracked); this.#closeAwaiters.delete(tracked); if (jobKey !== "") this.#inflightJobKeys.delete(jobKey); settleTracked(); }); } /** Drain in-flight worker handlers to a fixpoint at the current virtual instant: flush macrotasks * until the in-flight count stops changing. A handler that completes (or fails) leaves the set * and may enqueue engine work; a handler parked on a *future* `app.wait` timer stops consuming * microtasks, so the count stabilises and this returns with that handler still in-flight — to be * resumed by a later `advanceTime`. Rethrows the first engine-completion error a tracked handler * surfaced, preserving the fail-loud semantics the previous inline `await` had. */ async #quiesce(): Promise { let prev = -1; while (this.#inflight.size !== prev) { prev = this.#inflight.size; await flushMacrotask(); } this.#throwInflightError(); } /** Await every in-flight fire-and-forget handler to FULL settlement. Unlike {@link #quiesce}'s * macrotask fixpoint, this awaits the actual tracked promises, so a handler parked on *real-time* * async work (e.g. a worker awaiting a spawned subprocess) — which holds `#inflight.size` steady * and is therefore invisible to the fixpoint — is genuinely waited for. Used by {@link close} so * the engine is never {@link TestEngine.free freed} while such a handler is still running (issue * #446 use-after-free). Loops to a fixpoint because a handler settling can, in principle, leave * a follow-on tracked handler behind; `Promise.allSettled` snapshots the set, so re-check until * it is empty. Bounded by {@link MAX_DRAIN_ITERATIONS} so a pathological handler that perpetually * re-creates in-flight work throws (leaving `#inflight` non-empty) rather than hanging teardown * forever — {@link close} treats that throw as "not settled" and pointedly does NOT free. A * handler that self-initiates close() (or a peer that re-enters it) is excluded from the wait (see * the body) so it does not dead-await itself. */ async #settleInflight(): Promise { // A worker handler can itself call close() (workers reach the engine via `AppApi`), so its OWN // tracked promise sits in `#inflight` while it awaits close(). That promise can only settle AFTER // close() returns, so waiting on it here would deadlock — close() awaiting the handler that is // awaiting close() — and never reach the iteration cap (it is merely pending, not re-creating // work). Exclude EVERY close-awaiter (see {@link #closeAwaiters}): the initiator and any peer that // re-entered the memoized run. A peer can join `#closeAwaiters` AFTER a pass already began awaiting // it via `Promise.allSettled`, so race that wait against `#closeAwaiterWake` — resolved by close() // whenever it adds an awaiter — and re-filter the moment a peer parks on close(). For an external // close() the set is empty and nothing is excluded — the original whole-set wait, unchanged. for (let i = 0; i < MAX_DRAIN_ITERATIONS; i++) { const peers = [...this.#inflight].filter((p) => !this.#closeAwaiters.has(p)); if (peers.length === 0) break; const woken = new Promise((resolve) => { this.#closeAwaiterWake = resolve; }); await Promise.race([Promise.allSettled(peers), woken]); this.#closeAwaiterWake = undefined; } if ([...this.#inflight].some((p) => !this.#closeAwaiters.has(p))) { throw new Error( `close() did not settle in-flight handlers after ${MAX_DRAIN_ITERATIONS} iterations ` + "(a worker handler perpetually re-creating in-flight work?)", ); } } /** Surface (once) the first engine-completion error any tracked handler captured, clearing it so * a later drain does not re-throw the same error. Single source of truth for {@link #inflightError} * — shared by {@link #quiesce} and {@link close} so the two paths cannot drift. */ #throwInflightError(): void { if (this.#inflightError !== undefined) { const err = this.#inflightError; this.#inflightError = undefined; throw err; } } /** The job types to activate on a drain pass: every registered worker, plus every mock-only type * that actually carries at least one clause (an empty/reset mock induces no dispatch). * De-duplicated so a type that lands in more than one source is only activated once per pass. */ #dispatchableJobTypes(): string[] { const types = new Set(this.#workers.keys()); for (const [jobType, mock] of this.#workerMocks) { if (!this.#workers.has(jobType) && mock.hasClauses) types.add(jobType); } return [...types]; } async #runJob( worker: RegisteredWorker, hasRealWorker: boolean, raw: Record, store: HandlerStore, ): Promise { const jobKey = str(raw.key); // A keyless job cannot be completed/failed/errored; skip it rather than // issue an invalid completeJob("") (mirrors the live SDK adapter, which // logs and leaves such a job for redelivery). if (jobKey === "") return; const jobType = str(raw.type); const allVariables = isRecord(raw.variables) ? raw.variables : {}; const job: EngineJob = { jobKey, jobType, processInstanceKey: raw.instanceKey == null ? undefined : str(raw.instanceKey), elementId: typeof raw.elementId === "string" ? raw.elementId : undefined, // Honour fetchVariables: surface only the requested subset (intersected // with what is present), matching the live SDK adapter's server-side fetch. variables: worker.fetchVariables === undefined ? allVariables : pick(allVariables, worker.fetchVariables), }; // Resolve any registered mock for this type against the constructed `EngineJob` (whose // `variables` already reflect any `fetchVariables` filtering applied above) BEFORE notifying the // coverage observer, so the observer learns whether this dispatch is mock-satisfied. A // mock whose `when(...)` clauses don't match yields `undefined` here and falls through to // the real handler below, exactly like an un-mocked type. Mock resolution is a pure lookup // over the in-memory registry — it introduces no timers/real-time, so `drain()` still // reaches a fixpoint deterministically. const mockOutcome = this.#workerMocks.get(jobType)?.resolve(job); // Will this dispatch actually be serviced? Only when a mock clause matched (`mockOutcome`) OR a // real handler exists to run it. A mock-only type whose clauses don't match is left locked with // nothing run (see below), so it is NOT serviced — and must not be recorded as exercised, or a // job no mock and no handler touched would fabricate coverage and hide a genuine gap. const willBeServiced = mockOutcome !== undefined || hasRealWorker; // Notify the coverage observer (if any) that a job of this type was dispatched — before the // handler runs (or the mock applies), so an exercised-but-failing worker, and a mocked type // whose real handler never runs, both still count as exercised. The `mocked` flag lets the // gate record the type as covered yet flag it as mock-satisfied. Gated on `willBeServiced` so // an unserviceable dispatch never fabricates coverage. Isolated in its own try/catch: this is a // non-invasive test seam, so a throwing observer must never abort the drain nor alter // job/engine semantics (it would otherwise propagate out of #runJob). if (this.#onJob !== undefined && willBeServiced) { try { this.#onJob(jobType, mockOutcome !== undefined); } catch { // Swallow: an observer is a passive spectator of dispatch, never a participant. } } // A matching mock shadows the real handler: apply its outcome via the shared applier — the // exact same engine completion calls the real path below uses — and return without running // (or even needing) the app's handler. `applyOutcome` runs the same engine calls a real // handler resolves through (notably `JSON.stringify(variables)` for a `complete` outcome), // so it can throw exactly like the real path's completion does; route that throw through the // same error-to-`failJob`/`throwError` handling rather than letting it escape `#runJob` and // abort the whole drain. if (mockOutcome !== undefined) { try { applyOutcome(this.#engine, jobKey, mockOutcome); } catch (err) { this.#failFromError(raw, jobKey, err); } return; } // Fell through the mock (no match / no clause). With no real worker registered for this type // there is nothing to run — the job was activated by a mock-only type. Leave it (locked for // this virtual instant) rather than fabricate a completion; the drain still quiesces because a // locked job won't re-activate. With a real worker, run it exactly as an un-mocked type. if (!hasRealWorker) return; try { // Scope the handler `AsyncLocalStorage` context to the actual worker-handler invocation ONLY. // Everything above (the `#onJob` coverage observer, mock resolution/`applyOutcome`) is // dispatch-time machinery, NOT the handler — running it under this store would let a dispatch // callback that re-enters close() (e.g. an `observeJobs` observer) read THIS still-parked // handler's tracked promise as `store.own`, exclude it from `#settleInflight()`, and free the // engine under the live handler (#446). The store still propagates through every async // continuation of `worker.handler` — including a nested self-close() — so a handler that // genuinely initiates teardown is still identified and excluded from its own wait. const out = await this.#handlerContext.run(store, () => worker.handler(job)); // A handler can self-initiate teardown by calling close() (workers reach the engine through // `AppApi`); close() then frees the engine while this handler is parked on its await. On resume // the engine handle is gone, so completing the job would drive a freed pointer — the very issue // #446 use-after-free. `#settleInflight` deliberately excludes every close-awaiter from its wait // so a self-close does not deadlock; the price is that we must NOT touch the freed engine after. // (Any OTHER engine call a resumed self-closer makes — `publishMessage`, `createInstance`, … — // is guarded categorically by the `#liveEngine` accessor, which throws after `free()`; this // narrow `#freed` return only silences the job's own final `completeJob`.) if (this.#freed) return; this.#engine.completeJob(jobKey, JSON.stringify(out ?? {})); } catch (err) { // Same guard on the failure path: a handler that entered close() and then threw has no live // engine to fail the job against. (Only a handler parked INSIDE close() can run past `free()`: // `#settleInflight` awaits every OTHER in-flight handler before close() frees, and excludes only // close-awaiters, so no peer doing real work is ever mid-flight here.) if (this.#freed) return; this.#failFromError(raw, jobKey, err); } } /** Map a thrown handler/outcome error to the engine's completion surface — the single * canonical error→completion mapping shared by the real-handler path and the mock-apply path. * A {@link isBpmnError} throw raises a BPMN error (drives the modelled boundary); any other * throw fails the job, decrementing its redelivery budget (`retries - 1`, floored at 0, so the * last attempt raises an incident) exactly as the live SDK adapter does. */ #failFromError(raw: Record, jobKey: string, err: unknown): void { if (isBpmnError(err)) { this.#engine.throwError(jobKey, err.errorCode, err.message ?? err.errorCode); return; } const retries = typeof raw.retries === "number" ? raw.retries : 1; this.#engine.failJob( jobKey, Math.max(0, retries - 1), err instanceof Error ? err.message : String(err), ); } #snapshot(): Record { return this.#parseObj(this.#liveEngine.snapshot()); } #instanceVariables(key: string): Record { const inst = records(this.#snapshot().instances).find((i) => str(i.key) === key); return isRecord(inst?.variables) ? inst.variables : {}; } #parseObj(json: string): Record { const v: unknown = JSON.parse(json); return isRecord(v) ? v : {}; } #parseArray(json: string): Record[] { const v: unknown = JSON.parse(json); return records(v); } } /** Boot the wasm engine (idempotent) and return a fresh {@link WasmEngineClient}. */ export function createWasmEngineClient(): Promise { return WasmEngineClient.create(); } function requireCreated(created: unknown): string { if (created == null || created === "") { throw new Error("engine createInstance response missing `created` instance key"); } return String(created); } /** The `Record` elements of an unknown value (non-records dropped). */ function records(v: unknown): Record[] { return Array.isArray(v) ? v.filter(isRecord) : []; } /** Normalize the WASM snapshot's PascalCase job state (`"Created"`, `"TimedOut"`) to the REST/v2 * `JobStateEnum` spelling the gateway-backed `SdkEngineClient` reports (`"CREATED"`, * `"TIMED_OUT"`), by inserting `_` at camel boundaries and upper-casing — so the two adapters * report the same `state` string (No Drift Surfaces). Idempotent on an already-SCREAMING_SNAKE * value; `""` for a blank/absent state (the caller drops such a row). */ function normalizeJobState(raw: unknown): string { if (typeof raw !== "string") return ""; const t = raw.trim(); if (t === "") return ""; return t.replace(/([a-z0-9])([A-Z])/g, "$1_$2").toUpperCase(); } /** Build a `(processInstanceKey, elementId) → elementInstanceKey` index from each instance's * active elements. A snapshot park row (job/message/timer/signal) carries its `elementId` + * owning process instance but not the element-instance key, so it resolves the key through * this index — the same key {@link deriveElementInstances} reports, so the two cannot drift. * * When a single `(processInstanceKey, elementId)` maps to *more than one* active element * instance (a multi-instance activity's parallel tokens), the join is ambiguous — a park row * carries only `elementId` + process instance, not the element-instance key, so it can't be * paired to the right token. Such a key is marked ambiguous and left *absent* from the index, * so {@link waitStateIdentity} drops the park rather than attaching an arbitrary (wrong) key. */ function activeElementKeyIndex(snapshot: Record): Map { const index = new Map(); const ambiguous = new Set(); for (const inst of records(snapshot.instances)) { const processInstanceKey = presentKey(inst.key); if (processInstanceKey === undefined) continue; for (const el of records(inst.activeElements)) { const elementInstanceKey = presentKey(el.key); const elementId = presentString(el.elementId); if (elementInstanceKey === undefined || elementId === undefined) continue; const mapKey = `${processInstanceKey}\u0000${elementId}`; if (index.has(mapKey) || ambiguous.has(mapKey)) { index.delete(mapKey); ambiguous.add(mapKey); continue; } index.set(mapKey, elementInstanceKey); } } return index; } /** Derive the ACTIVE element instances the WASM snapshot records (see * {@link WasmEngineClient.searchElementInstances}). Applies the `processInstanceKey`/ * `elementId`/`state` selectors client-side, matching the effective behaviour of the * gateway-backed `SdkEngineClient`. */ export function deriveElementInstances( snapshot: Record, filter?: ElementInstanceFilter, ): ElementInstanceSummary[] { const out: ElementInstanceSummary[] = []; // Only ACTIVE element instances are retained (with keys) in the snapshot; a request for any // other lifecycle state cannot be satisfied here, so it yields nothing rather than a wrong row. if (filter?.state !== undefined && filter.state !== "ACTIVE") return out; for (const inst of records(snapshot.instances)) { const processInstanceKey = presentKey(inst.key); if (processInstanceKey === undefined) continue; if (filter?.processInstanceKey !== undefined && processInstanceKey !== filter.processInstanceKey) { continue; } for (const el of records(inst.activeElements)) { const elementInstanceKey = presentKey(el.key); const elementId = presentString(el.elementId); if (elementInstanceKey === undefined || elementId === undefined) continue; if (filter?.elementId !== undefined && elementId !== filter.elementId) continue; out.push({ elementInstanceKey, processInstanceKey, elementId, state: "ACTIVE" }); } } return out; } /** Derive element-instance wait states from the snapshot's park collections (see * {@link WasmEngineClient.searchElementInstanceWaitStates}). Synthesizes only the deployed * floor (`JOB | MESSAGE | USER_TASK`, {@link DEPLOYED_WAIT_STATE_TYPES}) — the park kinds the * deployed gateway's read model serves — so the emulation cannot report a park a live engine * would not. A `waitStateType` filter outside the floor is rejected with * {@link UnsupportedWaitStateTypeError} (the gateway answers HTTP 422). A JOB/MESSAGE park's * element-instance key is resolved through {@link activeElementKeyIndex}, while a USER_TASK * park carries its `elementInstanceKey` directly on the task row; the * `processInstanceKey`/`elementId`/`waitStateType` selectors are applied client-side. */ export function deriveWaitStates( snapshot: Record, filter?: ElementInstanceWaitStateFilter, ): ElementInstanceWaitState[] { // Reject an out-of-floor `waitStateType` up front, mirroring the gateway-backed // `SdkEngineClient` (which fails the same filter with HTTP 422). Enforced through the shared // guard so the floor has one canonical definition (No Drift Surfaces). assertDeployedWaitStateType(filter?.waitStateType); const keyIndex = activeElementKeyIndex(snapshot); const out: ElementInstanceWaitState[] = []; const accept = (w: ElementInstanceWaitState): void => { if (filter?.processInstanceKey !== undefined && w.processInstanceKey !== filter.processInstanceKey) { return; } if (filter?.elementId !== undefined && w.elementId !== filter.elementId) return; if (filter?.waitStateType !== undefined && w.waitStateType !== filter.waitStateType) return; out.push(w); }; // JOB parks — a service task awaiting a worker. A JOB without a jobType is malformed; skip it. for (const job of records(snapshot.jobs)) { const identity = waitStateIdentity(job, keyIndex); if (identity === undefined) continue; const jobType = presentString(job.jobType); if (jobType === undefined) continue; const jobKey = presentKey(job.key); accept({ ...identity, waitStateType: "JOB", jobType, ...(jobKey ? { jobKey } : {}) }); } // MESSAGE parks — an event awaiting message correlation. A MESSAGE without a messageName is // malformed; skip it. for (const sub of records(snapshot.messageSubscriptions)) { const identity = waitStateIdentity(sub, keyIndex); if (identity === undefined) continue; const messageName = presentString(sub.messageName); if (messageName === undefined) continue; const correlationKey = presentString(sub.correlationKey); accept({ ...identity, waitStateType: "MESSAGE", messageName, ...(correlationKey ? { correlationKey } : {}), }); } // TIMER/SIGNAL/CONDITION parks are deliberately *not* synthesized: the deployed gateway's // read model does not serve them yet (`assertDeployedWaitStateType` rejects a filter for // them, and an unfiltered search must not surface a park a live engine would omit) — the // 8.10 follow-on tracked in Magikcraft/nano-bpm#1042. // // USER_TASK parks — the snapshot carries the element-instance key directly on the task row. // A park exists only while the user task is OPEN: added on the user-task CREATED record and // removed when it completes/cancels (canonical lifecycle). The snapshot RETAINS completed/ // cancelled tasks (with `state: "Completed"`/`"Canceled"`), so gate on the open state rather // than the row's mere presence — otherwise a park would linger past completion, a drift a // live gateway would never show. The park's `userTaskKey` is the task row's own key. // Magikcraft/nano-bpm#1042 shipped this park in the gateway's read model, so re-synthesizing // it here keeps the emulation on the deployed floor rather than under-reporting it. for (const task of records(snapshot.userTasks)) { if (!isOpenUserTaskState(task.state)) continue; const processInstanceKey = presentKey(task.instanceKey); const elementId = presentString(task.elementId); const elementInstanceKey = presentKey(task.elementInstanceKey); const userTaskKey = presentKey(task.key); if ( processInstanceKey === undefined || elementId === undefined || elementInstanceKey === undefined || userTaskKey === undefined ) { continue; } accept({ elementInstanceKey, processInstanceKey, elementId, waitStateType: "USER_TASK", userTaskKey, }); } return out; } /** Whether a WASM-snapshot user-task `state` denotes an OPEN (answerable) task — the only * state a `USER_TASK` wait-state park exists in. The snapshot spells the state in PascalCase * (`"Created"`) and RETAINS completed/cancelled tasks (`"Completed"`/`"Canceled"`), so a park * must be gated on the open state rather than the row's presence; the check is * case-insensitive so a spelling drift does not silently re-open a closed task. Mirrors the * canonical lifecycle (added on user-task CREATED, removed on COMPLETED/CANCELED). */ function isOpenUserTaskState(state: unknown): boolean { return typeof state === "string" && state.trim().toUpperCase() === "CREATED"; } /** Resolve the `{ elementInstanceKey, processInstanceKey, elementId }` identity of a park row * (job/message/timer/signal) whose `instanceKey`/`elementId` join to an active element, or * `undefined` when the row is incomplete or its element is no longer active. */ function waitStateIdentity( row: Record, keyIndex: Map, ): { elementInstanceKey: string; processInstanceKey: string; elementId: string } | undefined { const processInstanceKey = presentKey(row.instanceKey); const elementId = presentString(row.elementId); if (processInstanceKey === undefined || elementId === undefined) return undefined; const elementInstanceKey = keyIndex.get(`${processInstanceKey}\u0000${elementId}`); if (elementInstanceKey === undefined) return undefined; return { elementInstanceKey, processInstanceKey, elementId }; } function str(v: unknown): string { return v == null ? "" : String(v); } /** A trimmed identifier, or `undefined` when blank/whitespace-only — the shared presence rule * the urban form contract (`resolveFormIdentifier`/`pickFormLinkage`) applies, so a blank * `formKey`/`externalFormReference` is treated as absent rather than a present-but-empty value. * Re-declared locally (not imported) because the kit depends only on urban's long-published * public API — see the re-declaration note at the top of this file. */ function present(v: string | undefined): string | undefined { const t = v?.trim(); return t ? t : undefined; } /** Presence-check a possibly-numeric form key under the shared trim rule — mirrors the urban * form contract's `presentKey` (`packages/urban/src/runtime/core/form-contract.ts`). The * read-model body may carry a `formKey` as a number, so coerce a *number* to a string; a string * is taken as-is; **any other type is absent**. It deliberately never `String(...)`-coerces an * arbitrary value, so a non-string (e.g. an object) can't leak in as a truthy `"[object Object]"` * identifier. Re-declared locally (not imported) for the same long-published-API-floor reason as * `present`/`str` above. Exported for the coercion-defect-class guard in the test suite. */ export function presentKey(v: unknown): string | undefined { if (typeof v === "number") return present(String(v)); if (typeof v === "string") return present(v); return undefined; } /** Presence-check a string-only identifier (`externalFormReference`/`formId`) under the shared * trim rule — mirrors the urban form contract, which treats these as present only when a string. * A number or any other type is absent, and it never `String(...)`-coerces, so a non-string can't * leak in as a garbage identifier. Exported for the coercion-defect-class guard in the tests. */ export function presentString(v: unknown): string | undefined { return typeof v === "string" ? present(v) : undefined; } /** The non-empty, trimmed string form of a *required* engine key, or throws when the value is * absent/blank (including a whitespace-only or padded string). A mutating seam op addresses a * single entity by key, so a blank key is a caller bug that must fail fast with an actionable * message rather than reach the WASM engine as a garbage identifier and surface an opaque error — * matching `SdkEngineClient`'s `requireEngineKey` so the two adapters stay behaviour-identical (No * Drift Surfaces). Reuses {@link presentKey}'s trim-and-coerce rule so a padded key like `" 5 "` * is normalized identically to how read paths resolve it. Exported for the guard in the tests. */ export function requireKey(value: unknown, name: string): string { const key = presentKey(value); if (key === undefined) { throw new Error(`${name} must be a non-empty key`); } return key; } /** Parse a form-js schema from a read-model `FormResult.schema`: the engine serializes it as a * JSON string (the REST wire shape), but tolerate an already-parsed object too. Returns `null` * when the value is neither, mirroring urban's `parseFormSchema`. */ function parseFormSchema(raw: unknown): Record | null { if (isRecord(raw)) return raw; if (typeof raw !== "string") return null; try { const parsed: unknown = JSON.parse(raw); return isRecord(parsed) ? parsed : null; } catch { return null; } } /** Build the typed `getForm` result from an untyped read-model body (the `getFormByKey` JSON * boundary). The `FormResult` DTO annotation on a `JSON.parse`d body is a shape *claim*, not a * runtime guarantee, so guard the body with `isRecord` — which, unlike `typeof === "object"`, * also excludes arrays — before reading its fields. A missing/invalid schema is treated as "no * such form" (`null`). Identifier presence is type-aware (mirrors the shared form contract's * `buildFormSchema`): a `formKey` counts only when a string/number, a `formId` only when a string, * so a non-string value can never coerce into a garbage `"[object Object]"` identifier. The single * source of truth `getForm` extracts through, so the boundary's tolerance cannot drift. */ export function parseForm( body: FormResult | null, ): { formKey?: string; formId?: string; version?: number; schema: Record } | null { if (!isRecord(body)) return null; const schema = parseFormSchema(body.schema); if (!schema) return null; const formKey = presentKey(body.formKey); const formId = presentString(body.formId); return { ...(formKey ? { formKey } : {}), ...(formId ? { formId } : {}), ...(typeof body.version === "number" ? { version: body.version } : {}), schema, }; }