gateway

package
v0.0.0-...-4e73138 Latest Latest
Warning

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

 Go to latest Published: Aug 24, 2026 License: MIT Imports: 27 Imported by: 0

Documentation

Overview

Package gateway exposes the canonical Speko Voice Protocol to local cross-language clients. It is intentionally a thin transport wrapper around runtime.Engine: provider routing and credentials remain in the signed plan.

Turn events are the conversation profiler's content-free markers: a worker probe observes one framework conversation, mints opaque conversation/turn identifiers, and posts small marker batches to this local route. The gateway validates a closed marker vocabulary, enriches conversation.started with the identity only the gateway knows, and forwards each marker through the existing bounded telemetry exporter.

Envelope reuse, deliberately: a marker travels as a runtime.TelemetryEvent whose SessionID carries the conversation_id and whose AttemptID carries the turn_id ("" for conversation-scoped markers). EventID is deterministic (tev_<conversation_id>_<seq>) so at-least-once delivery dedupes server-side exactly like session telemetry. Markers are never Required: disabling optional telemetry suppresses the profiler entirely.

Index

Constants

View Source 
const (
	// WebSocketSubprotocol is the only streaming wire format supported by this
	// development revision.
	WebSocketSubprotocol = "speko.voice.v0.r3"
)

Variables

View Source 
var (
	ErrLocalModeRequiresBYOK = errors.New("gateway: local routing requires provider-direct BYOK execution")
	ErrLocalPlanSignature    = errors.New("gateway: invalid local session plan signature")
)

Functions

func ListenUnix 

func ListenUnix(socketPath string) (net.Listener, error)

ListenUnix creates an owner-only Unix socket listener. It will not remove an existing filesystem entry, avoiding an accidental unlink of an unexpected path during service startup.

Types

type CatalogEntry 

type CatalogEntry struct {
	Provider     string               `json:"provider"`
	Kind         protocol.SessionKind `json:"kind"`
	Adapter      string               `json:"adapter"`
	DefaultModel string               `json:"default_model"`
	// Models lists every concrete model this adapter accepts when a provider
	// exposes several models on the same wire surface. Empty means the default
	// is the only published model. Local planning still uses DefaultModel when
	// the caller sends auto; /v1/models expands this list so explicit choices do
	// not disappear behind a single provider row.
	Models []string `json:"models,omitempty"`
	// DefaultVoice is used only in standalone BYOK mode. Several TTS vendors
	// REFUSE to open without a voice — Rime, Gradium and Google among them — and
	// the local planner has no benchmark board to choose from, so without this a
	// perfectly good adapter fails at open with a vendor error the operator cannot
	// act on. Empty means the vendor supplies its own default.
	//
	// Google is deliberately empty: its voice names embed the language
	// (hi-IN-Chirp3-HD-Aoede), so no single value is correct and the caller must
	// pass one. The control plane fills this from the measured board instead.
	DefaultVoice string             `json:"default_voice,omitempty"`
	Transport    protocol.Transport `json:"transport"`
	Endpoint     string             `json:"endpoint"`
	// ModelRoutes selects a different vendor endpoint for model families that
	// share one provider/kind adapter but are served on a versioned path. It is
	// local-planner metadata, not another published provider row. The first
	// matching prefix wins; unmatched models retain Endpoint.
	ModelRoutes []CatalogModelRoute `json:"-"`
	// RequiresDeploymentConfig, when non-empty, says this row cannot be dialled as
	// written and names what an operator must supply. Google STT is the case: its
	// path embeds the caller's own GCP project
	// (/v2/projects/{project}/locations/{location}/recognizers/_:recognize), so no
	// static endpoint is correct for anybody.
	//
	// The row stays published because the catalog answers "what does this build
	// implement", and hiding a supported vendor would be its own lie. But the
	// planner refuses it until configured rather than dialling a literal
	// PROJECT_ID and handing back a vendor 404 an operator cannot act on.
	RequiresDeploymentConfig string `json:"requires_deployment_config,omitempty"`
}

CatalogEntry is one thing this build can actually do: a provider, a modality, the adapter that implements it, and the default model and endpoint used when a caller names neither.

