stages

type ArenaAssertionStage ¶

type ArenaAssertionStage struct {
	stage.BaseStage
	// contains filtered or unexported fields
}

ArenaAssertionStage validates assertions after LLM responses.

func NewArenaAssertionStage(
	assertionConfigs []assertions.AssertionConfig,
) *ArenaAssertionStage

func (s *ArenaAssertionStage) Process(
	ctx context.Context,
	input <-chan stage.StreamElement,
	output chan<- stage.StreamElement,
) error

func (s *ArenaAssertionStage) WithTurnEvalRunner(runner TurnEvalRunner, sessionID string) *ArenaAssertionStage

type ArenaStateStoreSaveStage ¶

type ArenaStateStoreSaveStage struct {
	stage.BaseStage
	// contains filtered or unexported fields
}

ArenaStateStoreSaveStage saves conversation state with telemetry to ArenaStateStore. This stage captures validation results, turn metrics, and cost information for Arena testing and analysis.

func NewArenaStateStoreSaveStage(config *pipeline.StateStoreConfig) *ArenaStateStoreSaveStage

func NewArenaStateStoreSaveStageWithTurnState(
	config *pipeline.StateStoreConfig, turnState *stage.TurnState,
) *ArenaStateStoreSaveStage

func (s *ArenaStateStoreSaveStage) Process(
	ctx context.Context, input <-chan stage.StreamElement, output chan<- stage.StreamElement,
) error

func (s *ArenaStateStoreSaveStage) WithEmitter(emitter *events.Emitter) *ArenaStateStoreSaveStage

type AssistantTTSFilterStage ¶ added in v1.5.3

type AssistantTTSFilterStage struct {
	stage.BaseStage
}

AssistantTTSFilterStage sits between the ArenaStateStoreSaveStage and the TTSStageWithInterruption in the VAD composed pipeline. It drops any Message element whose Role is not "assistant", preventing the user's own transcript (wrapped as a user Message by STTUserMessageStage) from being spoken back through TTS.

All non-Message elements — audio, text, EndOfStream, and control elements — are forwarded unchanged so the TTS stage receives everything it needs.

This is a Filter stage: it passes through assistant Messages and non-Message elements, and drops user/system/tool Messages.

func NewAssistantTTSFilterStage() *AssistantTTSFilterStage

func (s *AssistantTTSFilterStage) Process(
	ctx context.Context,
	input <-chan stage.StreamElement,
	output chan<- stage.StreamElement,
) error

type CompletionInstructionStage ¶ added in v1.3.16

type CompletionInstructionStage struct {
	stage.BaseStage
	// contains filtered or unexported fields
}

CompletionInstructionStage appends natural termination instructions to the system prompt assembled by PersonaAssemblyStage. The instruction is appended exactly once per Turn — TurnState.SystemPrompt is the source of truth and the stage is idempotent across the elements that flow through it.

func NewCompletionInstructionStageWithTurnState(
	instruction string, turnState *stage.TurnState,
) *CompletionInstructionStage

func (s *CompletionInstructionStage) Process(ctx context.Context, input <-chan stage.StreamElement, output chan<- stage.StreamElement) error

type HistoryInjectionStage ¶

type HistoryInjectionStage struct {
	stage.BaseStage
	// contains filtered or unexported fields
}

HistoryInjectionStage prepends conversation history to the pipeline stream. This is useful for self-play scenarios where history needs to be explicitly provided before the LLM generates the next user message.

The stage emits all history messages first, then forwards any incoming elements.

When swapRoles is true, user↔assistant roles are swapped so that the self-play LLM sees its own prior outputs as "assistant" and the target's responses as "user". This prevents the self-play LLM from confusing itself with the target assistant.

func NewHistoryInjectionStage(history []types.Message) *HistoryInjectionStage

func NewHistoryInjectionStageSwapped(history []types.Message) *HistoryInjectionStage

func (s *HistoryInjectionStage) Process(ctx context.Context, input <-chan stage.StreamElement, output chan<- stage.StreamElement) error

type MockScenarioContextStage ¶

type MockScenarioContextStage struct {
	stage.BaseStage
	// contains filtered or unexported fields
}

