sdk

func BatchExitCode ¶ added in v0.1.2

func BatchExitCode(err error) int

BatchExitCode returns the recommended container exit code for an error. Timeout errors map to 124 (GNU timeout), all other non-nil errors default to 1.

func DebugModeEnabled ¶ added in v0.1.2

func DebugModeEnabled() bool

DebugModeEnabled reports whether verbose logging should be forced regardless of logger level.

func EmitSequencedSignal ¶ added in v0.1.2

func EmitSequencedSignal(ctx context.Context, key string, seq uint64, value any) error

EmitSequencedSignal emits a signal wrapped with a sequence number and timestamp.

func EmitSignal ¶ added in v0.1.2

func EmitSignal(ctx context.Context, key string, value any) error

EmitSignal patches the current StepRun status with a small JSON payload so that controllers and CEL expressions can react to live metadata. When value is nil, the signal is cleared. The latest value stored in status.signals is always a compact summary so raw payloads are not persisted in StepRun status. Signal events are sequenced by default.

func EmitSignalWithSequence ¶ added in v0.1.2

func EmitSignalWithSequence(ctx context.Context, key string, seq *SignalSequence, value any) error

EmitSignalWithSequence emits a signal using the provided sequence generator.

func EmitTextSignal ¶ added in v0.1.2

func EmitTextSignal(ctx context.Context, key string, text string, opts TextSignalOptions) error

EmitTextSignal emits a metadata summary for text payloads. By default it keeps metadata-only behavior; callers can opt into bounded inline samples via SampleBytes or SampleExtras. The full payload should still be passed via outputs/storage references when downstream steps need the complete content.

func ExecuteEffectOnce ¶ added in v0.1.2

func ExecuteEffectOnce(ctx context.Context, key string, fn func(context.Context) (any, error)) (any, bool, error)

ExecuteEffectOnce runs fn only if the effect key has not been recorded yet. It records a successful effect with the returned details. When the effect already exists, it returns `already=true` with `ErrEffectAlreadyRecorded`.

func ExecutionMode ¶ added in v0.1.2

func ExecutionMode() string

ExecutionMode returns the execution mode supplied by the operator (e.g. "batch", "job", "deployment"). Defaults to "batch" when not set so local runs behave sensibly.

func ExtractTraceContext ¶ added in v0.1.2

func ExtractTraceContext(ctx context.Context, msg *engram.StreamMessage) context.Context

ExtractTraceContext restores tracing context from a StreamMessage's metadata so Engrams can start child spans that are linked to upstream steps.

func HasEffect ¶ added in v0.1.2

func HasEffect(ctx context.Context, key string) (bool, error)

HasEffect returns true if the current StepRun already recorded an effect for the key.

func LoggerFromContext ¶

func LoggerFromContext(ctx context.Context) *slog.Logger

LoggerFromContext retrieves a slog.Logger from the context, or returns a default JSON logger.

If no logger was previously stored via WithLogger, this function returns a new JSON logger writing to stdout with default settings. This ensures the SDK always has a valid logger without requiring explicit configuration for simple use cases.

Thread-safe and idempotent.

func NewStreamErrorMessage ¶ added in v0.1.2

func NewStreamErrorMessage(
	errObj runsv1alpha1.StructuredError,
	opts ...StreamMessageOption) (engram.StreamMessage,
	error,
)

NewStreamErrorMessage wraps a StructuredError into a StreamMessage payload with Kind "error".

func NewStreamMessage ¶ added in v0.1.2

func NewStreamMessage(kind string, opts ...StreamMessageOption) engram.StreamMessage

NewStreamMessage constructs an engram.StreamMessage pre-populated with the provided options. It trims the kind identifier and applies any options in order. Note that metadata-only messages are invalid at send time; at least one of audio/video/binary payloads, JSON payload, inputs, or transports must be populated for the message to be published.

func NewStructuredError ¶ added in v0.1.2

func NewStructuredError(typ runsv1alpha1.StructuredErrorType, message string, opts ...StructuredErrorOption) error

NewStructuredError returns an error that carries a StructuredError payload.

func ParseStreamErrorMessage ¶ added in v0.1.2

func ParseStreamErrorMessage(msg engram.StreamMessage) (runsv1alpha1.StructuredError, bool, error)

ParseStreamErrorMessage extracts StructuredError payloads from StreamMessage Kind "error".

func PublishLogFile ¶ added in v0.1.2

func PublishLogFile(ctx context.Context, path string, contentType string) error

PublishLogFile reads the provided file and publishes its contents as logs.

func PublishLogs ¶ added in v0.1.2

func PublishLogs(ctx context.Context, payload []byte) error

PublishLogs uploads log bytes to storage (when enabled) and patches StepRun.status.logs with a storage reference. If storage is disabled, it returns ErrLogsUnavailable.

func PublishLogsWithContentType ¶ added in v0.1.2

func PublishLogsWithContentType(ctx context.Context, payload []byte, contentType string) error

PublishLogsWithContentType is like PublishLogs but lets callers set a content type.

func RecordEffect ¶ added in v0.1.2

