mirror of
https://github.com/hanzoai/tasks.git
synced 2026-10-04 02:04:03 +00:00
zeekay
e0171b4015
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.
22 lines
819 B
Go
22 lines
819 B
Go
package workflow
|
|
|
|
// GetSignalChannel returns the ReceiveChannel for the named signal.
|
|
// Two calls with the same name return handles that observe the
|
|
// same queue — signals are keyed by name, not handle identity.
|
|
// This matches Temporal's semantics and is what base/plugins/tasks
|
|
// relies on for claim/complete/fail/cancel fan-in.
|
|
//
|
|
// Signals are durable: the worker persists them and re-delivers
|
|
// them on replay. Sending a signal is the client's job
|
|
// (pkg/sdk/client); this helper only exposes the workflow-side read.
|
|
func GetSignalChannel(ctx Context, name string) ReceiveChannel {
|
|
if ctx == nil {
|
|
// Return an empty, closed channel so the caller's next
|
|
// Receive returns ok=false rather than deadlocking.
|
|
c := newChan(name, 0)
|
|
c.Close()
|
|
return c
|
|
}
|
|
return ctx.env().GetSignalChannel(name)
|
|
}
|