This table is the single source for BOTH standalone route construction and the published catalog. Keeping them separate is what lets a catalog claim a provider the binary cannot open — the failure mode is a customer reading the list, wiring the id, and getting a route error. Adding a provider here makes it routable and published in one edit, or neither.

func Catalog 

func Catalog() []CatalogEntry

Catalog returns every (provider, modality) this build implements, ordered so the published list is stable across restarts.

type CatalogModelRoute 

type CatalogModelRoute struct {
	ModelPrefix string
	Endpoint    string
}

type Config 

type Config struct {
	Engine *runtimepkg.Engine
	Plans  PlanClient
	// WarmPlans is optional. When set, a managed provider-direct create takes
	// an already-signed plan out of memory and dials the provider immediately,
	// with no control-plane call on the path. A miss falls through to Plans.
	WarmPlans      *PlanPool
	Runtime        protocol.RuntimeDescriptor
	Workload       *protocol.Workload
	LocalAuthToken string
	MaxSessions    int
	// AttachTimeout bounds the time a newly-created provider session may wait
	// for its local WebSocket client. A zero value uses a conservative default.
	AttachTimeout time.Duration
	// StreamWriteTimeout bounds one canonical event write to a local client.
	// A zero value uses a conservative default.
	StreamWriteTimeout time.Duration
	// SetupTimeout bounds control-plane plan issuance and provider opening.
	// A zero value uses a conservative default.
	SetupTimeout time.Duration
	// Now exists for deterministic lease scheduling tests. Production uses the
	// process wall clock.
	Now func() time.Time
	// Telemetry receives validated turn markers from POST /v1/turn-events.
	// A nil sink discards them, keeping the route harmless for embedders that
	// never wire an exporter.
	Telemetry runtimepkg.TelemetrySink
	// TurnEvents selects the export destination for turn markers. Empty
	// destinations leave the exporter's own anonymous default in charge.
	TurnEvents TurnEventDestinations
}

Config builds a local gateway service. LocalAuthToken is mandatory even for a Unix socket: a mounted socket can otherwise be reached by another process in the same host or container namespace.

type CreateSessionRequest 

type CreateSessionRequest struct {
	Kind        protocol.SessionKind      `json:"kind"`
	Integration *protocol.Integration     `json:"integration,omitempty"`
	Execution   protocol.ExecutionRequest `json:"execution"`
	Request     protocol.RequestOptions   `json:"request"`
	Media       *protocol.MediaFormat     `json:"media,omitempty"`
}

CreateSessionRequest contains caller-selectable, provider-neutral data. Runtime identity is configured by the gateway and cannot be supplied by the local caller.

type CredentialSourcePlanner 

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

CredentialSourcePlanner lets one Gateway serve local BYOK sessions and hosted managed sessions at the same time. BYOK requests never leave the process; all other requests retain the hosted control-plane behavior.

func NewCredentialSourcePlanner 

func NewCredentialSourcePlanner(byok, managed PlanClient) (*CredentialSourcePlanner, error)

func (*CredentialSourcePlanner) CreateSessionPlan 

func (p *CredentialSourcePlanner) CreateSessionPlan(ctx context.Context, request protocol.SessionPlanRequest, options controlplane.CreateOptions) (protocol.SessionPlan, string, error)

func (*CredentialSourcePlanner) CreateSessionPlanBatch 

func (p *CredentialSourcePlanner) CreateSessionPlanBatch(ctx context.Context, request protocol.SessionPlanRequest, count int, options controlplane.CreateOptions) ([]protocol.SessionPlan, string, error)

func (*CredentialSourcePlanner) ExchangeFallbackPlan 

func (p *CredentialSourcePlanner) ExchangeFallbackPlan(ctx context.Context, plan protocol.SessionPlan, request controlplane.FallbackRequest, idempotencyKey string) (protocol.SessionPlan, string, error)

type LocalPlanner 

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

LocalPlanner issues and verifies process-local BYOK plans. It is used both by a BYOK-only Gateway and by explicit BYOK requests on a mixed managed/BYOK Gateway. HMAC signing preserves the same verify-before-key-injection invariant used by connected plans without contacting Speko.

func NewLocalPlanner 

