loop

package
v0.10.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jun 15, 2026 License: AGPL-3.0 Imports: 12 Imported by: 0

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, and runs registered artifact handlers on the complete response.

Step is a thin orchestrator composed of two internal components:

  • EventBus — owns the broadcast infrastructure (events channel, FanOut, OnEmit callbacks, and bound state for auto-append). It is the single gateway for all observable mutations emitted by loop components.
  • Pipeline — the single-turn execution engine. It runs transforms, invokes the provider, accumulates streaming artifacts (with delta merging), and executes registered handlers.

Step sequences events around Pipeline.Turn(): emits "submitted" lifecycle event, runs the pipeline with an artifact callback that emits streaming events, then emits the final TurnCompleteEvent and runs handlers.

Public interfaces enable middleware composition without depending on the concrete *Step type:

  • TurnRunner — runs a single inference turn (Turn).
  • TurnSubmitter — records a non-inference turn (Submit).
  • TurnExecutor — combines both TurnRunner and TurnSubmitter.

Architecture overview: 

      Step (thin orchestrator)
       │
Emit() │          Turn()
       │            │
       ▼            ▼
  EventBus ◄── Pipeline (transforms, provider, accumulate, handlers)
       │            │
       │     onArtifact callback
       │            │
       │            ▼
       │      subscribers (channels)
       │
       └──► FanOut (broadcast)

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.Prepend or state.NewView 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.
  • WithState — bind a mutable state.State so that TurnCompleteEvent is automatically appended by Emit before OnEmit callbacks run. This is the canonical mechanism for state persistence; use WithOnEmit only for custom side-effects that do not mutate state.

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 a context.Context 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.

Subscribe is live-only: it delivers events from the point of subscription onward and does not replay historical events. Conduits that need historical state should fetch it directly from the bound state.State (via Turns() or the store) rather than relying on the event stream.

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) and "streaming" (first artifact arrived). Callers (e.g. session.Stream.Process) may emit "done" to signal pipeline completion. 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.

See also: cognitive package — composable middleware patterns (ReAct, WithVerification) that depend on the TurnRunner and TurnSubmitter interfaces.

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

func ProvenanceFrom added in v0.5.1 

func ProvenanceFrom(ctx context.Context) (string, bool)

ProvenanceFrom extracts the provenance name from a context, if present.

func ThreadIDFrom added in v0.5.1 

func ThreadIDFrom(ctx context.Context) (string, bool)

ThreadIDFrom extracts the thread ID from a context, if present.

func WithProvenance added in v0.5.1 

func WithProvenance(ctx context.Context, name string) context.Context

WithProvenance attaches a provenance name to the context.

func WithThreadID added in v0.5.1 

func WithThreadID(ctx context.Context, id string) context.Context

WithThreadID attaches a thread ID to the context.

Types

type ActivityEvent added in v0.7.3 

type ActivityEvent struct {
	Active      bool
	Description string

	// Ctx carries routing metadata for the event, such as provenance
	// information for echo suppression.
	Ctx context.Context
}

ActivityEvent signals that long-running operational work is happening. It is orthogonal to the inference lifecycle and is used for slash commands, tool execution, or any other background work that conduits should surface.

func (ActivityEvent) Context added in v0.7.3 

func (e ActivityEvent) Context() context.Context

Context returns the event context.

func (ActivityEvent) Kind added in v0.7.3 

func (e ActivityEvent) Kind() string

Kind returns the event kind identifier.

func (ActivityEvent) MarshalJSON added in v0.7.3 

func (e ActivityEvent) MarshalJSON() ([]byte, error)

MarshalJSON serializes the event to JSON.

type ArtifactEvent 

type ArtifactEvent struct {
	Artifact artifact.Artifact

	// Ctx carries routing metadata for the event, such as provenance
	// information for echo suppression.
	Ctx context.Context
}

ArtifactEvent wraps an artifact.Artifact with a context.Context so it can be emitted as an OutputEvent without polluting the artifact type with routing metadata.

func (ArtifactEvent) Context 

func (e ArtifactEvent) Context() context.Context

Context returns the event context.

func (ArtifactEvent) Kind 

func (e ArtifactEvent) Kind() string

