temporal

package
v0.3.0 Latest Latest
Warning

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

 Go to latest Published: Jul 26, 2026 License: Apache-2.0 Imports: 34 Imported by: 0

Documentation

Overview

Package temporal contains helpers for integrating this SDK with go.temporal.io (client, worker, etc.). This file adapts pkg/logger.Logger to go.temporal.io/sdk/log.Logger for client.Options. It is internal to this module; callers who build their own Temporal client should wire sdk/log.Logger themselves (or copy this small bridge).

If a file imports both this package and go.temporal.io/sdk/temporal, use an import alias on one of them (e.g. sdktrace "go.temporal.io/sdk/temporal") — import paths differ, but package names would both default to "temporal".

Index

Constants

View Source 
const QueryIsApprovalPending = "is_approval_pending"

QueryIsApprovalPending is the workflow query name that reports whether a ToolCallID still has a pending approval activity. Arg: toolCallID (string). Result: bool. Used by the forwarder on Events/reconnect to skip CUSTOM events for approvals that were already resolved while the subscriber was disconnected.

Variables

View Source 
var ErrAgentFingerprintMismatch = errors.New("temporal: agent fingerprint mismatch (caller vs worker); redeploy worker or align config/registries or retry run")

ErrAgentFingerprintMismatch is returned when the per-run agent fingerprint does not match the worker.

View Source 
var ErrNotApprovalCustomEvent = errors.New("temporal: custom event is not a recognized approval kind")

ErrNotApprovalCustomEvent means the CUSTOM event name is not tool or delegation approval.

Functions

func ComputeAgentFingerprint 

func ComputeAgentFingerprint(m AgentFingerprintPayload) string

ComputeAgentFingerprint returns a stable SHA-256 hex digest of the payload (identity, prompts, tools, sampling, limits, policy, optional MCP, A2A, and observability wiring). Use the same digests from pkg/agent on both the process that issues runs and the worker process.

func NewLogAdapter 

func NewLogAdapter(l logger.Logger) tlog.Logger

NewLogAdapter returns a Temporal sdk/log.Logger that delegates to the SDK logger.Logger. Calls use context.Background() because Temporal's logger API is not context-aware.

func ToolNamesFromTools 

func ToolNamesFromTools(tools []interfaces.Tool) []string

ToolNamesFromTools returns sorted tool names for fingerprinting.

Types

type AddConversationMessagesInput 

type AddConversationMessagesInput struct {
	ConversationID   string               `json:"conversation_id,omitempty"`
	Messages         []interfaces.Message `json:"messages,omitempty"`
	AgentFingerprint string               `json:"agent_fingerprint,omitempty"`
}

AddConversationMessagesInput is the input to AddConversationMessagesActivity.

type AgentFingerprintPayload 

type AgentFingerprintPayload struct {
	Version int `json:"v"`

	Name           string `json:"name"`
	Description    string `json:"description"`
	SystemPrompt   string `json:"system_prompt"`
	ResponseFormat *struct {
		Type   string         `json:"type"`
		Name   string         `json:"name,omitempty"`
		Schema map[string]any `json:"schema,omitempty"`
	} `json:"response_format,omitempty"`

	ToolNames []string `json:"tool_names"`

	// PolicyFingerprint is an opaque string from pkg/agent (same on caller and worker Temporal config).
	PolicyFingerprint string `json:"policy_fp"`

	// MCPFingerprint is the pkg/agent MCP wiring digest over transports (no secrets), timeouts, filters,
	// and extra MCP client names. Tool names already appear in ToolNames; this catches same tools
	// pointing at a different endpoint or policy. Omitted when empty.
	MCPFingerprint string `json:"mcp_fingerprint,omitempty"`

	// A2AFingerprint is the pkg/agent A2A wiring digest over server URLs, auth type (no secrets),
	// header keys, timeouts, skill filters, and extra A2A client names. Omitted when empty.
	A2AFingerprint string `json:"a2a_fingerprint,omitempty"`

	// ObservabilityFingerprint is the pkg/agent digest from [WithObservabilityConfig] (OTLP endpoint
	// and trace/metrics disable flags). Omitted when empty.
	ObservabilityFingerprint string `json:"observability_fingerprint,omitempty"`

	// AgentMode is the execution mode (e.g. interactive vs autonomous); must match pkg/agent WithAgentMode on caller and worker.
	AgentMode string `json:"agent_mode"`

	// AgentToolExecutionMode is the tool execution mode (e.g. sequential vs parallel); must match pkg/agent WithAgentToolExecutionMode on caller and worker.
	AgentToolExecutionMode string `json:"agent_tool_execution_mode"`

	// RetrieverFingerprint is the pkg/agent digest of retriever mode and registered retriever names.
	// Omitted when empty. Must match pkg/agent [retrieverConfigFingerprint] on caller and worker.
	RetrieverFingerprint string `json:"retriever_fingerprint,omitempty"`

	// HooksFingerprint is the pkg/agent digest of registered hook group names (sorted).
	// Omitted when empty. Must match pkg/agent [hookGroupsFingerprint] on caller and worker.
	HooksFingerprint string `json:"hooks_fingerprint,omitempty"`

	Sampling *sdkruntime.LLMSampling `json:"sampling,omitempty"`

	SessionSize int `json:"session_size"`

	MaxIterations     int   `json:"max_iterations"`
	TimeoutNs         int64 `json:"timeout_ns"`
	ApprovalTimeoutNs int64 `json:"approval_timeout_ns"`
}