func NewLocalPlanner(config LocalPlannerConfig) (*LocalPlanner, error)

func (*LocalPlanner) CreateSessionPlan 

func (p *LocalPlanner) CreateSessionPlan(_ context.Context, request protocol.SessionPlanRequest, _ controlplane.CreateOptions) (protocol.SessionPlan, string, error)

func (*LocalPlanner) CreateSessionPlanBatch 

func (p *LocalPlanner) CreateSessionPlanBatch(context.Context, protocol.SessionPlanRequest, int, controlplane.CreateOptions) ([]protocol.SessionPlan, string, error)

CreateSessionPlanBatch is deliberately unsupported. Local planning is already free — no network, no signing service, just an HMAC over a struct — so prefetching it would add bookkeeping to save nothing.

func (*LocalPlanner) ExchangeFallbackPlan 

func (p *LocalPlanner) ExchangeFallbackPlan(context.Context, protocol.SessionPlan, controlplane.FallbackRequest, string) (protocol.SessionPlan, string, error)

func (*LocalPlanner) Verify 

func (p *LocalPlanner) Verify(_ context.Context, plan protocol.SessionPlan) error

type LocalPlannerConfig 

type LocalPlannerConfig struct {
	Providers          []string
	MaxSessionDuration time.Duration
	// RouteOverrides supplies deployment-specific values that cannot live in
	// the public catalog. Keys are adapter IDs. Google STT's project-scoped
	// recognize URL and operator-selected TTS voices are the common cases.
	RouteOverrides map[string]LocalRouteOverride
	Now            func() time.Time
}

LocalPlannerConfig describes the credentials and limits available to a locally routed gateway. Provider key values are intentionally not accepted; only provider names are needed to make a local routing decision.

type LocalRouteOverride 

type LocalRouteOverride struct {
	Endpoint     string
	DefaultVoice string
}

LocalRouteOverride customizes one catalog route in process-local BYOK mode. Empty fields retain the catalog value.

type PlanClient 

type PlanClient interface {
	CreateSessionPlan(context.Context, protocol.SessionPlanRequest, controlplane.CreateOptions) (protocol.SessionPlan, string, error)
	// CreateSessionPlanBatch issues several plans in one round trip so a
	// PlanPool can keep them warm. A planner with no hosted control plane
	// behind it — LocalPlanner — returns an error and is simply never pooled.
	CreateSessionPlanBatch(context.Context, protocol.SessionPlanRequest, int, controlplane.CreateOptions) ([]protocol.SessionPlan, string, error)
	// ExchangeFallbackPlan performs the signed one-per-attempt fallback
	// exchange when provider opening fails before any output was produced.
	ExchangeFallbackPlan(context.Context, protocol.SessionPlan, controlplane.FallbackRequest, string) (protocol.SessionPlan, string, error)
}

PlanClient is the setup-only planning contract consumed by a gateway. It is implemented by the Speko client for managed routing and LocalPlanner for local routing, and is deliberately absent from runtime.Engine's hot path.

type PlanPool 

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

PlanPool keeps signed session plans warm so creating a session costs no network round trip.

This is the piece that makes the zero-overhead claim true rather than approximately true. Everything else in the fast path shaves milliseconds off a control-plane call; this removes the call. What remains between a caller's first audio frame and the provider socket is the provider dial itself.

It is strictly a cache. A route nobody has asked for yet, an exhausted pool, or an unreachable control plane all fall through to the ordinary synchronous create, so nothing fails that would not have failed before.

func NewPlanPool 

func NewPlanPool(config PlanPoolConfig) (*PlanPool, error)

func (*PlanPool) Metrics 

func (p *PlanPool) Metrics() PlanPoolMetrics

Metrics reports pool effectiveness. A miss rate that does not fall toward zero after warm-up means prefetching is not working and sessions are paying the control-plane round trip they were promised they would not.

func (*PlanPool) Run 

func (p *PlanPool) Run(ctx context.Context)

Run keeps every recently used route stocked until ctx is canceled.

func (*PlanPool) Take 

func (p *PlanPool) Take(request protocol.SessionPlanRequest) (protocol.SessionPlan, bool)

