runnable

README

runnable

A generic Runnable[In, Out] interface and small combinators (Pipe, Parallel, Branch, Map, Retry) for composing typed pipelines. Stdlib-only.

import "github.com/buckedunicorn/grok/runnable"

When to use

runnable is the SDK's universal connector. Wrap any function with runnable.Func[In, Out] and stitch together prompts, chat calls, parsers, post-processing, and retries into one typed pipeline.

It is opt-in sugar. Calling chat.Client directly stays the simplest path for one-off requests.

Surface

Type / function	Purpose
Runnable[In, Out]	Interface: Invoke(ctx, In) (Out, error)
Func[In, Out](fn)	Wraps a plain function as a Runnable
Pipe, Pipe3, Pipe4	Sequential composition
Parallel	Concurrent execution; collects results
Branch	Conditional dispatch
Map	Element-wise; fan-out over a slice
ParallelMap	Concurrent fan-out
Retry	Re-invoke on error with backoff

Example

type req struct{ Phrase, Language string }

renderPrompt := runnable.Func[req, []chat.Message](func(ctx context.Context, in req) ([]chat.Message, error) {
    return []chat.Message{{Role: "user", Content: fmt.Sprintf("Translate '%s' to %s.", in.Phrase, in.Language)}}, nil
})
callModel := runnable.Func[[]chat.Message, *chat.Completion](func(ctx context.Context, msgs []chat.Message) (*chat.Completion, error) {
    return client.Chat.Create(ctx, &chat.CreateRequest{Model: "grok-4-1-fast-non-reasoning", Messages: msgs})
})
extractText := runnable.Func[*chat.Completion, string](func(_ context.Context, comp *chat.Completion) (string, error) {
    s, _ := comp.Choices[0].Message.Content.(string)
    return s, nil
})

pipeline := runnable.Pipe3(renderPrompt, callModel, extractText)
out, _ := pipeline.Invoke(ctx, req{Phrase: "hello", Language: "French"})

Examples directory

• examples/sugar (Pipe3 pipeline)
• examples/workflow

Documentation

Overview

Package runnable provides a generic Runnable[In, Out] interface and small combinators (Pipe, Parallel, Branch, Map, Retry) for composing typed pipelines.

Runnable is the SDK's universal connector: prompts, chat calls, parsers, and tools can all satisfy the interface, so they compose without glue code. The package is opt-in, using *chat.Client directly remains the simplest path. Reach for runnable when you have a multi-step pipeline you want typed end-to-end:

type Summary struct{ Text string }

render := runnable.Func[map[string]any, []chat.Message](func(_ context.Context, vars map[string]any) ([]chat.Message, error) {
 return tmpl.Render(vars)
})
call := runnable.Func[[]chat.Message, *chat.Completion](func(ctx context.Context, msgs []chat.Message) (*chat.Completion, error) {
 return client.Chat.Create(ctx, &chat.CreateRequest{Model: "...", Messages: msgs})
})
parse := runnable.Func[*chat.Completion, Summary](func(_ context.Context, c *chat.Completion) (Summary, error) {
 return chat.Decode[Summary](c)
})

pipeline := runnable.Pipe3(render, call, parse)
result, err := pipeline.Run(ctx, map[string]any{"text": "..."})

Index

• Variables
• type Func
  • func (f Func[In, Out]) Run(ctx context.Context, in In) (Out, error)
• type MapOption
  • func WithMapConcurrency(n int) MapOption
• type RetryOptions
• type Runnable
  • func Branch[In, Out any](choose func(In) int, rs ...Runnable[In, Out]) Runnable[In, Out]
  • func Map[In, Out any](r Runnable[In, Out]) Runnable[[]In, []Out]
  • func Parallel[In, Out any](rs ...Runnable[In, Out]) Runnable[In, []Out]
  • func ParallelMap[In, Out any](r Runnable[In, Out], opts ...MapOption) Runnable[[]In, []Out]
  • func Pipe[A, B, C any](a Runnable[A, B], b Runnable[B, C]) Runnable[A, C]
  • func Pipe3[A, B, C, D any](a Runnable[A, B], b Runnable[B, C], c Runnable[C, D]) Runnable[A, D]
  • func Pipe4[A, B, C, D, E any](a Runnable[A, B], b Runnable[B, C], c Runnable[C, D], d Runnable[D, E]) Runnable[A, E]
  • func Retry[In, Out any](opts RetryOptions, r Runnable[In, Out]) Runnable[In, Out]