AgentFingerprintPayload is the JSON-serializable snapshot hashed by ComputeAgentFingerprint. Agent callers and Temporal workers must supply the same fields so client and worker digests match.

func BuildAgentFingerprintPayload 

func BuildAgentFingerprintPayload(
	spec sdkruntime.AgentSpec,
	toolNames []string,
	policyFingerprint string,
	sampling *sdkruntime.LLMSampling,
	sessionSize int,
	limits sdkruntime.AgentLimits,
	mcpFingerprint string,
	a2aFingerprint string,
	observabilityFingerprint string,
	agentMode string,
	agentToolExecutionMode types.AgentToolExecutionMode,
	retrieverFingerprint string,
	hooksFingerprint string,
) AgentFingerprintPayload

BuildAgentFingerprintPayload builds a payload from spec and execution fields shared by caller and worker.

type AgentLLMInput 

type AgentLLMInput struct {
	AgentName        string               `json:"agent_name,omitempty"`
	ConversationID   string               `json:"conversation_id,omitempty"`
	Messages         []interfaces.Message `json:"messages,omitempty"`
	SkipTools        bool                 `json:"skip_tools,omitempty"`
	AgentFingerprint string               `json:"agent_fingerprint,omitempty"`
	MessageID        string               `json:"message_id,omitempty"`
	StreamWorkflowID string               `json:"stream_workflow_id,omitempty"`
	MemoryContext    string               `json:"memory_context,omitempty"`
	RetrieverContext string               `json:"retriever_context,omitempty"`
	RunID            string               `json:"run_id,omitempty"`
	Iteration        int                  `json:"iteration,omitempty"`
}

AgentLLMInput is the input to AgentLLMActivity and AgentLLMStreamActivity. When ConversationID is set, the activity loads history from the store. MessageID is the assistant text id for TEXT_MESSAGE_* (and stream ordering with REASONING_*); the workflow sets it each turn. RetrieverContext is the pre-fetched RAG context from AgentRetrieverActivity (prefetch / hybrid modes). MemoryContext is the pre-fetched long-term memory context from AgentMemoryRecallActivity. StreamWorkflowID is the Temporal workflow ID of the stream the activity publishes events to; empty disables activity-side event publishing (non-streaming / non-approval runs).

type AgentLLMResult 

type AgentLLMResult struct {
	Content    string               `json:"content"`
	ToolCalls  []ToolCallRequest    `json:"tool_calls"`
	Usage      *interfaces.LLMUsage `json:"usage,omitempty"`
	RetryCount int32                `json:"retry_count,omitempty"` // number of Temporal retries before this successful attempt (Attempt - 1)
}

AgentLLMResult is the return value of AgentLLMActivity. Workflow uses it to decide: return content or execute tools.

type AgentMemoryRecallInput added in v0.2.2 

type AgentMemoryRecallInput struct {
	AgentFingerprint string                 `json:"agent_fingerprint,omitempty"`
	RunID            string                 `json:"run_id,omitempty"`
	UserPrompt       string                 `json:"user_prompt"`
	MemoryScope      interfaces.MemoryScope `json:"memory_scope,omitempty"`
}

AgentMemoryRecallInput is the input to AgentMemoryRecallActivity.

type AgentMemoryRecallResult added in v0.2.2 

type AgentMemoryRecallResult struct {
	MemoryContext string `json:"memory_context,omitempty"`
	TotalRecalls  int64  `json:"total_recalls,omitempty"`
	FailedRecalls int64  `json:"failed_recalls,omitempty"`
}

AgentMemoryRecallResult is the return value of AgentMemoryRecallActivity.

type AgentMemoryStoreInput added in v0.2.2 

type AgentMemoryStoreInput struct {
	AgentFingerprint string                 `json:"agent_fingerprint,omitempty"`
	RunID            string                 `json:"run_id,omitempty"`
	MemoryScope      interfaces.MemoryScope `json:"memory_scope,omitempty"`
	Messages         []interfaces.Message   `json:"messages,omitempty"`
}

AgentMemoryStoreInput is the input to AgentMemoryStoreActivity.

type AgentRetrieverInput added in v0.1.11 

type AgentRetrieverInput struct {
	AgentFingerprint string `json:"agent_fingerprint,omitempty"`
	RunID            string `json:"run_id,omitempty"`
	UserPrompt       string `json:"user_prompt"`
}

AgentRetrieverInput is the input to AgentRetrieverActivity.

type AgentRetrieverResult added in v0.1.11 

type AgentRetrieverResult struct {
	RetrieverContext string `json:"retriever_context,omitempty"`
	TotalSearches    int64  `json:"total_searches,omitempty"`
	FailedSearches   int64  `json:"failed_searches,omitempty"`
}

AgentRetrieverResult is the return value of AgentRetrieverActivity. RetrieverContext is the combined, formatted document context from all retrievers; empty when no documents were found. BuildLLMRequest injects it as a labeled "Relevant Context" message ahead of conversation history; the system prompt itself is left untouched.

type AgentToolApprovalInput 

type AgentToolApprovalInput struct {
	AgentName        string         `json:"agent_name"`
	ToolCallID       string         `json:"tool_call_id"`
	ToolName         string         `json:"tool_name"`
	ToolDisplayName  string         `json:"tool_display_name,omitempty"`
	Args             map[string]any `json:"args"`
	StreamWorkflowID string         `json:"stream_workflow_id"`
	SubAgentName     string         `json:"sub_agent_name,omitempty"`
	AgentFingerprint string         `json:"agent_fingerprint,omitempty"`
}

AgentToolApprovalInput is the input to AgentToolApprovalActivity. StreamWorkflowID is the workflow whose WorkflowStream receives the approval event; the activity publishes it via workflowstreams so the client can display the approval request.

type AgentToolAuthorizeInput added in v0.1.3 

type AgentToolAuthorizeInput struct {
	ToolName         string         `json:"tool_name"`
	Args             map[string]any `json:"args"`
	ToolCallID       string         `json:"tool_call_id"`
	AgentFingerprint string         `json:"agent_fingerprint,omitempty"`
}

type AgentToolAuthorizeResult added in v0.1.3 

type AgentToolAuthorizeResult struct {
	Allowed bool   `json:"allowed"`
	Reason  string `json:"reason,omitempty"`
}

type AgentToolExecuteInput 

type AgentToolExecuteInput struct {
	ToolName         string                 `json:"tool_name"`
	Args             map[string]any         `json:"args"`
	ConversationID   string                 `json:"conversation_id,omitempty"`
	Messages         []interfaces.Message   `json:"messages,omitempty"`
	ToolCallID       string                 `json:"tool_call_id,omitempty"`
	RunID            string                 `json:"run_id,omitempty"`
	Iteration        int                    `json:"iteration,omitempty"`
	AgentFingerprint string                 `json:"agent_fingerprint,omitempty"`
	MemoryScope      interfaces.MemoryScope `json:"memory_scope,omitempty"`
}

AgentToolExecuteInput is the input to AgentToolExecuteActivity.

type AgentWorkflowCleanupInput added in v0.3.0 

type AgentWorkflowCleanupInput struct {
	ApprovalActivityIDs []string `json:"approval_activity_ids,omitempty"`
}

