temporalruntime

package
v0.1.0-alpha.74 Latest Latest
Warning

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

 Go to latest Published: Sep 4, 2026 License: MIT Imports: 34 Imported by: 0

Documentation

Overview

Package temporalruntime dispatches durable agent runs to compiler-generated Temporal workflows. Workflow and activity worker registration remains owned by generated project code.

Index

Constants

View Source 
const (
	EnvAddress          = "GOBEYOND_TEMPORAL_ADDRESS"
	EnvNamespace        = "GOBEYOND_TEMPORAL_NAMESPACE"
	EnvEnvironment      = "GOBEYOND_TEMPORAL_ENVIRONMENT"
	EnvHostedRuntime    = "GOBEYOND_HOSTED_RUNTIME"
	EnvHostReportSocket = "GOBEYOND_HOST_REPORT_SOCKET"

	DefaultAddress   = "localhost:7233"
	DefaultNamespace = "default"
)
View Source 
const (
	VoiceSessionWorkflowName = "gobeyond.agents.voice_session.v1"

	// VoiceSessionExecuteToolUpdate is the Maglev/API → workflow Update that
	// runs a Live tool as a LocalActivity on the agent worker (P2 remote path
	// and P3 colocated LocalActivity tools).
	VoiceSessionExecuteToolUpdate = "gobeyond.agents.voice_session.execute_tool.v1"
)

Variables

View Source 
var (
	ErrClosed = errors.New("agent Temporal dispatcher is closed")
)

Functions

func ActivityName 

func ActivityName(agentID string) (string, error)

ActivityName returns the stable name generated workers must register for an agent's typed execution activity.

func Register 

func Register[Input any, Output any](registry worker.Registry, agentID string, definition agents.Definition[Input, Output]) error

Register installs the stable workflow and typed execution activity for one compiler-discovered durable agent. Generated worker code calls this helper, keeping Temporal types and serialization out of authored agent packages.

func RegisterAI 

func RegisterAI(_ worker.Worker, runtimes *AIRegistry, agentID string, definition agents.AIDefinition) error

func RegisterVoice 

func RegisterVoice(registry *VoiceRegistry, agentID string, definition agents.AIDefinition) error

RegisterVoice installs the Gemini Live adapter for an AI agent that declares LiveModel. Agents without LiveModel are a no-op so shared registration helpers stay safe.

func RegisterVoiceSessionWorkflow 

func RegisterVoiceSessionWorkflow(w worker.Registry)

RegisterVoiceSessionWorkflow registers the platform voice-session workflow and its LocalActivity tool executor.

func RetainVoiceRegistry 

func RetainVoiceRegistry(registry *VoiceRegistry)

RetainVoiceRegistry keeps the worker's voice registry reachable for the process lifetime so G5 / in-process Live dispatch can Lookup adapters.

func VoiceSessionWorkflow 

func VoiceSessionWorkflow(ctx workflow.Context, in VoiceSessionInput) error

VoiceSessionWorkflow is the lifecycle workflow for an AI phone/softphone call. Media may stay on Maglev (P0–P2) or colocate on the realtime RoleWorker (P3); this workflow is the hosted Agents SoR handle and the LocalActivity home for Live tool Updates.

func WorkflowID 

func WorkflowID(sessionID, runID string) (string, error)

WorkflowID returns the stable Temporal execution ID for one session run.

func WorkflowName 

func WorkflowName(agentID string) (string, error)

WorkflowName returns the stable name generated workers must register for an agent's durable workflow.

Types

type AIRegistry 

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

AIRegistry is the queue-wide worker runtime resolver for compiled AI agents. The Temporal AI SDK registers its stable workflow and activity names once per polling process, then selects provider/tools by AgentID and revision.

func NewAIRegistry 

func NewAIRegistry() *AIRegistry

func (*AIRegistry) Register 

func (runtimes *AIRegistry) Register(registry worker.Worker) error

Register installs the stable SDK workflow and activities exactly once for a generated queue worker. It is a no-op when the queue contains no AI agents.

func (*AIRegistry) ResolveAgentRuntime 

func (runtimes *AIRegistry) ResolveAgentRuntime(_ context.Context, scope activities.RuntimeScope) (activities.AgentRuntime, error)

type Client 

type Client interface {
	ExecuteWorkflow(context.Context, client.StartWorkflowOptions, interface{}, ...interface{}) (client.WorkflowRun, error)
	SignalWorkflow(context.Context, string, string, string, interface{}) error
	CancelWorkflow(context.Context, string, string) error
	Close()
}

Client is the narrow Temporal client surface used by Dispatcher. The SDK's client.Client implements this interface directly.

type ClientFactory 

type ClientFactory func(context.Context, client.Options) (Client, error)