func RecordEffect(ctx context.Context, key, status string, details any) error

RecordEffect appends an effect record to the current StepRun status ledger. The effect sequence is assigned server-side when Seq is 0.

func ReplaySignals ¶ added in v0.1.2

func ReplaySignals(
	ctx context.Context,
	stepRunName,
	namespace string,
	sinceSeq uint64) ([]runsv1alpha1.SignalEvent,
	error,
)

ReplaySignals returns the signal events for a StepRun, optionally filtered by sequence. When stepRunName or namespace are empty, environment defaults are used.

func RunBatch ¶

func RunBatch[C any, I any](ctx context.Context, e engram.BatchEngram[C, I]) error

RunBatch is the primary entry point for a BatchEngram. It provides a fully type-safe execution environment, handling all the boilerplate of context loading, data hydration, and status patching.

func RunImpulse ¶

func RunImpulse[C any](ctx context.Context, i engram.Impulse[C]) error

RunImpulse is the type-safe entry point for impulses (Kubernetes Deployments that trigger Stories).

This function infers config type C from the impulse implementation, providing compile-time type safety. It orchestrates the complete lifecycle:

1. Load execution context from environment (BUBU_STEP_CONFIG, BUBU_TRIGGER_DATA, etc.)
2. Unmarshal config into type C
3. Call impulse.Init with typed config and secrets
4. Create pre-configured Kubernetes client with namespace resolution
5. Call impulse.Run with client, transferring control to long-running process

The impulse's Run method should block until work completes or context is canceled. Typical use cases: webhook listeners, message queue consumers, schedulers, event watchers.

Respects context cancellation for graceful shutdown on SIGTERM. The impulse is responsible for handling shutdown signals within its Run implementation (e.g., draining in-flight requests).

Example:

type MyConfig struct {
    WebhookPort int    `mapstructure:"webhookPort"`
    SecretToken string `mapstructure:"secretToken"`
}

type MyImpulse struct { /* ... */ }

func (m *MyImpulse) Init(ctx context.Context, cfg MyConfig, secrets *engram.Secrets) error {
    // Setup webhook server, validate token, etc.
    return nil
}

func (m *MyImpulse) Run(ctx context.Context, client *k8s.Client) error {
    // Listen for webhooks, trigger stories via client.TriggerStory(...)
    <-ctx.Done()
    return ctx.Err()
}

func main() {
    if err := sdk.RunImpulse(context.Background(), &MyImpulse{}); err != nil {
        panic(err)
    }
}

func Start ¶ added in v0.1.2

func Start[C any, I any](ctx context.Context, e DualEngram[C, I]) error

Start launches the provided engram in either batch or streaming mode based on ExecutionMode. This lets engram entrypoints avoid direct environment inspection and stick to the SDK abstraction.

func StartBatch ¶

func StartBatch[C any, I any](ctx context.Context, e engram.BatchEngram[C, I]) error

StartBatch is the type-safe entry point for batch engrams (Kubernetes Jobs).

This function infers both config type C and input type I from the engram implementation, providing full compile-time type safety. It orchestrates the complete lifecycle:

1. Load execution context from environment (BUBU_STEP_CONFIG, BUBU_TRIGGER_DATA, etc.)
2. Unmarshal config and inputs into types C and I
3. Call engram.Init with typed config and secrets
4. Hydrate inputs from storage if needed
5. Call engram.Process with typed inputs and execution context
6. Dehydrate outputs to storage if they exceed size limits
7. Patch StepRun status with result, timing, and exit code

Enforces timeout via BUBU_STEP_TIMEOUT with context cancellation. On timeout, patches status with exit code 124 (retryable) and forcefully exits to prevent zombie Jobs. On logic errors, patches with exit code 1 (terminal). On success, patches with exit code 0.

Example:

type MyConfig struct { APIKey string `mapstructure:"apiKey"` }
type MyInputs struct { UserID string `mapstructure:"userId"` }

func main() {
    ctx := context.Background()
    if err := sdk.StartBatch(ctx, NewMyEngram()); err != nil {
        panic(err)  // Ensure non-zero exit for Job failure detection
    }
}

func StartStory ¶

func StartStory(ctx context.Context, storyName string, inputs map[string]any) (*runsv1alpha1.StoryRun, error)

StartStory triggers a new StoryRun for the named Story with the provided inputs. For cross-namespace executions, use StartStoryInNamespace to explicitly set the Story namespace.

This is the primary mechanism for programmatically initiating workflows, typically used from within an Impulse. The SDK automatically resolves the correct namespace from environment variables (BUBU_TARGET_STORY_NAMESPACE or fallbacks), creates a Kubernetes client, and submits the StoryRun resource.

Inputs are marshaled to JSON and stored in the StoryRun spec. The operator watches for new StoryRuns and orchestrates their execution.

Returns the created StoryRun on success, or an error if client creation or StoryRun creation fails. Respects context cancellation and deadlines.

Example:

sr, err := sdk.StartStory(ctx, "my-workflow", map[string]any{
    "userId": "12345",
    "action": "process",
})
if err != nil {
    return fmt.Errorf("failed to trigger story: %w", err)
}
log.Printf("Triggered StoryRun: %s", sr.Name)

