core

type Block ¶

type Block struct {
	Name      string
	Type      string
	Processor MessageProcessor
}

Block is a configured, named stage in a flow wrapping one MessageProcessor. The Processor is either a leaf (built from a BlockFactory) or a composite that embeds sub-flows (built by the flow builder). The block itself stays a thin record; any embedded flows live inside the composite processor.

type BlockDeps ¶

type BlockDeps struct {
	Connector func(name string) (connector Connector, ok bool)
	Flows     FlowCaller
	Env       map[string]string
	Services  RuntimeServices
	// Resources loads resources (templates, env files) a block may need, e.g. the
	// template-resource block reading a template by id. It is nil when no loader is
	// wired; a block must guard against that (or the caller supplies a Noop).
	Resources ResourceLoader
	// Breakpoint collects the message at an addressed block and halts the flow, for
	// the CLI's `invoke --break-at`. It is nil in every normal run — only the
	// implicit breakpoint block reads it, and it refuses to build without one, so a
	// flow can never carry a breakpoint that was not asked for.
	Breakpoint *Breakpoint
	// Spies collects what crosses each addressed block, for the CLI's `invoke
	// --spies`. Nil in every normal run, on the same terms as Breakpoint: only the
	// implicit spy block reads it, and it refuses to build without one.
	Spies *Spies
	// Mocks holds the canned outcomes each addressed block is replaced with, for the
	// CLI's `invoke --mocks`. Unlike the two above it collects nothing — it is an
	// input to the run — but it reaches the engine the same way, and the implicit
	// mock block refuses to build without it, so a flow can never carry a mock that
	// was not asked for.
	Mocks *Mocks
}

BlockDeps carries build-time services a block factory may need beyond its settings. Most blocks ignore it. Connector resolves a configured connector instance by name so a block can use a capability that connector provides — for example, a log block binding to a logger connector. ok is false when no connector with that name is configured. Flows lets a block call another flow by name (used by the flow-ref block); it is nil when no flow caller is wired. Env holds the config's resolved environment variables so a block can expose them to its expressions as env.NAME; it is nil when none are declared. Services exposes the runtime services (leader election, KV) to a block; it is nil for callers that do not wire them, so a block must guard against that.

type BlockFactory ¶

type BlockFactory func(settings types.Settings, deps BlockDeps) (MessageProcessor, error)

BlockFactory builds a leaf processor from its settings and build-time deps. Composite kinds (scope, fork) are not built through the block registry; the flow builder recognizes them and constructs their typed sub-flows directly.

type BlockMeta ¶

type BlockMeta struct {
	Type        string
	Label       string
	Category    string
	Group       string
	Icon        string
	Description string
	Config      reflect.Type
	// SrcDir is the source directory of the registering package, captured
	// automatically by RegisterBlockMeta for extension association and
	// doc-comment extraction. Do not set it by hand.
	SrcDir string
}

BlockMeta is the editor-facing description of a block. Type must match the name passed to MustRegisterBlock. Category is "processor" or "control-flow". Group and Icon fall back to the registering package's ExtensionMeta when empty. Config is the zero-value settings struct whose `octo`/`json` tags describe the block's fields; it is nil for composite (control-flow) blocks that have no settings struct and instead register a schema-only meta struct.

type BlockRegistry ¶

type BlockRegistry struct {
	// contains filtered or unexported fields
}

BlockRegistry holds leaf block factories keyed by block type. Composite kinds (scope, fork) are handled by the flow builder and are not registered here.

func DefaultBlockRegistry() *BlockRegistry

func NewBlockRegistry() *BlockRegistry

func (r *BlockRegistry) MustRegister(name string, factory BlockFactory)

func (r *BlockRegistry) Names() []string

func (r *BlockRegistry) New(name string, settings types.Settings, deps BlockDeps) (MessageProcessor, error)

func (r *BlockRegistry) Register(name string, factory BlockFactory) error

type Breakpoint ¶

type Breakpoint struct {
	// contains filtered or unexported fields
}

Breakpoint collects the message a flow was carrying when execution reached an addressed block. It is the debug seam behind the CLI's `invoke --break-at`: the runtime wraps the addressed block in an implicit breakpoint block, which runs the block, records the resulting message here, and halts the flow.

