web

Documentation

Overview

Package web provides the HTTP server and SSE event streaming for the Arena web UI.

Package web provides the HTTP server and SSE event streaming for the Arena web UI.

Index

• func LoadResultsIntoStore(outDir string, store *statestore.ArenaStateStore) int
• type EventAdapter
  • func NewEventAdapter() *EventAdapter
  • func (a *EventAdapter) AttachAudioRouter(runID string, router *arenaaudio.AudioRouter, rate int)
  • func (a *EventAdapter) BroadcastFullMessages(convID string, msgs []types.Message)
  • func (a *EventAdapter) HandleEvent(event *events.Event)
  • func (a *EventAdapter) Register() chan []byte
  • func (a *EventAdapter) RegisterAudio(ch chan []byte)
  • func (a *EventAdapter) Subscribe(bus events.Bus)
  • func (a *EventAdapter) Unregister(ch chan []byte)
  • func (a *EventAdapter) UnregisterAudio(ch chan []byte)
• type RunRequest
• type SSEEvent
• type Server
  • func NewServer(adapter *EventAdapter, eng *engine.Engine, store *statestore.ArenaStateStore, ...) *Server
  • func (s *Server) Handler() http.Handler
• type WorkflowGraph
  • func BuildWorkflowGraph(cfg *arenaconfig.Config) (WorkflowGraph, error)
• type WorkflowGraphEdge
• type WorkflowGraphNode

Functions

func LoadResultsIntoStore

func LoadResultsIntoStore(outDir string, store *statestore.ArenaStateStore) int

LoadResultsIntoStore scans outDir for run-result JSON files and loads each into the in-memory state store. Previously this delegated to JSONResultRepository.LoadResults, which reads run_ids out of index.json — and index.json is rewritten by each CLI invocation, so on server boot only the most recent batch's IDs were hydrated even though every prior <runID>.json sat right there on disk.

Scanning the directory directly means web-triggered runs (which never touch index.json) and old CLI runs both flow into Previous Runs on initial page load.

Types

type EventAdapter

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

EventAdapter subscribes to an events.Bus and fans out JSON-serialized events to registered SSE client channels.

func NewEventAdapter

func NewEventAdapter() *EventAdapter

NewEventAdapter creates a new EventAdapter.

func (*EventAdapter) AttachAudioRouter

func (a *EventAdapter) AttachAudioRouter(runID string, router *arenaaudio.AudioRouter, rate int)

AttachAudioRouter subscribes the adapter to a per-run AudioRouter. Frames are encoded as SSE "audio" events for clients that opted in via ?audio=1. The subscription lives until the router closes.

func (*EventAdapter) BroadcastFullMessages

func (a *EventAdapter) BroadcastFullMessages(convID string, msgs []types.Message)

BroadcastFullMessages emits a "message.full" SSE event per message in msgs, carrying the complete persisted types.Message (role, content, parts, tool_calls, tool_result, timestamp, latency_ms, cost_info, finish_reason, meta, validations) rather than the thin projection used by the live runtime-event stream. This lets the Inspector show metrics/meta/cost/raw JSON for messages as they're persisted.

Delivery is per-client delta: each client is sent only the messages whose payload differs from what that client last received. This keeps a long conversation from re-sending its whole history on every save (which cost O(N) SSE traffic per save, O(N^2) over a run) while preserving the two catch-up properties the full re-send used to provide for free:

• A newly registered or reconnecting client starts with an empty sent-set, so it still receives the complete history on the next save.
• A frame dropped because a client's buffer was full is not recorded as sent, so it is retried on the next save rather than lost.

The client reducer keys on index and upserts, so a partial send is safe.

func (*EventAdapter) HandleEvent

func (a *EventAdapter) HandleEvent(event *events.Event)

HandleEvent converts a runtime event to JSON and broadcasts to all clients.

func (*EventAdapter) Register

func (a *EventAdapter) Register() chan []byte

Register adds a new SSE client and returns its event channel.

func (*EventAdapter) RegisterAudio

