temporal

package
v0.2.0 Latest Latest
Warning

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

Go to latest
Published: Aug 29, 2026 License: Apache-2.0 Imports: 28 Imported by: 0

Documentation

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

func Dial

func Dial(opts client.Options) (client.Client, error)

Dial is client.Dial with Temporal's OpenTelemetry v2 plugin prepended. Call telemetry.Init first so the global TracerProvider is ReplaySafe; NewPlugin reads otel.GetTracerProvider() and otel.GetMeterProvider(). Default: context propagation only (no SDK auto-spans). SessionWorkflow starts tacklr.turn via temporalotel.Tracer.

func NewWorker

func NewWorker(c client.Client, taskQueue string, opts WorkerOptions) worker.Worker

NewWorker returns a Temporal worker with EnableSessionWorker and SessionWorkflow plus Inference, Tool, and CommitToolOutput activities. Worker sessions are created only when WorkflowInput.TurnLocalityTimeout is set (Runtime option WithTurnLocality). Create the client with Dial so Temporal's OTEL v2 plugin propagates trace context.

func SessionWorkflow

func SessionWorkflow(ctx workflow.Context, in WorkflowInput) (string, error)

SessionWorkflow is the harness wait loop: one Temporal workflow per agent session. It is the primary OpenTelemetry instrumentor: one tacklr.turn span per prompt or resume, with Inference/Tool activities as children via OTEL v2 propagation.

Types

type Activities

type Activities struct {
	Catalog        durable.Catalog
	Snapshots      durable.SnapshotStore
	Projection     vfs.Projection
	Fallback       durable.EventLog
	DisableStreams bool
}

Activities are the Inference and Tool bodies registered on the worker.

func (*Activities) CommitToolOutput

func (a *Activities) CommitToolOutput(ctx context.Context, in CommitToolInput) (ToolOutput, error)

func (*Activities) Inference

func (a *Activities) Inference(ctx context.Context, in InferenceInput) (InferenceOutput, error)

func (*Activities) Tool

func (a *Activities) Tool(ctx context.Context, in ToolInput) (ToolOutput, error)

type CommitToolInput

type CommitToolInput struct {
	SessionID  durable.SessionID
	AgentID    string
	MCPServers []mcp.MCPConfig
	Call       tacklr.ToolCall
	Output     string
	Auth       durable.AuthContext
	Mounts     []durable.MountRecipe
	Specialist string
	State      map[string]any
}

CommitToolInput records a tool output on the staged batch without executing the tool. SessionWorkflow uses this after spawn_specialist child completion.

type InferenceInput

type InferenceInput struct {
	SessionID     durable.SessionID
	AgentID       string
	MCPServers    []mcp.MCPConfig
	User          *tacklr.Message
	HadToolRound  bool
	ModelRequests int
	Resume        map[string][]byte
	Auth          durable.AuthContext
	Mounts        []durable.MountRecipe
	Specialist    string
	// State is host userState merged after checkpoint restore.
	State map[string]any
}

InferenceInput is the typed Inference activity argument.

type InferenceOutput

type InferenceOutput struct {
	Complete  bool
	ToolCalls []tacklr.ToolCall
	Result    string
}

InferenceOutput is the typed Inference activity result.

type Option

type Option func(*Runtime)

Option configures New.

func WithDisableStreams

func WithDisableStreams() Option

WithDisableStreams publishes and subscribes only via the fallback EventLog. The Temporal testsuite mock cannot serve Workflow Streams (GetSystemInfo); use this in unit tests. The dev-server integration test leaves streams on.

func WithEventLog

func WithEventLog(l *inprocess.MemoryEventLog) Option

WithEventLog sets the fallback EventLog used when Workflow Streams is disabled.

func WithSnapshotStore

func WithSnapshotStore(s durable.SnapshotStore) Option

WithSnapshotStore replaces the memory SnapshotStore.

func WithTurnLocality

func WithTurnLocality(d time.Duration) Option