It is a collector rather than a return value because a block can sit inside a composite that discards its sub-flow's message — a fork branch runs on a clone, an enrich body runs on a clone — so the snapshot has to leave the flow out-of-band. Address is carried along so a caller that only holds the Breakpoint can report which block it was waiting on.

A flow runs its blocks across several workers, and a fork runs branches concurrently, so Record may be called from more than one goroutine. The first call wins: a breakpoint reports the first time execution reached the block, and later arrivals are ignored rather than overwriting it.

func NewBreakpoint(address string) *Breakpoint

func (b *Breakpoint) Address() string

func (b *Breakpoint) Record(msg *types.Message)

func (b *Breakpoint) Snapshot() (*types.Message, bool)

type ChangeFunc ¶

type ChangeFunc func(kind ResourceKind, id string)

ChangeFunc is invoked by a watching loader when a resource changes, carrying the kind and id of the resource that changed so the runtime knows what changed, not just that something did.

type Connector ¶

type Connector interface {
	Start(ctx context.Context, config types.ConnectorConfig) error
	Stop(ctx context.Context) error
}

Connector is a runtime component that can be started and stopped.

type ConnectorMeta ¶

type ConnectorMeta struct {
	Type     string
	Label    string
	Icon     string
	Category string
	Settings reflect.Type
	Sources  []SourceMeta
	// SrcDir is captured automatically by RegisterConnectorMeta.
	SrcDir string
}

ConnectorMeta is the editor-facing description of a connector. Settings is the zero-value connector-settings struct whose tags describe its settings fields. Category groups a family of connectors (e.g. "llm") so a field can reference any member. Icon falls back to the package ExtensionMeta when empty.

type Entry ¶

type Entry struct {
	Value   []byte
	Version int64
}

Entry is a versioned KV value. Version is monotonic per key: a freshly created object has version 1, and each successful write increments it. A reader passes the version it last saw to the next write to detect concurrent updates.

type EventBus ¶

type EventBus struct {
	// contains filtered or unexported fields
}

EventBus is a process-wide, fan-out pub/sub dedicated to flow events. Every subscriber receives every event. It is safe for concurrent Publish and Subscribe.

func DefaultEventBus() *EventBus

func NewEventBus() *EventBus

func (b *EventBus) Publish(event types.FlowEvent)

func (b *EventBus) Subscribe(handler FlowEventHandler) (unsubscribe func())

type ExtensionMeta ¶

type ExtensionMeta struct {
	Group string
	Icon  string
	// SrcDir is captured automatically by RegisterExtension.
	SrcDir string
}

ExtensionMeta declares package-level defaults inherited by every block and connector registered from the same package (directory). It is registered once per package init(); a block/connector value overrides the default it carries.

type Factory ¶

type Factory func() Connector

Factory constructs a new Connector instance.

type FlowCaller ¶

type FlowCaller interface {
	// Call sends msg to the named flow and waits for its terminal outcome.
	// completed -> (result, nil); dropped -> (nil, nil); failed -> (nil, err).
	Call(ctx context.Context, name string, msg *types.Message) (*types.Message, error)
	// Send delivers msg to the named flow without awaiting a result (one-way).
	Send(ctx context.Context, name string, msg *types.Message) error
}

FlowCaller invokes a registered flow by name. It is the contract behind direct invocation (the CLI) and the flow-ref block: a flow without an external source registers its input channel under its name, and callers push messages into it, correlating the result through the flow-event bus.

Outcomes follow the flow's terminal event: a completed flow returns its result message, a dropped flow returns (nil, nil), and a failed flow returns the error.

type FlowEventHandler ¶

type FlowEventHandler func(event types.FlowEvent)

FlowEventHandler reacts to a published flow event. Handlers must not block; long-running work should be dispatched to the handler's own goroutine.

type KV ¶

type KV interface {
	// Get returns the entry for key in namespace. ok is false when the key is absent.
	Get(ctx context.Context, namespace, key string) (entry Entry, ok bool, err error)
	// Set stores value and returns the new version.
	Set(ctx context.Context, namespace, key string, value []byte, expectedVersion int64) (newVersion int64, err error)
	// Delete removes key in namespace. expectedVersion 0 deletes unconditionally; a
	// positive value must match the stored version or Delete returns ErrVersionConflict.
	Delete(ctx context.Context, namespace, key string, expectedVersion int64) error
}

KV is a deployment-scoped key/value store available to connectors and blocks. Every operation takes a namespace: keys are isolated per namespace, so state an internal component writes under its own namespace is invisible to a key read or written under another. A component using the store is expected to confine itself to a namespace it owns (e.g. a user-facing cache block writes under a "user" namespace), keeping internal state out of reach. In the k8s module the store is backed by the orchestrator API and scoped to the deployment; in the standalone module it is an in-process map.

Writes use optimistic concurrency: expectedVersion 0 creates the key (and fails if it already exists), while a positive expectedVersion must equal the stored version. A mismatch returns ErrVersionConflict. Successful writes return the new version.

type LLMClient ¶

type LLMClient interface {
	// Complete runs a single chat/completion turn. The request carries the full
	// conversation so far (system + messages) and any tool definitions; the
	// response is either assistant text, a set of tool calls the model wants run,
	// or both. Callers drive multi-turn tool loops by appending the assistant
	// turn (LLMResponse.Raw) and the tool results, then calling again.
	Complete(ctx context.Context, req LLMRequest) (*LLMResponse, error)
}

LLMClient is the provider-agnostic capability the AI elements (ai-router, ai-agent, ai-mapping, ai-retry) depend on. The provider connectors (llm-anthropic, llm-openai, llm-gemini) satisfy it and translate these DTOs to and from their SDK types.

The interface lives in core (not a connector package) on purpose: the AI composites are built by the flow builder in core/internal/engine, and core cannot import the connector packages without a cycle. So an AI element resolves a connector by name through BlockDeps.Connector and type-asserts the result to LLMClient — the interface, not a concrete connector type. This is the deliberate divergence from the concrete-type assertion other blocks use (e.g. the rest block asserting *httpclient.Connector), forced by the requirement that an AI element work with any provider.

Implementations must be safe for concurrent use: one connector instance is shared across all flows that reference it.

type LLMMessage ¶

type LLMMessage struct {
	Role        LLMRole
	Text        string
	ToolCalls   []LLMToolCall
	ToolResults []LLMToolResult
}

LLMMessage is one turn in the conversation. Which fields are populated depends on Role: user/assistant turns carry Text, an assistant turn may also carry ToolCalls, and a tool turn carries ToolResults.

type LLMRequest ¶

type LLMRequest struct {
	// System is the system prompt. It is provider-routed to the dedicated
	// system slot rather than prepended as a message. May be empty.
	System string
	// Messages is the ordered conversation: user turns, prior assistant turns
	// (which may carry ToolCalls), and tool turns (which carry ToolResults).
	Messages []LLMMessage
	// Tools are the function definitions the model may call. May be empty for a
	// plain text completion (e.g. ai-mapping).
	Tools []LLMTool
	// ToolChoice constrains whether and which tool the model must call. The zero
	// value is auto (the model decides).
	ToolChoice LLMToolChoice
	// MaxTokens caps the response length. Zero means the connector's default.
	MaxTokens int
}

LLMRequest is one completion turn. The shape mirrors the Anthropic Messages tool-use loop (system separate from the conversation, explicit tool-call IDs, tool results as their own turn) because it is the most expressive of the three providers and maps cleanly onto OpenAI and Gemini.

type LLMResponse ¶

type LLMResponse struct {
	Text       string
	ToolCalls  []LLMToolCall
	StopReason LLMStopReason
	Raw        LLMMessage
}

LLMResponse is the result of one completion turn. Text is the assembled text output; ToolCalls is set when StopReason is LLMStopToolUse. Raw is the assistant turn as an LLMMessage, ready to append back onto LLMRequest.Messages when driving a tool loop.

type LLMRole ¶

type LLMRole string

LLMRole identifies who produced a message.