ClientFactory creates a client from non-secret connection options. It is injectable so hosted construction and tests do not require a package-global Temporal connection.

type Dispatcher 

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

Dispatcher implements httpruntime.Dispatcher for compiler-generated durable agent workflows.

func New 

func New(ctx context.Context, options Options) (*Dispatcher, error)

New constructs a dispatcher and dials Temporal when Client is not supplied.

func NewLazyFromEnv 

func NewLazyFromEnv(ctx context.Context) (*Dispatcher, error)

NewLazyFromEnv chooses the site-bound host broker in a hosted slot and the lazy plaintext client in local development. Hosted web sandboxes never need Temporal API keys or mTLS leaf material.

func NewLazyLocalFromEnv 

func NewLazyLocalFromEnv(ctx context.Context) (*Dispatcher, error)

NewLazyLocalFromEnv constructs the local dispatcher without requiring Temporal to be reachable at site startup. The first durable invocation establishes the connection; direct agents and the rest of the site remain usable while a local Temporal server is starting.

func NewLocalFromEnv 

func NewLocalFromEnv(ctx context.Context) (*Dispatcher, error)

NewLocalFromEnv dials a local/plaintext Temporal client. It reads only the address, namespace, and queue environment slug; it never reads or logs API keys or certificate material.

func (*Dispatcher) Cancel 

func (dispatcher *Dispatcher) Cancel(ctx context.Context, _ httpruntime.Adapter, call httpruntime.CancelCall, _ httpruntime.EventEmitter) error

Cancel requests cancellation of the stable workflow execution for the run.

func (*Dispatcher) Close 

func (dispatcher *Dispatcher) Close()

Close releases the Temporal client. It is safe to call more than once.

func (*Dispatcher) Respond 

func (dispatcher *Dispatcher) Respond(ctx context.Context, adapter httpruntime.Adapter, call httpruntime.RespondCall, _ httpruntime.EventEmitter) error

Respond signals a pending durable AI tool approval. Typed handler agents keep their legacy unsupported response behavior.

func (*Dispatcher) Start 

func (dispatcher *Dispatcher) Start(ctx context.Context, adapter httpruntime.Adapter, call httpruntime.StartCall, emit httpruntime.EventEmitter) error

Start launches the generated workflow, waits for its JSON result, and emits the canonical agent.output event through the HTTP runtime.

type GeminiLiveAdapter 

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

GeminiLiveAdapter implements voice.Adapter with google.golang.org/genai Live.

func NewGeminiLiveAdapter 

func NewGeminiLiveAdapter(definition agents.AIDefinition) *GeminiLiveAdapter

NewGeminiLiveAdapter returns the production Live adapter for definition.

func (*GeminiLiveAdapter) Start 

func (adapter *GeminiLiveAdapter) Start(ctx context.Context, cfg voice.StartConfig, pcmIn <-chan []byte, pcmOut chan<- voice.AudioFrame) (voice.SessionHandle, voice.StartResult, error)

Start opens a Live session and returns a handle that pumps PCM and tools.

type GrokLiveAdapter 

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

func NewGrokLiveAdapter 

func NewGrokLiveAdapter(definition agents.AIDefinition) *GrokLiveAdapter

func (*GrokLiveAdapter) Start 

func (adapter *GrokLiveAdapter) Start(ctx context.Context, cfg voice.StartConfig, audioIn <-chan []byte, audioOut chan<- voice.AudioFrame) (voice.SessionHandle, voice.StartResult, error)

type HostedPCMStream 

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

HostedPCMStream is a duplex length-prefixed PCM exchange over HTTP UDS.

func (*HostedPCMStream) Close 

func (stream *HostedPCMStream) Close() error

Close ends the uplink pipe and response body. Safe to call multiple times. 

func (stream *HostedPCMStream) CloseUplink() error

CloseUplink closes the request body so servers that buffer the full upload can finish and send the response. Optional for true duplex servers.

func (*HostedPCMStream) ReadAudioFrame 

func (stream *HostedPCMStream) ReadAudioFrame() (voice.AudioFrame, error)

ReadAudioFrame reads one downlink audio frame, including interruption and turn-completion control markers when the endpoint uses v2 framing.

func (*HostedPCMStream) ReadPCM 

func (stream *HostedPCMStream) ReadPCM() ([]byte, error)

ReadPCM reads one downlink PCM payload.

func (*HostedPCMStream) WritePCM 

func (stream *HostedPCMStream) WritePCM(pcm []byte) error

WritePCM encodes and sends one uplink PCM payload.

func (*HostedPCMStream) WritePlayoutAck 

func (stream *HostedPCMStream) WritePlayoutAck(barrierID uint64) error

WritePlayoutAck acknowledges a v3 Flush barrier after telephone playout has drained. It is intentionally unavailable on the legacy framing versions.