AgentWorkflowCleanupInput is the input to AgentWorkflowCleanupActivity. ApprovalActivityIDs are Temporal activity IDs of still-pending AgentToolApprovalActivity tasks.

type AgentWorkflowInput 

type AgentWorkflowInput struct {
	UserPrompt       string                               `json:"user_prompt,omitempty"`
	RootWorkflowID   string                               `json:"root_workflow_id,omitempty"`
	StreamState      *workflowstreams.WorkflowStreamState `json:"stream_state,omitempty"`
	StreamingEnabled bool                                 `json:"streaming_enabled,omitempty"`
	ConversationID   string                               `json:"conversation_id,omitempty"`
	AgentFingerprint string                               `json:"agent_fingerprint,omitempty"`
	RunID            string                               `json:"run_id,omitempty"`
	MemoryScope      interfaces.MemoryScope               `json:"memory_scope,omitempty"`
	EventTypes       []events.AgentEventType              `json:"event_types,omitempty"`
	SubAgentDepth    int                                  `json:"sub_agent_depth,omitempty"`
	SubAgentRoutes   map[string]SubAgentRoute             `json:"sub_agent_routes,omitempty"`
	MaxSubAgentDepth int                                  `json:"max_sub_agent_depth,omitempty"`
	State            *AgentWorkflowState                  `json:"state,omitempty"`
}

AgentWorkflowInput is the input to AgentWorkflow.

RootWorkflowID is empty for the top-level agent run (the root). Sub-agent child workflows set it to the root workflow's ID so their events are published to the root's WorkflowStream instead of a separate stream that the client is not subscribed to.

StreamState carries the WorkflowStream's durable log across ContinueAsNew boundaries (root only). StreamingEnabled enables partial LLM token streaming (set by Agent.Stream). ConversationID is set when conversation is used; workflow fetches messages and writes via activities. SubAgentDepth is 0 for a top-level user run; each child workflow increments it (runtime cap vs maxSubAgentDepth). SubAgentRoutes maps sub-agent tool name -> route; built from WithSubAgents when the run starts. MemoryScope is resolved before the workflow starts and passed through for recall/store activities. EventTypes controls whether events are published to the stream: empty = no events, ["*"] = all events. AgentFingerprint is the per-run digest (config + resolved tools). Caller and worker compute it at resolve time.

type AgentWorkflowState added in v0.1.9 

type AgentWorkflowState struct {
	Iteration int                   `json:"iteration"`
	Messages  []interfaces.Message  `json:"messages"`
	LLMUsage  *interfaces.LLMUsage  `json:"llm_usage,omitempty"`
	Telemetry *types.AgentTelemetry `json:"telemetry,omitempty"`
}

AgentWorkflowState is the state of the agent workflow. It is used to store the state of the agent workflow on continue-as-new.

type Option 

type Option func(*TemporalRuntime)

Option configures a TemporalRuntime.

func WithA2AFingerprint added in v0.1.9 

func WithA2AFingerprint(fp string) Option

WithA2AFingerprint sets the A2A wiring digest used with ComputeAgentFingerprint. Must match pkg/agent's a2aConfigFingerprint for the same WithA2AConfig / WithA2AClients wiring.

func WithAgentConfig added in v0.2.2 

func WithAgentConfig(cfg sdkruntime.AgentConfig) Option

WithAgentConfig sets static LLM, session, limits, and tool approval policy on the worker runtime.

func WithAgentMode added in v0.1.3 

func WithAgentMode(mode string) Option

WithAgentMode sets the agent mode string (e.g. "interactive", "autonomous") used with ComputeAgentFingerprint. Must match pkg/agent WithAgentMode on both caller and worker.

func WithAgentSpec 

func WithAgentSpec(spec sdkruntime.AgentSpec) Option

WithAgentSpec sets identity and response format (same shape as sdkruntime.ExecuteRequest.AgentSpec).

func WithAgentToolExecutionMode added in v0.1.7 

func WithAgentToolExecutionMode(mode types.AgentToolExecutionMode) Option

WithAgentToolExecutionMode sets the tool execution mode. The value is stored in the embedded base.Runtime.ToolExecutionMode so it drives both fingerprinting and workflow-level tool execution.

func WithApprovalHandler added in v0.3.0 