const (
	// LLMRoleUser is an end-user / caller turn.
	LLMRoleUser LLMRole = "user"
	// LLMRoleAssistant is a model turn; it may carry ToolCalls.
	LLMRoleAssistant LLMRole = "assistant"
	// LLMRoleTool is a turn carrying the results of tool calls (ToolResults).
	LLMRoleTool LLMRole = "tool"
)

type LLMStopReason ¶

type LLMStopReason string

LLMStopReason is why the model stopped generating.

const (
	// LLMStopEndTurn is a normal completion.
	LLMStopEndTurn LLMStopReason = "end_turn"
	// LLMStopToolUse means the model wants tools run; ToolCalls is populated.
	LLMStopToolUse LLMStopReason = "tool_use"
	// LLMStopMaxTokens means the response hit the token cap.
	LLMStopMaxTokens LLMStopReason = "max_tokens"
	// LLMStopRefusal means the model declined to answer.
	LLMStopRefusal LLMStopReason = "refusal"
)

type LLMTool ¶

type LLMTool struct {
	Name        string
	Description string
	InputSchema json.RawMessage
}

LLMTool is a function the model may call. InputSchema is a JSON Schema object describing the arguments; it is passed through to the provider verbatim.

type LLMToolCall ¶

type LLMToolCall struct {
	ID    string
	Name  string
	Input json.RawMessage
	// Signature is an opaque, provider-specific continuation token the model
	// attaches to a tool call that must be echoed back verbatim on the next turn
	// for a multi-turn tool conversation to stay valid (Gemini 3.x thought
	// signatures). It is empty for providers that do not use one; callers treat it
	// as opaque and never inspect or construct it — they only carry it back via
	// LLMResponse.Raw.
	Signature []byte
}

LLMToolCall is a request from the model to run a tool. ID correlates the call with its later LLMToolResult. Providers that do not supply IDs (Gemini) have their connector synthesize a stable one. Input is the arguments as a JSON object.

type LLMToolChoice ¶

type LLMToolChoice struct {
	Mode LLMToolChoiceMode
	Name string
}

LLMToolChoice constrains tool calling. Name is used only when Mode is LLMToolChoiceTool.

type LLMToolChoiceMode ¶

type LLMToolChoiceMode string

LLMToolChoiceMode selects the tool-calling policy for a request.

const (
	// LLMToolChoiceAuto lets the model decide whether to call a tool. Zero value.
	LLMToolChoiceAuto LLMToolChoiceMode = ""
	// LLMToolChoiceAny forces the model to call some tool (its choice which).
	LLMToolChoiceAny LLMToolChoiceMode = "any"
	// LLMToolChoiceNone forbids tool calls.
	LLMToolChoiceNone LLMToolChoiceMode = "none"
	// LLMToolChoiceTool forces the model to call the tool named in
	// LLMToolChoice.Name.
	LLMToolChoiceTool LLMToolChoiceMode = "tool"
)

type LLMToolResult ¶

type LLMToolResult struct {
	ToolCallID string
	Content    string
	IsError    bool
}

LLMToolResult is the outcome of a tool call fed back to the model. ToolCallID must match the originating LLMToolCall.ID. Content is the serialized result; IsError marks it as a failure the model should react to rather than an answer.

type LeaderElection ¶

type LeaderElection interface {
	//nolint:ireturn // returns the Leadership interface the caller gates work on
	Acquire(ctx context.Context, key string) (Leadership, error)
}

LeaderElection lets a connector run work on exactly one replica across a cluster. Acquire starts campaigning for key in the background and returns a Leadership handle whose IsLeader tracks the current status. In the standalone module every Acquire is immediately and permanently the leader (a single process).

func NoopLeaderElection() LeaderElection

type Leadership ¶

type Leadership interface {
	// IsLeader reports whether this replica currently holds the key. It is safe to
	// call concurrently and cheap (it reads cached state, it does not block on the
	// backend).
	IsLeader() bool
	// Close stops campaigning for the key and releases it if held.
	Close() error
}

Leadership is a handle to an ongoing campaign for a leader-election key. Its IsLeader reports whether this replica currently holds leadership; Close stops campaigning and releases the key (best-effort). A connector typically acquires one per unit of exclusive work and gates that work on IsLeader.