Kind returns the underlying artifact's kind.

func (ArtifactEvent) MarshalJSON added in v0.6.0 

func (e ArtifactEvent) MarshalJSON() ([]byte, error)

MarshalJSON serializes the event to JSON. It merges the artifact's JSON with an optional context envelope.

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 context.Context
}

ErrorEvent is emitted when a turn fails due to a provider or handler error.

func (ErrorEvent) Context 

func (e ErrorEvent) Context() context.Context

Context returns the event context.

func (ErrorEvent) Kind 

func (e ErrorEvent) Kind() string

Kind returns the event kind identifier.

func (ErrorEvent) MarshalJSON added in v0.6.0 

func (e ErrorEvent) MarshalJSON() ([]byte, error)

MarshalJSON serializes the event to JSON.

type EventBus added in v0.6.2 

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

EventBus owns the broadcast infrastructure: event channel, FanOut, synchronous OnEmit callbacks, and the bound state for auto-append. It is the single gateway for all observable mutations emitted by loop components. Step delegates Emit, Subscribe, and Close to it.

func (*EventBus) Close added in v0.6.2 

func (eb *EventBus) Close() error

Close stops the EventBus's FanOut and closes all subscriber channels.

func (*EventBus) Emit added in v0.6.2 

func (eb *EventBus) 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. When a state has been bound via WithState, only TurnCompleteEvent is automatically appended to that state before OnEmit callbacks run. Other event types are passed through unchanged.

func (*EventBus) Subscribe added in v0.6.2 

func (eb *EventBus) 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 EventBus's FanOut is closed. Events are delivered non-blocking; slow subscribers may drop events.

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.

FanOut is a pure live broadcast dispatcher. Only subscribers that are registered at the time an event is sent receive that event. There is no replay buffer; callers that need historical state must use stream.Turns() or another explicit state mechanism.

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) Close 

func (f *FanOut) Close() error

Close stops the FanOut and closes all subscriber channels.

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.

Subscribe subscribes to live events only; it does not replay historical events. Callers that need history must use stream.Turns() or another explicit state mechanism.

Events are sent with a fixed buffer of 100000. If a subscriber falls behind and its buffer fills, send() applies backpressure to the entire FanOut rather than dropping events. The caller must read from the channel promptly to prevent head-of-line blocking for other subscribers.

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 FeedbackEvent added in v0.7.0 

type FeedbackEvent struct {
	Content string

	// Ctx carries routing metadata for the event, such as provenance
	// information for echo suppression.
	Ctx context.Context
}

FeedbackEvent is emitted for ephemeral, user-visible messages that bypass the LLM pipeline and are not persisted to state. Conduits render them as system-styled messages in the chat body — scrollable, but invisible to the LLM on the next turn.

func (FeedbackEvent) Context added in v0.7.0 

func (e FeedbackEvent) Context() context.Context

Context returns the event context.

func (FeedbackEvent) Kind added in v0.7.0 

func (e FeedbackEvent) Kind() string

Kind returns the event kind identifier.

func (FeedbackEvent) MarshalJSON added in v0.7.0 

func (e FeedbackEvent) MarshalJSON() ([]byte, error)

MarshalJSON serializes the event to JSON.

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"

	// Ctx carries routing metadata for the event, such as provenance
	// information for echo suppression.
	Ctx context.Context
}

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.

func (LifecycleEvent) Context added in v0.2.0 

func (e LifecycleEvent) Context() context.Context

Context returns the event context.

func (LifecycleEvent) Kind added in v0.2.0 

func (e LifecycleEvent) Kind() string

Kind returns the event kind identifier.

func (LifecycleEvent) MarshalJSON added in v0.6.0 

func (e LifecycleEvent) MarshalJSON() ([]byte, error)

MarshalJSON serializes the event to JSON.

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 

func WithHandlers(handlers ...Handler) Option

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 

func WithOnEmit(fns ...OnEmit) Option

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 WithState added in v0.5.2 

func WithState(st state.State) Option

WithState binds a mutable state.State to the Step so that every TurnCompleteEvent emitted by the Step (including from finalizeTurn and from artifact handlers) is automatically appended to the state before OnEmit callbacks run. The supplied state is mutated in-place. When no state is bound, TurnCompleteEvent has no automatic side effects on state.