func WithApprovalHandler(fn types.ApprovalHandler) Option

WithApprovalHandler sets the Run-path approval callback (from agent WithApprovalHandler). Stream uses CUSTOM events + Approve/OnApproval instead.

func WithDisableFingerprintCheck added in v0.1.6 

func WithDisableFingerprintCheck(disable bool) Option

WithDisableFingerprintCheck disables activity-time caller-vs-worker fingerprint verification. Break-glass only: use temporarily during rollout incidents; keep false in production for safety.

func WithDisableLocalWorker added in v0.1.3 

func WithDisableLocalWorker(disable bool) Option

WithDisableLocalWorker mirrors pkg/agent DisableLocalWorker. When false, the client embeds a worker and the runtime skips DescribeTaskQueue poller checks before starting workflows.

func WithHooksFingerprint added in v0.2.3 

func WithHooksFingerprint(fp string) Option

WithHooksFingerprint sets the hook group names digest used with ComputeAgentFingerprint. Must match pkg/agent [hookGroupsFingerprint] for the same WithHooks wiring.

func WithInstanceId 

func WithInstanceId(instanceId string) Option

WithInstanceId appends a suffix to the task queue (e.g. "myq-pod1") so multiple instances of the same agent can run on isolated queues.

func WithLogger 

func WithLogger(l logger.Logger) Option

WithLogger sets the logger.Logger used by the runtime. A nil value is silently ignored so the safe logger.NoopLogger default is preserved.

func WithMCPFingerprint added in v0.1.2 

func WithMCPFingerprint(fp string) Option

WithMCPFingerprint sets the MCP wiring digest used with ComputeAgentFingerprint. Must match pkg/agent's mcpConfigFingerprint for the same WithMCPConfig / WithMCPClients wiring.

func WithMetrics added in v0.1.10 

func WithMetrics(m interfaces.Metrics) Option

WithMetrics sets the optional interfaces.Metrics for this runtime.

func WithObservabilityFingerprint added in v0.1.10 

func WithObservabilityFingerprint(fp string) Option

WithObservabilityFingerprint sets the OTLP observability digest used with ComputeAgentFingerprint. Must match pkg/agent observabilityConfigFingerprint for the same WithObservabilityConfig wiring.

func WithPolicyFingerprint 

func WithPolicyFingerprint(fp string) Option

WithPolicyFingerprint sets the opaque policy digest used with ComputeAgentFingerprint. Must match pkg/agent's toolPolicyFingerprint for the same agent options.

func WithRemoteWorker 

func WithRemoteWorker(remote bool) Option

WithRemoteWorker marks the runtime as a remote worker (true for [NewAgentWorker], false for client [Agent] runtimes).

func WithRetrieverFingerprint added in v0.1.11 

func WithRetrieverFingerprint(fp string) Option

WithRetrieverFingerprint sets the retriever wiring digest (mode + retriever names). Must match pkg/agent [retrieverConfigFingerprint] for the same agent.

func WithTemporalClient 

func WithTemporalClient(tc client.Client, taskQueue string) Option

WithTemporalClient injects a caller-managed Temporal client and task queue. The runtime does NOT close this client on TemporalRuntime.Close.

func WithTemporalConfig 

func WithTemporalConfig(config *TemporalConfig) Option

WithTemporalConfig dials a new Temporal client from the supplied connection parameters. The runtime owns the resulting client and closes it on TemporalRuntime.Close.

func WithToolsResolver added in v0.2.2 

func WithToolsResolver(fn ToolsResolver) Option

WithToolsResolver sets the callback that resolves tools at activity time on the worker runtime.

func WithTracer added in v0.1.10 

func WithTracer(t interfaces.Tracer) Option

WithTracer sets the optional interfaces.Tracer. When the runtime dials its own Temporal client (WithTemporalConfig) and the tracer implements interfaces.OTelTracer, a Temporal OpenTelemetry client interceptor is attached automatically.

type PublishStreamEventInput added in v0.3.0 

type PublishStreamEventInput struct {
	StreamWorkflowID string          `json:"stream_workflow_id"`
	EventJSON        json.RawMessage `json:"event_json"`
}

PublishStreamEventInput is the activity input for PublishStreamEventActivity. StreamWorkflowID is the Temporal workflow ID whose WorkflowStream should receive the event.