type HostedVoiceClient 

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

HostedVoiceClient is the G5 client for gbhost voice start/stop/PCM over the slot-private host-report UDS. It does not implement the Live adapter; it only brokers sessions and binary PCM frames for voice-worker / G5b.

func NewHostedVoiceClient 

func NewHostedVoiceClient(socketPath string) *HostedVoiceClient

NewHostedVoiceClient builds a client that dials the given Unix socket path.

func NewHostedVoiceClientFromEnv 

func NewHostedVoiceClientFromEnv() (*HostedVoiceClient, error)

NewHostedVoiceClientFromEnv dials GOBEYOND_HOST_REPORT_SOCKET like the hosted Temporal agent dispatcher.

func (*HostedVoiceClient) Close 

func (client *HostedVoiceClient) Close()

Close releases idle UDS connections.

func (*HostedVoiceClient) OpenPCM 

func (client *HostedVoiceClient) OpenPCM(ctx context.Context, sessionToken string, spec voice.PCMEndpointSpec) (*HostedPCMStream, error)

OpenPCM opens a bidirectional length-prefixed PCM stream to the endpoint returned by Start. Uplink frames are written on the request body; downlink frames are read from the response body (POST /v1/agents/voice/pcm/{token}).

func (*HostedVoiceClient) Start 

func (client *HostedVoiceClient) Start(ctx context.Context, request voice.HostedStartRequest) (voice.HostedStartResponse, error)

Start POSTs /v1/agents/voice/start and returns the session token plus PCM endpoint. The token is a secret capability — use voice.RedactToken in logs.

func (*HostedVoiceClient) Stop 

func (client *HostedVoiceClient) Stop(ctx context.Context, request voice.HostedStopRequest) error

Stop POSTs /v1/agents/voice/stop. Callers may retry; the server is idempotent.

type Options 

type Options struct {
	Client      Client
	Factory     ClientFactory
	Address     string
	Namespace   string
	Environment string
}

Options configures a dispatcher. Client and Factory are mutually exclusive. Dispatcher takes ownership of Client and closes it from Close.

type VoiceRegistry 

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

VoiceRegistry holds Live voice adapters for AI agents on a realtime worker. Generated realtime-{agentId} mains register adapters alongside AIRegistry.

func NewVoiceRegistry 

func NewVoiceRegistry() *VoiceRegistry

NewVoiceRegistry returns an empty registry for generated worker mains.

func ProcessVoiceRegistry 

func ProcessVoiceRegistry() *VoiceRegistry

ProcessVoiceRegistry returns the registry retained by the generated worker main, if any.

func (*VoiceRegistry) Definition 

func (registry *VoiceRegistry) Definition(agentID string) (agents.AIDefinition, bool)

Definition returns the AI definition registered for voice, if any.

func (*VoiceRegistry) Lookup 

func (registry *VoiceRegistry) Lookup(agentID string) (voice.Adapter, bool)

Lookup returns the registered voice adapter for agentID.

type VoiceSessionExecuteToolInput 

type VoiceSessionExecuteToolInput struct {
	AgentID    string          `json:"agent_id"`
	ToolName   string          `json:"tool_name"`
	ToolCallID string          `json:"tool_call_id,omitempty"`
	Input      json.RawMessage `json:"input,omitempty"`
	ActorID    string          `json:"actor_id,omitempty"`
	ActorKind  string          `json:"actor_kind,omitempty"`
	NetworkID  string          `json:"network_id,omitempty"`
	// AllowedToolIDs is derived from the verified voice grant by the API. It is
	// optional for colocated/internal tests and older direct workflow callers.
	AllowedToolIDs []string `json:"allowed_tool_ids,omitempty"`
}

VoiceSessionExecuteToolInput is the Update / LocalActivity payload for one Gemini Live function call.

type VoiceSessionExecuteToolResult 

type VoiceSessionExecuteToolResult struct {
	Result json.RawMessage `json:"result,omitempty"`
	Error  string          `json:"error,omitempty"`
}

VoiceSessionExecuteToolResult is returned to Maglev so it can SendToolResponse.

func VoiceSessionExecuteToolActivity 

func VoiceSessionExecuteToolActivity(ctx context.Context, req VoiceSessionExecuteToolInput) (VoiceSessionExecuteToolResult, error)

VoiceSessionExecuteToolActivity resolves the agent tool from ProcessVoiceRegistry and runs it in-process on the realtime worker (LocalActivity). Maglev cannot hold customer tools; the agent worker can.

type VoiceSessionInput 

type VoiceSessionInput struct {
	AgentID     string `json:"agent_id"`
	CallID      string `json:"call_id"`
	SessionID   string `json:"session_id"`
	ExecutionID string `json:"execution_id"`
}

VoiceSessionInput is the durable voice-call workflow argument.

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.