func StartStoryInNamespace ¶ added in v0.1.2

func StartStoryInNamespace(
	ctx context.Context,
	storyName string,
	storyNamespace string,
	inputs map[string]any,
) (*runsv1alpha1.StoryRun, error)

StartStoryInNamespace is identical to StartStory but allows specifying the namespace of the referenced Story explicitly. Provide an empty namespace to use the default resolution (BUBU_TARGET_STORY_NAMESPACE or the pod namespace).

func StartStoryWithToken ¶ added in v0.1.2

func StartStoryWithToken(
	ctx context.Context,
	storyName string,
	token string,
	inputs map[string]any,
) (*runsv1alpha1.StoryRun, error)

StartStoryWithToken behaves like StartStory but accepts a per-call token for deterministic retries.

func StartStoryWithTokenInNamespace ¶ added in v0.1.2

func StartStoryWithTokenInNamespace(
	ctx context.Context,
	storyName string,
	storyNamespace string,
	token string,
	inputs map[string]any,
) (*runsv1alpha1.StoryRun, error)

StartStoryWithTokenInNamespace behaves like StartStoryInNamespace but accepts a per-call token for deterministic retries.

func StartStreamServer ¶

func StartStreamServer[C any](ctx context.Context, e engram.StreamingEngram[C]) error

StartStreamServer boots a StreamingEngram using the new transport connector contract. The Engram must have BUBU_TRANSPORT_BINDING set; no other transport modes are supported.

func StartStreaming ¶

func StartStreaming[C any](ctx context.Context, e engram.StreamingEngram[C]) error

StartStreaming is the type-safe entry point for streaming engrams (Kubernetes Deployments with gRPC).

This function infers config type C from the engram implementation, providing compile-time type safety. It orchestrates the complete lifecycle:

1. Load execution context from environment (BUBU_STEP_CONFIG, etc.)
2. Unmarshal config into type C
3. Call engram.Init with typed config and secrets
4. Start gRPC server on BUBU_GRPC_PORT (default 50051)
5. Register engram.Stream as the bidirectional streaming handler
6. Serve until context cancellation (SIGTERM) or error
7. Gracefully drain active streams before shutdown

The gRPC server implements:

• Transparent heartbeat sending/filtering to detect connection hangs
• Backpressure handling with configurable timeouts
• Graceful shutdown with BUBU_GRPC_GRACEFUL_SHUTDOWN_TIMEOUT drain phase
• Optional TLS via BUBU_GRPC_TLS_CERT_FILE and BUBU_GRPC_TLS_KEY_FILE
• Configurable message size limits via BUBU_GRPC_MAX_RECV_BYTES and BUBU_GRPC_MAX_SEND_BYTES

Example:

type MyConfig struct { BufferSize int `mapstructure:"bufferSize"` }

func main() {
    ctx := context.Background()
    if err := sdk.StartStreaming(ctx, NewMyStreamingEngram()); err != nil {
        panic(err)
    }
}

func StopStory ¶ added in v0.1.2

func StopStory(ctx context.Context, storyRunName string) error

StopStory cancels an in-flight StoryRun in the current namespace. Equivalent to StopStoryInNamespace with an empty namespace.

func StopStoryInNamespace ¶ added in v0.1.2

func StopStoryInNamespace(ctx context.Context, storyRunName, namespace string) error

StopStoryInNamespace cancels an in-flight StoryRun by marking it finished. If the StoryRun does not exist, ErrStoryRunNotFound is returned. Already-terminal StoryRuns are treated as a no-op. Active StoryRuns in phases the SDK will not force-finish return an invalid-transition error from the underlying k8s client.

func TriggerTokenFromContext ¶ added in v0.1.2

func TriggerTokenFromContext(ctx context.Context) string

TriggerTokenFromContext returns the trigger token stored in the context, if any.

func WithLogger ¶

func WithLogger(ctx context.Context, logger *slog.Logger) context.Context

WithLogger stores a slog.Logger in the context for SDK use.

This allows you to inject a custom configured logger (e.g., with specific log levels, handlers, or structured attributes) that the SDK will use for all internal logging. If not provided, the SDK defaults to JSON logging to stdout.

Example:

logger := slog.New(slog.NewJSONHandler(os.Stdout, &slog.HandlerOptions{Level: slog.LevelDebug}))
ctx := sdk.WithLogger(context.Background(), logger)
sdk.StartBatch(ctx, myEngram)

The logger is used for lifecycle events (init, shutdown), errors, and metrics. It does not intercept engram-specific logs; engrams should use their own loggers or retrieve the SDK logger via LoggerFromContext within their execution context.

func WithTriggerToken ¶ added in v0.1.2

func WithTriggerToken(ctx context.Context, token string) context.Context

WithTriggerToken attaches an idempotency token that StartStory passes through to the Kubernetes client. When provided, the SDK derives deterministic StoryRun names without relying on process-wide env vars. Nil contexts are accepted so callers can safely attach a token before choosing a base context.