func (a *EventAdapter) RegisterAudio(ch chan []byte)

RegisterAudio adds a client channel that receives audio SSE messages. Audio clients are distinct from regular event clients — only clients that explicitly opted in (via ?audio=1) receive audio frames. The same channel may also be registered via Register; the browser EventSource demuxes regular vs audio events by their `event:` line.

func (*EventAdapter) Subscribe

func (a *EventAdapter) Subscribe(bus events.Bus)

Subscribe subscribes the adapter to an event bus.

func (*EventAdapter) Unregister

func (a *EventAdapter) Unregister(ch chan []byte)

Unregister removes an SSE client. The caller is responsible for draining or discarding the channel after unregistering.

func (*EventAdapter) UnregisterAudio

func (a *EventAdapter) UnregisterAudio(ch chan []byte)

UnregisterAudio removes an audio client channel.

type RunRequest

type RunRequest struct {
	Providers []string `json:"providers,omitempty"`
	Scenarios []string `json:"scenarios,omitempty"`
	Regions   []string `json:"regions,omitempty"`
	// Runs is how many times to run the field (each a distinct sweep). Defaults
	// to 1; clamped to [1, maxFieldRuns].
	Runs int `json:"runs,omitempty"`
}

RunRequest is the JSON body for POST /api/run.

type SSEEvent

type SSEEvent struct {
	Type           string      `json:"type"`
	Timestamp      time.Time   `json:"timestamp"`
	ExecutionID    string      `json:"executionId,omitempty"`
	ConversationID string      `json:"conversationId,omitempty"`
	Data           interface{} `json:"data,omitempty"`
}

SSEEvent is the JSON structure sent to SSE clients.

type Server

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

Server is the Arena web UI HTTP server.

func NewServer

func NewServer(adapter *EventAdapter, eng *engine.Engine, store *statestore.ArenaStateStore, outputDir string) *Server

NewServer creates a new web server. eng and store may be nil (for testing SSE in isolation). outputDir is the path to the results directory (for DELETE /api/results).

When both eng and adapter are non-nil, the server registers an audio monitor hook on the engine so per-run AudioRouters automatically attach to the SSE relay. Audio monitoring still requires explicit opt-in via engine.EnableAudioMonitor — without it, the hook never fires.

func (*Server) Handler

func (s *Server) Handler() http.Handler

Handler returns the http.Handler for the server.

type WorkflowGraph

type WorkflowGraph struct {
	Nodes []WorkflowGraphNode `json:"nodes"`
	Edges []WorkflowGraphEdge `json:"edges"`
}

WorkflowGraph is the full workflow topology: its states (nodes) and transitions (edges).

func BuildWorkflowGraph

func BuildWorkflowGraph(cfg *arenaconfig.Config) (WorkflowGraph, error)

BuildWorkflowGraph turns a config's workflow spec into a topology graph. cfg == nil or cfg.Workflow == nil produces a single "default" node with no edges (Arena's implicit single-state workflow). Node and edge ordering is deterministic (sorted by state name, then event name).

type WorkflowGraphEdge

type WorkflowGraphEdge struct {
	From   string `json:"from"`
	To     string `json:"to"`
	Label  string `json:"label,omitempty"`
	Dashed bool   `json:"dashed,omitempty"`
}

WorkflowGraphEdge is a single transition between two states in the workflow topology graph.

type WorkflowGraphNode

type WorkflowGraphNode struct {
	ID       string `json:"id"`
	Label    string `json:"label"`
	Kind     string `json:"kind"` // entry|output|agent (composition kinds added in a later task)
	Entry    bool   `json:"entry"`
	Terminal bool   `json:"terminal"`
	// Parent is the owning workflow state's name for composition-step nodes
	// (including nested parallel-branch steps), letting the frontend
	// group/collapse a composition's steps under their state. Empty for
	// workflow state nodes themselves.
	Parent string `json:"parent,omitempty"`
}

WorkflowGraphNode is a single state in the workflow topology graph.