WithState is the canonical mechanism for state persistence. Use WithOnEmit only for custom side-effects that do not also append to the same state, or duplicate turns will result.

func WithTracer added in v0.5.1 

func WithTracer(tracer trace.Tracer) Option

WithTracer configures an OpenTelemetry tracer for the Step. When configured, Turn and Submit create spans for each inference turn.

func WithTransforms 

func WithTransforms(transforms ...Transform) Option

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() context.Context
}

OutputEvent represents any event emitted by a Step. All output events carry a context.Context so subscribers can access routing metadata uniformly. Events include wrapped artifacts (ArtifactEvent), turn completions (TurnCompleteEvent), and errors (ErrorEvent).

type Pipeline added in v0.6.2 

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

Pipeline is the single-turn execution engine. It runs transforms, invokes the provider, accumulates streaming artifacts (with delta merging), and executes registered handlers.

func (*Pipeline) RunHandlers added in v0.6.2 

func (p *Pipeline) RunHandlers(ctx context.Context, artifacts []artifact.Artifact, emitter Emitter) error

RunHandlers executes all registered handlers on each artifact, using the provided Emitter for any handler-side emissions.

func (*Pipeline) Turn added in v0.6.2 

func (p *Pipeline) Turn(ctx context.Context, st state.State, prov provider.Provider, onArtifact func(artifact.Artifact), opts ...provider.InvokeOption) (state.State, []artifact.Artifact, error)

Turn performs one inference turn: runs transforms, calls the provider, accumulates artifacts, and invokes onArtifact for each artifact (including deltas and flushed accumulated blocks). Returns the final accumulated artifacts and any error.

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 context.Context
}

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() context.Context

Context returns the event context.

func (PropertiesEvent) Kind added in v0.2.0 

func (e PropertiesEvent) Kind() string

Kind returns the event kind identifier.

func (PropertiesEvent) MarshalJSON added in v0.6.0 

func (e PropertiesEvent) MarshalJSON() ([]byte, error)

MarshalJSON serializes the event to JSON.

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 EventBus, and runs registered artifact handlers synchronously on the complete response.

func New 

func New(opts ...Option) *Step

New creates a Step with the given options.

func (*Step) Close 

func (s *Step) Close() error

Close stops the Step's FanOut and closes all subscriber channels.

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. When a state has been bound via WithState, only TurnCompleteEvent is automatically appended to that state before OnEmit callbacks run. Other event types are passed through unchanged.

func (*Step) SetEventContext 

func (s *Step) SetEventContext(ctx context.Context)

SetEventContext sets the context.Context 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 EventBus 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 

type Transform interface {
	Transform(ctx context.Context, st state.State) (state.State, error)
}

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 context.Context
}

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() context.Context

Context returns the event context.

func (TurnCompleteEvent) Kind 

func (e TurnCompleteEvent) Kind() string

Kind returns the event kind identifier.

func (TurnCompleteEvent) MarshalJSON added in v0.6.0 

func (e TurnCompleteEvent) MarshalJSON() ([]byte, error)

MarshalJSON serializes the event to JSON.

type TurnExecutor added in v0.6.2 

type TurnExecutor interface {
	TurnRunner
	TurnSubmitter
}

TurnExecutor combines TurnRunner and TurnSubmitter into a single interface for cognitive patterns that need both capabilities. Step satisfies this interface via delegation.

type TurnRunner added in v0.6.2 

type TurnRunner interface {
	Turn(ctx context.Context, st state.State, p provider.Provider, opts ...provider.InvokeOption) (state.State, error)
}

TurnRunner runs a single inference turn with a provider. Implementations invoke the provider, accumulate artifacts, and emit streaming events. The canonical implementation is Step.Turn.

type TurnSubmitter added in v0.6.2 

type TurnSubmitter interface {
	Submit(ctx context.Context, st state.State, role state.Role, artifacts ...artifact.Artifact) (state.State, error)
}

TurnSubmitter records a non-inference turn into state and emits a TurnCompleteEvent. The canonical implementation is Step.Submit.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.