type SubAgentRoute added in v0.1.12 

type SubAgentRoute struct {
	Name             string                   `json:"name"`
	TaskQueue        string                   `json:"task_queue,omitempty"`
	ChildRoutes      map[string]SubAgentRoute `json:"child_routes,omitempty"`
	AgentFingerprint string                   `json:"agent_fingerprint,omitempty"`
}

SubAgentRoute tells the Temporal runtime how to delegate to a sub-agent child workflow. It is serialised into AgentWorkflowInput and propagated to child workflows unchanged.

type TemporalConfig 

type TemporalConfig struct {
	Host      string
	Port      int
	Namespace string
	TaskQueue string
}

TemporalConfig holds the Temporal server connection parameters. Pass it to WithTemporalConfig when the runtime should dial its own client.

type TemporalRuntime 

type TemporalRuntime struct {
	base.Runtime // AgentSpec, AgentConfig, Tracer, Metrics, ToolExecutionMode
	// contains filtered or unexported fields
}

TemporalRuntime implements runtime.WorkerRuntime using Temporal workflows and activities as the execution backend. Agent event delivery goes through the workflowstreams WorkflowStream (see stream.go), not an in-process event bus (LocalRuntime shares its bus onto nested local sub-agent runtimes instead). It embeds base.Runtime for the common agent fields (AgentSpec, AgentConfig, Tracer, Metrics, ToolExecutionMode) and holds all Temporal-specific connection and fingerprint state as flat fields.

func NewTemporalRuntime 

func NewTemporalRuntime(opts ...Option) (*TemporalRuntime, error)

func (*TemporalRuntime) AddConversationMessagesActivity 

func (rt *TemporalRuntime) AddConversationMessagesActivity(ctx context.Context, input AddConversationMessagesInput) error

AddConversationMessagesActivity adds messages to the conversation memory.

func (*TemporalRuntime) AgentLLMActivity 

func (rt *TemporalRuntime) AgentLLMActivity(ctx context.Context, input AgentLLMInput) (*AgentLLMResult, error)

AgentLLMActivity calls the LLM and returns content plus any tool calls. When input.ConversationID is set, fetches from store and adds assistant message on completion. Events (e.g. reasoning) are published to the WorkflowStream when StreamWorkflowID is set.

func (*TemporalRuntime) AgentLLMStreamActivity 

func (rt *TemporalRuntime) AgentLLMStreamActivity(ctx context.Context, input AgentLLMInput) (*AgentLLMResult, error)

AgentLLMStreamActivity streams LLM response tokens. Event order: optional reasoning block (REASONING_*), then TEXT_MESSAGE_START → TEXT_MESSAGE_CONTENT* → TEXT_MESSAGE_END. When input.ConversationID is set, fetches messages from conversation and prepends to workflow messages. Events are published to the WorkflowStream via workflowstreams.Client with batched signals.

func (*TemporalRuntime) AgentMemoryRecallActivity added in v0.2.2 

func (rt *TemporalRuntime) AgentMemoryRecallActivity(ctx context.Context, input AgentMemoryRecallInput) (*AgentMemoryRecallResult, error)

AgentMemoryRecallActivity loads scoped long-term memories and returns formatted prompt context.

func (*TemporalRuntime) AgentMemoryStoreActivity added in v0.2.2 

func (rt *TemporalRuntime) AgentMemoryStoreActivity(ctx context.Context, input AgentMemoryStoreInput) error

AgentMemoryStoreActivity extracts and persists long-term memories from the run.

func (*TemporalRuntime) AgentRetrieverActivity added in v0.1.11 

func (rt *TemporalRuntime) AgentRetrieverActivity(ctx context.Context, input AgentRetrieverInput) (*AgentRetrieverResult, error)

AgentRetrieverActivity runs all configured retrievers in parallel using input.UserPrompt as the query, then returns a combined document context string for injection into the LLM system prompt. Called only for types.RetrieverModePrefetch and types.RetrieverModeHybrid. Partial failures (some retrievers fail) are logged and skipped; if all retrievers fail, the activity returns an error so Temporal can retry per the retry policy.

func (*TemporalRuntime) AgentToolApprovalActivity 

func (rt *TemporalRuntime) AgentToolApprovalActivity(ctx context.Context, input AgentToolApprovalInput) (types.ApprovalStatus, error)