MockScenarioContextStage adds scenario context to TurnState's ProviderRequestMetadata so MockProvider can select scenario-specific canned responses.

This stage should be placed before ProviderStage in the pipeline when using MockProvider to ensure scenario context is available.

func NewMockScenarioContextStageWithTurnState(
	scenario *config.Scenario, turnState *stage.TurnState,
) *MockScenarioContextStage

func (s *MockScenarioContextStage) Process(ctx context.Context, input <-chan stage.StreamElement, output chan<- stage.StreamElement) error

type PersonaAssemblyStage ¶

type PersonaAssemblyStage struct {
	stage.BaseStage
	// contains filtered or unexported fields
}

PersonaAssemblyStage assembles persona prompts using the same fragment/template system as PromptAssemblyStage.

This stage mirrors the behavior of PromptAssemblyMiddleware but for personas:

• Uses persona's BuildSystemPrompt() which handles fragment assembly
• Supports template variable substitution with {{variable}} syntax
• Injects persona-specific variables (goals, constraints, style)
• Optionally prepends a TTS characterization rubric when the persona has opted in via style.expressive (see issue #1130)
• Writes the rendered system prompt and base variables into TurnState

func NewPersonaAssemblyStageWithTurnState(
	persona *config.UserPersonaPack,
	region string,
	baseVariables map[string]string,
	turnState *stage.TurnState,
) *PersonaAssemblyStage

func (s *PersonaAssemblyStage) Process(ctx context.Context, input <-chan stage.StreamElement, output chan<- stage.StreamElement) error

func (s *PersonaAssemblyStage) WithProviderRubric(rubric string) *PersonaAssemblyStage

type STTUserMessageStage ¶ added in v1.5.3

type STTUserMessageStage struct {
	stage.BaseStage
}

STTUserMessageStage converts a transcribed text element (as produced by the STT stage) into a user Message element so the downstream prompt-assembly, provider, and state-store-save stages treat the spoken turn exactly like a typed user message. This is the bridge that makes the composed VAD voice pipeline materialize history identically to a text run: ProviderStage only accumulates Message elements, so a bare Text element from STT would otherwise never reach the provider or be persisted.

Text-only elements are wrapped; elements that already carry a Message, plus audio / EndOfStream / non-text elements, are forwarded unchanged.

This is a Transform stage: text element -> user message element (1:1).

func NewSTTUserMessageStage() *STTUserMessageStage

func (s *STTUserMessageStage) Process(
	ctx context.Context,
	input <-chan stage.StreamElement,
	output chan<- stage.StreamElement,
) error

type ScenarioContextExtractionStage ¶

type ScenarioContextExtractionStage struct {
	stage.BaseStage
	// contains filtered or unexported fields
}

ScenarioContextExtractionStage extracts context using scenario metadata and conversation history. This is designed for Arena use where rich scenario metadata is available.

When scenario metadata is present, it uses: - Scenario metadata variables (domain, user role from scenario definition) - Scenario description and task type - Message analysis as fallback

Extracted variables are merged into TurnState.Variables, where TemplateStage reads them when rendering the system prompt.

func NewScenarioContextExtractionStageWithTurnState(
	scenario *config.Scenario, turnState *stage.TurnState,
) *ScenarioContextExtractionStage

func (s *ScenarioContextExtractionStage) Process(ctx context.Context, input <-chan stage.StreamElement, output chan<- stage.StreamElement) error

type SelfPlayUserTurnContextStage ¶

type SelfPlayUserTurnContextStage struct {
	stage.BaseStage
	// contains filtered or unexported fields
}

SelfPlayUserTurnContextStage adds scenario context for the NEXT user turn (completed user turns + 1). Intended only for self-play user generation.

This stage writes scenario/turn coordination keys into TurnState's ProviderRequestMetadata so the mock provider can select the appropriate canned response by scenario id, turn number, and persona.

func NewSelfPlayUserTurnContextStageWithHintAndTurnState(
	scenario *config.Scenario,
	turnIndexHint int,
	personaID string,
	turnState *stage.TurnState,
) *SelfPlayUserTurnContextStage

func NewSelfPlayUserTurnContextStageWithTurnState(
	scenario *config.Scenario,
	personaID string,
	turnState *stage.TurnState,
) *SelfPlayUserTurnContextStage

func (s *SelfPlayUserTurnContextStage) Process(ctx context.Context, input <-chan stage.StreamElement, output chan<- stage.StreamElement) error

type SkillInstructionStage ¶ added in v1.4.5

type SkillInstructionStage struct {
	stage.BaseStage
	// contains filtered or unexported fields
}

SkillInstructionStage appends preloaded skill instructions to the system prompt so the model sees skills marked preload: true from turn 1 without having to call skill__activate. The instructions are appended exactly once per Turn into TurnState.SystemPrompt.

func NewSkillInstructionStageWithTurnState(
	instructions string, turnState *stage.TurnState,
) *SkillInstructionStage

func (s *SkillInstructionStage) Process(ctx context.Context, input <-chan stage.StreamElement, output chan<- stage.StreamElement) error

type StripToolMessagesStage ¶

type StripToolMessagesStage struct {
	stage.BaseStage
}

StripToolMessagesStage removes tool role messages from the stream. This is used in self-play scenarios before calling the self-play provider.

func NewStripToolMessagesStage() *StripToolMessagesStage

func (s *StripToolMessagesStage) Process(
	ctx context.Context,
	input <-chan stage.StreamElement,
	output chan<- stage.StreamElement,
) error

type TTSSentenceAggregatorStage ¶ added in v1.5.3

type TTSSentenceAggregatorStage struct {
	stage.BaseStage
	// contains filtered or unexported fields
}

TTSSentenceAggregatorStage turns a streaming provider's token-by-token text deltas into complete sentences for TTS, so the agent speaks in smooth phrases instead of "word … pause … word".

It sits between the AssistantTTSFilterStage (which has already dropped the user transcript) and the TTS stage. As assistant text deltas arrive it accumulates them and emits each complete sentence (ending in . ! or ?) the moment it forms — while the model is still generating the rest. That overlaps generation with speech: audio starts after the first sentence rather than after the whole reply.

The complete assistant Message that the provider emits at the end of a turn duplicates the deltas, so once any delta has been seen the Message is dropped (after flushing the trailing partial sentence). For a NON-streaming provider (no deltas), the Message's content is split into sentences instead.

Control signals pass through: Interrupt clears the buffer (barge-in discards the half-spoken reply); EndOfTurn / EndOfStream flush the trailing buffer.

func NewTTSSentenceAggregatorStage() *TTSSentenceAggregatorStage

func (s *TTSSentenceAggregatorStage) Process(
	ctx context.Context,
	input <-chan stage.StreamElement,
	output chan<- stage.StreamElement,
) error

type TurnEvalRunner ¶ added in v1.3.2

type TurnEvalRunner interface {
	RunAssertionsAsEvals(
		ctx context.Context,
		assertionConfigs []assertions.AssertionConfig,
		messages []types.Message,
		turnIndex int,
		sessionID string,
		trigger evals.EvalTrigger,
	) []evals.EvalResult
}

TurnEvalRunner is an interface for running assertions as evals during turn execution. EvalOrchestrator in the engine package implements this interface.

type TurnIndexStage ¶

type TurnIndexStage struct {
	stage.BaseStage
	// contains filtered or unexported fields
}

TurnIndexStage computes role-specific turn counters from accumulated messages. It writes the counters into TurnState's ProviderRequestMetadata so downstream stages (e.g. mock provider context) and persisted state.Metadata can read them. Keys written:

• arena_user_completed_turns: number of completed user messages
• arena_user_next_turn: completed user messages + 1 (next user turn to generate)
• arena_assistant_completed_turns: number of completed assistant messages
• arena_assistant_next_turn: completed assistant messages + 1

func NewTurnIndexStage() *TurnIndexStage

func NewTurnIndexStageWithTurnState(turnState *stage.TurnState) *TurnIndexStage

func (s *TurnIndexStage) Process(ctx context.Context, input <-chan stage.StreamElement, output chan<- stage.StreamElement) error