Variables

var ErrBranchOutOfRange = errors.New("runnable: Branch index out of range")

Branch picks one of rs based on the index returned by choose. If choose returns an out-of-range index, Branch returns ErrBranchOutOfRange.

Types

type Func

type Func[In, Out any] func(ctx context.Context, in In) (Out, error)

Func adapts a plain function into a Runnable.

func (Func[In, Out]) Run

func (f Func[In, Out]) Run(ctx context.Context, in In) (Out, error)

Run satisfies Runnable for Func.

type MapOption

type MapOption func(*mapConfig)

MapOption configures ParallelMap.

func WithMapConcurrency

func WithMapConcurrency(n int) MapOption

WithMapConcurrency caps the number of goroutines ParallelMap may have running at once. <= 0 means "one goroutine per input" (historical behaviour). For very large inputs (10k+) set a finite concurrency to avoid spawning that many goroutines at once .

type RetryOptions

type RetryOptions struct {
	MaxAttempts    int
	InitialBackoff time.Duration
	MaxBackoff     time.Duration
	Jitter         float64
	ShouldRetry    func(error) bool
}

RetryOptions configures Retry. MaxAttempts is the total number of tries (including the initial one); 0 or negative means a single attempt. InitialBackoff is the delay before the second attempt; subsequent delays double. MaxBackoff caps the per-attempt delay (default 30s); 0 or negative disables the cap (not recommended; see ). Jitter (0..1) randomizes each delay by up to ±jitter*delay. ShouldRetry returns true to retry on a given error; nil retries on every non-nil error.

type Runnable

type Runnable[In, Out any] interface {
	Run(ctx context.Context, in In) (Out, error)
}

Runnable is the universal connector: anything that takes In and produces Out via context-aware execution.

func Branch

func Branch[In, Out any](choose func(In) int, rs ...Runnable[In, Out]) Runnable[In, Out]

Branch runs the Runnable at rs[choose(in)] and returns its result.

func Map

func Map[In, Out any](r Runnable[In, Out]) Runnable[[]In, []Out]

Map applies r to each element of the input slice and returns the results in order. Elements are processed sequentially. Use ParallelMap for concurrent execution.

func Parallel

func Parallel[In, Out any](rs ...Runnable[In, Out]) Runnable[In, []Out]

Parallel runs every Runnable in rs concurrently with the same input and returns their outputs in input order. The first error aborts pending work (via context cancellation) and is returned.

func ParallelMap

func ParallelMap[In, Out any](r Runnable[In, Out], opts ...MapOption) Runnable[[]In, []Out]

ParallelMap is the concurrent variant of Map. The first element error cancels remaining work via the shared context.

func Pipe

func Pipe[A, B, C any](a Runnable[A, B], b Runnable[B, C]) Runnable[A, C]

Pipe composes A -> B sequentially: out := b(a(in)).

func Pipe3

func Pipe3[A, B, C, D any](a Runnable[A, B], b Runnable[B, C], c Runnable[C, D]) Runnable[A, D]

Pipe3 composes three Runnables A -> B -> C -> D. Sugar for nested Pipe.

func Pipe4

func Pipe4[A, B, C, D, E any](a Runnable[A, B], b Runnable[B, C], c Runnable[C, D], d Runnable[D, E]) Runnable[A, E]

Pipe4 composes four Runnables.

func Retry

func Retry[In, Out any](opts RetryOptions, r Runnable[In, Out]) Runnable[In, Out]

Retry wraps r with bounded exponential backoff retries per opts. The underlying context applies to every attempt; cancellation aborts pending retries immediately. The per-attempt delay is capped at MaxBackoff so a large MaxAttempts cannot wedge the goroutine for hours.