feat(sdk): TypeScript SDK (@hanzoai/tasks) — durable execution over ZAP (#7)

Native TS client + worker over the Hanzo Tasks ZAP wire, byte-compatible with pkg/sdk (Go) and luxfi/zap v0.2.0. Drop-in for @temporalio/* + nestjs-temporal-core so the social-orchestrator migration is a dependency swap.

Includes: byte-exact ZAP message codec + TCP node (handshake, reqID correlation, Send-push dispatch); full client (start/signal/signalWithStart/query/describe/cancel/terminate/list, namespaces, health); activity worker (subscribe/dispatch/heartbeat/respond); event-sourced workflow replay decider (proxyActivities, durable activity-backed timers, condition, defineSignal/setHandler); temporal failure envelope serde; @temporalio compat (Connection/Client/NativeConnection/Worker/ApplicationFailure).

Tests: 25 vitest specs green (byte-layout golden, decider unit, full durable loop through a faithful ZAP mock) + a live integration suite that passes against a running tasksd. tsc --noEmit clean.

Co-authored-by: Hanzo Dev <[email protected]>
This commit is contained in:
2026-07-04 13:39:01 -07:00
committed by GitHub
co-authored by Hanzo Dev
parent a9ac37abb9
commit 25f19a5a30
34 changed files with 6637 additions and 0 deletions
+4
View File
@@ -0,0 +1,4 @@
node_modules
dist
*.tsbuildinfo
coverage
+148
View File
@@ -0,0 +1,148 @@
# @hanzoai/tasks — TypeScript SDK
Durable workflow execution for [Hanzo Tasks](https://github.com/hanzoai/tasks)
over the native **ZAP** transport. Zero gRPC, zero `go.temporal.io`. Built as a
drop-in for `@temporalio/*` + `nestjs-temporal-core`, so a NestJS/Postiz app
(e.g. `social-orchestrator`) migrates off upstream Temporal as a dependency
swap rather than a rewrite.
The wire is byte-compatible with `pkg/sdk` (the Go SDK) and `luxfi/zap`
v0.2.0 — a `tasksd` server cannot tell a Go client from a TypeScript one.
## Install
```bash
npm add @hanzoai/tasks
```
## Client
```ts
import { TasksClient } from "@hanzoai/tasks";
const client = await TasksClient.connect({ address: "tasks:9999", namespace: "default" });
const handle = await client.startWorkflow("PublishPost", {
taskQueue: "social",
workflowId: "post-123",
args: [{ postId: "123" }],
memo: { org: "acme" },
});
await handle.signal("approve", { by: "editor" });
const state = await handle.query("getState");
await handle.result(); // blocks until terminal
await client.close();
```
`@temporalio/client`-shaped shim (`Connection` + `Client.workflow.*`):
```ts
import { Connection, Client } from "@hanzoai/tasks";
const connection = await Connection.connect({ address: "tasks:9999" });
const client = await Client.create({ connection, namespace: "default" });
await client.workflow.start("PublishPost", { taskQueue: "social", workflowId: "post-123", args: [{ postId: "123" }] });
```
## Worker
```ts
import { Worker } from "@hanzoai/tasks";
import * as activities from "./activities";
const worker = await Worker.create({
address: "tasks:9999",
namespace: "default",
taskQueue: "social",
activities,
workflows: { PublishPost, ScheduleThread },
});
await worker.run(); // blocks until shutdown()
```
## Workflow primitives
Import inside workflow functions — the API mirrors `@temporalio/workflow`:
```ts
import {
proxyActivities, sleep, condition, defineSignal, defineQuery,
setHandler, workflowInfo, getVersion,
} from "@hanzoai/tasks";
const { publish } = proxyActivities<typeof activities>({ startToCloseTimeout: "1 minute" });
const approve = defineSignal<[{ by: string }]>("approve");
export async function PublishPost(input: { postId: string }) {
let approved = false;
setHandler(approve, () => { approved = true; });
await condition(() => approved);
await sleep("5s");
return publish(input.postId);
}
```
### Execution model
Workflows are **event-sourced replay deciders** (identical to the Go SDK and
Temporal): on each workflow task the server delivers the run's full history and
the worker replays the function from the top. Durable primitives are assigned a
deterministic per-run sequence number in program order; if the outcome is
already in history it resolves, otherwise a command is emitted and the async
function suspends until the next task. **Workflow code must be deterministic** —
no `Date.now()`, `Math.random()`, `setTimeout`, or direct I/O; use activities
and the primitives above.
## Errors
```ts
import { ApplicationFailure, TaskError } from "@hanzoai/tasks";
throw ApplicationFailure.nonRetryable("bad input", "ValidationError");
```
Failures serialise to the exact `{ v, p }` envelope of `pkg/sdk/temporal`.
## Status — complete vs remaining
This is **PR #1 (CORE)**. What is fully wired and tested against the real ZAP
transport:
| Primitive | Status |
|---|---|
| Client: start / signal / signalWithStart / query / describe / cancel / terminate / list | ✅ complete |
| Client: namespace register/describe/list, health | ✅ complete |
| Worker: subscribe, activity dispatch + heartbeat, respond | ✅ complete |
| Workflow decider: `proxyActivities` (durable activities) | ✅ complete |
| Durable `sleep` / `startTimer` (activity-backed) | ✅ complete¹ |
| `condition` (parks until predicate holds) | ✅ complete |
| `defineSignal` / `setHandler` (replayed from history) | ✅ complete² |
| Retry policy, activity failure propagation, cancellation signal | ✅ complete |
| `@temporalio/*` shim: `Connection`, `Client`, `NativeConnection`, `Worker`, `ApplicationFailure` | ✅ complete |
Remaining (blocked on **server** wire support, not this SDK):
- **Durable timers as a native command.** `sleep`/`startTimer` are backed by an
internal `__hanzo_timer__` sleeper activity — durable and deterministic, but a
worker crash mid-sleep re-runs the delay from zero (fires late, never lost).
A native `TIMER_STARTED`/`TIMER_FIRED` server command removes the caveat. ¹
- **Signal-to-running-workflow delivery.** The API + history-replay path are
implemented; end-to-end delivery depends on the server appending
`WORKFLOW_EXECUTION_SIGNALED` to replayable history and re-dispatching a
workflow task (the Go SDK has the same gap today). ²
- **`continueAsNew`** — API present; throws `ContinueAsNewNotSupported` at
runtime (no server command yet).
- **Typed search attributes** — accepted on `StartWorkflowOptions` for API
compatibility but ignored on the v1 wire (`StartWorkflowRequest` has no
search-attribute field; only `memo` is carried).
- **`WorkflowHandle.result()` value** — resolves `undefined` on completion
because `DescribeWorkflowResponse` carries no result field on the v1 wire
(same gap as the Go SDK). It throws on failure/cancel/terminate.
- **Worker-side query handlers** — `defineQuery`/`setHandler(query)` register,
but answering a delivered query requires history in the query delivery
(server gap); the client-side `query()` RPC works.
## License
MIT.
+2770
View File
File diff suppressed because it is too large Load Diff
+50
View File
@@ -0,0 +1,50 @@
{
"name": "@hanzoai/tasks",
"version": "0.1.0",
"description": "TypeScript SDK for Hanzo Tasks — durable workflow execution over the native ZAP transport. Drop-in replacement for @temporalio/* + nestjs-temporal-core against a Hanzo Tasks (tasksd) server.",
"license": "MIT",
"author": "Hanzo AI",
"repository": {
"type": "git",
"url": "https://github.com/hanzoai/tasks.git",
"directory": "sdk/typescript"
},
"type": "commonjs",
"main": "./dist/index.cjs",
"module": "./dist/index.js",
"types": "./dist/index.d.ts",
"exports": {
".": {
"types": "./dist/index.d.ts",
"import": "./dist/index.js",
"require": "./dist/index.cjs"
}
},
"files": [
"dist",
"README.md"
],
"engines": {
"node": ">=18"
},
"scripts": {
"build": "tsup",
"typecheck": "tsc --noEmit",
"test": "vitest run",
"test:watch": "vitest"
},
"keywords": [
"hanzo",
"tasks",
"temporal",
"durable-execution",
"workflow",
"zap"
],
"devDependencies": {
"@types/node": "^20.14.0",
"tsup": "^8.0.0",
"typescript": "^5.4.0",
"vitest": "^1.6.0"
}
}
+305
View File
@@ -0,0 +1,305 @@
// Copyright © 2026 Hanzo AI. MIT License.
//
// TasksClient — the workflow client. JSON-over-ZAP round trips against the
// Hanzo Tasks frontend, wire-identical to pkg/sdk/client (Go). Every request
// body matches the corresponding *Request struct's JSON tags in
// pkg/sdk/client/{workflow,namespace}.go.
import { ZapNode, type Transport } from "../zap/node";
import { Opcode } from "../zap/opcodes";
import { decodeEnvelope } from "../zap/envelope";
import {
WorkflowStatus,
type ConnectionOptions,
type Namespace,
type RegisterNamespaceOptions,
type RetryPolicy,
type StartWorkflowOptions,
type WorkflowExecutionInfo,
} from "./options";
import { WorkflowHandle } from "./handle";
type WorkflowTypeArg = string | { name?: string };
function resolveType(workflow: WorkflowTypeArg): string {
if (typeof workflow === "string") {
if (!workflow) throw new Error("hanzo/tasks: workflow type name is empty");
return workflow;
}
const name = workflow?.name;
if (!name) throw new Error("hanzo/tasks: cannot resolve workflow type from an anonymous function");
return name;
}
function retryToWire(rp?: RetryPolicy): Record<string, unknown> | undefined {
if (!rp) return undefined;
return {
initial_interval_ms: rp.initialIntervalMs,
backoff_coefficient: rp.backoffCoefficient,
maximum_interval_ms: rp.maximumIntervalMs,
maximum_attempts: rp.maximumAttempts,
non_retryable_error_types: rp.nonRetryableErrorTypes,
};
}
function timeoutsToWire(o: StartWorkflowOptions): Record<string, unknown> {
return {
workflow_execution_ms: o.workflowExecutionTimeoutMs,
workflow_run_ms: o.workflowRunTimeoutMs,
workflow_task_ms: o.workflowTaskTimeoutMs,
};
}
interface RawWorkflowInfo {
workflow_id: string;
run_id: string;
workflow_type: string;
start_time?: string;
close_time?: string;
status: number;
history_length: number;
task_queue?: string;
memo?: Record<string, unknown>;
}
function toInfo(r: RawWorkflowInfo): WorkflowExecutionInfo {
return {
workflowId: r.workflow_id,
runId: r.run_id,
workflowType: r.workflow_type,
startTime: r.start_time,
closeTime: r.close_time,
status: (r.status ?? 0) as WorkflowStatus,
historyLength: r.history_length ?? 0,
taskQueue: r.task_queue,
memo: r.memo,
};
}
export interface HealthStatus {
service: string;
status: string;
}
export interface ListWorkflowsResult {
executions: WorkflowExecutionInfo[];
nextPageToken?: string;
}
export class TasksClient {
private constructor(
private readonly transport: Transport,
public readonly namespace: string,
public readonly identity: string,
private readonly owned: boolean,
) {}
/** Connect to a Hanzo Tasks frontend (or wrap an injected Transport). */
static async connect(opts: ConnectionOptions): Promise<TasksClient> {
const namespace = opts.namespace && opts.namespace.length > 0 ? opts.namespace : "default";
const identity = opts.identity && opts.identity.length > 0 ? opts.identity : "hanzo-tasks-ts-sdk";
if (opts.transport) {
return new TasksClient(opts.transport, namespace, identity, false);
}
const [host, portStr] = splitHostPort(opts.address);
const node = new ZapNode({
host,
port: portStr,
nodeId: identity,
dialTimeoutMs: opts.dialTimeoutMs,
callTimeoutMs: opts.callTimeoutMs,
});
await node.connect();
return new TasksClient(node, namespace, identity, true);
}
/** The underlying transport (shared connections; used by the worker). */
get rawTransport(): Transport {
return this.transport;
}
private async roundTrip<R = unknown>(opcode: number, req: unknown): Promise<R | undefined> {
const body = Buffer.from(JSON.stringify(req ?? {}), "utf8");
const respFrame = await this.transport.call(opcode, body);
const { status, detail, body: payload } = decodeEnvelope(respFrame);
if (status !== 0 && status !== 200) {
throw new Error(`hanzo/tasks: server status ${status}: ${detail}`);
}
if (payload.length === 0) return undefined;
return JSON.parse(payload.toString("utf8")) as R;
}
// ── workflow lifecycle ──
async startWorkflow(workflow: WorkflowTypeArg, opts: StartWorkflowOptions): Promise<WorkflowHandle> {
if (!opts.taskQueue) throw new Error("hanzo/tasks: StartWorkflowOptions.taskQueue is required");
const req = {
namespace: this.namespace,
workflow_id: opts.workflowId ?? "",
workflow_type: resolveType(workflow),
task_queue: opts.taskQueue,
input: opts.args ?? [],
retry_policy: retryToWire(opts.retryPolicy),
timeouts: timeoutsToWire(opts),
memo: opts.memo,
cron_schedule: opts.cronSchedule,
identity: this.identity,
};
const resp = await this.roundTrip<{ run_id: string }>(Opcode.StartWorkflow, req);
return new WorkflowHandle(this, opts.workflowId ?? "", resp?.run_id ?? "");
}
async signalWithStart(
workflow: WorkflowTypeArg,
signalName: string,
signalArg: unknown,
opts: StartWorkflowOptions,
): Promise<WorkflowHandle> {
if (!opts.workflowId) throw new Error("hanzo/tasks: workflowId is required for signalWithStart");
if (!signalName) throw new Error("hanzo/tasks: signalName is required");
if (!opts.taskQueue) throw new Error("hanzo/tasks: taskQueue is required");
const req = {
namespace: this.namespace,
workflow_id: opts.workflowId,
workflow_type: resolveType(workflow),
task_queue: opts.taskQueue,
input: opts.args ?? [],
retry_policy: retryToWire(opts.retryPolicy),
timeouts: timeoutsToWire(opts),
memo: opts.memo,
cron_schedule: opts.cronSchedule,
identity: this.identity,
signal_name: signalName,
signal_input: signalArg,
};
const resp = await this.roundTrip<{ run_id: string }>(Opcode.SignalWithStartWorkflow, req);
return new WorkflowHandle(this, opts.workflowId, resp?.run_id ?? "");
}
async signalWorkflow(workflowId: string, runId: string, signalName: string, arg?: unknown): Promise<void> {
if (!workflowId) throw new Error("hanzo/tasks: workflowId is required");
if (!signalName) throw new Error("hanzo/tasks: signalName is required");
await this.roundTrip(Opcode.SignalWorkflow, {
namespace: this.namespace,
workflow_id: workflowId,
run_id: runId || undefined,
signal_name: signalName,
input: arg,
});
}
async queryWorkflow<R = unknown>(
workflowId: string,
runId: string,
queryType: string,
args: unknown[] = [],
): Promise<R> {
if (!workflowId) throw new Error("hanzo/tasks: workflowId is required");
if (!queryType) throw new Error("hanzo/tasks: queryType is required");
const resp = await this.roundTrip<{ result?: string }>(Opcode.QueryWorkflow, {
namespace: this.namespace,
workflow_id: workflowId,
run_id: runId || undefined,
query_type: queryType,
query_args: args,
});
// result is a Go []byte → base64 on the wire → raw JSON of the value.
if (!resp?.result) return undefined as R;
const raw = Buffer.from(resp.result, "base64").toString("utf8").trim();
return (raw.length === 0 ? undefined : JSON.parse(raw)) as R;
}
async cancelWorkflow(workflowId: string, runId = ""): Promise<void> {
await this.roundTrip(Opcode.CancelWorkflow, {
namespace: this.namespace,
workflow_id: workflowId,
run_id: runId || undefined,
});
}
async terminateWorkflow(workflowId: string, runId = "", reason = ""): Promise<void> {
await this.roundTrip(Opcode.TerminateWorkflow, {
namespace: this.namespace,
workflow_id: workflowId,
run_id: runId || undefined,
reason: reason || undefined,
});
}
async describeWorkflow(workflowId: string, runId = ""): Promise<WorkflowExecutionInfo> {
const resp = await this.roundTrip<{ info: RawWorkflowInfo }>(Opcode.DescribeWorkflow, {
namespace: this.namespace,
workflow_id: workflowId,
run_id: runId || undefined,
});
if (!resp?.info) throw new Error(`hanzo/tasks: describe returned no info for ${workflowId}`);
return toInfo(resp.info);
}
async listWorkflows(query = "", pageSize = 0, nextPageToken?: string): Promise<ListWorkflowsResult> {
const resp = await this.roundTrip<{ executions?: RawWorkflowInfo[]; next_page_token?: string }>(
Opcode.ListWorkflows,
{
namespace: this.namespace,
query: query || undefined,
page_size: pageSize || undefined,
next_page_token: nextPageToken,
},
);
return {
executions: (resp?.executions ?? []).map(toInfo),
nextPageToken: resp?.next_page_token,
};
}
getHandle<T = unknown>(workflowId: string, runId = ""): WorkflowHandle<T> {
return new WorkflowHandle<T>(this, workflowId, runId);
}
// ── namespace ops ──
async registerNamespace(opts: RegisterNamespaceOptions): Promise<void> {
if (!opts.name) throw new Error("hanzo/tasks: namespace name is required");
await this.roundTrip(Opcode.RegisterNamespace, {
name: opts.name,
description: opts.description,
owner_email: opts.ownerEmail,
retention_ms: opts.retentionMs,
});
}
async describeNamespace(name: string): Promise<Namespace> {
const resp = await this.roundTrip<{ namespace: Namespace }>(Opcode.DescribeNamespace, { name });
if (!resp?.namespace) throw new Error(`hanzo/tasks: namespace ${name} not found`);
return resp.namespace;
}
async listNamespaces(pageSize = 0, nextPageToken?: string): Promise<{ namespaces: Namespace[]; nextPageToken?: string }> {
const resp = await this.roundTrip<{ namespaces?: Namespace[]; next_page_token?: string }>(Opcode.ListNamespaces, {
page_size: pageSize || undefined,
next_page_token: nextPageToken,
});
return { namespaces: resp?.namespaces ?? [], nextPageToken: resp?.next_page_token };
}
// ── health ──
async health(): Promise<HealthStatus> {
const resp = await this.roundTrip<HealthStatus>(Opcode.Health, {});
return { service: resp?.service ?? "", status: resp?.status ?? "" };
}
/** Close the underlying connection (only if this client owns it). */
async close(): Promise<void> {
if (this.owned) await this.transport.close();
}
}
function splitHostPort(addr: string): [string, number] {
const idx = addr.lastIndexOf(":");
if (idx < 0) throw new Error(`hanzo/tasks: address must be host:port, got ${addr}`);
const host = addr.slice(0, idx) || "127.0.0.1";
const port = Number(addr.slice(idx + 1));
if (!Number.isFinite(port) || port <= 0) throw new Error(`hanzo/tasks: invalid port in ${addr}`);
return [host, port];
}
+85
View File
@@ -0,0 +1,85 @@
// Copyright © 2026 Hanzo AI. MIT License.
import { WorkflowStatus, type WorkflowExecutionInfo } from "./options";
import { TaskError, FailureCode } from "../common/failure";
import type { TasksClient } from "./client";
export interface WorkflowResultOptions {
/** Poll cadence bounds; defaults 250ms → 5s. */
initialIntervalMs?: number;
maxIntervalMs?: number;
/** When true, stop at a ContinuedAsNew terminal instead of following it. */
followRuns?: boolean;
}
/**
* WorkflowHandle mirrors @temporalio/client's WorkflowHandle for the RPCs the
* v1 wire supports. Note: DescribeWorkflow does not yet return a workflow
* result on the v1 wire, so `result()` blocks until terminal and resolves
* `undefined` on success (throwing on failure/cancel/terminate).
*/
export class WorkflowHandle<T = unknown> {
constructor(
private readonly client: TasksClient,
public readonly workflowId: string,
public readonly runId: string = "",
) {}
/** Post a signal to this execution. */
signal(signalName: string, arg?: unknown): Promise<void> {
return this.client.signalWorkflow(this.workflowId, this.runId, signalName, arg);
}
/** Run a registered query and decode its result. */
query<R = unknown>(queryType: string, ...args: unknown[]): Promise<R> {
return this.client.queryWorkflow<R>(this.workflowId, this.runId, queryType, args);
}
cancel(): Promise<void> {
return this.client.cancelWorkflow(this.workflowId, this.runId);
}
terminate(reason = ""): Promise<void> {
return this.client.terminateWorkflow(this.workflowId, this.runId, reason);
}
describe(): Promise<WorkflowExecutionInfo> {
return this.client.describeWorkflow(this.workflowId, this.runId);
}
/**
* Block until the workflow reaches a terminal state. Resolves `undefined`
* on completion (the v1 wire does not surface the result value yet);
* throws a TaskError on failure/cancel/terminate/timeout.
*/
async result(opts: WorkflowResultOptions = {}): Promise<T | undefined> {
let backoff = opts.initialIntervalMs ?? 250;
const maxBackoff = opts.maxIntervalMs ?? 5000;
let runId = this.runId;
for (;;) {
const info = await this.client.describeWorkflow(this.workflowId, runId);
switch (info.status) {
case WorkflowStatus.Completed:
return undefined;
case WorkflowStatus.Failed:
throw new TaskError(`workflow failed: ${this.workflowId}`, FailureCode.Application, true);
case WorkflowStatus.Canceled:
throw new TaskError(`workflow canceled: ${this.workflowId}`, FailureCode.Canceled, true);
case WorkflowStatus.Terminated:
throw new TaskError(`workflow terminated: ${this.workflowId}`, FailureCode.Application, true);
case WorkflowStatus.TimedOut:
throw new TaskError(`workflow timed out: ${this.workflowId}`, FailureCode.Timeout, true);
case WorkflowStatus.ContinuedAsNew:
if (opts.followRuns === false) {
throw new TaskError(`workflow continued as new: ${this.workflowId}`, FailureCode.Application, true);
}
runId = "";
break;
default:
break;
}
await new Promise((r) => setTimeout(r, backoff));
if (backoff < maxBackoff) backoff = Math.min(backoff * 2, maxBackoff);
}
}
}
+5
View File
@@ -0,0 +1,5 @@
// Copyright © 2026 Hanzo AI. MIT License.
export * from "./options";
export * from "./handle";
export * from "./client";
+97
View File
@@ -0,0 +1,97 @@
// Copyright © 2026 Hanzo AI. MIT License.
import type { Transport } from "../zap/node";
/** ConnectionOptions configures a client/worker connection to tasksd. */
export interface ConnectionOptions {
/** "host:port" of the Hanzo Tasks ZAP frontend (default port 9999). */
address: string;
/** Namespace scoping every RPC. Empty is rewritten to "default". */
namespace?: string;
/** Caller identity sent on worker RPCs. Defaults to "hanzo-tasks-ts-sdk". */
identity?: string;
dialTimeoutMs?: number;
callTimeoutMs?: number;
/** Inject a Transport (tests / embedding). When set, `address` is ignored. */
transport?: Transport;
}
export interface RetryPolicy {
initialIntervalMs?: number;
backoffCoefficient?: number;
maximumIntervalMs?: number;
maximumAttempts?: number;
nonRetryableErrorTypes?: string[];
}
export interface StartWorkflowOptions {
/** Workflow ID. Empty asks the server to generate one. */
workflowId?: string;
/** Required — workers poll this queue. */
taskQueue: string;
/** Positional workflow arguments. */
args?: unknown[];
workflowExecutionTimeoutMs?: number;
workflowRunTimeoutMs?: number;
workflowTaskTimeoutMs?: number;
retryPolicy?: RetryPolicy;
/** Cron expression; empty runs once. */
cronSchedule?: string;
/** Opaque metadata attached to the execution. */
memo?: Record<string, unknown>;
/**
* Search attributes. Accepted for @temporalio API compatibility, but the
* v1 Hanzo Tasks wire (StartWorkflowRequest) does not yet carry a search-
* attribute field — see the SDK README "Remaining" section. Values are
* currently ignored on the wire.
*/
searchAttributes?: Record<string, unknown>;
}
export enum WorkflowStatus {
Unspecified = 0,
Running = 1,
Completed = 2,
Failed = 3,
Canceled = 4,
Terminated = 5,
ContinuedAsNew = 6,
TimedOut = 7,
}
export interface WorkflowExecutionInfo {
workflowId: string;
runId: string;
workflowType: string;
startTime?: string;
closeTime?: string;
status: WorkflowStatus;
historyLength: number;
taskQueue?: string;
memo?: Record<string, unknown>;
}
export enum NamespaceState {
Unspecified = 0,
Registered = 1,
Deprecated = 2,
Deleted = 3,
}
export interface Namespace {
info: {
name: string;
state: NamespaceState;
description?: string;
owner_email?: string;
id?: string;
};
config: { retention_ms?: number };
}
export interface RegisterNamespaceOptions {
name: string;
description?: string;
ownerEmail?: string;
retentionMs?: number;
}
+33
View File
@@ -0,0 +1,33 @@
// Copyright © 2026 Hanzo AI. MIT License.
//
// Payload conversion. The v1 Hanzo Tasks wire encodes activity/workflow
// arguments and results as JSON. Arguments always travel as a JSON array
// (json.Marshal(args) in the Go SDK); a single result travels as the bare
// JSON of the returned value.
/** JSON-encode an argument list into the array form the server records. */
export function encodeArgs(args: unknown[]): Buffer {
return Buffer.from(JSON.stringify(args ?? []), "utf8");
}
/** Decode a JSON args array. Empty/whitespace decodes to []. */
export function decodeArgs(bytes: Buffer | Uint8Array | null | undefined): unknown[] {
if (!bytes || bytes.length === 0) return [];
const s = Buffer.from(bytes).toString("utf8").trim();
if (s.length === 0) return [];
const v = JSON.parse(s);
return Array.isArray(v) ? v : [v];
}
/** JSON-encode a single value (activity/workflow result). */
export function encodeValue(v: unknown): Buffer {
return Buffer.from(JSON.stringify(v ?? null), "utf8");
}
/** Decode a single JSON value. Empty decodes to undefined. */
export function decodeValue<T = unknown>(bytes: Buffer | Uint8Array | null | undefined): T | undefined {
if (!bytes || bytes.length === 0) return undefined;
const s = Buffer.from(bytes).toString("utf8").trim();
if (s.length === 0) return undefined;
return JSON.parse(s) as T;
}
+47
View File
@@ -0,0 +1,47 @@
// Copyright © 2026 Hanzo AI. MIT License.
//
// Duration parsing. Accepts a number of milliseconds, or a human string in
// the `ms`-package style used by @temporalio ("1 minute", "30s", "2 hours",
// "500ms"). Returns milliseconds.
export type Duration = number | string;
const UNIT_MS: Record<string, number> = {
ms: 1,
msec: 1,
msecs: 1,
millisecond: 1,
milliseconds: 1,
s: 1000,
sec: 1000,
secs: 1000,
second: 1000,
seconds: 1000,
m: 60_000,
min: 60_000,
mins: 60_000,
minute: 60_000,
minutes: 60_000,
h: 3_600_000,
hr: 3_600_000,
hrs: 3_600_000,
hour: 3_600_000,
hours: 3_600_000,
d: 86_400_000,
day: 86_400_000,
days: 86_400_000,
};
/** Convert a Duration to milliseconds. Undefined → 0. */
export function toMs(d: Duration | undefined): number {
if (d === undefined || d === null) return 0;
if (typeof d === "number") return Math.round(d);
const s = d.trim();
const m = /^(-?\d*\.?\d+)\s*([a-zA-Z]+)?$/.exec(s);
if (!m) throw new Error(`hanzo/tasks: invalid duration "${d}"`);
const value = parseFloat(m[1]);
const unit = (m[2] ?? "ms").toLowerCase();
const mult = UNIT_MS[unit];
if (mult === undefined) throw new Error(`hanzo/tasks: unknown duration unit "${unit}" in "${d}"`);
return Math.round(value * mult);
}
+126
View File
@@ -0,0 +1,126 @@
// Copyright © 2026 Hanzo AI. MIT License.
//
// Failure serde — byte-compatible with pkg/sdk/temporal/serde.go. The wire
// shape carried in the `failure` bytes of RespondActivityTaskFailed and the
// ACTIVITY_TASK_FAILED history attribute is:
//
// { "v": 1, "p": { message, code, nonRetryable?, details?, cause? } }
//
// A version other than 1 decodes to a non-retryable DecodeError. Encoding
// never throws — a workflow/activity must always be able to report failure.
export const FailureCode = {
Application: "ApplicationError",
Canceled: "CanceledError",
Timeout: "TimeoutError",
Decode: "DecodeError",
} as const;
const ENVELOPE_VERSION = 1;
/** TaskError is the single concrete error type this SDK emits and decodes. */
export class TaskError extends Error {
readonly code: string;
readonly nonRetryable: boolean;
readonly details: unknown[];
readonly cause?: TaskError;
constructor(
message: string,
code: string = FailureCode.Application,
nonRetryable = false,
details: unknown[] = [],
cause?: TaskError,
) {
super(message);
this.name = "TaskError";
this.code = code || FailureCode.Application;
this.nonRetryable = nonRetryable;
this.details = details;
this.cause = cause;
}
/** True when this is (or wraps) a cancellation. */
get isCanceled(): boolean {
return this.code === FailureCode.Canceled;
}
/** True when this is (or wraps) a timeout. */
get isTimeout(): boolean {
return this.code === FailureCode.Timeout;
}
}
interface FailurePayload {
message: string;
code: string;
nonRetryable?: boolean;
details?: unknown[];
cause?: FailurePayload;
}
interface FailureEnvelope {
v: number;
p: FailurePayload;
}
function toPayload(err: unknown): FailurePayload {
if (err instanceof TaskError) {
const p: FailurePayload = {
message: err.message,
code: err.code || FailureCode.Application,
};
if (err.nonRetryable) p.nonRetryable = true;
if (err.details && err.details.length > 0) p.details = err.details;
if (err.cause) p.cause = toPayload(err.cause);
return p;
}
if (err instanceof Error) {
return { message: err.message, code: FailureCode.Application };
}
return { message: String(err), code: FailureCode.Application };
}
function fromPayload(p: FailurePayload | undefined): TaskError | undefined {
if (!p) return undefined;
return new TaskError(
p.message,
p.code || FailureCode.Application,
p.nonRetryable ?? false,
Array.isArray(p.details) ? p.details : [],
fromPayload(p.cause),
);
}
/** Serialise any error into the on-wire failure bytes. Never throws. */
export function encodeFailure(err: unknown): Buffer {
let env: FailureEnvelope;
try {
env = { v: ENVELOPE_VERSION, p: toPayload(err) };
return Buffer.from(JSON.stringify(env), "utf8");
} catch {
env = {
v: ENVELOPE_VERSION,
p: { message: "failed to encode failure", code: FailureCode.Decode, nonRetryable: true },
};
return Buffer.from(JSON.stringify(env), "utf8");
}
}
/** Parse failure bytes back into a TaskError. Fail-secure on hostile input. */
export function decodeFailure(bytes: Buffer | Uint8Array | null | undefined): TaskError {
if (!bytes || bytes.length === 0) {
return new TaskError("empty failure payload", FailureCode.Decode, true);
}
let env: FailureEnvelope;
try {
env = JSON.parse(Buffer.from(bytes).toString("utf8"));
} catch (e) {
return new TaskError(`failure envelope unmarshal: ${(e as Error).message}`, FailureCode.Decode, true);
}
if (env.v !== ENVELOPE_VERSION) {
return new TaskError(`unsupported failure envelope version ${env.v}`, FailureCode.Decode, true);
}
const decoded = fromPayload(env.p);
return decoded ?? new TaskError("nil failure payload", FailureCode.Decode, true);
}
+183
View File
@@ -0,0 +1,183 @@
// Copyright © 2026 Hanzo AI. MIT License.
//
// @temporalio/* compatibility layer. These shims mirror the surface of
// @temporalio/client and @temporalio/worker that social-orchestrator (and
// nestjs-temporal-core) depend on, so migrating is a dependency swap rather
// than a rewrite. Semantics are the Hanzo Tasks event-sourced model.
//
// Migration map:
// @temporalio/client Connection, Client, WorkflowClient, WorkflowHandle
// @temporalio/worker NativeConnection, Worker
// @temporalio/common ApplicationFailure
// @temporalio/workflow proxyActivities, defineSignal, setHandler, condition,
// sleep, continueAsNew, workflowInfo (see ./workflow)
import { ZapNode, type Transport } from "./zap/node";
import { TasksClient } from "./client/client";
import { WorkflowHandle } from "./client/handle";
import { toMs, type Duration } from "./common/duration";
import { TaskError, FailureCode } from "./common/failure";
import type { RetryPolicy, StartWorkflowOptions } from "./client/options";
function splitHostPort(addr: string): [string, number] {
const idx = addr.lastIndexOf(":");
if (idx < 0) throw new Error(`hanzo/tasks: address must be host:port, got ${addr}`);
return [addr.slice(0, idx) || "127.0.0.1", Number(addr.slice(idx + 1))];
}
/** A connected ZAP transport shared by clients and workers. */
export class Connection {
protected constructor(
readonly transport: Transport,
readonly address: string,
) {}
static async connect(opts: { address: string; identity?: string }): Promise<Connection> {
const [host, port] = splitHostPort(opts.address);
const node = new ZapNode({ host, port, nodeId: opts.identity });
await node.connect();
return new Connection(node, opts.address);
}
async close(): Promise<void> {
await this.transport.close();
}
}
/** Worker-side connection. Same wire as Connection (single ZAP transport). */
export class NativeConnection extends Connection {}
interface CompatRetry {
initialInterval?: Duration;
backoffCoefficient?: number;
maximumInterval?: Duration;
maximumAttempts?: number;
nonRetryableErrorTypes?: string[];
}
export interface WorkflowStartOptions {
taskQueue: string;
workflowId?: string;
args?: unknown[];
workflowExecutionTimeout?: Duration;
workflowRunTimeout?: Duration;
workflowTaskTimeout?: Duration;
retry?: CompatRetry;
cronSchedule?: string;
memo?: Record<string, unknown>;
searchAttributes?: Record<string, unknown>;
}
function toRetryPolicy(r?: CompatRetry): RetryPolicy | undefined {
if (!r) return undefined;
return {
initialIntervalMs: r.initialInterval !== undefined ? toMs(r.initialInterval) : undefined,
backoffCoefficient: r.backoffCoefficient,
maximumIntervalMs: r.maximumInterval !== undefined ? toMs(r.maximumInterval) : undefined,
maximumAttempts: r.maximumAttempts,
nonRetryableErrorTypes: r.nonRetryableErrorTypes,
};
}
function toStartOptions(o: WorkflowStartOptions): StartWorkflowOptions {
return {
taskQueue: o.taskQueue,
workflowId: o.workflowId,
args: o.args,
workflowExecutionTimeoutMs: o.workflowExecutionTimeout !== undefined ? toMs(o.workflowExecutionTimeout) : undefined,
workflowRunTimeoutMs: o.workflowRunTimeout !== undefined ? toMs(o.workflowRunTimeout) : undefined,
workflowTaskTimeoutMs: o.workflowTaskTimeout !== undefined ? toMs(o.workflowTaskTimeout) : undefined,
retryPolicy: toRetryPolicy(o.retry),
cronSchedule: o.cronSchedule,
memo: o.memo,
searchAttributes: o.searchAttributes,
};
}
type WorkflowRef = string | { name?: string };
/** Mirrors @temporalio/client's WorkflowClient (Client.workflow). */
export class WorkflowClient {
constructor(private readonly client: TasksClient) {}
start(workflow: WorkflowRef, opts: WorkflowStartOptions): Promise<WorkflowHandle> {
return this.client.startWorkflow(workflow, toStartOptions(opts));
}
async execute<T = unknown>(workflow: WorkflowRef, opts: WorkflowStartOptions): Promise<T | undefined> {
const handle = await this.client.startWorkflow(workflow, toStartOptions(opts));
return handle.result() as Promise<T | undefined>;
}
getHandle<T = unknown>(workflowId: string, runId = ""): WorkflowHandle<T> {
return this.client.getHandle<T>(workflowId, runId);
}
signalWithStart(
workflow: WorkflowRef,
opts: WorkflowStartOptions & { signal: string; signalArgs?: unknown[] },
): Promise<WorkflowHandle> {
const signalArg = opts.signalArgs && opts.signalArgs.length > 0 ? opts.signalArgs[0] : undefined;
return this.client.signalWithStart(workflow, opts.signal, signalArg, toStartOptions(opts));
}
}
/** Mirrors @temporalio/client's Client. */
export class Client {
readonly workflow: WorkflowClient;
private readonly client: TasksClient;
private constructor(client: TasksClient) {
this.client = client;
this.workflow = new WorkflowClient(client);
}
static async create(opts: { connection: Connection; namespace?: string; identity?: string }): Promise<Client> {
const client = await TasksClient.connect({
address: opts.connection.address,
namespace: opts.namespace,
identity: opts.identity,
transport: opts.connection.transport,
});
return new Client(client);
}
/** Direct access to the underlying TasksClient. */
get connection(): TasksClient {
return this.client;
}
async close(): Promise<void> {
await this.client.close();
}
}
// Worker (with static create) is re-exported unchanged from ./worker — its
// WorkerOptions already accepts `connection`, `address`, or `transport`, so
// `Worker.create({ connection, namespace, taskQueue, workflows, activities })`
// works exactly like @temporalio/worker.
/** Mirrors @temporalio/common's ApplicationFailure. */
export class ApplicationFailure extends TaskError {
static create(opts: {
message: string;
type?: string;
nonRetryable?: boolean;
details?: unknown[];
}): ApplicationFailure {
return new ApplicationFailure(
opts.message,
opts.type ?? FailureCode.Application,
opts.nonRetryable ?? false,
opts.details ?? [],
);
}
static nonRetryable(message: string, type?: string, ...details: unknown[]): ApplicationFailure {
return new ApplicationFailure(message, type ?? FailureCode.Application, true, details);
}
static retryable(message: string, type?: string, ...details: unknown[]): ApplicationFailure {
return new ApplicationFailure(message, type ?? FailureCode.Application, false, details);
}
}
+27
View File
@@ -0,0 +1,27 @@
// Copyright © 2026 Hanzo AI. MIT License.
//
// @hanzoai/tasks — TypeScript SDK for Hanzo Tasks durable execution over the
// native ZAP transport. Drop-in for @temporalio/* + nestjs-temporal-core.
//
// Public surface:
// • Native client: TasksClient, WorkflowHandle
// • Worker: Worker (activities + workflow decider)
// • Workflow API: proxyActivities, sleep, condition, defineSignal,
// setHandler, workflowInfo, continueAsNew, ...
// • @temporalio shim: Connection, NativeConnection, Client, ApplicationFailure
// • Errors: TaskError, FailureCode
// • Low-level ZAP: exported under the `zap` namespace
export * from "./common/failure";
export * from "./common/converter";
export * from "./common/duration";
export * from "./client";
export * from "./worker";
export * from "./workflow";
export * from "./compat";
// Low-level ZAP wire (advanced / embedding / testing).
export * as zap from "./zap";
export type { Transport } from "./zap/node";
@@ -0,0 +1,68 @@
// Copyright © 2026 Hanzo AI. MIT License.
//
// Activity execution context — heartbeat, info, cancellation. Mirrors
// pkg/sdk/activity (Go) and @temporalio/activity's Context.
import { AsyncLocalStorage } from "node:async_hooks";
export interface ActivityInfo {
taskToken: Buffer;
workflowId: string;
runId: string;
activityId: string;
activityType: string;
taskQueue: string;
attempt: number;
scheduledTimeMs: number;
startedTimeMs: number;
}
export class ActivityContext {
readonly info: ActivityInfo;
private readonly heartbeatSink: (details: unknown[]) => void;
private readonly abort: AbortController;
constructor(info: ActivityInfo, heartbeatSink: (details: unknown[]) => void, abort: AbortController) {
this.info = info;
this.heartbeatSink = heartbeatSink;
this.abort = abort;
}
/** Record a heartbeat with optional details. */
heartbeat(...details: unknown[]): void {
this.heartbeatSink(details);
}
/** True once the server requested cancellation via a heartbeat response. */
get cancelled(): boolean {
return this.abort.signal.aborted;
}
/** AbortSignal that fires when the server requests cancellation. */
get cancellationSignal(): AbortSignal {
return this.abort.signal;
}
}
const storage = new AsyncLocalStorage<ActivityContext>();
/** The current activity's context. Throws outside an activity. */
export function activityContext(): ActivityContext {
const ctx = storage.getStore();
if (!ctx) throw new Error("hanzo/tasks: activityContext() called outside an activity");
return ctx;
}
/** Convenience: current activity info. */
export function activityInfo(): ActivityInfo {
return activityContext().info;
}
/** Record a heartbeat from the current activity. */
export function heartbeat(...details: unknown[]): void {
activityContext().heartbeat(...details);
}
export function runInActivityContext<T>(ctx: ActivityContext, fn: () => T): T {
return storage.run(ctx, fn);
}
+77
View File
@@ -0,0 +1,77 @@
// Copyright © 2026 Hanzo AI. MIT License.
//
// Workflow-task decider. One episode: seed a DeciderContext from history,
// replay the registered workflow function, drain microtasks, and translate
// the outcome into commands. Faithful async port of
// pkg/sdk/worker/dispatch.go:dispatchWorkflowTask + runWorkflowEpisode.
import {
completeWorkflowCommand,
failWorkflowCommand,
parseHistory,
workflowInput,
type RawCommand,
} from "./history";
import {
DeciderContext,
runWithContext,
type WorkflowInfo,
} from "../workflow/runtime";
export type WorkflowFn = (...args: any[]) => unknown | Promise<unknown>;
/**
* Replay `fn` for one workflow-task episode and return the commands to ship
* back. Returns only the schedule commands when the workflow parks on a
* durable primitive (activity/timer) whose result is not yet in history.
*/
export async function runWorkflowEpisode(
fn: WorkflowFn,
info: WorkflowInfo,
historyStr: string,
): Promise<RawCommand[]> {
const events = parseHistory(historyStr);
const ctx = new DeciderContext(info);
ctx.loadHistory(events);
const args = workflowInput(events);
let settled = false;
let isError = false;
let value: unknown;
let error: unknown;
let promise: Promise<unknown>;
try {
promise = Promise.resolve(runWithContext(ctx, () => fn(...args)));
} catch (e) {
// Synchronous throw before the first await.
return [...ctx.newCommands, failWorkflowCommand(e)];
}
promise.then(
(v) => {
settled = true;
value = v;
},
(e) => {
settled = true;
isError = true;
error = e;
},
);
// A history-resolved replay settles entirely through microtasks, which are
// fully drained before a `setImmediate` macrotask fires. Give a small
// margin; if still unsettled the workflow is parked on a durable primitive.
for (let i = 0; i < 3 && !settled; i++) {
await new Promise<void>((resolve) => setImmediate(resolve));
}
const commands = ctx.newCommands.slice();
if (!settled) {
return commands; // parked — advance on the next task with results appended
}
if (isError) {
return [...commands, failWorkflowCommand(error)];
}
return [...commands, completeWorkflowCommand(value)];
}
+128
View File
@@ -0,0 +1,128 @@
// Copyright © 2026 Hanzo AI. MIT License.
//
// History + command wire model, matching pkg/tasks/types.go (HistoryEvent),
// pkg/tasks/dispatch.go (delivery JSON) and pkg/sdk/worker/dispatch.go
// (commandsEnvelope / rawCommand).
//
// Encoding asymmetry to respect exactly:
// • History attributes (server → worker): `input`/`result`/`failure` are
// RAW JSON values (json.RawMessage); `seq` is a number.
// • Command fields (worker → server): `input`/`result`/`failure` are Go
// []byte, i.e. BASE64 strings in the commands JSON.
import { encodeFailure } from "../common/failure";
export const EventType = {
WorkflowStarted: "WORKFLOW_EXECUTION_STARTED",
ActivityScheduled: "ACTIVITY_TASK_SCHEDULED",
ActivityCompleted: "ACTIVITY_TASK_COMPLETED",
ActivityFailed: "ACTIVITY_TASK_FAILED",
WorkflowSignaled: "WORKFLOW_EXECUTION_SIGNALED",
} as const;
export interface HistoryEvent {
eventId: number;
eventTime?: string;
eventType: string;
attributes?: Record<string, unknown>;
}
export const CommandKind = {
CompleteWorkflow: 0,
FailWorkflow: 1,
ScheduleActivity: 2,
} as const;
export interface RawCommand {
kind: number;
result?: string; // base64
failure?: string; // base64
seq?: number;
activityType?: string;
input?: string; // base64
taskQueue?: string;
startToCloseMs?: number;
heartbeatMs?: number;
retryPolicy?: {
initial_interval_ms?: number;
backoff_coefficient?: number;
maximum_interval_ms?: number;
maximum_attempts?: number;
non_retryable_error_types?: string[];
};
}
export interface CommandsEnvelope {
v: number;
cmds: RawCommand[];
}
/** Parse a delivered history string into events. Empty → []. */
export function parseHistory(history: string | undefined): HistoryEvent[] {
if (!history || history.trim().length === 0) return [];
const parsed = JSON.parse(history);
return Array.isArray(parsed) ? (parsed as HistoryEvent[]) : [];
}
/** Numeric seq attribute, or undefined. */
export function eventSeq(ev: HistoryEvent): number | undefined {
const seq = ev.attributes?.["seq"];
return typeof seq === "number" ? seq : undefined;
}
/** Workflow args recorded on WORKFLOW_EXECUTION_STARTED. */
export function workflowInput(events: HistoryEvent[]): unknown[] {
for (const ev of events) {
if (ev.eventType === EventType.WorkflowStarted) {
const input = ev.attributes?.["input"];
if (Array.isArray(input)) return input;
return input === undefined ? [] : [input];
}
}
return [];
}
export function completeWorkflowCommand(result: unknown): RawCommand {
const cmd: RawCommand = { kind: CommandKind.CompleteWorkflow };
if (result !== undefined) {
cmd.result = Buffer.from(JSON.stringify(result), "utf8").toString("base64");
}
return cmd;
}
export function failWorkflowCommand(err: unknown): RawCommand {
return {
kind: CommandKind.FailWorkflow,
failure: encodeFailure(err).toString("base64"),
};
}
export interface ScheduleActivitySpec {
seq: number;
activityType: string;
args: unknown[];
taskQueue: string;
startToCloseMs?: number;
heartbeatMs?: number;
retryPolicy?: RawCommand["retryPolicy"];
}
export function scheduleActivityCommand(spec: ScheduleActivitySpec): RawCommand {
return {
kind: CommandKind.ScheduleActivity,
seq: spec.seq,
activityType: spec.activityType,
input: Buffer.from(JSON.stringify(spec.args ?? []), "utf8").toString("base64"),
taskQueue: spec.taskQueue,
startToCloseMs: spec.startToCloseMs || undefined,
heartbeatMs: spec.heartbeatMs || undefined,
retryPolicy: spec.retryPolicy,
};
}
export function encodeCommands(cmds: RawCommand[]): Buffer {
const env: CommandsEnvelope = { v: 1, cmds };
return Buffer.from(JSON.stringify(env), "utf8");
}
export const emptyCommands = (): Buffer => encodeCommands([]);
+12
View File
@@ -0,0 +1,12 @@
// Copyright © 2026 Hanzo AI. MIT License.
export * from "./worker";
export * from "./activity-context";
export {
EventType,
CommandKind,
type HistoryEvent,
type RawCommand,
type CommandsEnvelope,
} from "./history";
export { runWorkflowEpisode, type WorkflowFn } from "./decider";
+344
View File
@@ -0,0 +1,344 @@
// Copyright © 2026 Hanzo AI. MIT License.
//
// Worker runtime. Subscribes once per kind, receives server-pushed workflow
// and activity tasks over ZAP, dispatches each to a registered function, and
// ships the result back. Faithful port of pkg/sdk/worker (Go).
import { ZapNode, type Transport } from "../zap/node";
import { Opcode } from "../zap/opcodes";
import {
decodeEnvelope,
decodeHeartbeatResponse,
encodeFieldObject,
} from "../zap/envelope";
import { encodeCommands, failWorkflowCommand } from "./history";
import { runWorkflowEpisode, type WorkflowFn } from "./decider";
import {
ActivityContext,
runInActivityContext,
type ActivityInfo,
} from "./activity-context";
import { encodeFailure, TaskError } from "../common/failure";
import { encodeValue, decodeArgs } from "../common/converter";
import { TIMER_ACTIVITY_TYPE, type WorkflowInfo } from "../workflow/runtime";
export type ActivityFn = (...args: any[]) => unknown | Promise<unknown>;
export interface WorkerOptions {
/** A shared connection (from Connection.connect). Preferred. */
connection?: { transport: Transport };
/** "host:port" of tasksd. Ignored when `connection`/`transport` is set. */
address?: string;
namespace?: string;
/** Required — the queue this worker serves. */
taskQueue: string;
identity?: string;
workflows?: Record<string, WorkflowFn>;
activities?: Record<string, ActivityFn>;
/** Inject a Transport directly (tests). */
transport?: Transport;
dialTimeoutMs?: number;
callTimeoutMs?: number;
}
interface WorkflowDelivery {
task_token: string;
workflow_id: string;
run_id: string;
workflow_type_name: string;
history?: string;
}
interface ActivityDelivery {
task_token: string;
workflow_id: string;
run_id: string;
activity_id: string;
activity_type_name: string;
input?: string;
scheduled_time_ms: number;
start_to_close_timeout_ms?: number;
heartbeat_timeout_ms?: number;
}
const delay = (ms: number): Promise<void> => new Promise((r) => setTimeout(r, ms));
export class Worker {
private readonly workflows = new Map<string, WorkflowFn>();
private readonly activities = new Map<string, ActivityFn>();
private readonly inflight = new Set<Promise<void>>();
private wfSubId = "";
private actSubId = "";
private started = false;
private stopped = false;
private resolveDone: (() => void) | null = null;
private donePromise: Promise<void> | null = null;
private constructor(
private readonly transport: Transport,
private readonly namespace: string,
private readonly taskQueue: string,
private readonly identity: string,
private readonly ownsTransport: boolean,
) {
// Durable timers are backed by an internal sleeper activity.
this.activities.set(TIMER_ACTIVITY_TYPE, async (ms: number) => {
await delay(Math.max(0, Number(ms) || 0));
});
}
static async create(opts: WorkerOptions): Promise<Worker> {
if (!opts.taskQueue) throw new Error("hanzo/tasks: WorkerOptions.taskQueue is required");
const namespace = opts.namespace && opts.namespace.length > 0 ? opts.namespace : "default";
const identity = opts.identity && opts.identity.length > 0 ? opts.identity : "hanzo-tasks-ts-worker";
let transport: Transport;
let owns = false;
if (opts.transport) {
transport = opts.transport;
} else if (opts.connection) {
transport = opts.connection.transport;
} else {
if (!opts.address) throw new Error("hanzo/tasks: WorkerOptions requires connection, address, or transport");
const [host, port] = splitHostPort(opts.address);
const node = new ZapNode({
host,
port,
nodeId: identity,
dialTimeoutMs: opts.dialTimeoutMs,
callTimeoutMs: opts.callTimeoutMs,
});
await node.connect();
transport = node;
owns = true;
}
const w = new Worker(transport, namespace, opts.taskQueue, identity, owns);
for (const [name, fn] of Object.entries(opts.workflows ?? {})) w.registerWorkflow(name, fn);
for (const [name, fn] of Object.entries(opts.activities ?? {})) w.registerActivity(name, fn);
return w;
}
registerWorkflow(name: string, fn: WorkflowFn): void {
if (this.workflows.has(name)) throw new Error(`hanzo/tasks: workflow ${name} already registered`);
this.workflows.set(name, fn);
}
registerActivity(name: string, fn: ActivityFn): void {
if (this.activities.has(name) && name !== TIMER_ACTIVITY_TYPE) {
throw new Error(`hanzo/tasks: activity ${name} already registered`);
}
this.activities.set(name, fn);
}
/** Install handlers and subscribe. Non-blocking. */
async start(): Promise<void> {
if (this.started) return;
this.started = true;
this.transport.handle(Opcode.DeliverWorkflowTask, (_from, body) => {
this.track(this.onWorkflowTask(body));
});
this.transport.handle(Opcode.DeliverActivityTask, (_from, body) => {
this.track(this.onActivityTask(body));
});
this.actSubId = await this.subscribeActivity();
this.wfSubId = await this.subscribeWorkflow();
}
/** Start, then block until shutdown() is called. Mirrors @temporalio run(). */
async run(): Promise<void> {
await this.start();
this.donePromise = new Promise<void>((resolve) => {
this.resolveDone = resolve;
});
await this.donePromise;
}
/** Unsubscribe, drain in-flight, and close the connection if owned. */
async shutdown(): Promise<void> {
if (this.stopped) return;
this.stopped = true;
try {
if (this.wfSubId) await this.unsubscribe(this.wfSubId);
if (this.actSubId) await this.unsubscribe(this.actSubId);
} catch {
// Best effort — the server treats unknown sub-ids as already gone.
}
await Promise.allSettled([...this.inflight]);
if (this.ownsTransport) await this.transport.close();
if (this.resolveDone) this.resolveDone();
}
private track(p: Promise<void>): void {
this.inflight.add(p);
p.finally(() => this.inflight.delete(p)).catch(() => {});
}
// ── workflow task ──
private async onWorkflowTask(body: Buffer): Promise<void> {
let msg: WorkflowDelivery;
try {
msg = JSON.parse(body.toString("utf8"));
} catch {
return;
}
if (!msg.task_token) return;
const token = Buffer.from(msg.task_token, "utf8");
const fn = this.workflows.get(msg.workflow_type_name);
if (!fn) {
await this.respondWorkflow(token, encodeCommands([]));
return;
}
const info: WorkflowInfo = {
workflowId: msg.workflow_id,
runId: msg.run_id,
workflowType: msg.workflow_type_name,
taskQueue: this.taskQueue,
namespace: this.namespace,
attempt: 1,
};
try {
const commands = await runWorkflowEpisode(fn, info, msg.history ?? "");
await this.respondWorkflow(token, encodeCommands(commands));
} catch (err) {
// A decider bug must not wedge the run — report a workflow failure.
await this.respondWorkflow(token, encodeCommands([failWorkflowCommand(err)]));
}
}
// ── activity task ──
private async onActivityTask(body: Buffer): Promise<void> {
let msg: ActivityDelivery;
try {
msg = JSON.parse(body.toString("utf8"));
} catch {
return;
}
if (!msg.task_token) return;
const token = Buffer.from(msg.task_token, "utf8");
const fn = this.activities.get(msg.activity_type_name);
if (!fn) {
await this.respondActivityFailed(
token,
encodeFailure(new TaskError(`activity ${msg.activity_type_name} not registered`, "NotFoundError", true)),
);
return;
}
const abort = new AbortController();
const info: ActivityInfo = {
taskToken: token,
workflowId: msg.workflow_id,
runId: msg.run_id,
activityId: msg.activity_id,
activityType: msg.activity_type_name,
taskQueue: this.taskQueue,
attempt: 1,
scheduledTimeMs: msg.scheduled_time_ms ?? 0,
startedTimeMs: Date.now(),
};
const ctx = new ActivityContext(
info,
(details) => {
void this.heartbeat(token, details).then((cancelRequested) => {
if (cancelRequested) abort.abort();
});
},
abort,
);
let stopHeartbeat: (() => void) | null = null;
if (msg.heartbeat_timeout_ms && msg.heartbeat_timeout_ms > 0) {
stopHeartbeat = this.autoHeartbeat(token, Math.max(100, msg.heartbeat_timeout_ms / 2), abort);
}
try {
const args = decodeArgs(msg.input ? Buffer.from(msg.input, "utf8") : null);
const result = await runInActivityContext(ctx, () => Promise.resolve(fn(...args)));
if (stopHeartbeat) stopHeartbeat();
await this.respondActivityCompleted(token, encodeValue(result));
} catch (err) {
if (stopHeartbeat) stopHeartbeat();
await this.respondActivityFailed(token, encodeFailure(err));
}
}
private autoHeartbeat(token: Buffer, intervalMs: number, abort: AbortController): () => void {
const timer = setInterval(() => {
void this.heartbeat(token, []).then((cancelRequested) => {
if (cancelRequested) abort.abort();
});
}, intervalMs);
return () => clearInterval(timer);
}
// ── worker transport RPCs ──
private async subscribeWorkflow(): Promise<string> {
const body = Buffer.from(
JSON.stringify({ namespace: this.namespace, task_queue: this.taskQueue, identity: this.identity }),
"utf8",
);
const frame = await this.transport.call(Opcode.SubscribeWorkflowTasks, body);
return this.decodeSubId(frame);
}
private async subscribeActivity(): Promise<string> {
const body = Buffer.from(
JSON.stringify({ namespace: this.namespace, task_queue: this.taskQueue, identity: this.identity }),
"utf8",
);
const frame = await this.transport.call(Opcode.SubscribeActivityTasks, body);
return this.decodeSubId(frame);
}
private async unsubscribe(subId: string): Promise<void> {
const body = Buffer.from(JSON.stringify({ subscription_id: subId }), "utf8");
await this.transport.call(Opcode.UnsubscribeTasks, body);
}
private decodeSubId(frame: Buffer): string {
const { status, detail, body } = decodeEnvelope(frame);
if (status !== 0 && status !== 200) throw new Error(`subscribe failed: status ${status}: ${detail}`);
if (body.length === 0) return "";
const resp = JSON.parse(body.toString("utf8")) as { subscription_id?: string };
return resp.subscription_id ?? "";
}
private async respondWorkflow(token: Buffer, commands: Buffer): Promise<void> {
await this.transport.call(Opcode.RespondWorkflowTaskCompleted, encodeFieldObject(token, commands));
}
private async respondActivityCompleted(token: Buffer, result: Buffer): Promise<void> {
await this.transport.call(Opcode.RespondActivityTaskCompleted, encodeFieldObject(token, result));
}
private async respondActivityFailed(token: Buffer, failure: Buffer): Promise<void> {
await this.transport.call(Opcode.RespondActivityTaskFailed, encodeFieldObject(token, failure));
}
private async heartbeat(token: Buffer, details: unknown[]): Promise<boolean> {
try {
const payload = details.length > 0 ? Buffer.from(JSON.stringify(details), "utf8") : Buffer.alloc(0);
const frame = await this.transport.call(Opcode.RecordActivityTaskHeartbeat, encodeFieldObject(token, payload));
return decodeHeartbeatResponse(frame);
} catch {
return false;
}
}
}
function splitHostPort(addr: string): [string, number] {
const idx = addr.lastIndexOf(":");
if (idx < 0) throw new Error(`hanzo/tasks: address must be host:port, got ${addr}`);
const host = addr.slice(0, idx) || "127.0.0.1";
const port = Number(addr.slice(idx + 1));
if (!Number.isFinite(port) || port <= 0) throw new Error(`hanzo/tasks: invalid port in ${addr}`);
return [host, port];
}
+163
View File
@@ -0,0 +1,163 @@
// Copyright © 2026 Hanzo AI. MIT License.
//
// Workflow-side primitives. Import these inside workflow functions. The API
// deliberately mirrors @temporalio/workflow so a Postiz/NestJS migration is a
// dependency swap. Semantics are the Hanzo Tasks event-sourced replay model
// (see runtime.ts).
import { randomUUID } from "node:crypto";
import { toMs, type Duration } from "../common/duration";
import {
currentContext,
ContinueAsNewNotSupported,
type ActivityOptions,
type WorkflowInfo,
} from "./runtime";
export {
ContinueAsNewNotSupported,
type ActivityOptions,
type WorkflowInfo,
type DeciderContext,
} from "./runtime";
// ── activities ──
/** Options accepted by proxyActivities (superset: ms numbers or durations). */
export interface ProxyActivityOptions {
startToCloseTimeout?: Duration;
scheduleToCloseTimeout?: Duration;
heartbeatTimeout?: Duration;
taskQueue?: string;
retry?: {
initialInterval?: Duration;
backoffCoefficient?: number;
maximumInterval?: Duration;
maximumAttempts?: number;
nonRetryableErrorTypes?: string[];
};
}
function normalizeActivityOptions(opts: ProxyActivityOptions): ActivityOptions {
return {
startToCloseTimeoutMs: opts.startToCloseTimeout !== undefined ? toMs(opts.startToCloseTimeout) : undefined,
scheduleToCloseTimeoutMs:
opts.scheduleToCloseTimeout !== undefined ? toMs(opts.scheduleToCloseTimeout) : undefined,
heartbeatTimeoutMs: opts.heartbeatTimeout !== undefined ? toMs(opts.heartbeatTimeout) : undefined,
taskQueue: opts.taskQueue,
retry: opts.retry
? {
initialIntervalMs: opts.retry.initialInterval !== undefined ? toMs(opts.retry.initialInterval) : undefined,
backoffCoefficient: opts.retry.backoffCoefficient,
maximumIntervalMs: opts.retry.maximumInterval !== undefined ? toMs(opts.retry.maximumInterval) : undefined,
maximumAttempts: opts.retry.maximumAttempts,
nonRetryableErrorTypes: opts.retry.nonRetryableErrorTypes,
}
: undefined,
};
}
/**
* Create typed activity stubs. Calling a stub schedules a durable activity
* and returns a Promise that resolves with its result (or rejects with a
* TaskError). `A` is an interface of activity functions.
*/
export function proxyActivities<A extends Record<string, (...args: any[]) => any>>(
opts: ProxyActivityOptions,
): A {
const normalized = normalizeActivityOptions(opts);
return new Proxy(
{},
{
get(_target, prop) {
if (typeof prop !== "string") return undefined;
return (...args: unknown[]) => currentContext().scheduleActivity(prop, args, normalized);
},
},
) as A;
}
// ── timers ──
/** Durable sleep. Backed by an internal sleeper activity (see README). */
export function sleep(duration: Duration): Promise<void> {
return currentContext().sleep(toMs(duration));
}
/** Alias matching @temporalio's Timer semantics. */
export function startTimer(duration: Duration): Promise<void> {
return sleep(duration);
}
// ── condition ──
/** Resolve when `fn()` becomes true, or false when the optional timeout fires. */
export function condition(fn: () => boolean, timeout?: Duration): Promise<boolean> {
return currentContext().condition(fn, timeout !== undefined ? toMs(timeout) : undefined);
}
// ── signals & queries ──
export interface SignalDefinition<Args extends any[] = []> {
type: "signal";
name: string;
readonly _args?: Args;
}
export interface QueryDefinition<Ret = unknown, Args extends any[] = []> {
type: "query";
name: string;
readonly _ret?: Ret;
readonly _args?: Args;
}
export function defineSignal<Args extends any[] = []>(name: string): SignalDefinition<Args> {
return { type: "signal", name };
}
export function defineQuery<Ret = unknown, Args extends any[] = []>(name: string): QueryDefinition<Ret, Args> {
return { type: "query", name };
}
/** Register a signal or query handler. */
export function setHandler<Args extends any[], Ret>(
def: SignalDefinition<Args> | QueryDefinition<Ret, Args>,
handler: (...args: Args) => Ret | void,
): void {
const ctx = currentContext();
if (def.type === "signal") {
ctx.registerSignalHandler(def.name, (arg: unknown) => {
const args = (Array.isArray(arg) ? arg : [arg]) as Args;
handler(...args);
});
} else {
ctx.registerQueryHandler(def.name, (...args: unknown[]) => handler(...(args as Args)) as Ret);
}
}
// ── info & versioning ──
export function workflowInfo(): WorkflowInfo {
return currentContext().info;
}
export function getVersion(changeId: string, minSupported: number, maxSupported: number): number {
return currentContext().getVersion(changeId, minSupported, maxSupported);
}
export function upsertSearchAttributes(attrs: Record<string, unknown>): void {
currentContext().upsertSearchAttributes(attrs);
}
/**
* continueAsNew — API-present, not yet durable on the v1 wire (no server
* command). Throws ContinueAsNewNotSupported at runtime. See README.
*/
export function continueAsNew<_F extends (...args: any[]) => any>(..._args: unknown[]): Promise<never> {
return Promise.reject(new ContinueAsNewNotSupported(currentContext().info.workflowType));
}
/** Deterministic-friendly UUID (delegates to crypto; document determinism). */
export function uuid4(): string {
return randomUUID();
}
+237
View File
@@ -0,0 +1,237 @@
// Copyright © 2026 Hanzo AI. MIT License.
//
// Workflow replay runtime. Faithful async port of pkg/sdk/worker/env.go's
// event-sourced replay decider.
//
// Model: on each workflow task the server delivers the run's full history.
// The registered workflow function is re-run from the top. Each durable
// primitive (activity, timer) is assigned a deterministic per-run sequence
// number in program order. If its outcome is already in history it resolves
// immediately; otherwise the decider emits a command and the primitive's
// promise stays PENDING FOREVER — the async function suspends, the episode
// ends with the emitted commands, and the run advances when the server
// delivers the next task with the outcome appended.
//
// Suspending on a never-settling promise (rather than throwing a sentinel)
// makes blocking survive user `try/catch`, `Promise.all`, and `Promise.race`
// exactly as real Temporal workflows do.
//
// Determinism contract (identical to Temporal): a workflow function must be
// pure w.r.t. its input + history. Do not call Date.now(), Math.random(),
// setTimeout, or perform I/O directly — use sleep()/activities/side-effects.
import { AsyncLocalStorage } from "node:async_hooks";
import { TaskError } from "../common/failure";
import {
EventType,
eventSeq,
scheduleActivityCommand,
type HistoryEvent,
type RawCommand,
} from "../worker/history";
import { decodeFailure } from "../common/failure";
export interface WorkflowInfo {
workflowId: string;
runId: string;
workflowType: string;
taskQueue: string;
namespace: string;
attempt: number;
}
export interface ActivityOptions {
/** Maximum time for a single activity attempt (required in practice). */
startToCloseTimeoutMs?: number;
/** Accepted for API compatibility; the v1 wire schedules on startToClose. */
scheduleToCloseTimeoutMs?: number;
heartbeatTimeoutMs?: number;
/** Override the workflow's task queue for this activity. */
taskQueue?: string;
retry?: {
initialIntervalMs?: number;
backoffCoefficient?: number;
maximumIntervalMs?: number;
maximumAttempts?: number;
nonRetryableErrorTypes?: string[];
};
}
/** Internal activity type that backs durable timers (auto-registered). */
export const TIMER_ACTIVITY_TYPE = "__hanzo_timer__";
const TIMER_SLACK_MS = 60_000;
interface ActivityOutcome {
failed: boolean;
result?: unknown;
failure?: TaskError;
}
/** Sentinel surfaced when a workflow calls the not-yet-supported continueAsNew. */
export class ContinueAsNewNotSupported extends Error {
constructor(public readonly workflowType: string) {
super(
`continueAsNew is not yet supported by the Hanzo Tasks server wire ` +
`(no ContinueAsNew command). Workflow type: ${workflowType}. See SDK README "Remaining".`,
);
this.name = "ContinueAsNewNotSupported";
}
}
/** DeciderContext holds one episode's replay state. Mirrors workerEnv. */
export class DeciderContext {
readonly info: WorkflowInfo;
private readonly activityResults = new Map<number, ActivityOutcome>();
private readonly scheduledSeqs = new Set<number>();
private nextSeq = 0;
readonly newCommands: RawCommand[] = [];
/** Signal payloads by name, in history order. */
private readonly signalsByName = new Map<string, unknown[]>();
/** Registered signal handlers; a matching name drains buffered payloads. */
private readonly signalHandlers = new Map<string, (arg: unknown) => void>();
readonly queryHandlers = new Map<string, (...args: unknown[]) => unknown>();
/** Local, non-durable (mirrors Go: no wire command yet). */
readonly searchAttributes: Record<string, unknown> = {};
private readonly versions = new Map<string, number>();
constructor(info: WorkflowInfo) {
this.info = info;
}
/** Seed from the run's history: scheduled seqs, outcomes, buffered signals. */
loadHistory(events: HistoryEvent[]): void {
for (const ev of events) {
if (ev.eventType === EventType.WorkflowSignaled) {
const name = String(ev.attributes?.["signal"] ?? "");
if (name) {
const arr = this.signalsByName.get(name) ?? [];
arr.push(ev.attributes?.["input"]);
this.signalsByName.set(name, arr);
}
continue;
}
const seq = eventSeq(ev);
if (seq === undefined) continue;
switch (ev.eventType) {
case EventType.ActivityScheduled:
this.scheduledSeqs.add(seq);
break;
case EventType.ActivityCompleted:
this.activityResults.set(seq, { failed: false, result: ev.attributes?.["result"] });
break;
case EventType.ActivityFailed: {
const failureAttr = ev.attributes?.["failure"];
const failure = decodeFailure(Buffer.from(JSON.stringify(failureAttr ?? null), "utf8"));
this.activityResults.set(seq, { failed: true, failure });
break;
}
}
}
}
/** Durable activity call. Resolves from history or emits a schedule + parks. */
scheduleActivity(activityType: string, args: unknown[], opts: ActivityOptions): Promise<unknown> {
if (!activityType) {
return Promise.reject(new TaskError("activity type is empty", "ConfigError", true));
}
const seq = this.nextSeq++;
const outcome = this.activityResults.get(seq);
if (outcome) {
return outcome.failed
? Promise.reject(outcome.failure ?? new TaskError("activity failed", "ApplicationError", true))
: Promise.resolve(outcome.result);
}
if (!this.scheduledSeqs.has(seq)) {
this.newCommands.push(
scheduleActivityCommand({
seq,
activityType,
args,
taskQueue: opts.taskQueue || this.info.taskQueue,
startToCloseMs: opts.startToCloseTimeoutMs,
heartbeatMs: opts.heartbeatTimeoutMs,
retryPolicy: opts.retry
? {
initial_interval_ms: opts.retry.initialIntervalMs,
backoff_coefficient: opts.retry.backoffCoefficient,
maximum_interval_ms: opts.retry.maximumIntervalMs,
maximum_attempts: opts.retry.maximumAttempts,
non_retryable_error_types: opts.retry.nonRetryableErrorTypes,
}
: undefined,
}),
);
this.scheduledSeqs.add(seq);
}
return parkForever();
}
/** Durable timer, backed by an internal sleeper activity. */
sleep(ms: number): Promise<void> {
if (ms <= 0) return Promise.resolve();
return this.scheduleActivity(TIMER_ACTIVITY_TYPE, [ms], {
startToCloseTimeoutMs: ms + TIMER_SLACK_MS,
}) as Promise<void>;
}
/** Wait until predicate() is true. Re-evaluated each episode as signals land. */
condition(predicate: () => boolean, timeoutMs?: number): Promise<boolean> {
if (predicate()) return Promise.resolve(true);
if (timeoutMs && timeoutMs > 0) {
// Race a durable timer: when it fires, the predicate is re-checked and
// condition resolves false if still unmet.
const timer = this.sleep(timeoutMs).then(() => predicate());
return timer;
}
return parkForever();
}
registerSignalHandler(name: string, fn: (arg: unknown) => void): void {
this.signalHandlers.set(name, fn);
const buffered = this.signalsByName.get(name);
if (buffered) {
for (const payload of buffered) fn(payload);
this.signalsByName.delete(name);
}
}
registerQueryHandler(name: string, fn: (...args: unknown[]) => unknown): void {
this.queryHandlers.set(name, fn);
}
getVersion(changeId: string, minSupported: number, maxSupported: number): number {
const existing = this.versions.get(changeId);
if (existing !== undefined) return existing;
// No durable marker command on the v1 wire — record maxSupported locally.
void minSupported;
this.versions.set(changeId, maxSupported);
return maxSupported;
}
upsertSearchAttributes(attrs: Record<string, unknown>): void {
Object.assign(this.searchAttributes, attrs);
}
}
/** A promise that never settles — parks the async workflow on suspension. */
export function parkForever(): Promise<never> {
return new Promise<never>(() => {});
}
const storage = new AsyncLocalStorage<DeciderContext>();
/** The active decider context. Throws if called outside a workflow. */
export function currentContext(): DeciderContext {
const ctx = storage.getStore();
if (!ctx) {
throw new Error("hanzo/tasks: workflow primitive called outside a workflow context");
}
return ctx;
}
/** Run `fn` bound to `ctx` for one replay episode. */
export function runWithContext<T>(ctx: DeciderContext, fn: () => T): T {
return storage.run(ctx, fn);
}
+126
View File
@@ -0,0 +1,126 @@
// Copyright © 2026 Hanzo AI. MIT License.
//
// Frame codecs matching pkg/sdk/client/client.go and worker_transport.go.
//
// Two body framings exist on the Hanzo Tasks wire:
//
// 1. JSON envelope — a struct StartObject(24) with:
// field 0 Bytes JSON body
// field 8 Uint32 status (0/200 = ok)
// field 12 Bytes error detail
// Used by every client RPC, Subscribe/Unsubscribe, StartChildWorkflow,
// and every server-pushed delivery.
//
// 2. Field object — a struct StartObject(32) with:
// field 0 Bytes task token
// field 8 Bytes payload (commands / result / failure / details)
// Used by RespondWorkflowTaskCompleted / RespondActivity* / Heartbeat.
import { Builder, ZapMessage } from "./wire";
import { Field } from "./opcodes";
const ENVELOPE_BODY = 0;
const ENVELOPE_STATUS = 8;
const ENVELOPE_ERROR = 12;
const ENVELOPE_OBJECT_SIZE = 24;
const FIELD_OBJECT_SIZE = 32;
const ZAP_MAGIC_PREFIX = Buffer.of(0x5a, 0x41, 0x50, 0x00);
/** Wrap a JSON body in the single-field ZAP envelope, flagged with opcode. */
export function encodeEnvelope(opcode: number, body: Buffer | Uint8Array | null): Buffer {
const bodyBuf = body ? Buffer.from(body) : Buffer.alloc(0);
const b = new Builder(bodyBuf.length + ENVELOPE_OBJECT_SIZE + 32);
const obj = b.startObject(ENVELOPE_OBJECT_SIZE);
obj.setBytes(ENVELOPE_BODY, bodyBuf);
obj.finishAsRoot();
return b.finishWithFlags((opcode << 8) & 0xffff);
}
/** Build a server-style response envelope (used by the test mock + parity). */
export function encodeResponseEnvelope(
opcode: number,
body: Buffer | Uint8Array | null,
status = 0,
errDetail = "",
): Buffer {
const bodyBuf = body ? Buffer.from(body) : Buffer.alloc(0);
const b = new Builder(bodyBuf.length + ENVELOPE_OBJECT_SIZE + 64);
const obj = b.startObject(ENVELOPE_OBJECT_SIZE);
obj.setBytes(ENVELOPE_BODY, bodyBuf);
if (status !== 0) obj.setUint32(ENVELOPE_STATUS, status >>> 0);
if (errDetail) obj.setText(ENVELOPE_ERROR, errDetail);
obj.finishAsRoot();
return b.finishWithFlags((opcode << 8) & 0xffff);
}
export interface DecodedEnvelope {
status: number;
detail: string;
body: Buffer;
}
/** Read status / error / body out of a response frame. */
export function decodeEnvelope(frame: Buffer): DecodedEnvelope {
const msg = ZapMessage.parse(frame);
const root = msg.root();
return {
status: root.uint32(ENVELOPE_STATUS),
detail: root.bytes(ENVELOPE_ERROR).toString("utf8"),
body: root.bytes(ENVELOPE_BODY),
};
}
/** Read just the JSON body from a delivery/ack envelope. */
export function envelopeBody(frame: Buffer): Buffer {
return ZapMessage.parse(frame).root().bytes(ENVELOPE_BODY);
}
/**
* Produce a complete ZAP frame for `opcode`. If `body` is already a framed
* ZAP object (worker field-object path), re-stamp its flags; otherwise wrap
* the raw bytes in the JSON envelope. Mirrors client.FrameBody.
*/
export function frameBody(opcode: number, body: Buffer): Buffer {
if (body.length >= 8 && body.subarray(0, 4).equals(ZAP_MAGIC_PREFIX)) {
const out = Buffer.from(body);
out.writeUInt16LE((opcode << 8) & 0xffff, 6);
return out;
}
return encodeEnvelope(opcode, body);
}
/** Build a field-object frame: token @0, payload @8. */
export function encodeFieldObject(token: Buffer, payload: Buffer): Buffer {
const b = new Builder(token.length + payload.length + 64);
const obj = b.startObject(FIELD_OBJECT_SIZE);
obj.setBytes(Field.TaskToken, token);
obj.setBytes(Field.Payload, payload);
obj.finishAsRoot();
return b.finish();
}
export interface FieldObject {
token: Buffer;
payload: Buffer;
}
/** Read a field-object frame (server side / tests). */
export function decodeFieldObject(frame: Buffer): FieldObject {
const root = ZapMessage.parse(frame).root();
return { token: root.bytes(Field.TaskToken), payload: root.bytes(Field.Payload) };
}
/** Encode the heartbeat response carrying the cancelRequested flag. */
export function encodeHeartbeatResponse(cancelRequested: boolean): Buffer {
const b = new Builder(64);
const obj = b.startObject(FIELD_OBJECT_SIZE);
obj.setBool(Field.RespCancelRequested, cancelRequested);
obj.finishAsRoot();
return b.finish();
}
/** Read the cancelRequested flag from a heartbeat response frame. */
export function decodeHeartbeatResponse(frame: Buffer): boolean {
return ZapMessage.parse(frame).root().bool(Field.RespCancelRequested);
}
+6
View File
@@ -0,0 +1,6 @@
// Copyright © 2026 Hanzo AI. MIT License.
export * from "./wire";
export * from "./opcodes";
export * from "./envelope";
export * from "./node";
+283
View File
@@ -0,0 +1,283 @@
// Copyright © 2026 Hanzo AI. MIT License.
//
// ZAP node — the TCP client half of luxfi/zap's node.go, in TypeScript.
//
// Socket framing: [u32 LE length][message bytes]
// Handshake: both sides send one message = StartObject(64) with the
// node id as raw bytes at offset 0 and its length at offset
// 60. The client sends first, then reads the server's.
// Call: request bytes are prefixed with [u32 reqID][u32 flag=1]
// before length-framing; the response carries flag=2 and the
// same reqID. Server-pushed deliveries arrive as bare frames
// (Send) and are routed to handlers by opcode.
import net from "node:net";
import { randomBytes } from "node:crypto";
import { Builder, ZapMessage } from "./wire";
import { encodeEnvelope, envelopeBody, frameBody } from "./envelope";
const REQ_FLAG_REQ = 1;
const REQ_FLAG_RESP = 2;
const MAX_MESSAGE = 10 * 1024 * 1024;
/** Transport is the request/response + push abstraction (mirrors Go Transport). */
export interface Transport {
/** Issue one request for `opcode`; resolve with the response frame. */
call(opcode: number, body: Buffer): Promise<Buffer>;
/** Register a callback for server-pushed messages on `opcode`. */
handle(opcode: number, fn: (from: string, body: Buffer) => void): void;
/** Release transport resources. */
close(): Promise<void>;
}
export interface ZapNodeOptions {
host: string;
port: number;
/** Stable-ish client node id; a random suffix is appended for uniqueness. */
nodeId?: string;
dialTimeoutMs?: number;
callTimeoutMs?: number;
}
interface Pending {
resolve: (frame: Buffer) => void;
reject: (err: Error) => void;
timer?: NodeJS.Timeout;
}
/** Build a handshake message: node id bytes @0, length @60, StartObject(64). */
function encodeHandshake(nodeId: string): Buffer {
const b = new Builder(128);
const obj = b.startObject(64);
const idBytes = Buffer.from(nodeId, "utf8");
const n = Math.min(idBytes.length, 60);
for (let i = 0; i < n; i++) obj.setUint8(i, idBytes[i]);
obj.setUint32(60, idBytes.length >>> 0);
obj.finishAsRoot();
return b.finish();
}
function decodeHandshakePeer(frame: Buffer): string {
const root = ZapMessage.parse(frame).root();
const len = root.uint32(60);
if (len === 0 || len > 60) return "";
const bytes = Buffer.alloc(len);
for (let i = 0; i < len; i++) bytes[i] = root.uint8(i);
return bytes.toString("utf8");
}
/** ZapNode is a single-peer ZAP client over one TCP connection. */
export class ZapNode implements Transport {
private sock: net.Socket | null = null;
private readonly nodeId: string;
private peerId = "";
private handshakeDone = false;
private closed = false;
private reqId = 0;
private readonly pending = new Map<number, Pending>();
private readonly handlers = new Map<number, (from: string, body: Buffer) => void>();
private acc: Buffer = Buffer.alloc(0);
private connectPromise: Promise<void> | null = null;
constructor(private readonly opts: ZapNodeOptions) {
const base = opts.nodeId && opts.nodeId.length > 0 ? opts.nodeId : "hanzo-tasks-sdk";
this.nodeId = `${base}-${randomBytes(4).toString("hex")}`;
}
/** Establish the TCP connection and complete the ZAP handshake. */
connect(): Promise<void> {
if (this.connectPromise) return this.connectPromise;
this.connectPromise = new Promise<void>((resolve, reject) => {
const sock = net.createConnection({ host: this.opts.host, port: this.opts.port });
sock.setNoDelay(true);
this.sock = sock;
const dialTimeout = this.opts.dialTimeoutMs ?? 10000;
const timer = setTimeout(() => {
sock.destroy(new Error(`zap dial timeout after ${dialTimeout}ms`));
}, dialTimeout);
let settled = false;
const finishOk = () => {
if (settled) return;
settled = true;
clearTimeout(timer);
resolve();
};
const finishErr = (err: Error) => {
if (settled) {
this.failAll(err);
return;
}
settled = true;
clearTimeout(timer);
reject(err);
};
sock.on("connect", () => {
// Client sends its handshake first (mirrors ConnectDirect).
this.writeFrame(encodeHandshake(this.nodeId));
});
sock.on("data", (chunk) => this.onData(chunk, finishOk));
sock.on("error", (err) => finishErr(err));
sock.on("close", () => {
this.closed = true;
this.failAll(new Error("zap connection closed"));
});
});
return this.connectPromise;
}
private onData(chunk: Buffer, onHandshake: () => void): void {
this.acc = this.acc.length === 0 ? chunk : Buffer.concat([this.acc, chunk]);
for (;;) {
if (this.acc.length < 4) return;
const len = this.acc.readUInt32LE(0);
if (len > MAX_MESSAGE) {
this.sock?.destroy(new Error("zap: message too large"));
return;
}
if (this.acc.length < 4 + len) return;
const payload = this.acc.subarray(4, 4 + len);
this.acc = this.acc.subarray(4 + len);
if (!this.handshakeDone) {
this.handshakeDone = true;
try {
this.peerId = decodeHandshakePeer(Buffer.from(payload));
} catch {
this.peerId = "";
}
onHandshake();
continue;
}
this.route(Buffer.from(payload));
}
}
private route(data: Buffer): void {
if (data.length >= 8) {
const flag = data.readUInt32LE(4);
if (flag === REQ_FLAG_RESP) {
const reqId = data.readUInt32LE(0);
const p = this.pending.get(reqId);
if (p) {
this.pending.delete(reqId);
if (p.timer) clearTimeout(p.timer);
p.resolve(data.subarray(8));
}
return;
}
if (flag === REQ_FLAG_REQ) {
// Server pushed a Call (waits for a correlated ack).
const reqId = data.readUInt32LE(0);
this.dispatch(data.subarray(8), reqId);
return;
}
}
// Bare frame (Send push) — no ack expected.
this.dispatch(data, null);
}
private dispatch(frame: Buffer, reqId: number | null): void {
let msg: ZapMessage;
try {
msg = ZapMessage.parse(frame);
} catch {
return;
}
const opcode = msg.opcode();
const handler = this.handlers.get(opcode);
if (handler) {
let body: Buffer;
try {
body = envelopeBody(frame);
} catch {
body = Buffer.alloc(0);
}
try {
handler(this.peerId, body);
} catch {
// Handlers must not crash the receive loop.
}
}
if (reqId !== null) {
// Ack a Call-style push so the server's Call completes.
const ack = encodeEnvelope(opcode, null);
this.writeCorrelated(reqId, REQ_FLAG_RESP, ack);
}
}
call(opcode: number, body: Buffer): Promise<Buffer> {
if (this.closed || !this.sock) {
return Promise.reject(new Error("zap: transport closed"));
}
const frame = frameBody(opcode, body);
this.reqId = (this.reqId + 1) >>> 0;
const id = this.reqId;
return new Promise<Buffer>((resolve, reject) => {
const p: Pending = { resolve, reject };
const timeout = this.opts.callTimeoutMs ?? 30000;
if (timeout > 0) {
p.timer = setTimeout(() => {
if (this.pending.delete(id)) {
reject(new Error(`zap call 0x${opcode.toString(16)} timed out after ${timeout}ms`));
}
}, timeout);
}
this.pending.set(id, p);
try {
this.writeCorrelated(id, REQ_FLAG_REQ, frame);
} catch (err) {
this.pending.delete(id);
if (p.timer) clearTimeout(p.timer);
reject(err as Error);
}
});
}
handle(opcode: number, fn: (from: string, body: Buffer) => void): void {
this.handlers.set(opcode, fn);
}
async close(): Promise<void> {
this.closed = true;
const s = this.sock;
this.sock = null;
if (s) {
await new Promise<void>((resolve) => {
s.end(() => resolve());
s.once("close", () => resolve());
setTimeout(() => {
s.destroy();
resolve();
}, 1000);
});
}
this.failAll(new Error("zap: transport closed"));
}
private writeCorrelated(reqId: number, flag: number, frame: Buffer): void {
const wrapped = Buffer.allocUnsafe(8 + frame.length);
wrapped.writeUInt32LE(reqId >>> 0, 0);
wrapped.writeUInt32LE(flag >>> 0, 4);
frame.copy(wrapped, 8);
this.writeFrame(wrapped);
}
private writeFrame(data: Buffer): void {
if (!this.sock) throw new Error("zap: not connected");
const len = Buffer.allocUnsafe(4);
len.writeUInt32LE(data.length >>> 0, 0);
this.sock.write(len);
this.sock.write(data);
}
private failAll(err: Error): void {
for (const [, p] of this.pending) {
if (p.timer) clearTimeout(p.timer);
p.reject(err);
}
this.pending.clear();
}
}
+72
View File
@@ -0,0 +1,72 @@
// Copyright © 2026 Hanzo AI. MIT License.
//
// ZAP opcodes owned by the Hanzo Tasks wire. Authoritative source:
// pkg/sdk/client/{client,transport}.go and pkg/tasks/dispatch.go. These
// values are stable across releases and never reused — mirror the Go
// constants exactly.
export const Opcode = {
// ── client workflow lifecycle (0x0060–0x006F) ──
StartWorkflow: 0x0060,
SignalWorkflow: 0x0061,
CancelWorkflow: 0x0062,
TerminateWorkflow: 0x0063,
DescribeWorkflow: 0x0064,
ListWorkflows: 0x0065,
SignalWithStartWorkflow: 0x0066,
QueryWorkflow: 0x0067,
// in-workflow scheduling
ScheduleActivity: 0x006b,
StartChildWorkflow: 0x006d,
// ── client schedule ops (0x0070–0x007F) ──
CreateSchedule: 0x0070,
ListSchedules: 0x0071,
DeleteSchedule: 0x0072,
PauseSchedule: 0x0073,
UpdateSchedule: 0x0074,
TriggerSchedule: 0x0075,
DescribeSchedule: 0x0076,
// ── client namespace ops (0x0080–0x008F) ──
RegisterNamespace: 0x0080,
DescribeNamespace: 0x0081,
ListNamespaces: 0x0082,
// ── health + meta (0x0090–0x009F) ──
Health: 0x0090,
// ── worker subscribe / respond (0x00A0–0x00AF) ──
SubscribeWorkflowTasks: 0x00a0,
SubscribeActivityTasks: 0x00a1,
RespondWorkflowTaskCompleted: 0x00a2,
RespondActivityTaskCompleted: 0x00a3,
RespondActivityTaskFailed: 0x00a4,
RecordActivityTaskHeartbeat: 0x00a5,
UnsubscribeTasks: 0x00a6,
// ── server → worker deliveries (Send; 0x00B0–0x00BF) ──
DeliverWorkflowTask: 0x00b0,
DeliverActivityTask: 0x00b1,
DeliverCancelRequest: 0x00b3,
DeliverQuery: 0x00b4,
// worker → server query response
RespondQuery: 0x00c4,
// generic error
Error: 0x00ff,
} as const;
export type OpcodeValue = (typeof Opcode)[keyof typeof Opcode];
// Field offsets for the field-object framing used by worker respond /
// heartbeat RPCs (pkg/sdk/client/transport.go). token at 0; the single
// payload (commands / result / failure / details) at 8; cancel flag at 8.
export const Field = {
TaskToken: 0,
Payload: 8,
RespStatus: 0,
RespCancelRequested: 8,
} as const;
+273
View File
@@ -0,0 +1,273 @@
// Copyright © 2026 Hanzo AI. MIT License.
//
// Byte-exact TypeScript port of the luxfi/zap (v0.2.0) message wire format —
// the "Zero-copy Application Protocol". Every byte this module emits is
// identical to what github.com/luxfi/zap's Builder/Object produce in Go, so a
// Hanzo Tasks server (tasksd) cannot tell a Go client from this one.
//
// Wire format:
//
// Header (16 bytes):
// Magic [0:4] "ZAP\0"
// Version [4:6] u16 LE (== 1)
// Flags [6:8] u16 LE (opcode is carried in the high byte)
// RootOffset [8:12] u32 LE offset to the root object
// Size [12:16] u32 LE total message size including header
// Data segment (8-byte aligned): Cap'n-Proto-style structs / bytes.
//
// A struct ("object") is a fixed region of `dataSize` bytes. Scalar fields
// live inline at their byte offset. A bytes/text field is a {relOffset u32,
// length u32} pair; relOffset is measured from the field's own position and
// points at the payload appended after the fixed section.
export const ZAP_MAGIC = Uint8Array.of(0x5a, 0x41, 0x50, 0x00); // "ZAP\0"
export const ZAP_VERSION = 1;
export const HEADER_SIZE = 16;
const ALIGNMENT = 8;
/** Builder constructs a single ZAP message. Mirrors zap.Builder. */
export class Builder {
private buf: Buffer;
private pos: number;
private rootOffset = 0;
constructor(capacity = 256) {
if (capacity < HEADER_SIZE) capacity = 256;
this.buf = Buffer.alloc(capacity);
this.pos = HEADER_SIZE;
this.buf[0] = ZAP_MAGIC[0];
this.buf[1] = ZAP_MAGIC[1];
this.buf[2] = ZAP_MAGIC[2];
this.buf[3] = ZAP_MAGIC[3];
this.buf.writeUInt16LE(ZAP_VERSION, 4);
}
/** Start an object with a fixed section of `dataSize` bytes. */
startObject(dataSize: number): ObjectBuilder {
this.align(ALIGNMENT);
return new ObjectBuilder(this, this.pos, dataSize);
}
finish(): Buffer {
this.buf.writeUInt32LE(this.rootOffset >>> 0, 8);
this.buf.writeUInt32LE(this.pos >>> 0, 12);
return this.buf.subarray(0, this.pos);
}
finishWithFlags(flags: number): Buffer {
this.buf.writeUInt16LE(flags & 0xffff, 6);
return this.finish();
}
// ── internals shared with ObjectBuilder ──
/** Current backing buffer. Re-fetch after any growth. */
raw(): Buffer {
return this.buf;
}
position(): number {
return this.pos;
}
setPosition(p: number): void {
this.pos = p;
}
setRoot(off: number): void {
this.rootOffset = off;
}
/** Ensure capacity for `n` more bytes past the current position. */
grow(n: number): void {
if (this.pos + n <= this.buf.length) return;
let newCap = this.buf.length * 2;
if (newCap < this.pos + n) newCap = this.pos + n;
const nb = Buffer.alloc(newCap);
this.buf.copy(nb, 0, 0, this.pos);
this.buf = nb;
}
/** Grow + zero-fill so the buffer covers up to absolute `needed`. */
ensureAbs(needed: number): void {
if (needed > this.pos) {
this.grow(needed - this.pos);
this.buf.fill(0, this.pos, needed);
this.pos = needed;
}
}
private align(a: number): void {
const padding = (a - (this.pos % a)) % a;
this.grow(padding);
for (let i = 0; i < padding; i++) {
this.buf[this.pos] = 0;
this.pos++;
}
}
}
interface DeferredBytes {
fieldOffset: number;
data: Buffer;
}
/** ObjectBuilder writes fields into a struct region. Mirrors zap.ObjectBuilder. */
export class ObjectBuilder {
private deferred: DeferredBytes[] = [];
constructor(
private readonly b: Builder,
public readonly startPos: number,
private readonly dataSize: number,
) {}
private ensureField(endOffset: number): void {
this.b.ensureAbs(this.startPos + endOffset);
}
setUint8(fieldOffset: number, v: number): void {
this.ensureField(fieldOffset + 1);
this.b.raw()[this.startPos + fieldOffset] = v & 0xff;
}
setUint16(fieldOffset: number, v: number): void {
this.ensureField(fieldOffset + 2);
this.b.raw().writeUInt16LE(v & 0xffff, this.startPos + fieldOffset);
}
setUint32(fieldOffset: number, v: number): void {
this.ensureField(fieldOffset + 4);
this.b.raw().writeUInt32LE(v >>> 0, this.startPos + fieldOffset);
}
setInt8(fieldOffset: number, v: number): void {
this.setUint8(fieldOffset, v & 0xff);
}
setBool(fieldOffset: number, v: boolean): void {
this.setUint8(fieldOffset, v ? 1 : 0);
}
/** Set a bytes field. The payload is appended during finish(). */
setBytes(fieldOffset: number, v: Buffer | Uint8Array | null | undefined): void {
this.ensureField(fieldOffset + 8);
const raw = this.b.raw();
if (!v || v.length === 0) {
raw.writeUInt32LE(0, this.startPos + fieldOffset);
raw.writeUInt32LE(0, this.startPos + fieldOffset + 4);
return;
}
const data = Buffer.from(v);
this.deferred.push({ fieldOffset, data });
raw.writeUInt32LE(data.length >>> 0, this.startPos + fieldOffset + 4);
}
setText(fieldOffset: number, s: string): void {
this.setBytes(fieldOffset, Buffer.from(s, "utf8"));
}
finish(): number {
this.ensureField(this.dataSize);
for (const e of this.deferred) {
const dataPos = this.b.position();
this.b.grow(e.data.length);
e.data.copy(this.b.raw(), this.b.position());
this.b.setPosition(this.b.position() + e.data.length);
const fieldAbsPos = this.startPos + e.fieldOffset;
const relOffset = dataPos - fieldAbsPos;
this.b.raw().writeInt32LE(relOffset, fieldAbsPos);
}
return this.startPos;
}
finishAsRoot(): number {
const off = this.finish();
this.b.setRoot(off);
return off;
}
}
/** ZapMessage is a zero-copy reader over a complete frame. Mirrors zap.Message. */
export class ZapMessage {
private constructor(public readonly data: Buffer) {}
static parse(input: Buffer | Uint8Array): ZapMessage {
const data = Buffer.isBuffer(input) ? input : Buffer.from(input);
if (data.length < HEADER_SIZE) throw new Error("zap: buffer too small");
if (data[0] !== 0x5a || data[1] !== 0x41 || data[2] !== 0x50 || data[3] !== 0x00) {
throw new Error("zap: invalid magic");
}
const version = data.readUInt16LE(4);
if (version !== ZAP_VERSION) throw new Error(`zap: unsupported version ${version}`);
const size = data.readUInt32LE(12);
if (size > data.length) throw new Error("zap: declared size exceeds buffer");
return new ZapMessage(data.subarray(0, size));
}
bytes(): Buffer {
return this.data;
}
flags(): number {
return this.data.readUInt16LE(6);
}
/** Opcode carried in the flags high byte. */
opcode(): number {
return this.flags() >> 8;
}
root(): ZapObject {
const offset = this.data.readUInt32LE(8);
return new ZapObject(this.data, offset);
}
}
/** ZapObject is a zero-copy view into a struct. Mirrors zap.Object. */
export class ZapObject {
constructor(
private readonly data: Buffer,
public readonly offset: number,
) {}
isNull(): boolean {
return this.offset === 0;
}
uint8(fieldOffset: number): number {
const pos = this.offset + fieldOffset;
return pos >= this.data.length ? 0 : this.data[pos];
}
uint16(fieldOffset: number): number {
const pos = this.offset + fieldOffset;
return pos + 2 > this.data.length ? 0 : this.data.readUInt16LE(pos);
}
uint32(fieldOffset: number): number {
const pos = this.offset + fieldOffset;
return pos + 4 > this.data.length ? 0 : this.data.readUInt32LE(pos);
}
bool(fieldOffset: number): boolean {
return this.uint8(fieldOffset) !== 0;
}
bytes(fieldOffset: number): Buffer {
const pos = this.offset + fieldOffset;
if (pos + 4 > this.data.length) return Buffer.alloc(0);
const relOffset = this.data.readInt32LE(pos);
if (relOffset === 0) return Buffer.alloc(0);
const lenPos = pos + 4;
if (lenPos + 4 > this.data.length) return Buffer.alloc(0);
const length = this.data.readUInt32LE(lenPos);
const absPos = pos + relOffset;
if (absPos < 0 || absPos + length > this.data.length) return Buffer.alloc(0);
return this.data.subarray(absPos, absPos + length);
}
text(fieldOffset: number): string {
return this.bytes(fieldOffset).toString("utf8");
}
}
+68
View File
@@ -0,0 +1,68 @@
// Copyright © 2026 Hanzo AI. MIT License.
import { describe, it, expect, beforeAll, afterAll } from "vitest";
import { MockTasksServer } from "./mock-server";
import { TasksClient } from "../src/client";
import { WorkflowStatus } from "../src/client/options";
describe("TasksClient — client RPCs over the real ZAP transport", () => {
let server: MockTasksServer;
let client: TasksClient;
beforeAll(async () => {
server = new MockTasksServer({ queryResult: { count: 7 } });
const port = await server.listen();
client = await TasksClient.connect({ address: `127.0.0.1:${port}`, namespace: "default" });
});
afterAll(async () => {
await client.close();
await server.close();
});
it("connects and reports health over the wire", async () => {
const h = await client.health();
expect(h.service).toBe("hanzo-tasks");
expect(h.status).toBe("ok");
});
it("starts a workflow and reads it back via describe (memo round-trips)", async () => {
const handle = await client.startWorkflow("Greeter", {
taskQueue: "q",
workflowId: "wf-describe",
args: ["bob"],
memo: { source: "test" },
// search attributes are accepted for API compatibility (ignored on v1 wire)
searchAttributes: { CustomKeyword: "abc" },
});
expect(handle.runId).not.toBe("");
const info = await client.describeWorkflow("wf-describe", handle.runId);
expect(info.workflowType).toBe("Greeter");
expect(info.status).toBe(WorkflowStatus.Running);
expect(info.taskQueue).toBe("q");
expect(info.memo).toEqual({ source: "test" });
});
it("queries a workflow and decodes the base64 result", async () => {
const handle = await client.startWorkflow("Greeter", { taskQueue: "q", workflowId: "wf-query" });
const result = await handle.query<{ count: number }>("getState");
expect(result).toEqual({ count: 7 });
});
it("signals a workflow without error", async () => {
const handle = await client.startWorkflow("Greeter", { taskQueue: "q", workflowId: "wf-signal" });
await expect(handle.signal("go", { n: 1 })).resolves.toBeUndefined();
});
it("cancels a workflow and observes the terminal status", async () => {
const handle = await client.startWorkflow("Greeter", { taskQueue: "q", workflowId: "wf-cancel" });
await handle.cancel();
const info = await client.describeWorkflow("wf-cancel", handle.runId);
expect(info.status).toBe(WorkflowStatus.Canceled);
});
it("registers a namespace", async () => {
await expect(client.registerNamespace({ name: "playground", retentionMs: 60000 })).resolves.toBeUndefined();
});
});
+68
View File
@@ -0,0 +1,68 @@
// Copyright © 2026 Hanzo AI. MIT License.
//
// Live integration test against a running tasksd ZAP endpoint. Skipped unless
// TASKS_ZAP_ADDR is set (e.g. "127.0.0.1:9999"). Run tasksd with:
// ./tasksd --allow-no-auth start
// then: TASKS_ZAP_ADDR=127.0.0.1:9999 npx vitest run test/integration.test.ts
import { describe, it, expect, afterAll } from "vitest";
import { TasksClient } from "../src/client";
import { Worker } from "../src/worker";
import { proxyActivities } from "../src/workflow";
import { WorkflowStatus } from "../src/client/options";
const ADDR = process.env.TASKS_ZAP_ADDR;
const NS = process.env.TASKS_NAMESPACE ?? "default";
const suite = ADDR ? describe : describe.skip;
suite("integration — live tasksd over ZAP", () => {
const cleanup: Array<() => Promise<void>> = [];
afterAll(async () => {
for (const fn of cleanup) await fn().catch(() => {});
});
it("reports health", async () => {
const client = await TasksClient.connect({ address: ADDR!, namespace: NS });
cleanup.push(() => client.close());
const h = await client.health();
expect(typeof h.status).toBe("string");
});
it("runs an activity workflow end to end", async () => {
const tq = `ts-sdk-it-${Date.now()}`;
const acts = { echo: async (s: string) => `echo:${s}` };
const worker = await Worker.create({
address: ADDR!,
namespace: NS,
taskQueue: tq,
activities: acts,
workflows: {
Echoer: async (s: string) => {
const a = proxyActivities<typeof acts>({ startToCloseTimeout: "10s" });
return a.echo(s);
},
},
});
await worker.start();
cleanup.push(() => worker.shutdown());
const client = await TasksClient.connect({ address: ADDR!, namespace: NS });
cleanup.push(() => client.close());
const handle = await client.startWorkflow("Echoer", {
taskQueue: tq,
workflowId: `it-${Date.now()}`,
args: ["hi"],
});
const deadline = Date.now() + 20000;
let status = WorkflowStatus.Running;
while (Date.now() < deadline) {
const info = await client.describeWorkflow(handle.workflowId, handle.runId);
status = info.status;
if (status !== WorkflowStatus.Running) break;
await new Promise((r) => setTimeout(r, 250));
}
expect(status).toBe(WorkflowStatus.Completed);
});
});
+394
View File
@@ -0,0 +1,394 @@
// Copyright © 2026 Hanzo AI. MIT License.
//
// MockTasksServer — a faithful, in-memory reimplementation of the tasksd ZAP
// frontend + durable engine, sufficient to exercise the SDK end to end. It
// speaks the exact wire luxfi/zap does: length-framed messages, a node-id
// handshake, reqID/flag correlation for Calls, and bare `Send` frames for
// server-pushed task deliveries. The engine mirrors pkg/tasks: per-run
// event-sourced history, seq-keyed activity scheduling, and re-delivery of
// workflow tasks as progress lands.
import net from "node:net";
import { randomBytes } from "node:crypto";
import {
Builder,
ZapMessage,
encodeResponseEnvelope,
envelopeBody,
decodeFieldObject,
encodeEnvelope,
encodeHeartbeatResponse,
Opcode,
} from "../src/zap";
const REQ_FLAG_REQ = 1;
const REQ_FLAG_RESP = 2;
interface Run {
workflowId: string;
runId: string;
workflowType: string;
taskQueue: string;
history: Array<{ eventId: number; eventType: string; attributes?: Record<string, unknown> }>;
status: number; // 1 running, 2 completed, 3 failed, 4 canceled, 5 terminated
scheduledSeqs: Set<number>;
memo?: Record<string, unknown>;
}
interface Subscriber {
socket: net.Socket;
taskQueue: string;
}
const WorkflowStatus = { Running: 1, Completed: 2, Failed: 3, Canceled: 4, Terminated: 5 };
function encodeHandshake(nodeId: string): Buffer {
const b = new Builder(128);
const obj = b.startObject(64);
const idBytes = Buffer.from(nodeId, "utf8");
for (let i = 0; i < Math.min(idBytes.length, 60); i++) obj.setUint8(i, idBytes[i]);
obj.setUint32(60, idBytes.length >>> 0);
obj.finishAsRoot();
return b.finish();
}
export interface MockOptions {
/** Optional canned query result value returned by QueryWorkflow. */
queryResult?: unknown;
}
export class MockTasksServer {
private readonly server: net.Server;
private readonly runs = new Map<string, Run>();
private readonly byWorkflowId = new Map<string, string>(); // workflowId → latest runId
private readonly wfTokens = new Map<string, string>(); // token → runId
private readonly actTokens = new Map<string, { runId: string; seq: number; activityType: string }>();
private readonly wfSubs: Subscriber[] = [];
private readonly actSubs: Subscriber[] = [];
private readonly sockets = new Set<net.Socket>();
/** Records every RPC opcode the server received (for assertions). */
readonly seenOpcodes: number[] = [];
constructor(private readonly opts: MockOptions = {}) {
this.server = net.createServer((sock) => this.onConnection(sock));
}
listen(): Promise<number> {
return new Promise((resolve) => {
this.server.listen(0, "127.0.0.1", () => {
const addr = this.server.address() as net.AddressInfo;
resolve(addr.port);
});
});
}
async close(): Promise<void> {
for (const s of this.sockets) s.destroy();
await new Promise<void>((resolve) => this.server.close(() => resolve()));
}
getRun(runId: string): Run | undefined {
return this.runs.get(runId);
}
// ── connection handling (server side of node.go) ──
private onConnection(sock: net.Socket): void {
this.sockets.add(sock);
sock.setNoDelay(true);
let acc: Buffer = Buffer.alloc(0);
let handshakeDone = false;
sock.on("data", (chunk) => {
acc = acc.length === 0 ? chunk : Buffer.concat([acc, chunk]);
for (;;) {
if (acc.length < 4) return;
const len = acc.readUInt32LE(0);
if (acc.length < 4 + len) return;
const payload = Buffer.from(acc.subarray(4, 4 + len));
acc = acc.subarray(4 + len);
if (!handshakeDone) {
handshakeDone = true;
this.writeFrame(sock, encodeHandshake("mock-tasksd"));
continue;
}
this.onFrame(sock, payload);
}
});
sock.on("error", () => this.cleanup(sock));
sock.on("close", () => this.cleanup(sock));
}
private cleanup(sock: net.Socket): void {
this.sockets.delete(sock);
const drop = (arr: Subscriber[]) => {
for (let i = arr.length - 1; i >= 0; i--) if (arr[i].socket === sock) arr.splice(i, 1);
};
drop(this.wfSubs);
drop(this.actSubs);
}
private onFrame(sock: net.Socket, data: Buffer): void {
if (data.length < 8) return;
const flag = data.readUInt32LE(4);
if (flag !== REQ_FLAG_REQ) return; // acks / responses from client are ignored
const reqId = data.readUInt32LE(0);
const frame = Buffer.from(data.subarray(8));
let msg: ZapMessage;
try {
msg = ZapMessage.parse(frame);
} catch {
return;
}
const opcode = msg.opcode();
this.seenOpcodes.push(opcode);
const response = this.dispatch(sock, opcode, frame);
this.writeCorrelated(sock, reqId, REQ_FLAG_RESP, response);
}
private dispatch(sock: net.Socket, opcode: number, frame: Buffer): Buffer {
switch (opcode) {
case Opcode.StartWorkflow:
return this.handleStart(opcode, frame);
case Opcode.SignalWorkflow:
return this.handleSignal(opcode, frame);
case Opcode.SignalWithStartWorkflow:
return this.handleStart(opcode, frame, true);
case Opcode.DescribeWorkflow:
return this.handleDescribe(opcode, frame);
case Opcode.QueryWorkflow:
return this.handleQuery(opcode, frame);
case Opcode.CancelWorkflow:
return this.handleTerminal(opcode, frame, WorkflowStatus.Canceled);
case Opcode.TerminateWorkflow:
return this.handleTerminal(opcode, frame, WorkflowStatus.Terminated);
case Opcode.Health:
return encodeResponseEnvelope(opcode, Buffer.from(JSON.stringify({ service: "hanzo-tasks", status: "ok" })));
case Opcode.RegisterNamespace:
return encodeResponseEnvelope(opcode, Buffer.alloc(0));
case Opcode.SubscribeWorkflowTasks:
return this.handleSubscribe(sock, opcode, frame, this.wfSubs);
case Opcode.SubscribeActivityTasks:
return this.handleSubscribe(sock, opcode, frame, this.actSubs);
case Opcode.UnsubscribeTasks:
return encodeResponseEnvelope(opcode, Buffer.alloc(0));
case Opcode.RespondWorkflowTaskCompleted:
return this.handleRespondWorkflow(opcode, frame);
case Opcode.RespondActivityTaskCompleted:
return this.handleRespondActivity(opcode, frame, false);
case Opcode.RespondActivityTaskFailed:
return this.handleRespondActivity(opcode, frame, true);
case Opcode.RecordActivityTaskHeartbeat:
return encodeHeartbeatResponse(false);
default:
return encodeResponseEnvelope(opcode, Buffer.alloc(0), 400, `unknown opcode 0x${opcode.toString(16)}`);
}
}
// ── client RPCs ──
private handleStart(opcode: number, frame: Buffer, withSignal = false): Buffer {
const req = JSON.parse(envelopeBody(frame).toString("utf8"));
const runId = randomHex();
const workflowId = req.workflow_id || `wf-${runId}`;
const run: Run = {
workflowId,
runId,
workflowType: req.workflow_type,
taskQueue: req.task_queue,
history: [
{ eventId: 1, eventType: "WORKFLOW_EXECUTION_STARTED", attributes: { input: req.input ?? [] } },
],
status: WorkflowStatus.Running,
scheduledSeqs: new Set(),
memo: req.memo,
};
if (withSignal && req.signal_name) {
run.history.push({
eventId: run.history.length + 1,
eventType: "WORKFLOW_EXECUTION_SIGNALED",
attributes: { signal: req.signal_name, input: req.signal_input },
});
}
this.runs.set(runId, run);
this.byWorkflowId.set(workflowId, runId);
setImmediate(() => this.pushWorkflowTask(run));
return encodeResponseEnvelope(opcode, Buffer.from(JSON.stringify({ run_id: runId })));
}
private handleSignal(opcode: number, frame: Buffer): Buffer {
const req = JSON.parse(envelopeBody(frame).toString("utf8"));
const run = this.resolveRun(req.workflow_id, req.run_id);
if (run && run.status === WorkflowStatus.Running) {
run.history.push({
eventId: run.history.length + 1,
eventType: "WORKFLOW_EXECUTION_SIGNALED",
attributes: { signal: req.signal_name, input: req.input },
});
setImmediate(() => this.pushWorkflowTask(run));
}
return encodeResponseEnvelope(opcode, Buffer.alloc(0));
}
private handleDescribe(opcode: number, frame: Buffer): Buffer {
const req = JSON.parse(envelopeBody(frame).toString("utf8"));
const run = this.resolveRun(req.workflow_id, req.run_id);
if (!run) return encodeResponseEnvelope(opcode, Buffer.alloc(0), 404, "not found");
const info = {
workflow_id: run.workflowId,
run_id: run.runId,
workflow_type: run.workflowType,
status: run.status,
history_length: run.history.length,
task_queue: run.taskQueue,
memo: run.memo,
};
return encodeResponseEnvelope(opcode, Buffer.from(JSON.stringify({ info })));
}
private handleQuery(opcode: number, frame: Buffer): Buffer {
// Canned response: result is a Go []byte (base64) holding the value JSON.
const value = this.opts.queryResult ?? null;
const resultB64 = Buffer.from(JSON.stringify(value), "utf8").toString("base64");
return encodeResponseEnvelope(opcode, Buffer.from(JSON.stringify({ result: resultB64 })));
}
private handleTerminal(opcode: number, frame: Buffer, status: number): Buffer {
const req = JSON.parse(envelopeBody(frame).toString("utf8"));
const run = this.resolveRun(req.workflow_id, req.run_id);
if (run) run.status = status;
return encodeResponseEnvelope(opcode, Buffer.alloc(0));
}
private handleSubscribe(sock: net.Socket, opcode: number, frame: Buffer, list: Subscriber[]): Buffer {
const req = JSON.parse(envelopeBody(frame).toString("utf8"));
const subId = randomHex();
list.push({ socket: sock, taskQueue: req.task_queue });
return encodeResponseEnvelope(opcode, Buffer.from(JSON.stringify({ subscription_id: subId })));
}
// ── worker responses ──
private handleRespondWorkflow(opcode: number, frame: Buffer): Buffer {
const { token, payload } = decodeFieldObject(frame);
const runId = this.wfTokens.get(token.toString("utf8"));
const run = runId ? this.runs.get(runId) : undefined;
if (!run) return encodeResponseEnvelope(opcode, Buffer.alloc(0));
const env = JSON.parse(payload.toString("utf8")) as { v: number; cmds: any[] };
for (const cmd of env.cmds ?? []) {
if (cmd.kind === 2) {
const seq: number = cmd.seq ?? 0;
if (run.scheduledSeqs.has(seq)) continue;
run.scheduledSeqs.add(seq);
const argsJSON = cmd.input ? Buffer.from(cmd.input, "base64").toString("utf8") : "[]";
run.history.push({
eventId: run.history.length + 1,
eventType: "ACTIVITY_TASK_SCHEDULED",
attributes: { seq, activityType: cmd.activityType, input: JSON.parse(argsJSON) },
});
this.pushActivityTask(run, seq, cmd.activityType, argsJSON, cmd.taskQueue || run.taskQueue, cmd);
} else if (cmd.kind === 0) {
run.status = WorkflowStatus.Completed;
} else if (cmd.kind === 1) {
run.status = WorkflowStatus.Failed;
}
}
return encodeResponseEnvelope(opcode, Buffer.alloc(0));
}
private handleRespondActivity(opcode: number, frame: Buffer, failed: boolean): Buffer {
const { token, payload } = decodeFieldObject(frame);
const info = this.actTokens.get(token.toString("utf8"));
const run = info ? this.runs.get(info.runId) : undefined;
if (!run || !info) return encodeResponseEnvelope(opcode, Buffer.alloc(0));
if (failed) {
const failureVal = payload.length > 0 ? JSON.parse(payload.toString("utf8")) : null;
run.history.push({
eventId: run.history.length + 1,
eventType: "ACTIVITY_TASK_FAILED",
attributes: { seq: info.seq, failure: failureVal },
});
} else {
const resultVal = payload.length > 0 ? JSON.parse(payload.toString("utf8")) : null;
run.history.push({
eventId: run.history.length + 1,
eventType: "ACTIVITY_TASK_COMPLETED",
attributes: { seq: info.seq, result: resultVal },
});
}
if (run.status === WorkflowStatus.Running) setImmediate(() => this.pushWorkflowTask(run));
return encodeResponseEnvelope(opcode, Buffer.alloc(0));
}
// ── deliveries (Send) ──
private pushWorkflowTask(run: Run): void {
if (run.status !== WorkflowStatus.Running) return;
const sub = this.wfSubs.find((s) => s.taskQueue === run.taskQueue);
if (!sub) return;
const token = randomHex();
this.wfTokens.set(token, run.runId);
const body = {
task_token: token,
workflow_id: run.workflowId,
run_id: run.runId,
workflow_type_name: run.workflowType,
history: JSON.stringify(run.history),
};
this.writeFrame(sub.socket, encodeEnvelope(Opcode.DeliverWorkflowTask, Buffer.from(JSON.stringify(body))));
}
private pushActivityTask(
run: Run,
seq: number,
activityType: string,
argsJSON: string,
taskQueue: string,
cmd: any,
): void {
const sub = this.actSubs.find((s) => s.taskQueue === taskQueue);
if (!sub) return;
const token = randomHex();
this.actTokens.set(token, { runId: run.runId, seq, activityType });
const body = {
task_token: token,
workflow_id: run.workflowId,
run_id: run.runId,
activity_id: `act-${seq}`,
activity_type_name: activityType,
input: argsJSON,
scheduled_time_ms: Date.now(),
start_to_close_timeout_ms: cmd.startToCloseMs ?? 0,
heartbeat_timeout_ms: cmd.heartbeatMs ?? 0,
};
this.writeFrame(sub.socket, encodeEnvelope(Opcode.DeliverActivityTask, Buffer.from(JSON.stringify(body))));
}
private resolveRun(workflowId: string, runId?: string): Run | undefined {
if (runId) return this.runs.get(runId);
const latest = this.byWorkflowId.get(workflowId);
return latest ? this.runs.get(latest) : undefined;
}
// ── framing ──
private writeFrame(sock: net.Socket, data: Buffer): void {
const len = Buffer.allocUnsafe(4);
len.writeUInt32LE(data.length >>> 0, 0);
sock.write(len);
sock.write(data);
}
private writeCorrelated(sock: net.Socket, reqId: number, flag: number, frame: Buffer): void {
const wrapped = Buffer.allocUnsafe(8 + frame.length);
wrapped.writeUInt32LE(reqId >>> 0, 0);
wrapped.writeUInt32LE(flag >>> 0, 4);
frame.copy(wrapped, 8);
this.writeFrame(sock, wrapped);
}
}
function randomHex(): string {
return randomBytes(12).toString("hex");
}
+146
View File
@@ -0,0 +1,146 @@
// Copyright © 2026 Hanzo AI. MIT License.
//
// End-to-end durable loop over the real ZAP transport: client starts a
// workflow, the mock server delivers a workflow task, the worker replays and
// schedules an activity, the server delivers the activity task, the worker
// runs it and responds, the server appends the result and re-delivers the
// workflow task, and the workflow completes.
import { describe, it, expect, beforeEach, afterEach } from "vitest";
import { MockTasksServer } from "./mock-server";
import { TasksClient } from "../src/client";
import { Worker } from "../src/worker";
import { proxyActivities, sleep } from "../src/workflow";
async function waitForStatus(server: MockTasksServer, runId: string, status: number, timeoutMs = 8000): Promise<void> {
const deadline = Date.now() + timeoutMs;
for (;;) {
const run = server.getRun(runId);
if (run && run.status === status) return;
if (Date.now() > deadline) {
throw new Error(`timeout waiting for status ${status}; got ${run?.status}`);
}
await new Promise((r) => setTimeout(r, 20));
}
}
describe("Worker — full durable activity round-trip", () => {
let server: MockTasksServer;
let client: TasksClient;
let worker: Worker;
let port: number;
beforeEach(async () => {
server = new MockTasksServer();
port = await server.listen();
client = await TasksClient.connect({ address: `127.0.0.1:${port}` });
});
afterEach(async () => {
await worker?.shutdown();
await client.close();
await server.close();
});
it("runs an activity and completes the workflow", async () => {
const greeting = { greet: async (name: string) => `hello ${name}` };
worker = await Worker.create({
address: `127.0.0.1:${port}`,
taskQueue: "q",
activities: greeting,
workflows: {
Greeter: async (name: string) => {
const acts = proxyActivities<typeof greeting>({ startToCloseTimeout: "5s" });
return acts.greet(name);
},
},
});
await worker.start();
const handle = await client.startWorkflow("Greeter", { taskQueue: "q", workflowId: "g1", args: ["bob"] });
await waitForStatus(server, handle.runId, 2 /* Completed */);
const run = server.getRun(handle.runId)!;
const completed = run.history.find((e) => e.eventType === "ACTIVITY_TASK_COMPLETED");
expect(completed?.attributes?.result).toBe("hello bob");
});
it("propagates an activity failure to the workflow", async () => {
worker = await Worker.create({
address: `127.0.0.1:${port}`,
taskQueue: "q",
activities: {
boom: async () => {
throw new Error("kaboom");
},
},
workflows: {
Boomer: async () => {
const acts = proxyActivities<{ boom(): Promise<void> }>({ startToCloseTimeout: "5s" });
await acts.boom();
return "unreached";
},
},
});
await worker.start();
const handle = await client.startWorkflow("Boomer", { taskQueue: "q", workflowId: "b1" });
await waitForStatus(server, handle.runId, 3 /* Failed */);
const run = server.getRun(handle.runId)!;
const failed = run.history.find((e) => e.eventType === "ACTIVITY_TASK_FAILED");
expect(failed).toBeDefined();
});
it("fires a durable timer (activity-backed sleep) and completes", async () => {
worker = await Worker.create({
address: `127.0.0.1:${port}`,
taskQueue: "q",
workflows: {
Sleeper: async () => {
await sleep(30);
return "awake";
},
},
});
await worker.start();
const handle = await client.startWorkflow("Sleeper", { taskQueue: "q", workflowId: "s1" });
await waitForStatus(server, handle.runId, 2 /* Completed */);
const run = server.getRun(handle.runId)!;
const timer = run.history.find(
(e) => e.eventType === "ACTIVITY_TASK_SCHEDULED" && e.attributes?.activityType === "__hanzo_timer__",
);
expect(timer).toBeDefined();
});
it("chains two activities in program order (seq 0 then seq 1)", async () => {
const acts = {
double: async (n: number) => n * 2,
inc: async (n: number) => n + 1,
};
worker = await Worker.create({
address: `127.0.0.1:${port}`,
taskQueue: "q",
activities: acts,
workflows: {
Pipe: async (n: number) => {
const a = proxyActivities<typeof acts>({ startToCloseTimeout: "5s" });
const d = await a.double(n);
return a.inc(d);
},
},
});
await worker.start();
const handle = await client.startWorkflow("Pipe", { taskQueue: "q", workflowId: "p1", args: [10] });
await waitForStatus(server, handle.runId, 2 /* Completed */);
const run = server.getRun(handle.runId)!;
const results = run.history
.filter((e) => e.eventType === "ACTIVITY_TASK_COMPLETED")
.map((e) => e.attributes?.result);
expect(results).toEqual([20, 21]);
});
});
+154
View File
@@ -0,0 +1,154 @@
// Copyright © 2026 Hanzo AI. MIT License.
import { describe, it, expect } from "vitest";
import { runWorkflowEpisode } from "../src/worker";
import {
proxyActivities,
sleep,
condition,
defineSignal,
setHandler,
} from "../src/workflow";
import { CommandKind } from "../src/worker";
import { TIMER_ACTIVITY_TYPE } from "../src/workflow/runtime";
import type { WorkflowInfo } from "../src/workflow";
const info: WorkflowInfo = {
workflowId: "w1",
runId: "r1",
workflowType: "T",
taskQueue: "q",
namespace: "default",
attempt: 1,
};
function started(input: unknown[]) {
return [{ eventId: 1, eventType: "WORKFLOW_EXECUTION_STARTED", attributes: { input } }];
}
function b64(s: string): string {
return Buffer.from(s, "utf8").toString("base64");
}
describe("workflow decider — event-sourced replay produces exact wire commands", () => {
it("emits a ScheduleActivity command when the activity is not yet in history", async () => {
const wf = async (n: number) => {
const acts = proxyActivities<{ stepA(n: number): Promise<number> }>({ startToCloseTimeout: 1000 });
const a = await acts.stepA(n);
return a + 1;
};
const cmds = await runWorkflowEpisode(wf, info, JSON.stringify(started([5])));
expect(cmds).toHaveLength(1);
expect(cmds[0].kind).toBe(CommandKind.ScheduleActivity);
expect(cmds[0].seq).toBe(0);
expect(cmds[0].activityType).toBe("stepA");
expect(Buffer.from(cmds[0].input!, "base64").toString("utf8")).toBe("[5]");
});
it("completes the workflow once the activity result is in history", async () => {
const wf = async (n: number) => {
const acts = proxyActivities<{ stepA(n: number): Promise<number> }>({ startToCloseTimeout: 1000 });
const a = await acts.stepA(n);
return a + 1;
};
const history = [
...started([5]),
{ eventId: 2, eventType: "ACTIVITY_TASK_SCHEDULED", attributes: { seq: 0 } },
{ eventId: 3, eventType: "ACTIVITY_TASK_COMPLETED", attributes: { seq: 0, result: 10 } },
];
const cmds = await runWorkflowEpisode(wf, info, JSON.stringify(history));
expect(cmds).toHaveLength(1);
expect(cmds[0].kind).toBe(CommandKind.CompleteWorkflow);
expect(Buffer.from(cmds[0].result!, "base64").toString("utf8")).toBe("11");
});
it("does not re-schedule an activity already scheduled in history", async () => {
const wf = async () => {
const acts = proxyActivities<{ stepA(): Promise<number> }>({ startToCloseTimeout: 1000 });
await acts.stepA();
return "done";
};
const history = [
...started([]),
{ eventId: 2, eventType: "ACTIVITY_TASK_SCHEDULED", attributes: { seq: 0 } },
];
const cmds = await runWorkflowEpisode(wf, info, JSON.stringify(history));
// scheduled but not yet complete → parked, no new commands
expect(cmds).toHaveLength(0);
});
it("propagates an activity failure into a workflow failure", async () => {
const wf = async () => {
const acts = proxyActivities<{ boom(): Promise<void> }>({ startToCloseTimeout: 1000 });
await acts.boom();
return "unreached";
};
const history = [
...started([]),
{ eventId: 2, eventType: "ACTIVITY_TASK_SCHEDULED", attributes: { seq: 0 } },
{
eventId: 3,
eventType: "ACTIVITY_TASK_FAILED",
attributes: { seq: 0, failure: { v: 1, p: { message: "kaboom", code: "ApplicationError", nonRetryable: true } } },
},
];
const cmds = await runWorkflowEpisode(wf, info, JSON.stringify(history));
expect(cmds).toHaveLength(1);
expect(cmds[0].kind).toBe(CommandKind.FailWorkflow);
const failure = JSON.parse(Buffer.from(cmds[0].failure!, "base64").toString("utf8"));
expect(failure.p.message).toBe("kaboom");
});
it("schedules a durable timer via the internal sleeper activity", async () => {
const wf = async () => {
await sleep(1000);
return "woke";
};
const cmds = await runWorkflowEpisode(wf, info, JSON.stringify(started([])));
expect(cmds).toHaveLength(1);
expect(cmds[0].kind).toBe(CommandKind.ScheduleActivity);
expect(cmds[0].activityType).toBe(TIMER_ACTIVITY_TYPE);
expect(Buffer.from(cmds[0].input!, "base64").toString("utf8")).toBe("[1000]");
});
it("parks on condition until a signal flips the predicate", async () => {
const goSignal = defineSignal("go");
const wf = async () => {
let done = false;
setHandler(goSignal, () => {
done = true;
});
await condition(() => done);
return "ok";
};
// Episode 1: no signal → parked, no commands.
const first = await runWorkflowEpisode(wf, info, JSON.stringify(started([])));
expect(first).toHaveLength(0);
// Episode 2: signal delivered → condition resolves → workflow completes.
const withSignal = [
...started([]),
{ eventId: 2, eventType: "WORKFLOW_EXECUTION_SIGNALED", attributes: { signal: "go", input: undefined } },
];
const second = await runWorkflowEpisode(wf, info, JSON.stringify(withSignal));
expect(second).toHaveLength(1);
expect(second[0].kind).toBe(CommandKind.CompleteWorkflow);
});
it("survives user try/catch around a parked activity (block escapes catch)", async () => {
const wf = async () => {
const acts = proxyActivities<{ stepA(): Promise<number> }>({ startToCloseTimeout: 1000 });
try {
await acts.stepA();
} catch {
return "swallowed"; // must NOT run on a block
}
return "reached";
};
const cmds = await runWorkflowEpisode(wf, info, JSON.stringify(started([])));
// Parked on the activity — no completion command leaked through the catch.
expect(cmds.some((c) => c.kind === CommandKind.ScheduleActivity)).toBe(true);
expect(cmds.some((c) => c.kind === CommandKind.CompleteWorkflow)).toBe(false);
});
});
+93
View File
@@ -0,0 +1,93 @@
// Copyright © 2026 Hanzo AI. MIT License.
import { describe, it, expect } from "vitest";
import {
Builder,
ZapMessage,
encodeEnvelope,
decodeEnvelope,
encodeResponseEnvelope,
encodeFieldObject,
decodeFieldObject,
frameBody,
encodeHeartbeatResponse,
decodeHeartbeatResponse,
} from "../src/zap";
describe("ZAP wire — byte-exact layout (parity with luxfi/zap v0.2.0)", () => {
it("encodes a JSON envelope with the exact Go byte layout", () => {
const body = Buffer.from("{}", "utf8");
const opcode = 0x0060;
const frame = encodeEnvelope(opcode, body);
// header
expect(frame.subarray(0, 4).toString("latin1")).toBe("ZAP\x00");
expect(frame.readUInt16LE(4)).toBe(1); // version
expect(frame.readUInt16LE(6)).toBe((opcode << 8) & 0xffff); // flags carry opcode
expect(frame.readUInt32LE(8)).toBe(16); // root offset = HeaderSize
expect(frame.readUInt32LE(12)).toBe(42); // total size = 16 header + 24 obj + 2 body
// object: bytes field 0 → relOffset 24, length 2; payload appended at 40
expect(frame.readInt32LE(16)).toBe(24);
expect(frame.readUInt32LE(20)).toBe(2);
expect(frame.subarray(40, 42).toString("utf8")).toBe("{}");
expect(frame.length).toBe(42);
});
it("round-trips a response envelope with status + error detail", () => {
const frame = encodeResponseEnvelope(0x0064, Buffer.from('{"ok":true}'), 500, "boom");
const { status, detail, body } = decodeEnvelope(frame);
expect(status).toBe(500);
expect(detail).toBe("boom");
expect(JSON.parse(body.toString("utf8"))).toEqual({ ok: true });
});
it("round-trips a field object (token @0, payload @8)", () => {
const token = Buffer.from("abc123", "utf8");
const payload = Buffer.from('{"v":1,"cmds":[]}', "utf8");
const frame = encodeFieldObject(token, payload);
const decoded = decodeFieldObject(frame);
expect(decoded.token.toString("utf8")).toBe("abc123");
expect(decoded.payload.toString("utf8")).toBe('{"v":1,"cmds":[]}');
});
it("frameBody re-stamps the opcode flags on an already-framed body", () => {
const framed = encodeFieldObject(Buffer.from("t"), Buffer.from("p"));
const restamped = frameBody(0x00a2, framed);
expect(ZapMessage.parse(restamped).opcode()).toBe(0x00a2);
// still decodes as a field object
const d = decodeFieldObject(restamped);
expect(d.token.toString()).toBe("t");
expect(d.payload.toString()).toBe("p");
});
it("frameBody wraps a raw JSON body in the envelope", () => {
const wrapped = frameBody(0x0060, Buffer.from('{"a":1}'));
expect(decodeEnvelope(wrapped).body.toString("utf8")).toBe('{"a":1}');
});
it("empty bytes fields become null pointers", () => {
const frame = encodeEnvelope(0x0090, null);
const { body } = decodeEnvelope(frame);
expect(body.length).toBe(0);
});
it("heartbeat response carries the cancelRequested bool", () => {
expect(decodeHeartbeatResponse(encodeHeartbeatResponse(true))).toBe(true);
expect(decodeHeartbeatResponse(encodeHeartbeatResponse(false))).toBe(false);
});
it("Builder/Reader round-trip scalars and text at fixed offsets", () => {
const b = new Builder(64);
const obj = b.startObject(24);
obj.setUint32(8, 0xdeadbeef);
obj.setText(0, "hello");
obj.setBool(16, true);
obj.finishAsRoot();
const msg = ZapMessage.parse(b.finish());
const root = msg.root();
expect(root.uint32(8)).toBe(0xdeadbeef);
expect(root.text(0)).toBe("hello");
expect(root.bool(16)).toBe(true);
});
});
+21
View File
@@ -0,0 +1,21 @@
{
"compilerOptions": {
"target": "ES2022",
"lib": ["ES2022"],
"module": "ESNext",
"moduleResolution": "Bundler",
"declaration": true,
"strict": true,
"noUncheckedIndexedAccess": false,
"esModuleInterop": true,
"skipLibCheck": true,
"forceConsistentCasingInFileNames": true,
"resolveJsonModule": true,
"isolatedModules": true,
"verbatimModuleSyntax": false,
"types": ["node"],
"outDir": "dist"
},
"include": ["src", "test"],
"exclude": ["dist", "node_modules"]
}
+14
View File
@@ -0,0 +1,14 @@
import { defineConfig } from "tsup";
export default defineConfig({
entry: ["src/index.ts"],
format: ["esm", "cjs"],
dts: true,
sourcemap: true,
clean: true,
target: "es2022",
platform: "node",
outExtension({ format }) {
return { js: format === "cjs" ? ".cjs" : ".js" };
},
});
+10
View File
@@ -0,0 +1,10 @@
import { defineConfig } from "vitest/config";
export default defineConfig({
test: {
include: ["test/**/*.test.ts"],
environment: "node",
testTimeout: 15000,
hookTimeout: 15000,
},
});