type LogShipper ¶

type LogShipper interface {
	// LogSink returns a handler that forwards records to the central sink, or nil
	// when this module ships no logs.
	LogSink() slog.Handler
}

LogShipper is an optional capability a RuntimeServices module may implement to ship log records to a central destination (e.g. a NATS subject) in addition to the process's normal output. Callers type-assert the active RuntimeServices to LogShipper and, when the assertion holds and LogSink returns a non-nil handler, tee their slog handler through it. The standalone module ships nothing, so LogSink may return nil and callers MUST nil-check before wiring it.

type MessageProcessor ¶

type MessageProcessor interface {
	Process(ctx context.Context, msg *types.Message) (*types.Message, error)
}

MessageProcessor transforms a single message. It is the unit of work a Block wraps. Because one processor instance is shared across all workers in a flow, implementations must be safe for concurrent use.

Returning (nil, nil) drops the message: it is filtered out and the rest of the chain is skipped. A non-nil error aborts the message.

type MessageSource ¶

type MessageSource interface {
	Start(ctx context.Context) error
	Stop(ctx context.Context) error
}

MessageSource is a flow's entry point, created and owned by a connector. It responds to connector events by building a *types.Message and sending it on the output channel handed to it at construction.

The runtime owns the channel's lifecycle: Start must not send after Stop returns, and the runtime closes the channel only after Stop completes. Start must not block; it acquires resources and returns, doing its work on its own goroutine(s).

type MockCase ¶

type MockCase struct {
	When string `json:"when,omitempty" yaml:"when,omitempty"`
	// Body replaces the message body. Vars are set on the message alongside it, for
	// the blocks that report through a variable rather than the body (an http call
	// setting vars.status). Vars without a Body is rejected: it would be a case that
	// says nothing about what the block returned.
	Body any            `json:"body,omitempty" yaml:"body,omitempty"`
	Vars map[string]any `json:"vars,omitempty" yaml:"vars,omitempty"`
	// Error fails the block with this message, which is how an error path is tested
	// without arranging for the real block to fail.
	Error string `json:"error,omitempty" yaml:"error,omitempty"`
	// Drop filters the message out, as a filter block would.
	Drop bool `json:"drop,omitempty" yaml:"drop,omitempty"`
}

MockCase is one canned outcome of a mocked block: a CEL condition, and what the block should do when it holds. Exactly one of Body, Error and Drop is set — a block either produces a message, fails, or filters the message out, and a mock stands in for a block.

When is evaluated against the message the block *received*, so a mock dispatches on its input the way the real block would: `body.amount > 100`, `vars.tier == "vip"`. It is the only expression a mock carries — Body is a literal value, not an expression, because a mock is a canned response and the dispatch already happened in When.

type MockSpec ¶

type MockSpec struct {
	Cases   []MockCase `json:"cases,omitempty"   yaml:"cases,omitempty"`
	Default *MockCase  `json:"default,omitempty" yaml:"default,omitempty"`
}

MockSpec is what one addressed block is mocked with: the cases, tried in order, and an optional default for when none of them matched.

With no default, a message that matches no case fails the block. That is deliberate, and it is not the same as "fall through to the real block": the mock replaced the block, so there is no real block left to fall through to. Keeping one wired would mean a mocked HTTP call or LLM call could still reach the network and spend money on a run the user believes is mocked — which is the failure mocking exists to prevent. An unmatched message is a hole in the mock, and saying so is more useful than silently doing nothing. The escape hatches are one line: a `default`, or a trailing case with `when: "true"`.

type Mocks ¶

type Mocks struct {
	// contains filtered or unexported fields
}

Mocks is the set of blocks to stand in for, keyed by address. Unlike Breakpoint and Spies it collects nothing: a mock is an input to the run, not an observation of it. It reaches the engine the same way they do — through BlockDeps — so that a `type: mock` block cannot be authored in a flow, only injected.

func NewMocks(specs map[string]MockSpec) (*Mocks, error)

func (m *Mocks) Addresses() []string

func (m *Mocks) For(address string) (MockSpec, bool)

type NoopResourceLoader ¶

