Documentation
¶
Overview ¶
Package loop implements the single-turn execution primitive for ore. It provides a Step type that invokes a provider, distributes all artifacts (both delta and complete) to subscribers via an embedded FanOut, and runs registered artifact handlers on the complete response.
Why use transforms?
Transforms inject virtual content (system prompts, guardrails, dynamic context) into the provider's view at inference time without mutating the persistent conversation history. This keeps the buffer append-only while still shaping every model call.
Options include:
- WithTransforms — modify the state view presented to the provider during inference. Transforms run before each provider call in Turn(), composing in registration order. They must not mutate the underlying persistent buffer; use state.NewVirtualTurnState to create derived views that prepend virtual turns. See x/systemprompt for a reusable transform that injects a system prompt without touching history.
- WithHandlers — run artifact handlers on the complete response.
Step is the single canonical single-turn execution primitive with optional, opt-in capabilities via functional options. A Step with no options is valid for non-streaming, non-handler use cases.
Conduits subscribe to specific artifact kinds via Step.Subscribe(), receiving ArtifactEvent wrappers (which satisfy OutputEvent via Kind()) as they are emitted by the provider. Each ArtifactEvent carries the underlying artifact and an EventContext for routing metadata. The artifact.Delta marker interface controls whether an artifact is persisted to state; it does NOT filter event-stream visibility. All artifacts are forwarded to subscribers.
Output event taxonomy:
- ArtifactEvent wraps a provider artifact (text, reasoning, tool_call, etc.) and is emitted for every chunk received from the provider.
- TurnCompleteEvent fires after each individual turn is fully accumulated, carrying the complete turn for state persistence and non-streaming delivery (e.g. Slack, Telegram).
- LifecycleEvent signals phase transitions: "submitted" (message accepted, waiting for provider), "streaming" (first artifact arrived), and "done" (turn complete). Conduits observe this to drive UI state without inferring lifecycle from data events.
- ErrorEvent fires when an individual turn fails inside the provider or a registered handler.
- PropertiesEvent carries ambient, persistent metadata as a map of key-value pairs (e.g. thread_id, token counts, model name). It is emitted by any component holding a *session.Stream and delivered to all subscribers through the per-session FanOut.
Index ¶
- type ArtifactEvent
- type Emitter
- type ErrorEvent
- type EventContext
- type FanOut
- type Handler
- type LifecycleEvent
- type OnEmit
- type Option
- type OutputEvent
- type PropertiesEvent
- type Step
- func (s *Step) Close() error
- func (s *Step) Emit(ctx context.Context, event OutputEvent)
- func (s *Step) SetEventContext(ctx EventContext)
- func (s *Step) Submit(ctx context.Context, st state.State, role state.Role, ...) (state.State, error)
- func (s *Step) Subscribe(kinds ...string) <-chan OutputEvent
- func (s *Step) Turn(ctx context.Context, st state.State, p provider.Provider, ...) (state.State, error)
- type Transform
- type TurnCompleteEvent
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type ArtifactEvent ¶
type ArtifactEvent struct {
Artifact artifact.Artifact
// Ctx carries routing metadata for the event, such as provenance
// information for echo suppression.
Ctx EventContext
}
ArtifactEvent wraps an artifact.Artifact with an EventContext so it can be emitted as an OutputEvent without polluting the artifact type with routing metadata.
func (ArtifactEvent) Context ¶
func (e ArtifactEvent) Context() EventContext
Context returns the event context.
func (ArtifactEvent) Kind ¶
func (e ArtifactEvent) Kind() string
Kind returns the underlying artifact's kind.
type Emitter ¶ added in v0.1.1
type Emitter interface {
Emit(ctx context.Context, event OutputEvent)
}
Emitter is the interface exposed to artifact handlers for emitting events back into the Step's output stream.
type ErrorEvent ¶
type ErrorEvent struct {
Err error
// Ctx carries routing metadata for the event, such as provenance
// information for echo suppression.
Ctx EventContext
}
ErrorEvent is emitted when a turn fails due to a provider or handler error.
func (ErrorEvent) Context ¶
func (e ErrorEvent) Context() EventContext
Context returns the event context.
type EventContext ¶
type EventContext struct {
Provenance string
}
EventContext carries metadata for an event, analogous to context.Context. It travels with an event through the event stream so subscribers can access routing metadata (provenance, trace IDs, etc.) uniformly.
type FanOut ¶
type FanOut struct {
// contains filtered or unexported fields
}
FanOut distributes OutputEvent values from a source channel to multiple subscribers, filtered by event kind.
func NewFanOut ¶
func NewFanOut(src <-chan outputEventEnvelope) *FanOut
NewFanOut creates a FanOut that reads from src and distributes events. The FanOut starts a background goroutine that reads from src until it is closed or the FanOut is closed.
func (*FanOut) Subscribe ¶
func (f *FanOut) Subscribe(kinds ...string) <-chan OutputEvent
Subscribe returns a receive-only channel that receives all OutputEvents whose Kind() matches any of the given kinds. If no kinds are provided, the channel receives all events regardless of kind. The channel is closed when the FanOut is closed.
Events are sent non-blocking with a fixed buffer of 100. If a subscriber falls behind and its buffer fills, subsequent matching events are dropped for that subscriber. The caller must read from the channel promptly to avoid missing events.
Subscribing to multiple kinds on one channel preserves ordering across those event types — events are delivered in the order they were received from the source.
type Handler ¶
type Handler interface {
// Handle processes a single artifact. It may emit events via the
// provided Emitter (e.g., emit a TurnCompleteEvent with RoleTool and
// tool results) or perform other side effects. Using Emitter instead
// of mutating state directly keeps all observable mutations on the same
// event stream, ensuring UI conduits receive tool results without
// special-casing.
Handle(ctx context.Context, art artifact.Artifact, e Emitter) error
}
Handler processes individual artifacts from an assistant turn. Multiple handlers may be registered on a Step; each handler inspects the artifact Kind() and acts only on types it understands.
type LifecycleEvent ¶ added in v0.2.0
type LifecycleEvent struct {
Phase string // "submitted", "streaming", "done"
// Ctx carries routing metadata for the event, such as provenance
// information for echo suppression.
Ctx EventContext
}
LifecycleEvent is emitted at structural boundaries of a single inference turn to signal phase transitions. Phases are linear per-pipeline:
- "submitted": the user message has been accepted and the provider call is about to start (after transforms).
- "streaming": the first artifact has arrived from the provider.
- "done": the turn (including accumulation and handler execution) is complete. For tool-execution loops, the framework returns to "submitted" for each subsequent provider call.
func (LifecycleEvent) Context ¶ added in v0.2.0
func (e LifecycleEvent) Context() EventContext
Context returns the event context.
func (LifecycleEvent) Kind ¶ added in v0.2.0
func (e LifecycleEvent) Kind() string
Kind returns the event kind identifier.
type OnEmit ¶ added in v0.1.1
type OnEmit func(ctx context.Context, event OutputEvent)
OnEmit is a synchronous callback invoked by Emit before the event is forwarded to the async FanOut. OnEmit callbacks are blocking, ordered, and zero-drop. They replace previous direct state.Append calls, ensuring lossless state updates while keeping the event stream observable for UI conduits. This is the canonical mechanism for wiring state persistence.
type Option ¶
type Option func(*Step)
Option configures a Step.
func WithHandlers ¶
WithHandlers configures artifact handlers to run after each turn.
func WithInvokeOptions ¶
func WithInvokeOptions(opts ...provider.InvokeOption) Option
WithInvokeOptions configures pre-bound provider invocation options that are automatically passed to every provider call made by this Step.
func WithOnEmit ¶ added in v0.1.1
WithOnEmit configures synchronous callbacks that run before the FanOut. OnEmit callbacks receive every OutputEvent emitted by the Step, including TurnCompleteEvent, ArtifactEvent, ErrorEvent, and LifecycleEvent. They are invoked in registration order, blocking, and zero-drop. This is the single place to wire state persistence, replacing previous patterns that mutated state directly inside Turn().
func WithTransforms ¶
WithTransforms configures inference assembly transforms that run before each provider call in Turn(). Transforms receive the state after any user/system/tool submissions and before the provider serializes it. They must not mutate the underlying buffer.
type OutputEvent ¶
type OutputEvent interface {
Kind() string
Context() EventContext
}
OutputEvent represents any event emitted by a Step. All output events carry an EventContext so subscribers can access routing metadata uniformly. Events include wrapped artifacts (ArtifactEvent), turn completions (TurnCompleteEvent), and errors (ErrorEvent).
type PropertiesEvent ¶ added in v0.2.0
type PropertiesEvent struct {
Properties map[string]string
// Ctx carries routing metadata for the event, such as provenance
// information for echo suppression.
Ctx EventContext
}
PropertiesEvent carries ambient, persistent metadata as a map of key-value pairs. It is emitted by any producer holding a *session.Stream and flows through the per-session FanOut so all conduits receive it simultaneously.
func (PropertiesEvent) Context ¶ added in v0.2.0
func (e PropertiesEvent) Context() EventContext
Context returns the event context.
func (PropertiesEvent) Kind ¶ added in v0.2.0
func (e PropertiesEvent) Kind() string
Kind returns the event kind identifier.
type Step ¶
type Step struct {
// contains filtered or unexported fields
}
Step executes a single complete inference turn: it invokes the provider, distributes streaming artifacts to subscribers via an embedded FanOut, and runs registered artifact handlers synchronously on the complete response.
func (*Step) Emit ¶
func (s *Step) Emit(ctx context.Context, event OutputEvent)
Emit runs all registered OnEmit callbacks synchronously, then sends the event to the FanOut and blocks until it has been delivered.
func (*Step) SetEventContext ¶
func (s *Step) SetEventContext(ctx EventContext)
SetEventContext sets the EventContext that will be attached to all subsequent output events emitted by this Step. It is used by Stream.Process to thread context from the input event through the turn pipeline. Callers must ensure this is called before Turn or Submit and cleared after (typically via defer).
func (*Step) Submit ¶
func (s *Step) Submit(ctx context.Context, st state.State, role state.Role, artifacts ...artifact.Artifact) (state.State, error)
Submit records a non-inference turn into state, runs registered handlers, and emits a TurnCompleteEvent to all subscribers. It is the canonical mechanism for user, system, or tool turns to enter the same artifact stream as assistant responses from Turn().
func (*Step) Subscribe ¶
func (s *Step) Subscribe(kinds ...string) <-chan OutputEvent
Subscribe returns a receive-only channel of OutputEvents whose Kind() matches any of the given kinds. The channel is closed when the Step's FanOut is closed. Events are delivered non-blocking; slow subscribers may drop events.
func (*Step) Turn ¶
func (s *Step) Turn(ctx context.Context, st state.State, p provider.Provider, opts ...provider.InvokeOption) (state.State, error)
Turn performs one inference turn with the given provider. The provider emits artifacts to a channel; all artifacts are forwarded to the Step's FanOut subscribers immediately as they arrive. Deltas implementing Accumulable are merged into blocks keyed by AccumulatorKey, so non-adjacent deltas of the same kind are combined into a single block. Accumulated blocks are flushed on non-delta boundaries and at stream end. The accumulated turn is appended to state once the provider returns. After the turn completes, all registered handlers are invoked on each artifact from the assistant turn. The operation is fully synchronous and blocking.
type Transform ¶
Transform modifies the state view presented to the provider during inference. Implementations must not mutate the underlying persistent buffer; they may return a derived state.State wrapper instead.
Multiple transforms compose in registration order. Each transform receives the state returned by the previous one. An error from any transform aborts the turn before the provider is invoked.
type TurnCompleteEvent ¶
type TurnCompleteEvent struct {
Turn state.Turn
// Ctx carries routing metadata for the event, such as provenance
// information for echo suppression.
Ctx EventContext
}
TurnCompleteEvent is emitted when a turn (assistant, user, system, or tool) has been fully constructed. OnEmit callbacks fire synchronously before the event reaches the async FanOut and may mutate persistent state; handlers run after OnEmit completes.
func (TurnCompleteEvent) Context ¶
func (e TurnCompleteEvent) Context() EventContext
Context returns the event context.
func (TurnCompleteEvent) Kind ¶
func (e TurnCompleteEvent) Kind() string
Kind returns the event kind identifier.
Source Files