AgentToolApprovalActivity blocks until the driver completes it via CompleteActivity. Publishes a CUSTOM (tool_approval / delegation) event to the WorkflowStream so the client can display the approval UI and call OnApproval with the task token. StreamWorkflowID is the workflow whose stream receives the approval event (root or sub-agent root).

func (*TemporalRuntime) AgentToolAuthorizeActivity added in v0.1.3 

func (rt *TemporalRuntime) AgentToolAuthorizeActivity(ctx context.Context, input AgentToolAuthorizeInput) (AgentToolAuthorizeResult, error)

AgentToolAuthorizeActivity checks optional programmatic authorization before approval/execute.

func (*TemporalRuntime) AgentToolExecuteActivity 

func (rt *TemporalRuntime) AgentToolExecuteActivity(ctx context.Context, input AgentToolExecuteInput) (string, error)

AgentToolExecuteActivity executes a tool by name and adds tool message to conversation when ConversationID is set.

func (*TemporalRuntime) AgentWorkflow 

func (rt *TemporalRuntime) AgentWorkflow(ctx workflow.Context, input AgentWorkflowInput) (*types.AgentRunResult, error)

AgentWorkflow runs the agent loop: LLM → tool calls → approval/execute → feed results back to LLM → repeat. Stops when LLM returns no tool calls, or max iterations reached.

When input.RootWorkflowID is empty, this is the root workflow: it creates a WorkflowStream and publishes all agent events directly to it (no activity overhead). When non-empty, this is a sub-agent: events are forwarded to the root's stream via PublishStreamEventActivity.

ContinueAsNew: when workflow history length or size (GetInfo) exceeds agentWorkflowHistory*, after tool results are merged into messages for that iteration. The root workflow detaches the stream's pollers, waits for all handlers to finish, captures the stream state, and carries it forward in StreamState.

func (*TemporalRuntime) AgentWorkflowCleanupActivity added in v0.3.0 

func (rt *TemporalRuntime) AgentWorkflowCleanupActivity(ctx context.Context, input AgentWorkflowCleanupInput) error

AgentWorkflowCleanupActivity cancels leftover pending approval activities by ID. Invoked from AgentWorkflow defer via a disconnected context when the workflow exits with pendingApprovals still populated (typically cancel while waiting for user approval). Already-resolved / not-found activities are ignored; other errors are returned for retry.

func (*TemporalRuntime) Approve 

func (rt *TemporalRuntime) Approve(ctx context.Context, approvalToken string, status types.ApprovalStatus) error

func (*TemporalRuntime) Close 

func (rt *TemporalRuntime) Close()

func (*TemporalRuntime) GetRunHandle added in v0.3.0 

func (rt *TemporalRuntime) GetRunHandle(ctx context.Context, runID string) (runtime.RunHandle, error)

GetRunHandle reconnects to an existing non-stream agent run by runID.

It derives the Temporal workflow ID (agent-run-…), describes the execution, and returns a runtime.RunHandle. Same-runtime: if the run is already in activeRuns, that handle is returned. Otherwise it re-attaches driveRun on an independent ctx (Background + Limits.Timeout) — the caller's ctx is only used for status/describe and does not CancelWorkflow when cancelled. Stop the run with runtime.RunHandle.Cancel. Approvals reattach when approvalHandler is set (QueryIsApprovalPending skips already-resolved CUSTOM events). Crash reconnect assumes the same agent config (see version/fingerprint TODO below).

Returns types.ErrRunNotFound when runID is empty or the workflow is unknown. Returns types.ErrRunAlreadyCompleted when the workflow is already terminal.

func (*TemporalRuntime) GetStreamHandle added in v0.3.0 

func (rt *TemporalRuntime) GetStreamHandle(ctx context.Context, runID string) (runtime.StreamHandle, error)

GetStreamHandle reconnects to an existing stream agent run by runID.

It derives the Temporal stream workflow ID (agent-stream-…), describes the execution, and returns a runtime.StreamHandle. Same-runtime: if the stream is already in activeStreams, that handle is returned. Otherwise it registers a new handle (cleanup on awaitCompletion) and starts driveStream on an independent ctx (Background + Limits.Timeout) — the caller's ctx is only used for status/describe and does not CancelWorkflow when cancelled. Stop the run with runtime.StreamHandle.Cancel. Call runtime.StreamHandle.Events (optionally with fromOffset > 0) to subscribe. Crash reconnect assumes the same agent config (see version/fingerprint TODO below).