type NoopResourceLoader struct{}

NoopResourceLoader is the fallback ResourceLoader used when none is wired: every Load reports the resource missing. It does not implement ResourceWatcher, so it is never watched. It is also what the byte-parse config entrypoint uses, where declared env resources simply never load — which is correct, since a missing env resource is not a runtime error.

func (NoopResourceLoader) Load(context.Context, ResourceKind, string) ([]byte, error)

type QueueHandler ¶

type QueueHandler func(ctx context.Context, msg types.Message) (reply types.Message, err error)

QueueHandler processes one inbound message and may return a reply. The reply is sent only when the sender requested one (a Request, not a Publish); for a fire-and-forget message it is ignored. A non-nil error is logged by the module and, for a Request, surfaces to the requester as a failed request.

type Queues ¶

type Queues interface {
	// Publish sends msg to subject for exactly one competing consumer. It does not
	// wait for, or expect, a reply.
	Publish(ctx context.Context, subject string, msg types.Message) error
	// Request sends msg to subject and waits for one reply, bounded by ctx and the
	// configured timeout (WithTimeout, else DefaultRequestTimeout).
	Request(ctx context.Context, subject string, msg types.Message, opts ...RequestOption) (types.Message, error)
	// Subscribe joins the competing-consumer group on subject. The handler runs on
	// listeners concurrent goroutines (WithListeners, else DefaultListeners). When
	// the inbound message carries a reply destination, the reply the handler
	// returns is delivered to the requester; otherwise it is dropped. The returned
	// Subscription stops consuming when closed.
	//nolint:ireturn // returns the Subscription interface the caller stores
	Subscribe(ctx context.Context, subject string, handler QueueHandler, opts ...SubscribeOption) (Subscription, error)
}

Queues is a deployment-scoped message queue available to connectors and blocks. It offers two send shapes over one competing-consumer model: Publish is point-to-point (exactly one consumer handles each message, no reply), and Request is request-reply (the caller waits for one reply). The consumer side is the same Subscribe for both — whether a reply is sent is decided by whether the inbound message carries a reply destination.

In the standalone module queues are in-process (buffered channels); in the k8s module they are backed by NATS (queue-group subscriptions and native request- reply). Delivery is at-most-once: a message published with no live consumer is dropped.

func NoopQueues() Queues

type Registry ¶

type Registry struct {
	// contains filtered or unexported fields
}

Registry holds connector factories keyed by connector type.

func DefaultRegistry() *Registry

func NewRegistry() *Registry

func (r *Registry) Has(name string) bool

func (r *Registry) MustRegister(name string, factory Factory)

func (r *Registry) Names() []string

func (r *Registry) New(name string) (Connector, error)

func (r *Registry) Register(name string, factory Factory) error

type RequestConfig ¶

type RequestConfig struct {
	// Timeout bounds the wait for a reply when the request context has no deadline.
	Timeout time.Duration
}

RequestConfig is the resolved configuration for a request. Modules build it from the caller's options with NewRequestConfig.

func NewRequestConfig(opts ...RequestOption) RequestConfig

type RequestOption ¶

type RequestOption func(*RequestConfig)

RequestOption configures a Request call.

func WithTimeout(d time.Duration) RequestOption

type ResourceKind ¶

type ResourceKind string

ResourceKind identifies the kind of resource a ResourceLoader serves. It lets a loader route kinds to different backends (a k8s loader might read env from a ConfigMap and templates from another store); the filesystem loader ignores it because the id alone maps to a path.

const (
	// ResourceKindEnv is a .env-convention file combined into the runtime environment.
	ResourceKindEnv ResourceKind = "env"
	// ResourceKindTemplate is a text template that may embed {{ }} CEL expressions.
	ResourceKindTemplate ResourceKind = "template"
)

type ResourceLoader ¶

type ResourceLoader interface {
	// Load returns the bytes for the resource id of the given kind, or
	// ErrResourceNotFound when it does not exist.
	Load(ctx context.Context, kind ResourceKind, id string) ([]byte, error)
}

ResourceLoader reads resources into the runtime by id. The runtime only ever reads: creating, listing, and writing resource files is host-side (done from outside the runtime against disk or the orchestrator), so those operations are deliberately not on this interface.

