The tasks client and the SDK client dial the Tasks server and are never
started, so they bind no socket. Responses and server-pushed deliveries
arrive on the connection the client opened; with discovery off nothing
dials back.
TestZAPClientsListenNowhere asks the OS which sockets the process
listens on: the server adds one, both clients add none.
Co-authored-by: Hanzo Dev <[email protected]>
The engine took a PORT and the SDK an Options.HostPort, so neither end
could use a unix socket for traffic that never leaves the host. The
field name was not the only obstacle: luxfi/zap bound net.Listen("tcp",
":port") and reached a peer with net.DialTimeout("tcp", addr), so a path
was refused at the dialer with "missing port in address" and could not
be bound at all. That is fixed upstream (luxfi/zap Address + Network);
this pins the commit until it is tagged.
EmbedConfig.Address, Embedded.Address, ServeGated(addr) and
client.Options.Address each take ONE address, because a port is a detail
of one transport rather than a thing every listener has. A path binds or
dials a unix socket; anything else is host:port.
TestDialRequiresHostPort became TestDialRequiresAddress: Dial needs an
address, and asserts a socket path is refused for being absent rather
than for being portless. TestE2E_UnixSocket runs a real workflow over a
socket end to end, and TestE2E_UnixSocketRebind proves the socket is
unlinked on stop so a restart binds cleanly.
Co-authored-by: Hanzo Dev <[email protected]>
Enables consolidating the standalone tasksd onto an in-process embedded engine
(cloud) by exposing it cluster-wide under mandatory identity gating, and by
fixing two gaps that made identity-gated ZAP durable execution impossible.
Embedded.ServeGated(ctx, port, *auth.Validator): starts a SECOND ZAP listener
that exposes the SAME engine to out-of-process callers with RequireIdentity --
every request must carry an auth_token that validates against IAM, org-scoped to
the token owner. The loopback listener from Embed is unchanged (in-process,
trusted). Server-push is routed to the listener holding the worker's
subscription, so gated and loopback workers both receive delivery. A nil
validator is refused (never expose durable execution ungated).
SDK: Options.Token TokenSource attaches auth_token to every JSON-body RPC
(client lifecycle + worker Subscribe) at the transport; binary respond frames
stay task_token-gated. This is the missing client half -- the frontend already
read req["auth_token"], but no SDK ever set it, so a RequireIdentity frontend
was unreachable (every caller 401'd). StaticToken helper for fixed bearers;
client_credentials callers return a fresh token per call so long-lived workers
never present an expired identity.
Org-scoped completion: the respond handlers ran on rootEn (unscoped) while
org-scoped runs live in an org-prefixed store partition, so an org-scoped
workflow was accepted (RUNNING) but never advanced. Thread orgID through the
dispatcher's pending tasks and apply completion via en.WithOrg(orgID). orgID==""
(loopback/unscoped) makes WithOrg a no-op -- the in-process path is unchanged.
RegisterNamespace over ZAP: accept the SDK's flat wire ({"name",...}); the nested
{"namespaceInfo":{...}} unmarshal never matched what the client sends, so an
org-scoped caller could not register the namespace ExecuteWorkflow requires.
Test: TestE2E_ServeGated_DurableWorkflow_RequiresIdentity drives a real durable
workflow (schedules an activity, returns its result) to COMPLETED over the gated
listener with a JWKS-signed token, and asserts an unauthenticated caller is
refused. Full suite green.
Co-authored-by: Hanzo Dev <[email protected]>
Replace the coroutine-push activity path with deterministic, history-backed
replay. One way: the old push-result mechanism is removed.
Server (pkg/tasks):
- durable.go: seq-keyed activity records (wfact/ns/wf/run/seq) — the
exactly-once anchor. applyScheduleActivity is idempotent per (run,seq);
completeWorkflowActivity records the terminal outcome and schedules a new
workflow task so the worker replays and advances. Per-run keyedMutex
serializes advancing a run. Server-side retry+backoff per RetryPolicy
(temporal.RetryPolicy, finite default bound). Recover() rebuilds in-flight
dispatch from the store at boot (called from Embed).
- workflow tasks now carry the run's full event-sourced history.
- dispatch.go: ScheduleActivity/CompleteActivityTask(push) -> DispatchActivity
(seq-bound token) + ResolveActivityToken. workflowPeer/pendingActivity and
OpcodeDeliverActivityResult retired.
- respondWorkflowHandler applies ScheduleActivity{seq} commands under the run
lock; respondActivity* resolve the token and advance the run.
Worker (pkg/sdk/worker):
- workerEnv is now a replay decider: ExecuteActivity resolves from history by
seq (recorded result -> no re-dispatch; else emit ScheduleActivity{seq} and
end the episode via a blockedFuture). Stateless between episodes.
- removed OnActivityResult/pendingActivities push machinery.
Client (pkg/sdk/client): drop OnActivityResult + ScheduleActivity from
WorkerTransport (dead in the replay model).
Tests: e2e (exactly-once, real workers) green; new worker replay determinism
tests (unresolved->schedule+block, resolved->recorded result, 2-activity
determinism). Full 'go test -race ./pkg/tasks/... ./pkg/sdk/...' green.
- newZAPTransport now adds a random suffix to client NodeID so
multiple worker pods don't collide on the server's peer table.
- dispatcher.go gains DISPATCH_TRACE=1 env-gated stderr logs for
Subscribe and EnqueueWorkflowTask, helping triage delivery
routing without recompiling.
pkg/sdk/examples/embed.go — boots tasks.Embed in-process on an
ephemeral port, dials it via pkg/sdk/client, round-trips Health →
ExecuteWorkflow → ListWorkflows. This is the canonical "single
binary" pattern Hanzo Base apps use when TASKS_EMBED=true.
pkg/sdk/examples/realtime.go — exercises the full realtime pipeline:
embedded daemon → engine broker → SSE handler → client.SubscribeEvents
→ caller channel. Starts 3 workflows, cancels even-indexed, asserts
events arrived on the SDK channel. Proves the realtime path works
end-to-end without HTTP proxy or gRPC.
Tests:
TestHealth PASS
TestHello PASS
TestSignal PASS
TestCron PASS
TestEmbed PASS (NEW — Base TASKS_EMBED path)
TestRealtime PASS (NEW — SSE → SDK channel pipeline)
All over native ZAP. Zero gRPC. Zero go.temporal.io. Zero proxy.
ui/src/lib/events.ts — useRealtime() React hook subscribes to
/v1/tasks/events via EventSource and invalidates SWR caches when
workflow / schedule / batch / namespace transitions affect the
current view. No polling. No proxy. Native SSE only.
Wired into:
Workflows page — live workflow.{started,canceled,terminated,signaled}
Schedules page — live schedule.{created,paused,resumed,deleted}
Batches page — live batch.started
Namespaces page — live namespace.registered
pkg/sdk/client/events.go — Go callers get the same stream natively
via client.SubscribeEvents(ctx, httpAddr) → <-chan Event. No SDK
spin-up, no JSON-RPC overhead — long-lived TCP read with inline SSE
parsing. End-to-end verified with a /tmp demo: two POST workflow
starts hit the channel in real time as workflow.started events.
README.md — drop the stale gRPC compat reference. The wire is ZAP
plus an HTTP/JSON shim plus SSE + MCP for browsers and agents. No
gRPC, no go.temporal.io, anywhere.
Bundle 320.7 KB (+0.6 from 320.1). Binary 7.6 MB unchanged.
4/4 integration tests still pass.
pkg/sdk/examples/{hello,signal,cron}.go drive the engine end-to-end
through pkg/sdk/client over the ZAP wire — zero HTTP, zero
go.temporal.io. Tests boot tasks.Embed in-process, dial the local
ZAP node, and round-trip every workflow + schedule opcode.
Engine fixes (pkg/tasks/embed.go):
- ZAP handlers now read the SDK's snake_case JSON shape
(workflow_id / workflow_type / task_queue / signal_name / schedule_id)
in addition to the camelCase shape used by the HTTP shim
- DescribeWorkflow returns {info: WorkflowExecutionInfo} (SDK shape)
- WorkflowExecutionInfo emits status as int8 (1=RUNNING / 4=CANCELED / 5=TERMINATED)
- StartWorkflow returns {run_id, workflow_id}
- PauseSchedule respects the requested paused flag (was always true)
- Schedule list response goes through scheduleSDK marshaller
- Empty time.Time fields are omitted (SDK can't parse "")
Store fix:
- store.list() snapshots under RLock then iterates without lock,
preventing the RWMutex upgrade deadlock that hung GET /v1/tasks/...
whenever the cron sweeper's put() was waiting on a Lock that
competed with reader RLocks
go test ./pkg/sdk/examples/... → 4/4 PASS:
TestHealth, TestHello, TestSignal, TestCron
pkg/sdk/inproc.Transport implements pkg/sdk/client.Transport by
dispatching synchronously into a frontend.ZAPHandler. No network
listener bind, no ZAP node start, no encode→frame→decode round trip.
Wire-compatible with the network path: every opcode reaches the same
handler function.
This is the Wave 1.5 keystone groundwork. The cascade — replacing every
common/sdk.ClientFactory consumer (history self-dial, system workers,
temporaltest) and deleting the bound :7233 gRPC listener in
service/frontend/service.go — lands in a follow-up.
Tests:
- pkg/sdk/inproc: 5 unit tests covering envelope/framed bodies,
close idempotency, dispatcher error propagation, transport conflict
rejection.
- service/frontend: 4 integration tests wiring inproc.NewClient against
the real *ZAPHandler (Health, StartWorkflow, DescribeWorkflow,
unknown-opcode error path).
go test ./pkg/sdk/inproc/... ./service/frontend/... — green.
ZAP binary protocol duplex RPC standardizes on :9999 across every
service. Lux validator staking/RPC ports (9630/9631 mainnet, 9650/9651
devnet) are unchanged — those are luxd network protocol ports, not ZAP.
One ZAP port everywhere (was a mix of 9651/9652/9653/9633/9900).
Three new opcodes (0x0074-0x0076) wired through the frontend ZAP
handler, along with a thin Unpause alias over PauseSchedule(false).
Update replaces the schedule definition wholesale (partial updates
not on the v1 wire — same shape as upstream UpdateSchedule).
Trigger fires immediately, ignoring the schedule's spec; overlap_policy
maps to schedulepb.ScheduleOverlapPolicy via triggerOverlapEnum so all
six policy enums round-trip.
Describe returns the live schedule body + ScheduleInfo (action_count,
missed_catchup, overlap_skipped, buffer counters, running workflows).
Frontend handlers reuse CreateSchedule's spec/action transform path
so update / create stay in lockstep on encoding.
Tests assert opcodes, request body shape, response decoding, alias
semantics, and ID validation rejects.
Both interfaces now carry an unexported sentinel method sealed() so
external packages cannot implement them. A fork that shadows the
workflow package will fail to satisfy the interface at compile
time, preventing silent divergence in selector / coroutine-scheduler
semantics.
The worker package's signalChan wrapper had to go — it lived
outside the workflow package and so cannot satisfy sealed().
Replaced with workflow.NewSignalChannel(name, buffer), which
returns the canonical sealed chanImpl directly. The name is
carried on the channel itself, so the wrapper was cosmetic.
Added pkg/sdk/workflow/channel_sealed_test.go asserting the
canonical impl satisfies the sealed interfaces.
The v1 broker issued raw "zapact:<id>" tokens, so any caller who
learned the activityTaskID string could complete/fail an activity
belonging to another workflow. v2 mints an opaque HMAC-SHA256-signed
token bound to {namespace, taskQueue, workflowID, runID,
activityTaskID, exp}; Respond* handlers verify the signature,
expiry, and scope-match before delivering.
Signing key loads from TASKS_ACTIVITY_TOKEN_KEY (base64); unset /
invalid falls back to an ephemeral random key with a WARN log so
tokens no longer validate across restarts (acceptable for dev,
not prod).
The schedule-activity response now carries the signed TaskToken
alongside ActivityTaskID so the worker that executes the activity
presents the signed token on respond.
v1 tokens are still accepted as legacy (prefix "zapact:") so any
in-flight tokens at the upgrade boundary drain; new registrations
always mint v2.
Tests cover: forged token (different signing key) rejected, expired
token rejected, cross-workflow token mismatch rejected, malformed
token rejected, happy-path round-trip, env-var key load, claims
inspection.
Adds every symbol hanzoai/base/plugins/tasks and hanzoai/commerce
call today that was missing from pkg/sdk/*: SignalWithStartWorkflow,
GetWorkflow, QueryWorkflow, CheckHealth, WorkflowRunGetOptions +
WorkflowRun.GetWithOptions, ExecuteChildWorkflow + ChildWorkflowFuture,
LocalActivityOptions + WithLocalActivityOptions + ExecuteLocalActivity
(Phase-1 remote-path), worker.InterruptCh(), worker.Options rate-limit
fields wired via golang.org/x/time/rate.
New opcodes 0x0066 signalWithStartWorkflow / 0x0067 queryWorkflow
land here; frontend-side handlers were wired in PR-A via the
translate.go dispatch table.
Closes red-review §2 gap table. Dry-run base migration compiles.
NewTimer used to collapse any duration to zero — production workflows
with schedule-level timers (24h dunning, weekly roll-up) fired on
the next dispatch loop tick. Selector busy-spun at 1 kHz.
Real timer.Timer behind each workflow.NewTimer; Selector.Await
blocks on fan-in over Future.readyCh channels. StubEnv becomes
test-only; the production dispatch env in pkg/sdk/worker uses the
real impl.
Closes red-review §5.2 and §5.9.
Replaces StubEnv use in dispatch.go with workerEnv, a real
CoroutineEnv implementation. Activity dispatch now goes over the
wire via ScheduleActivity (opcode 0x006B) + long-poll
WaitActivityResult (0x006C), not a pre-populated in-memory map.
NewTimer uses time.NewTimer in a per-timer goroutine; Select uses
a fan-in wake channel with a 20ms watcher poll (no per-iteration
allocation). Root cancel scope propagates to timers, activity polls,
and child scopes on return.
Workflow return errors are no longer swallowed — non-nil fn returns
get encoded via temporal.Encode into a single FailureCommand entry
in the commands envelope so the frontend can fail the execution
per its retry policy.
Phase-1 note documented in struct comment: every dispatch re-runs
the workflow from the top with the same input — activities must be
idempotent. Phase 2 replaces this with event-sourced replay
without changing the public surface.
Adds WorkerTransport.ScheduleActivity / WaitActivityResult /
StartChildWorkflow with v1 JSON-in-envelope bodies. Schema
schema/tasks.zap gains the 0x006B/0x006C struct definitions.
Implements pkg/sdk/worker — the Temporal-replacement worker runtime
used by base / commerce / ta after the import swap. Transport is
luxfi/zap via pkg/sdk/client.WorkerTransport; zero go.temporal.io/*
and zero google.golang.org/grpc imports.
Public surface (unchanged from upstream — callers only change the
import path):
worker.New(c client.Client, taskQueue string, opts Options) Worker
worker.Options{MaxConcurrentActivityExecutionSize,
MaxConcurrentWorkflowTaskPollers,
Identity, Logger}
Worker.Start() / Stop() / Run(interruptCh <-chan any)
Worker.RegisterWorkflow(fn) / RegisterWorkflowWithOptions
Worker.RegisterActivity(fn) / RegisterActivityWithOptions
Files:
- worker.go — Worker interface, New, Start/Stop/Run lifecycle,
defaults, identity + namespace resolution from
the attached Client.
- registry.go — reflect-based function registry with both
explicit-name and reflected-name registration;
double-registration panics unless disabled.
- pollers.go — two long-poll loops (workflow + activity) with
transient-error backoff and graceful shutdown.
- dispatch.go — workflow + activity task dispatch. Workflow re-
runs the fn synchronously against a
workflow.StubEnv (Phase 1 determinism); activity
builds an activity.Scope, injects via
activity.NewContext, decodes JSON args, invokes
fn(ctx, args...), ships result or failure via
temporal.Encode. Panic recovery on both paths.
Auto-heartbeat goroutine fires at half the
server-configured HeartbeatTimeout.
- env.go — workerEnv satisfies workflow.CoroutineEnv by
delegating to a workflow.StubEnv for Phase 1;
Phase 2 swaps the delegate for a history-driven
replay scheduler.
- worker_test.go — in-memory fake WorkerTransport; covers
registration, dispatch for workflow / activity /
unregistered / user-error paths, end-to-end
workflow + activity poll loops, and the Run ->
interrupt -> Stop lifecycle. 15/15 pass under -race.
Determinism (Phase 1)
The worker re-runs each registered workflow function from the
start on every workflow-task dispatch. The function must be pure
w.r.t. its inputs: same arguments => same sequence of primitives.
Activities are expected to be idempotent (same contract Temporal
imposes). Phase 2 lands event-sourced replay (history log +
replay decider) without changing the package's public surface.
Concurrency
MaxConcurrentWorkflowTaskPollers goroutines long-poll
OpcodePollWorkflowTask; MaxConcurrentActivityExecutionSize
goroutines long-poll OpcodePollActivityTask. No goroutine
hand-off: each poller dispatches the task it claimed synchronously
and re-polls. Defaults 4 workflow + 8 activity.
Verification:
go vet ./pkg/sdk/worker/... — clean
go test ./pkg/sdk/worker/... — 15/15 pass, race-clean
Implements pkg/sdk/client — the Temporal-replacement surface base,
commerce, and ta call when migrating off go.temporal.io/sdk/client.
Transport is luxfi/zap; zero go.temporal.io/* and zero
google.golang.org/grpc imports.
Opcode allocation (append-only, never reuse):
0x0050-0x005F pkg/tasks legacy (submit / schedule), reserved
0x0060-0x006F workflow lifecycle (start, signal, cancel,
terminate, describe, list)
0x0070-0x007F schedule ops (create, list, delete, pause)
0x0080-0x008F namespace ops (register, describe, list)
0x0090-0x009F health + meta
0x00A0-0x00AF worker poll / respond (moved up from 0x0070
to avoid collision with schedule ops)
Wire (v1): request and response bodies are a JSON document wrapped
in a ZAP envelope with three fields:
field 0 (Bytes) — body
field 8 (Uint32) — status (0/200 = ok)
field 12 (Bytes) — error detail on non-zero status
Opcode is stamped on the ZAP frame flags (high byte). Native ZAP
serde replaces JSON in a follow-up without changing opcodes.
Files:
- client.go — Client interface (Close/ExecuteWorkflow/SignalWorkflow/
CancelWorkflow/Terminate/Describe/List + schedule +
namespace + health), Dial, Transport abstraction,
luxfi/zap default transport, envelope codec.
- options.go — Options, StartWorkflowOptions (RetryPolicy via
pkg/sdk/temporal, no upstream import).
- workflow.go — ExecuteWorkflow, Signal/Cancel/Terminate/Describe/List,
WorkflowRun with reflect-based type-name resolution.
- namespace.go — Register/Describe/List + wire duration-ms codec.
- schedule.go — Create/List/Delete/Pause + wire duration-ms codec.
- transport.go — WorkerTransport + PollWorkflowTaskRequest etc. +
the worker-opcode constants (now 0x00A0+).
- client_test.go — Dial/Close test, ExecuteWorkflow round-trip with a
fake Transport capturing the request and replaying
a canned response (same pattern as pkg/tasks).
Worker bridge helpers (worker_bridge.go, worker_transport.go) were
added concurrently on feat/zap-internal; they import and wrap the
Transport this commit defines.
WorkflowRun.Get gap: the v1 schema does not yet declare a
GetWorkflowHistory / GetWorkflowResult RPC, so Get polls
DescribeWorkflow and returns ErrNotImplementedLocally when the
caller asks for a result from a completed workflow. This is marked
in the doc comment and will be filled in when pkg/sdk/worker lands
the history-fetch opcode.
Verification:
go vet ./pkg/sdk/client/... — clean
go test ./pkg/sdk/client/... — 7/7 pass
go list -deps ./pkg/sdk/client/... — no go.temporal.io/*, no
google.golang.org/grpc
Ports the workflow-code-facing API that base/plugins/tasks and
commerce/infra rely on today, with zero go.temporal.io or grpc
imports. Wire shape mirrors Temporal exactly so call-site migration
is a pure import-path swap.
Package layout (pkg/sdk/workflow):
env.go — CoroutineEnv interface, the one seam the worker
(pkg/sdk/worker, separately) must satisfy; Info,
SelectCase, CancelFunc. NewContextFromEnv is the
single entry point user ctx flows through.
context.go — Context interface (Deadline/Done/Err/Value), ctx
chain with WithCancel / WithValue / WithActivityOptions;
GetLogger, Now, Sleep, GetInfo.
future.go — Future + settleable impl. NewFuture / Settle split
so both the worker and the in-memory stub drive
completion through the same concrete. EncodePayload
is the Phase-1 JSON codec; Phase 2 swaps for ZAP.
channel.go — ReceiveChannel / Channel; unbuffered + buffered
chanImpl backing both user channels and signal
queues. Send/Receive/ReceiveAsync/Close.
selector.go — Selector with AddFuture / AddReceive / AddDefault;
Select delegates to env.Select and dispatches to
the matched callback.
timer.go — NewTimer schedules a durable timer Future.
signal.go — GetSignalChannel returns shared-by-name signal queues.
activity.go — ActivityOptions mirror of Temporal shape;
WithActivityOptions attaches per-call opts;
ExecuteActivity reads opts from ctx and dispatches.
reflect.go — Sole reflect use site (activity name resolution).
stub.go — StubEnv: synchronous, in-memory CoroutineEnv for
unit-testing workflow code without a worker.
OnActivity(...).Return(...), SignalAsync, AdvanceClock,
Cancel. Used throughout workflow_test.go.
workflow_test.go — 27 tests exercising Context, cancel scopes,
Now/Sleep, timers, activities (success/error/nil),
signal channels, user channels, Selector (future/
receive/default), and an end-to-end AgentTaskWorkflow-
shaped test that mirrors base/plugins/tasks.
Phase 1 determinism caveats (documented in env.go):
* No event-sourced replay yet — crashes restart from the top with
the original input. Same idempotency contract as Temporal.
* Payloads encoded as JSON, not the canonical ZAP codec.
* time.Now / rand / goroutines inside workflow code remain banned;
CI linter comes in a follow-up.
go vet clean. 27 tests pass including -race.
go list -deps reports zero go.temporal.io or google.golang.org/grpc.
Implements pkg/sdk/temporal: the call-site surface that base, commerce,
and ta use when migrating off go.temporal.io/sdk/temporal. Stdlib-only
imports; go list -deps confirms zero go.temporal.io/* in the transitive
tree.
retry.go
- RetryPolicy{InitialInterval, BackoffCoefficient, MaximumInterval,
MaximumAttempts, NonRetryableErrorTypes} — field shape matches the
existing callers in base/plugins/tasks and commerce/billing/workflows
so the migration is a pure import swap.
- DefaultRetryPolicy, Normalize (non-mutating), ShouldRetry,
NextInterval helpers. Exponential backoff is computed in float64
with early clamp at MaximumInterval; no overflow on large attempt
counts. Pure: no sleep, no log, no network.
errors.go
- Single concrete error type (*Error) with Message, Code, NonRetryable,
Details, Cause. errors.Is / errors.As friendly.
- NewError(msg, code, nonRetryable, details...) is the one canonical
constructor; upstream NewApplicationError maps to it with
nonRetryable=false. NewErrorWithCause covers the one call site that
used NewApplicationErrorWithCause.
- NewCanceledError, NewTimeoutError construct non-retryable sentinels.
Canceled and Timeout are package-level errors.Is targets matched by
Code so cross-process comparisons (decoded from the wire) work.
- IsApplicationError, IsCanceledError, IsTimeoutError classification
helpers unwrap through errors.As chains.
serde.go
- Encode/Decode between *Error and the bytes carried in
ActivityTask.failure / RespondActivityTaskFailedRequest.failure
(schema/tasks.zap). JSON body inside a versioned envelope
({v:1, p:...}); the envelope decouples body format from callers, so
we swap to native ZAP once zapc codegen lands without breaking
anything above.
- Fail-secure: nil, empty, malformed, or wrong-version bytes decode to
a non-retryable DecodeError — never panic, never nil pointer.
- Non-JSON-serialisable details (e.g. channels) degrade to their
string form so workers can always report a failure.
temporal_test.go
- Table tests for RetryPolicy (defaults, Normalize immutability,
ShouldRetry, NextInterval backoff/clamp/nil-policy path).
- Error constructors, empty-code default, Error() string contract,
Unwrap, sentinel match via errors.Is.
- Classification helpers across nil / plain / application / canceled /
timeout and wrapped forms.
- Encode/Decode round-trip across plain, application, non-retryable,
canceled, timeout, with-cause, no-details, 3-level nested cause.
- Hostile-input coverage: non-JSON-serialisable details, nil, empty,
garbage, wrong version, truncated payload.
Passes: go vet ./pkg/sdk/temporal/... and
go test ./pkg/sdk/temporal/...
Implements pkg/sdk/activity: GetInfo, GetLogger, RecordHeartbeat plus the
Scope / NewContext / FromContext contract that workers use to inject a
per-task execution context.
- Info, WorkflowExecution types matching the caller surface in
base/commerce/ta (no upstream type aliases).
- Scope is worker-owned; exposes Logger, HeartbeatSink, and a
goroutine-safe Heartbeats() inspector for tests.
- Context key is a private distinct-type struct (collision + impersonation
safe).
- GetInfo returns a deep copy (TaskToken []byte copied) -- activity code
cannot mutate the scope's stored token.
- RecordHeartbeat defensive-copies caller details before capture and
forwards to the worker's HeartbeatSink (luxfi/zap transport at that layer).
- Nil-safe fail-closed defaults: GetLogger -> log.Noop, GetInfo -> zero,
RecordHeartbeat -> no-op when ctx carries no scope.
Tests cover round-trip, TaskToken isolation, nil scope paths, logger field
propagation (asserts JSON output contains bound fields), sink forwarding,
defensive copy, concurrent heartbeat recording, and Heartbeats() snapshot
independence. All 14 tests pass under -race.
Imports: stdlib + github.com/luxfi/log only. Zero go.temporal.io, zero
google.golang.org/grpc.
schema/tasks.zap — one wire contract for all task RPCs:
* Namespace, Workflow lifecycle, Schedule, TaskQueue (matching)
* 20 RPCs as the v1 surface, mirroring what base/commerce/ta
actually use today. Opcodes are appended on extension; never
reused. Request/response types carry only the fields active
handlers need — no pass-through of Temporal internals.
pkg/sdk/doc.go — commits the migration contract:
go.temporal.io/sdk/{client,worker,workflow,activity,temporal}
→ github.com/hanzoai/tasks/pkg/sdk/{same}
Implementation fans out across blue/red agent pairs per package
(client, worker, workflow, activity, temporal). Each pair ports
one facet of the surface with no temporal.io imports. CI check
to enforce this lands after the first two packages merge.
zeekayandHanzo Dev