The social-orchestrator migration off upstream Temporal onto the embedded
Hanzo Tasks engine needs five workflow primitives its digest/sendEmail/post
workflows depend on. Implemented server-side (event-sourced, replay-safe) plus
the @hanzoai/tasks TS SDK (the worker social runs):
- Signal → running workflow: engine.signalWorkflow now re-dispatches a workflow
task after appending WORKFLOW_EXECUTION_SIGNALED, so the stateless worker
replays the extended history and fires the handler (digest/sendEmail are
entirely signal-driven). TS decider read the wrong history attr key
("signal" vs "signalName") — fixed.
- continueAsNew (kind=4): closes the run CONTINUED_AS_NEW + starts a successor
under the same workflowId carrying input/search-attrs/memo. TS continueAsNew
emits the command instead of throwing ContinueAsNewNotSupported.
- startChild (kind=5): detached (ABANDON) child, seq-keyed + idempotent; retires
the 0x006D 501 for the command path. TS startChild/executeChild added.
- Typed search attributes: carried on StartWorkflow/SignalWithStart wire, stored
on the execution, visibility-queryable (already read by fieldValue). TS adds
TypedSearchAttributes/defineSearchAttributeKey/SearchAttributeType.
- workflowIdConflictPolicy USE_EXISTING/TERMINATE_EXISTING/FAIL.
- ActivityFailure wraps ApplicationFailure (with .type) so workflow error
matching (err.cause.type === 'refresh_token') works.
Also fixes a pre-existing wire defect: the server registered
SignalWithStartWorkflow at 0x0069 while every client sends 0x0066 (schema +
Go/TS SDK) — signalWithStart 404'd against the real engine. Swapped with the
HTTP-only GetWorkflowHistory (no ZAP client).
DescribeWorkflow(workflowId, "") now resolves the OPEN run (Temporal's current
execution) so a continueAsNew successor is the one signalled/described.
Tests: engine unit (search attrs, conflict policy, continueAsNew, startChild)
+ TS e2e vs real tasksd (signals-across-continueAsNew, startChild, search attrs
run to COMPLETED). go test -race ./pkg/... green; TS vitest green.
5.5 KiB
@hanzoai/tasks — TypeScript SDK
Durable workflow execution for Hanzo 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
npm add @hanzoai/tasks
Client
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.*):
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
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:
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
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
Wired and tested against the real ZAP transport (Go engine ≥ v1.49.0):
| Primitive | Status |
|---|---|
| Client: start / signal / signalWithStart / query / describe / cancel / terminate / list | ✅ complete |
| Client: namespace register/describe/list, health | ✅ complete |
Client: searchAttributes + workflowIdConflictPolicy on start / signalWithStart |
✅ 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 — signal delivered to a running workflow |
✅ complete |
continueAsNew — closes the run, server starts the successor |
✅ complete |
startChild / executeChild — detached (ABANDON) child workflow |
✅ complete |
Typed search attributes (TypedSearchAttributes, defineSearchAttributeKey) — stored + visibility-queryable |
✅ complete |
Retry policy, activity failure propagation (ActivityFailure→ApplicationFailure), cancellation |
✅ complete |
@temporalio/* shim: Connection, Client, NativeConnection, Worker, ApplicationFailure, ActivityFailure |
✅ complete |
Remaining:
- Durable timers as a native command.
sleep/startTimerare 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 nativeTIMER_STARTED/TIMER_FIREDserver command removes the caveat. ¹ - Child result await.
startChild/executeChildare ABANDON (detached): they resolve on the child's start, not its completion — the parent does not block on the child's result. This matches how social uses repeat-post children. WorkflowHandle.result()value — resolvesundefinedon completion becauseDescribeWorkflowResponsecarries 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-sidequery()RPC works.
License
MIT.