The active implementation is chosen at startup: the standalone module resolves an id to a file under the config directory; the k8s module is a no-op for now.

type ResourceWatcher ¶

type ResourceWatcher interface {
	OnChange(ctx context.Context, fn ChangeFunc) error
}

ResourceWatcher is implemented by loaders that choose to emit change events. Implementing it is the opt-in: a loader that does not implement ResourceWatcher is simply never watched, so its resources load once per generation and are re-read only on an explicit reload. OnChange registers fn to be called whenever a resource changes, until ctx is done (which also stops the underlying watch).

type RuntimeServices ¶

type RuntimeServices interface {
	//nolint:ireturn // returns the LeaderElection interface a connector depends on
	LeaderElection() LeaderElection
	//nolint:ireturn // returns the KV interface a connector depends on
	KV() KV
	//nolint:ireturn // returns the SecretStore interface a connector depends on
	Secrets() SecretStore
	//nolint:ireturn // returns the Queues interface a connector depends on
	Queues() Queues
	//nolint:ireturn // returns the Topics interface a connector depends on
	Topics() Topics
	//nolint:ireturn // returns the ResourceLoader interface blocks and env loading depend on
	Resources() ResourceLoader
	Close() error
}

RuntimeServices is the set of generally-available services wired into the runtime execution context. The active implementation is chosen at startup by the RUNTIME_SERVICES_MODULE environment variable (standalone or k8s). Close releases the implementation's resources; the process owner (the CLI) owns its lifecycle, not an individual Service generation.

func NoopRuntimeServices() RuntimeServices

func RuntimeServicesFromContext(ctx context.Context) RuntimeServices

type SchemaRegistry ¶

type SchemaRegistry struct {
	// contains filtered or unexported fields
}

SchemaRegistry holds the registered schema metadata in registration order (the order the editor renders them). It is safe for concurrent registration.

func DefaultSchemaRegistry() *SchemaRegistry

func NewSchemaRegistry() *SchemaRegistry

func (r *SchemaRegistry) AddBlock(m BlockMeta)

func (r *SchemaRegistry) AddConnector(m ConnectorMeta)

func (r *SchemaRegistry) AddExtension(m ExtensionMeta)

func (r *SchemaRegistry) Blocks() []BlockMeta

func (r *SchemaRegistry) Connectors() []ConnectorMeta

type SecretStore ¶

type SecretStore interface {
	// Get returns the (decrypted) entry for key in namespace; ok is false when absent.
	Get(ctx context.Context, namespace, key string) (entry Entry, ok bool, err error)
	// Set encrypts and stores value, returning the new version.
	Set(ctx context.Context, namespace, key string, value []byte, expectedVersion int64) (newVersion int64, err error)
	// Delete removes key (see KV.Delete for the version semantics).
	Delete(ctx context.Context, namespace, key string, expectedVersion int64) error
}

SecretStore is a store for sensitive values. It has the same namespaced, versioned API as KV and shares its backing store, but it routes every operation to a dedicated secret namespace (NamespaceSystemSecrets / NamespaceUserSecrets) whose values the backend encrypts at rest. So secrets never collide with plain keys and ordinary KV traffic pays no encryption cost, without a second table. The caller passes the same logical namespaces it uses for KV (system/user); the store maps them to their secret counterparts.

func NewSecretStore(kv KV) SecretStore

type SourceDeps ¶

type SourceDeps struct {
	Resources ResourceLoader
}

SourceDeps carries build-time services a source may need beyond its own config. Most sources ignore it. Resources loads the resources a source's expressions may reach — a cron payload calling templateResource(id) renders through it. It is never nil: the runtime substitutes a no-op loader when none is wired.

type SourceMeta ¶

type SourceMeta struct {
	Type     string
	Label    string
	Icon     string
	Settings reflect.Type
	// SrcDir is populated from the owning connector's directory when empty.
	SrcDir string
}

SourceMeta is the editor-facing description of a connector source (e.g. the http route source). Settings is the zero-value source-settings struct whose tags describe its fields; nil means the source takes no fields.