Returns types.ErrStreamNotFound when runID is empty or the workflow is unknown. Returns types.ErrRunAlreadyCompleted when the workflow is already terminal.

func (*TemporalRuntime) OnApproval 

func (rt *TemporalRuntime) OnApproval(ctx context.Context, approvalToken string, status types.ApprovalStatus) error

OnApproval completes a tool approval when using ExecuteStream. Pass the string from ev.Approval (see the streaming examples) along with the chosen status.

Returns types.ErrApprovalAlreadyResolved when the approval token refers to an activity that has already been completed. This can happen after Events (reconnect) replays a CUSTOM event for an approval that was resolved while the subscriber was disconnected. Treat it as informational.

func (*TemporalRuntime) PublishStreamEventActivity added in v0.3.0 

func (rt *TemporalRuntime) PublishStreamEventActivity(ctx context.Context, in PublishStreamEventInput) error

PublishStreamEventActivity publishes one agent event to a remote WorkflowStream. Used by sub-agent workflows to forward their events to the root workflow's stream so that the subscribing client sees a unified event sequence across the delegation tree. A new workflowstreams.Client is created per call with forceFlush=true to deliver immediately; the overhead is comparable to the prior SendAgentEventUpdateActivity approach.

func (*TemporalRuntime) Run added in v0.3.0 

func (rt *TemporalRuntime) Run(ctx context.Context, req *runtime.RunRequest) (runtime.RunHandle, error)

Run starts the agent workflow and returns a runtime.RunHandle immediately. Use runtime.RunHandle.Get or runtime.RunHandle.Done to wait for completion. When rt.approvalHandler is set, after ExecuteWorkflow succeeds a background goroutine subscribes to the WorkflowStream for CUSTOM approval events and completes them via CompleteActivity.

func (*TemporalRuntime) Start 

func (rt *TemporalRuntime) Start(ctx context.Context) error

Start starts the worker (blocks until Stop is called).

func (*TemporalRuntime) Stop 

func (rt *TemporalRuntime) Stop()

Stop stops the Temporal worker(s).

func (*TemporalRuntime) Stream added in v0.3.0 

func (rt *TemporalRuntime) Stream(ctx context.Context, req *runtime.RunRequest) (runtime.StreamHandle, error)

Stream starts the agent Temporal workflow and returns a runtime.StreamHandle immediately.

It does not open the event subscription here — call runtime.StreamHandle.Events (optionally with fromOffset > 0 to reconnect). The handle owns Status/Cancel/Events for this runID. Same-runtime reuse goes through activeStreams (see TemporalRuntime.GetStreamHandle).

Cancelling ctx cancels the agent run (same idea as TemporalRuntime.Run). Cancelling the context passed to runtime.StreamHandle.Events only stops that subscriber — it does not cancel the Temporal workflow. Use a separate Events ctx for reconnect / "subscriber gone". Agent Limits.Timeout still applies when the Stream ctx has no deadline.

Identifiers: runID is minted here; workflowID uses the stream naming convention (agent-stream-…); threadID prefers ConversationID for synthetic RUN_STARTED/FINISHED. Registers in activeStreams; cleanup deletes the entry from awaitCompletion when the workflow finishes. Cancel/timeout stop the workflow via driveStream.

type ToolCallRequest 

type ToolCallRequest struct {
	ToolCallID      string         `json:"tool_call_id"` // from LLM; used to match tool results
	ToolName        string         `json:"tool_name"`
	ToolDisplayName string         `json:"tool_display_name,omitempty"`
	ToolKind        types.ToolKind `json:"tool_kind"`
	Args            map[string]any `json:"args"`
	NeedsApproval   bool           `json:"needs_approval"`
}

ToolCallRequest is a tool invocation with approval flag. NeedsApproval is set by AgentLLMActivity.

type ToolsResolver added in v0.2.2 

type ToolsResolver func(ctx context.Context) ([]interfaces.Tool, error)

ToolsResolver resolves per-run tools from registries at activity entry (worker runtime).

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.