Take returns a warm plan for this request, if one is available, and always registers the route so the next request for the same shape is warm.

A miss is not an error. It means the caller pays what it paid before.

func (*PlanPool) Warm 

func (p *PlanPool) Warm(ctx context.Context, request protocol.SessionPlanRequest) error

Warm registers a route before any traffic has asked for it, so the very first session of a process is warm too. Without it the first call of every shape — which for a voice agent worker is the one a real person is waiting on — always pays full price.

type PlanPoolConfig 

type PlanPoolConfig struct {
	Plans PlanClient
	// Target is the number of plans kept warm per distinct route.
	Target int
	// MinRemaining refuses to hand out a plan that is about to expire. A plan
	// only has to be live at the provider handshake, but handing over one with
	// two seconds left would trade a control-plane round trip for a flaky one.
	MinRemaining time.Duration
	Interval     time.Duration
	// RefillTimeout bounds one background batch request. It is not the caller's
	// deadline — nobody is waiting on a refill — it exists so a control plane
	// that accepts a connection and then stalls cannot take the pool down with
	// it.
	RefillTimeout time.Duration
	IdleAfter     time.Duration
	MaxRoutes     int
	Runtime       protocol.RuntimeDescriptor
	Workload      *protocol.Workload
	Now           func() time.Time
}

PlanPoolConfig configures prefetching.

type PlanPoolMetrics 

type PlanPoolMetrics struct {
	Hits     uint64
	Misses   uint64
	Expired  uint64
	Refills  uint64
	Failures uint64
	Depth    int
	Routes   int
}

PlanPoolMetrics reports whether prefetching is actually absorbing demand.

type Server 

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

Server serves REST setup, canonical WebSocket streaming, readiness/drain, and Prometheus metrics. It holds no permanent provider credential.

func New 

func New(config Config) (*Server, error)

New validates static gateway configuration.

func (*Server) BeginDrain 

func (s *Server) BeginDrain()

BeginDrain atomically stops new session creation without waiting for active sessions. Callers can report the draining state before they wait or exit.

func (*Server) Drain 

func (s *Server) Drain(ctx context.Context) error

Drain stops new session creation, allows existing sessions to finish, and returns once all are gone or ctx expires. Existing WebSockets remain active.

func (*Server) Handler 

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

Handler returns the local HTTP handler. Bind it only to a Unix socket or loopback listener; Handler itself intentionally does not make a public network listener safe.

func (*Server) Serve 

func (s *Server) Serve(listener net.Listener) error

Serve serves listener connections. Calling Drain controls readiness and active sessions; callers own listener shutdown after draining completes.

func (*Server) Stats 

func (s *Server) Stats() Stats

Stats returns a point-in-time snapshot for instance heartbeats and local operational tooling.

type Stats 

type Stats struct {
	ActiveSessions  int64
	PendingSessions int
	SessionCapacity int
	SessionsTotal   uint64
	Draining        bool
}

Stats is the bounded, content-free process state safe to report to the hosted customer control plane. It intentionally excludes local socket paths, host resources, request bodies, and session identifiers.

type SttSupportError 

type SttSupportError struct {
	Provider string
	Option   string
	Detail   string
}

SttSupportError is a session refused because the routed provider cannot honor a canonical STT ask. Distinguished from provider open failures so the create handler can answer 422 with the option named instead of a generic bad-gateway, and so the fallback exchange is never spent on it — no retry changes what a vendor supports.

func (*SttSupportError) Error 

func (e *SttSupportError) Error() string

type TurnEventDestinations 

type TurnEventDestinations struct {
	// AuthenticatedEndpoint is the control-plane turn-event ingest URL. It must
	// be derived from the configured control-plane origin only, never from
	// request data. Used together with AuthenticatedToken when both are set.
	AuthenticatedEndpoint string
	// AuthenticatedToken is the Speko API key. It is a credential: never log it.
	AuthenticatedToken string
	// AnonymousEndpoint receives markers when no API key is configured. It
	// carries no token and no account linkage.
	AnonymousEndpoint string
}

TurnEventDestinations selects where validated turn markers are exported. The route chooses the destination; the exporter's anonymous fallback is intentionally not relied upon so the choice stays auditable in one place.

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.