type SourceProvider ¶

type SourceProvider interface {
	NewSource(cfg types.SourceConfig, out chan<- *types.Message, deps SourceDeps) (MessageSource, error)
}

SourceProvider is an optional capability a connector implements to supply sources for flows bound to it. The returned source closes over the connector's own resources (connections, transaction managers), which is why no separate globals registry is needed. cfg.Type selects which source the connector builds.

type Spies ¶

type Spies struct {
	// contains filtered or unexported fields
}

Spies collects what crossed each addressed block. It is the debug seam behind the CLI's `invoke --spies`: the runtime wraps each addressed block in an implicit spy block, which records every message through it and otherwise leaves the flow alone.

Like Breakpoint it is a collector rather than a return value, because a spied block can sit inside a composite that discards its sub-flow's message — a fork branch runs on a clone, an enrich body runs on a clone — so the records have to leave the flow out-of-band.

Unlike Breakpoint it appends rather than keeping the first write: a block inside a fork branch, a foreach body or an enrich runs many times, and every crossing is worth seeing. Those runs are concurrent, so Record is safe for concurrent use.

func NewSpies(addresses []string) (*Spies, error)

func (s *Spies) Addresses() []string

func (s *Spies) Record(address string, rec SpyRecord)

func (s *Spies) Records(address string) []SpyRecord

type SpyRecord ¶

type SpyRecord struct {
	// Seq orders records across every spy, not just this one. It is what makes a
	// multi-spy trace readable: block A ran before block B, this fork branch before
	// that one.
	Seq     int64          `json:"seq"`
	At      time.Time      `json:"at"`
	Input   *types.Message `json:"input"`
	Output  *types.Message `json:"output,omitempty"`
	Dropped bool           `json:"dropped,omitempty"`
	Error   string         `json:"error,omitempty"`
}

SpyRecord is one execution of a spied block: the message that went in, and what came out — including the two outcomes that are not a message, a drop and a failure. Exactly one of Output, Dropped and Error is set.

Input is the message as the block received it, captured before the block ran: blocks mutate the message in place and hand the same pointer back, so a record taken afterwards would show the input already overwritten by the output.

type SubscribeConfig ¶

type SubscribeConfig struct {
	// Listeners is the number of concurrent handler goroutines.
	Listeners int
}

SubscribeConfig is the resolved configuration for a subscription. Modules build it from the caller's options with NewSubscribeConfig.

func NewSubscribeConfig(opts ...SubscribeOption) SubscribeConfig

type SubscribeOption ¶

type SubscribeOption func(*SubscribeConfig)

SubscribeOption configures a Subscribe call.

func WithListeners(n int) SubscribeOption

type Subscription ¶

type Subscription interface {
	Close() error
}

Subscription is a handle to an active subscription. Close stops the handler goroutines and unsubscribes from the backend (best-effort).

type TopicHandler ¶

type TopicHandler func(ctx context.Context, msg types.Message) error

TopicHandler processes one broadcast message. A non-nil error is logged by the module and otherwise ignored — a topic has no requester to surface it to.

type Topics ¶

type Topics interface {
	// Publish broadcasts msg to every subscriber on subject. It does not wait for,
	// or expect, a reply.
	Publish(ctx context.Context, subject string, msg types.Message) error
	// Subscribe delivers every message on subject to handler, which runs on
	// listeners concurrent goroutines (WithListeners, else DefaultListeners). The
	// returned Subscription stops consuming when closed.
	//nolint:ireturn // returns the Subscription interface the caller stores
	Subscribe(ctx context.Context, subject string, handler TopicHandler, opts ...SubscribeOption) (Subscription, error)
}

Topics is a deployment-scoped broadcast pub/sub available to connectors and blocks. Unlike Queues (a competing-consumer model where each message reaches exactly one consumer), a Topics message is fanned out to every subscriber on the subject — the platform-events / topic model. There is no reply.

In the standalone module topics are in-process (fan-out to local subscribers); in the k8s module they are backed by NATS (a plain, non-queue subscription, so every replica's subscribers receive every message). Delivery is at-most-once: a message published with no live subscriber is dropped.

func NoopTopics() Topics