WithTurnLocality keeps a turn's activities on one Temporal worker for that duration so the turn's VFS stays on the same process. Zero (default) does not pin activities; they may run on any worker.

type Runtime

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

Runtime implements durable.Runtime with one Temporal workflow per session.

func New

func New(c client.Client, taskQueue string, catalog durable.Catalog, opts ...Option) *Runtime

New constructs a Temporal Runtime. The host must also run Worker on the same task queue with EnableSessionWorker. Tokens travel on Prompt/Resume payloads.

func (*Runtime) Cancel

func (r *Runtime) Cancel(ctx context.Context, sessionID durable.SessionID) error

Cancel implements durable.Runtime.

func (*Runtime) Children

func (r *Runtime) Children(ctx context.Context, parent durable.SessionID) ([]durable.SessionID, error)

Children implements durable.Runtime.

func (*Runtime) Close

func (r *Runtime) Close(ctx context.Context, sessionID durable.SessionID) error

Close implements durable.Runtime.

func (*Runtime) CreateSession

func (r *Runtime) CreateSession(ctx context.Context, req durable.CreateSession) (durable.SessionID, error)

CreateSession implements durable.Runtime.

func (*Runtime) Head

func (r *Runtime) Head(ctx context.Context, sessionID durable.SessionID) (durable.Seq, error)

Head implements durable.Runtime. When Workflow Streams is on, this is the stream's next offset so Subscribe(after Head) skips prior-turn events.

func (*Runtime) Prompt

func (r *Runtime) Prompt(ctx context.Context, sessionID durable.SessionID, msg durable.Prompt) error

Prompt implements durable.Runtime.

func (*Runtime) Resume

func (r *Runtime) Resume(ctx context.Context, sessionID durable.SessionID, resume durable.Resume) error

Resume implements durable.Runtime.

func (*Runtime) Status

Status implements durable.Runtime.

func (*Runtime) Subscribe

func (r *Runtime) Subscribe(ctx context.Context, sessionID durable.SessionID, after durable.Seq) (durable.Subscription, error)

Subscribe implements durable.Runtime.

type ToolInput

type ToolInput struct {
	SessionID  durable.SessionID
	AgentID    string
	MCPServers []mcp.MCPConfig
	Call       tacklr.ToolCall
	Auth       durable.AuthContext
	Mounts     []durable.MountRecipe
	Specialist string
	// Known is this session's child ids for list_children (not a ChildOp ledger).
	Known []durable.SessionID
	State map[string]any
}

ToolInput is the typed Tool activity argument.

type ToolOutput

type ToolOutput struct {
	Interrupted bool
	InterruptID string
	SpawnID     durable.SessionID
	SpawnSpec   string
	SpawnTask   string
	CancelID    durable.SessionID
	AwaitID     durable.SessionID
}

ToolOutput is the typed Tool activity result. Spawn/cancel/await are this call's Runtime intent; the workflow starts, cancels, or waits via ExecuteChildWorkflow.

type WorkerOptions

type WorkerOptions struct {
	Catalog        durable.Catalog
	Snapshots      durable.SnapshotStore
	Projection     vfs.Projection
	Fallback       durable.EventLog
	DisableStreams bool
}

WorkerOptions configures NewWorker.

type WorkflowInput

type WorkflowInput struct {
	SessionID  durable.SessionID
	AgentID    string
	MCPServers []mcp.MCPConfig
	Mounts     []durable.MountRecipe
	Auth       durable.AuthContext
	// TurnLocalityTimeout, when > 0, pins the turn's activities to one worker
	// (Temporal CreateSession). Zero skips worker sessions: activities can run
	// on any worker. There is no default timeout.
	TurnLocalityTimeout time.Duration
	// Prompt, when set, runs one turn then completes the workflow (spawn_specialist child).
	Prompt     string
	Parent     durable.SessionID
	Specialist string
	// State is CreateSession.State, already JSON-roundtripped by Runtime.CreateSession.
	State map[string]any
}

WorkflowInput is the typed start payload (no interface{}).

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL