The runtime for composable work

type UserRequest struct { ID string `json:"id"` }
type User       struct { ID, Name string }

lookupUser := action.New("user.lookup",
    func(ctx context.Context, req UserRequest) (User, error) {
        return User{ID: req.ID, Name: "Alice"}, nil
    }).
    Timeout(2 * time.Second).
    Retry(2, action.ExponentialBackoff(25*time.Millisecond, 2*time.Second)).
    Build()

user, err := lookupUser.Do(ctx, UserRequest{ID: "usr-42"})

act := action.New("user.lookup", lookupHandler).Build()

// 1. Direct type-erased call: 0 heap allocations on matching Req or *Req pointer
result, err := act.DoAny(ctx, UserRequest{ID: "usr-42"})

// 2. Universal Invoker: automatically decodes and coerces dynamic payloads
result, err = action.InvokeAny(ctx, act, map[string]any{"id": "usr-42"})

// 3. Coerce: converts dynamic structures (JSON, maps, strings) into target types
req, err := action.Coerce[UserRequest](untypedInput)

// 4. Dynamic lifting: lifts AnyAction into *Builder[any, any] for composition
dynamicAct := action.Dynamic(act).Build()

// Pipeline with zero-allocation inline field transform
pipeline := action.PipeWith("user.process",
    fetchUserAct,
    func(ctx context.Context, u User) (string, error) { return u.Email, nil },
    sendEmailAct,
).Build()

// Heterogeneous concurrent execution
reviewer := action.ParallelNamed[string]("review.suite", map[string]action.AnyAction{
    "security": checkSecurityAct,
    "linter":   runLinterAct,
}).Build()

results, err := reviewer.Do(ctx, "package main...")

act := action.New("fetch.data", coreHandler).
    Timeout(2 * time.Second).
    Retry(3, action.ExponentialJitter(100*time.Millisecond, 2*time.Second)).
    Idempotent().
    Cache(30*time.Second, keyFn, cacheStore).
    Dedup(keyFn).
    ConcurrencyLimit(10).
    RateLimit(100, 50).
    ExclusiveFenced(distLock, 10*time.Second, keyFn).
    Build()

stream := action.NewStream("telemetry.stream",
    func(ctx context.Context, req Query) (iter.Seq2[Record, error], error) {
        return func(yield func(Record, error) bool) {
            for i := 0; i < req.Limit; i++ {
                if ctx.Err() != nil {
                    yield(Record{}, ctx.Err())
                    return
                }
                if !yield(Record{ID: i}, nil) {
                    return // Consumer stopped early
                }
            }
        }, nil
    })

seq, err := stream.Do(ctx, Query{Limit: 100})
for record, err := range seq {
    // Zero heap allocation streaming
}

graph, err := dag.New("order.workflow").
    AddNode("validate", "val_out", validateAction).
    AddNode("reserve",  "res_out", reserveAction).
    AddNode("charge",   "chg_out", chargeAction).
    AddEdge("validate", "reserve").
    AddEdge("validate", "charge").
    Compile()

initialState := dag.AcquireState()
defer initialState.Release()

finalState, err := graph.Execute(ctx, initialState)
defer finalState.Release()

res, err := dag.GetNodeOutput[ReserveResult](finalState, "reserve")

// Categories: BadRequest, Unauthorized, Forbidden, NotFound, Conflict,
// Validation, TooManyRequests, Timeout, Unavailable, Internal, CircuitBreaker.
return xerr.NotFound("user not found")
return xerr.Unavailable("database connection dropped", originalErr)

if xerr.IsTransient(err) {
    // safe to retry — the kernel's Retry policy already checks this
}

appErr := xerr.From(err)
// Public returns a sanitized ErrorResponse suitable for API responses
// without leaking internal causes or stack traces.
writeJSON(w, appErr.Public(requestID))

type fanoutSink struct{ sinks []observe.Sink }
func (f *fanoutSink) Emit(ctx context.Context, e observe.Event) {
    for _, s := range f.sinks { s.Emit(ctx, e) }
}

sink := &fanoutSink{sinks: []observe.Sink{
    observe.NewSlogSink(slog.Default()),
    observe.NewMetricsSink(),
    observe.NewPrometheusSink(),
    observe.NewMemorySink(100),
    observe.NewJSONLSink(os.Stdout, 4096),
}}

act := action.New("user.find", findUser).
    AnyHook(observe.Hook(sink)).
    Build()

# Verify formatting
gofmt -l .

# Run static analysis
go vet ./...

# Run test suite with race detector
go test -race ./...

# Benchmark core dispatch and allocation profiles
go test ./... -run '^$' -bench=. -benchmem