Help us shape the future of Go! Take the 2026 Go Developer Survey

temporal

package
v0.1.0-alpha.4 Latest Latest
Warning

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

Go to latest
Published: Oct 18, 2026 License: Apache-2.0 Imports: 31 Imported by: 0

Documentation

Overview

Package temporal implements the platform-spine's durable session orchestration: a SessionWorkflow keyed by the public session ID, granular model/tool Activities behind a Workflow-owned agent loop, idempotent PostgreSQL completion, and the outbox relay that wakes the workflow with Signal-With-Start.

PostgreSQL remains the source of truth for public events; Temporal owns only in-flight orchestration. Signals carry wakeup metadata only — never event payloads.

Index

Constants

View Source
const (
	ActivityLoadEvents           = "LoadEvents"
	ActivityLoadInterrupt        = "LoadInterrupt"
	ActivityLoadPendingActions   = "LoadPendingActions"
	ActivityPrepareTurn          = "PrepareTurn"
	ActivityAdmitModelRequest    = "AdmitModelRequest"
	ActivityStartModelRequest    = "StartModelRequest"
	ActivityAppendWorkflowEvents = "AppendWorkflowEvents"
	ActivityRecordModelRetry     = "RecordModelRetry"
	ActivityResumeModelRetry     = "ResumeModelRetry"
	ActivityCallModel            = "CallModel"
	ActivityAccountModelRequest  = "AccountModelRequest"
	ActivityEvaluateOutcome      = "EvaluateOutcome"
	ActivityExecuteTool          = "ExecuteTool"
	ActivityCompleteWorkflowTurn = "CompleteWorkflowTurn"
)

Registered activity names. Referenced by the workflow through the exported symbols; named explicitly so a rename cannot silently break replay.

View Source
const (
	ActivityEnsureToolPermissionAttempt = "EnsureToolPermissionAttempt"
	ActivityEvaluateToolPermission      = "EvaluateToolPermission"
)
View Source
const (
	// TaskQueue is the single task queue the session worker listens on for this
	// slice. Split queues (session vs thread vs webhook) are a later concern.
	TaskQueue = "mango-session"

	// WakeupSignalName is the Signal the outbox relay (and the API fast path)
	// send to wake a SessionWorkflow. Its payload is metadata only.
	WakeupSignalName = "session-wakeup"

	// SessionWorkflowType is the registered workflow name. The Workflow ID is the
	// public session ID, so starting the same session twice is idempotent.
	SessionWorkflowType = "SessionWorkflow"

	// SessionThreadWorkflowType is the independent durable loop for one child
	// Thread. Its Workflow ID is derived solely from the public Thread id.
	SessionThreadWorkflowType = "SessionThreadWorkflow"
)

Variables

This section is empty.

Functions

func Dial

func Dial(cfg ClientConfig) (client.Client, error)

Dial opens a Temporal client. The caller owns its lifecycle (Close).

func NewWorker

func NewWorker(c client.Client, acts *Activities) worker.Worker

NewWorker builds a Temporal worker that serves the SessionWorkflow and its Activities on the session task queue. The caller runs it (worker.Run) and owns the client's lifecycle.

Activities are registered under stable names so a Go method rename cannot silently break workflow replay.

func NewWorkerOnTaskQueue

func NewWorkerOnTaskQueue(
	c client.Client,
	acts *Activities,
	taskQueue string,
) worker.Worker

func SessionThreadWorkflow

func SessionThreadWorkflow(
	ctx workflow.Context,
	in SessionThreadWorkflowInput,
) error

SessionThreadWorkflow is the independent durable loop for one child Agent. PostgreSQL owns its Thread identity and ledger; the Workflow owns only the in-flight cursor and model/tool command sequence.

func SessionWorkflow

func SessionWorkflow(ctx workflow.Context, in SessionWorkflowInput) error

SessionWorkflow is the durable, long-lived orchestrator for one session. Its Workflow ID is the public session ID, so Signal-With-Start is idempotent.

Design invariants:

  • PostgreSQL is the event source of truth. Signals are wakeups carrying only the highest known receipt sequence; the workflow loads authoritative events after its own durable cursor.
  • A pending client-action barrier is selected before ordinary receipt-order work. Partial resolution remains parked; a full barrier resumes as one logical model turn before queued messages.
  • The durable cursor advances monotonically. Duplicate or out-of-order wakeups are harmless because authoritative events at or below it are ignored.
  • State carried across Continue-As-New is small: just the cursor. Completed model and tool Activity results enter Workflow history, while PostgreSQL remains authoritative for the public event ledger and projection.

Types

type AccountModelRequestInput

type AccountModelRequestInput struct {
	SessionID      string            `json:"session_id"`
	ThreadID       string            `json:"thread_id,omitempty"`
	RequestEventID string            `json:"request_event_id"`
	Model          domain.Model      `json:"model"`
	Usage          domain.TokenUsage `json:"usage"`
	StopReason     string            `json:"stop_reason,omitempty"`
}

type Activities

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

Activities holds the I/O dependencies of the session Activities. Self-hosted built-ins execute through Environment Work; only control-plane tools such as MCP, coordinator, Advisor, and Skill activation execute here.

func NewActivities

func NewActivities(
	modelClient model.Client,
	source EventSource,
	journal JournalStore,
	ids domain.IDGenerator,
	previewPublisher ...PreviewPublisher,
) *Activities

func (*Activities) AccountModelRequest

func (a *Activities) AccountModelRequest(
	ctx context.Context,
	in AccountModelRequestInput,
) error

AccountModelRequest commits provider usage independently of public turn completion. This preserves billed usage when a later tool, grader, interrupt, or terminal error prevents the turn itself from completing.

func (*Activities) AdmitModelRequest

func (a *Activities) AdmitModelRequest(
	ctx context.Context,
	in AdmitModelRequestInput,
) (AdmitModelRequestResult, error)

func (*Activities) AppendWorkflowEvents

func (a *Activities) AppendWorkflowEvents(
	ctx context.Context,
	in AppendWorkflowEventsInput,
) error

AppendWorkflowEvents publishes a completed, non-terminal prefix before the next model request starts. Final status, pending barriers, usage, attempts, and provider transcript still commit through CompleteWorkflowTurn.

func (*Activities) CallModel

func (a *Activities) CallModel(ctx context.Context, in CallModelInput) (CallModelResult, error)

CallModel bounds the response without retrying completed inference.

func (*Activities) CompleteWorkflowTurn

func (a *Activities) CompleteWorkflowTurn(
	ctx context.Context,
	in CompleteWorkflowTurnInput,
) (RunTurnResult, error)

CompleteWorkflowTurn commits the Workflow's durable output and optional tool attempt through one idempotent PostgreSQL transaction.

func (*Activities) EnsureToolPermissionAttempt

func (a *Activities) EnsureToolPermissionAttempt(ctx context.Context, owner domain.ToolPermissionOwner) (EnsureToolPermissionAttemptResult, error)

func (*Activities) EvaluateOutcome

EvaluateOutcome preserves usage when the grader result cannot be transported.

func (*Activities) EvaluateToolPermission

EvaluateToolPermission bills only a committed response, including a response received after cancellation. The immutable decision is never execution authority.

func (*Activities) ExecuteTool

ExecuteTool bounds transport after preserving the durable tool journal.

func (*Activities) LoadEvents

func (a *Activities) LoadEvents(ctx context.Context, in LoadEventsInput) (LoadEventsResult, error)

LoadEvents returns the ordered public event references after a cursor. Only metadata (id, seq, type) crosses back into workflow history; payloads stay in PostgreSQL.

func (*Activities) LoadInterrupt

func (a *Activities) LoadInterrupt(
	ctx context.Context,
	in LoadInterruptInput,
) (LoadInterruptResult, error)

LoadInterrupt scans authoritative history after one active trigger for the first unprocessed interrupt. It runs once before a turn's first interruptible Activity, after later wakeups, or while a durable pending-action barrier is parked. Signals remain metadata and ordinary model/tool progress does not poll PostgreSQL continuously.

func (*Activities) LoadPendingActions

func (a *Activities) LoadPendingActions(
	ctx context.Context,
	in LoadPendingActionsInput,
) (LoadPendingActionsResult, error)

LoadPendingActions returns the durable requires_action barrier as a small selector projection. The Workflow uses only this recorded Activity result to choose between parking, resuming the full barrier, and consuming ordinary messages.

func (*Activities) PrepareTurn

PrepareTurn returns either bounded immutable context or a terminal failure.

func (*Activities) RecordModelRetry

func (a *Activities) RecordModelRetry(
	ctx context.Context,
	in RecordModelRetryInput,
) error

RecordModelRetry publishes the retrying error and rescheduled projection in one PostgreSQL transaction. Keeping them together prevents observers from seeing a retry error while the Session still claims to be running.

func (*Activities) ResumeModelRetry

func (a *Activities) ResumeModelRetry(
	ctx context.Context,
	in ResumeModelRetryInput,
) error

ResumeModelRetry publishes status_running at the same linearization point that the Session projection returns to running.

func (*Activities) StartModelRequest

func (a *Activities) StartModelRequest(
	ctx context.Context,
	in StartModelRequestInput,
) error

StartModelRequest durably publishes the span start before the long-running CallModel Activity can emit any best-effort preview. The explicit Workflow- owned ID makes Activity retries harmless.

func (*Activities) WithContextTokenBudget

func (a *Activities) WithContextTokenBudget(tokens int) *Activities

WithContextTokenBudget overrides the request-time message budget. It is primarily useful for deterministic conformance tests and smaller providers.

func (*Activities) WithMCPAuthSource

func (a *Activities) WithMCPAuthSource(source mcpclient.AuthSource) *Activities

func (*Activities) WithMCPClient

func (a *Activities) WithMCPClient(client mcpclient.Client) *Activities

WithMCPClient replaces the remote MCP adapter. Production uses the official Go SDK-backed client by default; tests can inject a deterministic fake.

func (*Activities) WithPermissionEvaluator

func (a *Activities) WithPermissionEvaluator(evaluator permission.Evaluator) *Activities

func (*Activities) WithSkillInstructionLoader

func (a *Activities) WithSkillInstructionLoader(loader SkillInstructionLoader) *Activities

func (*Activities) WithToolOutputStore

func (a *Activities) WithToolOutputStore(store ToolOutputStore) *Activities

type AdmitModelRequestInput

type AdmitModelRequestInput struct {
	SessionID string `json:"session_id"`
	ThreadID  string `json:"thread_id,omitempty"`
}

type AdmitModelRequestResult

type AdmitModelRequestResult struct {
	Allowed bool `json:"allowed"`
}

type AdvisorToolSource

type AdvisorToolSource interface {
	CompleteAdvisorToolStep(
		context.Context,
		string,
		string,
		string,
		string,
		domain.ToolStepResult,
		domain.AdvisorConsultation,
	) error
}

type AppendWorkflowEventsInput

type AppendWorkflowEventsInput struct {
	SessionID      string              `json:"session_id"`
	TriggerEventID string              `json:"trigger_event_id"`
	Events         []domain.EventDraft `json:"events"`
}

AppendWorkflowEventsInput carries a completed non-terminal event prefix that must become visible before another model request begins.

type CallModelInput

type CallModelInput struct {
	SessionID           string `json:"session_id"`
	ThreadID            string `json:"thread_id,omitempty"`
	ModelRequestStartID string `json:"model_request_start_id,omitempty"`
	ModelRequestEndID   string `json:"model_request_end_id,omitempty"`
	// HandleRetryableErrors opts new Workflow histories into the public retry
	// lifecycle. Older histories leave it false and retain Activity-level retry
	// behavior, which keeps replay compatible across the rollout.
	HandleRetryableErrors bool          `json:"handle_retryable_errors,omitempty"`
	Request               model.Request `json:"request"`
}

CallModelInput is one plan/observe step. Each call is its own Activity so its completed response is recorded independently in Workflow history.

type CallModelResult

type CallModelResult struct {
	Response             model.Response    `json:"response"`
	MessageEventID       string            `json:"message_event_id,omitempty"`
	ThinkingEventID      string            `json:"thinking_event_id,omitempty"`
	ModelRequestStartID  string            `json:"model_request_start_id,omitempty"`
	ModelRequestEndID    string            `json:"model_request_end_id,omitempty"`
	ToolSteps            []PlannedToolStep `json:"tool_steps,omitempty"`
	FatalError           string            `json:"fatal_error,omitempty"`
	FatalErrorType       string            `json:"fatal_error_type,omitempty"`
	ContextOverflow      bool              `json:"context_overflow,omitempty"`
	ContextOverflowError string            `json:"context_overflow_error,omitempty"`
	RetryError           *ModelRetryError  `json:"retry_error,omitempty"`
}

CallModelResult carries a normalized model response. Provider tool-use IDs remain untouched for exact replay; ToolSteps maps each one to server-owned public/journal IDs. MessageEventID names the public agent.message when the response contains non-empty text.

type ClientConfig

type ClientConfig struct {
	HostPort  string
	Namespace string
}

ClientConfig configures the Temporal client connection.

type CompleteWorkflowTurnInput

type CompleteWorkflowTurnInput struct {
	SessionID      string                 `json:"session_id"`
	ThreadID       string                 `json:"thread_id,omitempty"`
	IsChild        bool                   `json:"is_child,omitempty"`
	TriggerEventID string                 `json:"trigger_event_id"`
	Output         []domain.EventDraft    `json:"output"`
	Status         domain.Status          `json:"status"`
	AttemptID      string                 `json:"attempt_id,omitempty"`
	AttemptState   domain.RunAttemptState `json:"attempt_state,omitempty"`
	AttemptError   *string                `json:"attempt_error,omitempty"`
	// PendingActionEventIDs names action events in Output that park this turn
	// awaiting client input.
	PendingActionEventIDs []string `json:"pending_action_event_ids,omitempty"`
	// ResolutionEventIDs names every client event that closes the current
	// pending-action barrier. PostgreSQL validates the set atomically.
	ResolutionEventIDs []string `json:"resolution_event_ids,omitempty"`
	// TranscriptDelta and ToolUseMappings are provider-private continuation
	// state. PostgreSQL commits them atomically with Output when supported.
	TranscriptDelta       []domain.Message                `json:"transcript_delta,omitempty"`
	ToolUseMappings       []domain.ProviderToolUseMapping `json:"tool_use_mappings,omitempty"`
	Usage                 domain.TokenUsage               `json:"usage,omitempty"`
	UsageAlreadyAccounted bool                            `json:"usage_already_accounted,omitempty"`
}

CompleteWorkflowTurnInput atomically finalizes the optional tool attempt and commits the public turn output in PostgreSQL.

type CoordinatorToolSource

type CoordinatorToolSource interface {
	ExecuteCoordinatorToolStep(
		context.Context, string, string, string, string, string, map[string]any,
	) (domain.ToolStepResult, error)
}

type DeletionStore

type DeletionStore interface {
	ListDeletingSessionIDs(ctx context.Context, limit int) ([]string, error)
	FinalizeSessionMemoryResources(ctx context.Context, sessionID string) error
	FinalizeSessionDeletion(ctx context.Context, sessionID string) error
}

DeletionStore is the worker-side view of fenced Session deletion state. PostgreSQL remains authoritative; the reconciler only resumes the external cleanup and finalization that a crashed API process may have left unfinished.

type EnsureToolPermissionAttemptResult

type EnsureToolPermissionAttemptResult struct {
	Established bool `json:"established"`
}

type EvaluateOutcomeInput

type EvaluateOutcomeInput struct {
	SessionID    string             `json:"session_id"`
	StartEventID string             `json:"start_event_id,omitempty"`
	EndEventID   string             `json:"end_event_id,omitempty"`
	Model        string             `json:"model"`
	Effort       string             `json:"effort,omitempty"`
	Speed        string             `json:"speed,omitempty"`
	Outcome      domain.OutcomeSpec `json:"outcome"`
	Candidate    []domain.Message   `json:"candidate"`
	Iteration    int                `json:"iteration"`
	FinalCycle   bool               `json:"final_cycle"`
}

type EvaluateOutcomeResult

type EvaluateOutcomeResult struct {
	StartEventID string            `json:"start_event_id"`
	EndEventID   string            `json:"end_event_id"`
	Result       string            `json:"result"`
	Explanation  string            `json:"explanation"`
	Usage        domain.TokenUsage `json:"usage"`
	StopReason   string            `json:"stop_reason,omitempty"`
	FatalError   string            `json:"fatal_error,omitempty"`
}

type EvaluateToolPermissionInput

type EvaluateToolPermissionInput struct {
	Owner          domain.ToolPermissionOwner `json:"owner"`
	ToolUseEventID string                     `json:"tool_use_event_id"`
	Model          domain.Model               `json:"model"`
	Input          permission.Input           `json:"input"`
}

type EvaluateToolPermissionResult

type EvaluateToolPermissionResult struct {
	Decision     domain.ToolPermissionDecision `json:"decision"`
	BudgetPaused bool                          `json:"budget_paused"`
	Stopped      bool                          `json:"stopped"`
}

type EventRef

type EventRef struct {
	ID   string `json:"id"`
	Seq  int64  `json:"seq"`
	Type string `json:"type"`
}

EventRef is the minimal projection of a public event the workflow needs to decide what to do. Payloads never enter workflow history; the Activity holds the authoritative event.

type EventSource

type EventSource interface {
	EventsAfter(ctx context.Context, sessionID string, cursor int64, limit int) ([]domain.Event, error)
	FirstUnprocessedInterruptAfter(
		ctx context.Context,
		sessionID string,
		afterSeq int64,
	) (*domain.Event, error)
	HistoryThrough(ctx context.Context, sessionID, triggerEventID string, limit int) ([]domain.Event, error)
	GetSession(ctx context.Context, id string) (domain.Session, error)
	GetEvent(ctx context.Context, sessionID, id string) (domain.Event, error)
	UnresolvedPendingActions(ctx context.Context, sessionID string) ([]domain.PendingAction, error)
	CompleteWorkflowTurn(
		ctx context.Context,
		sessionID string,
		triggerEventID string,
		output []domain.EventDraft,
		status domain.Status,
		attemptID string,
		attemptState domain.RunAttemptState,
		attemptError *string,
		pendingActionEventIDs []string,
		resolutionEventIDs []string,
	) (TurnCompletionResult, error)
}

EventSource is the read side of the PostgreSQL ledger the Activities depend on. The concrete implementation is *pg.Store; the interface keeps the Activities testable with an in-memory fake.

func NewStoreSource

func NewStoreSource(store *pg.Store) EventSource

NewStoreSource wraps a PostgreSQL store as an Activity EventSource.

type ExecuteToolInput

type ExecuteToolInput struct {
	SessionID           string                     `json:"session_id"`
	ThreadID            string                     `json:"thread_id,omitempty"`
	TriggerEventID      string                     `json:"trigger_event_id"`
	AttemptID           string                     `json:"attempt_id"`
	Ordinal             int                        `json:"ordinal"`
	ToolUseEventID      string                     `json:"tool_use_event_id"`
	ToolStepID          string                     `json:"tool_step_id"`
	ToolName            string                     `json:"tool_name"`
	PermissionReceiptID string                     `json:"permission_receipt_id,omitempty"`
	ToolKind            TurnToolKind               `json:"tool_kind,omitempty"`
	MCPServer           domain.MCPServer           `json:"mcp_server,omitempty"`
	MCPToolName         string                     `json:"mcp_tool_name,omitempty"`
	Input               map[string]any             `json:"input"`
	SkillRuntimeRoot    string                     `json:"skill_runtime_root,omitempty"`
	SkillAlreadyLoaded  bool                       `json:"skill_already_loaded,omitempty"`
	AdvisorRequest      model.Request              `json:"advisor_request,omitempty"`
	AdvisorConsultation domain.AdvisorConsultation `json:"advisor_consultation,omitempty"`
}

ExecuteToolInput identifies one logical built-in tool step. ToolUseEventID is stable because it came from the completed CallModel Activity result.

type ExecuteToolResult

type ExecuteToolResult struct {
	Result     domain.ToolStepResult `json:"result"`
	Events     []domain.EventDraft   `json:"events,omitempty"`
	Ambiguous  bool                  `json:"ambiguous"`
	FatalError string                `json:"fatal_error,omitempty"`
}

ExecuteToolResult is the durable result of one tool Activity. Ambiguous is a successful Activity result (not a retryable error): the Workflow must terminate the turn honestly without scheduling the side effect again.

type JournalStore

type JournalStore interface {
	EnsureAttempt(ctx context.Context, sessionID, triggerEventID, attemptID string) error
	EnsureToolStep(ctx context.Context, attemptID, stepID string, ordinal int, toolUseEventID, toolName string, input map[string]any) (domain.ToolStep, error)
	StartToolStep(ctx context.Context, stepID string) error
	CompleteToolStep(ctx context.Context, stepID string, result domain.ToolStepResult) error
	MarkToolStepAmbiguous(ctx context.Context, stepID string) error
}

JournalStore is the durable tool-execution journal used by the granular ExecuteTool Activity. It preserves the prepared/started/completed/ambiguous boundary across Activity retries. *pg.Store implements it.

type LifecycleReconcileResult

type LifecycleReconcileResult struct {
	Deletions int
}

LifecycleReconcileResult reports successfully discharged durable obligations.

type LifecycleReconciler

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

LifecycleReconciler closes the process-crash window between a durable Session deletion fence and Workflow termination/finalization.

func NewLifecycleReconciler

func NewLifecycleReconciler(
	store DeletionStore,
	terminator SessionTerminator,
	cfg LifecycleReconcilerConfig,
) *LifecycleReconciler

func (*LifecycleReconciler) Run

func (*LifecycleReconciler) RunOnce

type LifecycleReconcilerConfig

type LifecycleReconcilerConfig struct {
	// PollInterval is the idle delay between scans. A scan that completes work
	// immediately repeats so a backlog drains without one interval per batch.
	PollInterval time.Duration
	// BatchSize bounds deleting-session scans.
	BatchSize int
	// AttemptTimeout prevents one unavailable provider from starving the rest of
	// the batch. The deterministic cleanup Workflow continues after this local
	// wait expires and a later scan joins it.
	AttemptTimeout time.Duration
}

type LoadEventsInput

type LoadEventsInput struct {
	SessionID string `json:"session_id"`
	ThreadID  string `json:"thread_id,omitempty"`
	Cursor    int64  `json:"cursor"`
	Limit     int    `json:"limit"`
}

LoadEventsInput requests the ordered public events after a cursor.

type LoadEventsResult

type LoadEventsResult struct {
	Events []EventRef `json:"events"`
}

LoadEventsResult carries the ordered event references after the cursor.

type LoadInterruptInput

type LoadInterruptInput struct {
	SessionID string `json:"session_id"`
	ThreadID  string `json:"thread_id,omitempty"`
	AfterSeq  int64  `json:"after_seq"`
}

LoadInterruptInput asks the Activity read side for the first still-unprocessed interrupt after a turn's trigger sequence. Wakeup Signals remain metadata only; PostgreSQL decides whether an interrupt was durably admitted.

type LoadInterruptResult

type LoadInterruptResult struct {
	Interrupt *EventRef `json:"interrupt,omitempty"`
}

type LoadPendingActionsInput

type LoadPendingActionsInput struct {
	SessionID string `json:"session_id"`
	ThreadID  string `json:"thread_id,omitempty"`
}

LoadPendingActionsInput asks PostgreSQL for the current durable requires_action barrier. The result, rather than mutable database state read directly by Workflow code, drives the Workflow selector.

type LoadPendingActionsResult

type LoadPendingActionsResult struct {
	Actions []PendingActionRef `json:"actions,omitempty"`
}

type MCPDiscoveryStore

type MCPDiscoveryStore interface {
	GetMCPDiscoverySnapshot(
		ctx context.Context,
		sessionID string,
		threadID string,
		server domain.MCPServer,
	) ([]mcpclient.Tool, bool, error)
	PutMCPDiscoverySnapshot(
		ctx context.Context,
		sessionID string,
		threadID string,
		server domain.MCPServer,
		tools []mcpclient.Tool,
	) ([]mcpclient.Tool, error)
}

MCPDiscoveryStore pins the discovered tool surface for each Thread/server. Threads share a Session sandbox, but their Agent configurations and remote tool surfaces remain private. The first PrepareTurn discovers remotely; later turns in that Thread reuse the durable snapshot.

type ModelRequestAdmissionSource

type ModelRequestAdmissionSource interface {
	AdmitModelRequest(context.Context, string, string) (bool, error)
}

type ModelRequestUsageSource

type ModelRequestUsageSource interface {
	AccountModelRequest(
		context.Context,
		string,
		string,
		string,
		domain.Model,
		domain.TokenUsage,
		string,
	) error
}

type ModelRetryError

type ModelRetryError struct {
	Type             string `json:"type"`
	Message          string `json:"message"`
	RetryAfterMillis int64  `json:"retry_after_millis,omitempty"`
}

ModelRetryError is a provider failure that may succeed without changing the logical request. It is an Activity result, not an Activity error, so the Workflow can publish the documented retry lifecycle deterministically.

type Orchestrator

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

Orchestrator bridges PostgreSQL admission to Temporal. It makes a best-effort post-commit fast-path signal to reduce latency and relies on the durable outbox relay for eventual, correct delivery.

func NewOrchestrator

func NewOrchestrator(store *pg.Store, signaler *Signaler) *Orchestrator

func (*Orchestrator) Admit

func (o *Orchestrator) Admit(ctx context.Context, sessionID string, drafts []domain.EventDraft) ([]domain.Event, error)

Admit admits a client event batch into PostgreSQL and returns the committed public events. The admission transaction has already written the coalescible outbox wakeup, so the returned events are durable and will be processed even if the fast-path signal below fails. The fast-path signal is best-effort: a failure is logged and left to the relay, because correctness depends on the outbox, not on this call.

func (*Orchestrator) CreateAPISession

func (o *Orchestrator) CreateAPISession(
	ctx context.Context,
	session domain.Session,
	initial []domain.EventDraft,
	resourceSets ...[]app.PreparedSessionResource,
) (domain.Session, []domain.Event, error)

CreateAPISession is the HTTP control-plane variant. In addition to the admission/outbox transaction it requires the exact Agent version and Environment to remain active through the session insert.

func (*Orchestrator) CreateDeploymentSession

func (o *Orchestrator) CreateDeploymentSession(
	ctx context.Context,
	session domain.Session,
	initial []domain.EventDraft,
	run domain.DeploymentRun,
	resourceSets ...[]app.PreparedSessionResource,
) (domain.Session, []domain.Event, error)

CreateDeploymentSession commits the Session and its immutable Deployment Run success record together. A retry can therefore never expose a successful Session without the audit row that names it.

func (*Orchestrator) CreateSession

func (o *Orchestrator) CreateSession(ctx context.Context, session domain.Session, initial []domain.EventDraft) (domain.Session, []domain.Event, error)

CreateSession creates a session and admits its initial events, then fast-path signals as above.

func (*Orchestrator) TerminateSession

func (o *Orchestrator) TerminateSession(ctx context.Context, sessionID string) error

type OutboxStore

type OutboxStore interface {
	ListWakeupsForDelivery(ctx context.Context, limit int) ([]pg.OutboxWakeup, error)
	DeleteWakeupIfUnchanged(ctx context.Context, sessionID string, maxSeq int64) (bool, error)
	RecordAttempt(ctx context.Context, sessionID string, cause string) error
}

OutboxStore is the relay's view of the PostgreSQL outbox. *pg.Store implements it; the narrow interface keeps the relay testable.

type PendingActionRef

type PendingActionRef struct {
	ActionEventID      string                   `json:"action_event_id"`
	ActionEventSeq     int64                    `json:"action_event_seq"`
	Kind               domain.PendingActionKind `json:"kind"`
	ResolutionEventID  string                   `json:"resolution_event_id,omitempty"`
	ResolutionEventSeq int64                    `json:"resolution_event_seq,omitempty"`
}

PendingActionRef is the minimal deterministic selector projection for one unresolved pending action. Payloads remain behind the Activity boundary.

type PermissionSource

type PermissionSource interface {
	EnsureToolPermissionAttempt(context.Context, domain.ToolPermissionOwner) error
	PermissionIntentThrough(context.Context, string, string) (domain.PermissionIntent, error)
	GetToolPermissionReceipt(context.Context, domain.ToolPermissionReceipt) (*domain.ToolPermissionReceipt, error)
	AdmitToolPermissionEvaluation(context.Context, domain.ToolPermissionReceipt) (bool, error)
	RecordToolPermissionReceipt(context.Context, domain.ToolPermissionReceipt) (domain.ToolPermissionReceipt, error)
	StartToolStepWithPermission(context.Context, string, string) error
}

PermissionSource keeps durable judgment facts separate from current action authority. PostgreSQL implements both checks without holding locks over I/O.

type PlannedToolStep

type PlannedToolStep struct {
	ToolUseEventID    string `json:"tool_use_event_id"`
	ProviderToolUseID string `json:"provider_tool_use_id"`
	ToolStepID        string `json:"tool_step_id"`
}

PlannedToolStep binds one public tool-use event to its internal journal step. Both ids are Activity output recorded in Workflow history; retries therefore reuse explicit operation ids rather than deriving one namespace from another.

type PrepareTurnInput

type PrepareTurnInput struct {
	SessionID          string   `json:"session_id"`
	TriggerEventID     string   `json:"trigger_event_id"`
	ResolutionEventIDs []string `json:"resolution_event_ids,omitempty"`
}

PrepareTurnInput identifies one public trigger whose model turn should run.

type PrepareTurnResult

type PrepareTurnResult struct {
	AlreadyCompleted bool           `json:"already_completed"`
	Terminated       bool           `json:"terminated"`
	FatalError       string         `json:"fatal_error,omitempty"`
	AttemptID        string         `json:"attempt_id,omitempty"`
	ThreadID         string         `json:"thread_id,omitempty"`
	IsChild          bool           `json:"is_child,omitempty"`
	SkillRuntimeRoot string         `json:"skill_runtime_root,omitempty"`
	Request          model.Request  `json:"request"`
	Tools            []TurnTool     `json:"tools,omitempty"`
	ResumeActions    []ResumeAction `json:"resume_actions,omitempty"`
	// PreludeEvents are recoverable setup diagnostics, such as one unavailable
	// MCP server. The Workflow commits them with the turn while continuing with
	// the remaining tool surface.
	PermissionIntent domain.PermissionIntent `json:"permission_intent"`
	PreludeEvents    []domain.EventDraft     `json:"prelude_events,omitempty"`
	// UsesProviderTranscript is true when Request.Messages came from the
	// lossless private transcript rather than the legacy public-event
	// projection. TranscriptDelta contains only the new input represented by
	// this turn; Workflow code appends provider responses and tool results.
	UsesProviderTranscript bool                     `json:"uses_provider_transcript,omitempty"`
	TranscriptDelta        []domain.Message         `json:"transcript_delta,omitempty"`
	Outcome                *domain.OutcomeSpec      `json:"outcome,omitempty"`
	ContextProjection      domain.ContextProjection `json:"context_projection"`
	ContextSnapshotID      string                   `json:"context_snapshot_id,omitempty"`
}

PrepareTurnResult is the immutable starting state for a Workflow-owned turn. The projected messages and tool definitions are Activity output, so Temporal records them in history and deterministic replay never rereads PostgreSQL.

type PreviewPublisher

type PreviewPublisher interface {
	PublishPreview(context.Context, string, domain.PreviewFrame) error
}

PreviewPublisher carries best-effort model deltas to live subscribers. It is never part of turn correctness and may be nil.

type ProviderTranscriptCompletionSource

type ProviderTranscriptCompletionSource interface {
	CompleteWorkflowTurnWithTranscript(
		ctx context.Context,
		sessionID string,
		triggerEventID string,
		output []domain.EventDraft,
		status domain.Status,
		attemptID string,
		attemptState domain.RunAttemptState,
		attemptError *string,
		pendingActionEventIDs []string,
		resolutionEventIDs []string,
		transcriptDelta []domain.Message,
		toolUseMappings []domain.ProviderToolUseMapping,
	) (TurnCompletionResult, error)
}

ProviderTranscriptCompletionSource atomically commits the public turn and the private provider transcript delta.

type ProviderTranscriptSource

type ProviderTranscriptSource interface {
	LoadProviderTranscript(
		ctx context.Context,
		sessionID string,
	) (domain.ProviderTranscript, error)
}

ProviderTranscriptSource is the optional private-context capability supplied by the PostgreSQL adapter. Tests and legacy stores that do not implement it continue to use the public-event projection.

type ProviderTranscriptUsageCompletionSource

type ProviderTranscriptUsageCompletionSource interface {
	CompleteWorkflowTurnWithTranscriptAndUsage(
		ctx context.Context,
		sessionID string,
		triggerEventID string,
		output []domain.EventDraft,
		status domain.Status,
		attemptID string,
		attemptState domain.RunAttemptState,
		attemptError *string,
		pendingActionEventIDs []string,
		resolutionEventIDs []string,
		transcriptDelta []domain.Message,
		toolUseMappings []domain.ProviderToolUseMapping,
		usage domain.TokenUsage,
	) (TurnCompletionResult, error)
}

ProviderTranscriptUsageCompletionSource is the full atomic completion capability used by the PostgreSQL adapter.

type RecordModelRetryInput

type RecordModelRetryInput struct {
	SessionID      string          `json:"session_id"`
	ThreadID       string          `json:"thread_id,omitempty"`
	IsChild        bool            `json:"is_child,omitempty"`
	TriggerEventID string          `json:"trigger_event_id"`
	ErrorEventID   string          `json:"error_event_id"`
	StatusEventID  string          `json:"status_event_id"`
	Error          ModelRetryError `json:"error"`
}

type Relay

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

Relay drains the PostgreSQL orchestration outbox and delivers each wakeup to Temporal with Signal-With-Start, retrying until the service accepts it.

Correctness rests on the outbox, not on any post-commit fast path:

  • A wakeup is deleted only after a successful signal AND only if no later admission raised its sequence in the meantime (DeleteWakeupIfUnchanged).
  • A crash after signaling but before the delete leaves the row; the next cycle re-delivers it, which the workflow treats as a harmless duplicate.
  • A delivery failure (Worker/Temporal unavailable) records the attempt and leaves the row for retry, so admission can outrun execution-plane outages.

func NewRelay

func NewRelay(store OutboxStore, deliverer WakeupDeliverer, cfg RelayConfig) *Relay

func (*Relay) Run

func (r *Relay) Run(ctx context.Context) error

Run drives the relay until ctx is canceled.

func (*Relay) RunOnce

func (r *Relay) RunOnce(ctx context.Context) (int, error)

RunOnce reads a batch of pending wakeups, delivers each, and reconciles the outbox. It returns the number of wakeups successfully delivered-and-removed. It is the unit the tests drive directly to exercise crash boundaries. Delivery is at-least-once: a concurrent relay may read and deliver the same wakeup, and the SessionWorkflow deduplicates by receipt sequence.

type RelayConfig

type RelayConfig struct {
	// PollInterval is how often the relay scans the outbox when the last scan
	// found no pending wakeups.
	PollInterval time.Duration
	// BatchSize bounds how many wakeups one scan reads.
	BatchSize int
}

RelayConfig tunes the relay loop.

type ResumeAction

type ResumeAction struct {
	ActionEventID string `json:"action_event_id"`
	// ActionEventType is the public event type PostgreSQL durably holds for the
	// parked call. The result that answers this action must be the matching
	// documented variant, and that decision cannot be made from Workflow code
	// alone: a barrier can outlive both a worker upgrade and a Continue-As-New,
	// so the answering execution may run newer code than the one that wrote the
	// call. An empty value predates this field and means the legacy
	// agent.tool_use spelling.
	ActionEventType     string                   `json:"action_event_type,omitempty"`
	EvaluatedPermission string                   `json:"evaluated_permission,omitempty"`
	Kind                domain.PendingActionKind `json:"kind"`
	ToolName            string                   `json:"tool_name"`
	Input               map[string]any           `json:"input"`
	ResolutionEventID   string                   `json:"resolution_event_id"`
	Content             []any                    `json:"content,omitempty"`
	IsError             bool                     `json:"is_error,omitempty"`
	Confirmation        string                   `json:"confirmation,omitempty"`
	DenyMessage         string                   `json:"deny_message,omitempty"`
	ToolStepID          string                   `json:"tool_step_id,omitempty"`
	ProviderToolUseID   string                   `json:"provider_tool_use_id,omitempty"`
}

ResumeAction is the server-owned reconstruction of one parked tool call and its admitted client result. The Activity validates and normalizes the raw event payloads before they enter Workflow history. Confirmations also carry a stable journal step id for an allowed built-in execution.

type ResumeModelRetryInput

type ResumeModelRetryInput struct {
	SessionID      string `json:"session_id"`
	ThreadID       string `json:"thread_id,omitempty"`
	IsChild        bool   `json:"is_child,omitempty"`
	TriggerEventID string `json:"trigger_event_id"`
	StatusEventID  string `json:"status_event_id"`
}

type RunTurnResult

type RunTurnResult struct {
	Disposition TurnDisposition `json:"disposition"`
}

RunTurnResult reports whether the workflow-owned turn completed, parked on a client-action barrier, or terminated the session.

type Runtime

type Runtime struct {
	Client    client.Client
	Worker    worker.Worker
	Relay     *Relay
	Lifecycle *LifecycleReconciler
	Store     *pg.Store
	Signal    *Signaler
}

Runtime bundles the worker and relay so cmd can run the execution plane with a single call.

func NewRuntime

func NewRuntime(config RuntimeConfig) *Runtime

NewRuntime wires the full Temporal execution plane. The store is both the event source and durable tool-execution journal. Self-hosted workers own their execution environments through the Environment Work protocol.

func (*Runtime) Orchestrator

func (r *Runtime) Orchestrator() *Orchestrator

Orchestrator returns an admission orchestrator sharing this runtime's store and signaler (post-commit fast path).

type RuntimeConfig

type RuntimeConfig struct {
	TemporalClient client.Client
	Store          *pg.Store
	ModelClient    model.Client
	// PermissionEvaluator defaults to the configured model adapter.
	PermissionEvaluator permission.Evaluator
	IDGenerator         domain.IDGenerator
	RelayConfig         RelayConfig
	TaskQueue           string
	SkillInstructions   SkillInstructionLoader
	ToolOutputs         ToolOutputStore
	MCPAuth             credentialruntime.AuthSource
	PreviewPublisher    PreviewPublisher
}

RuntimeConfig declares the execution-plane dependencies and optional capabilities assembled by NewRuntime. TaskQueue defaults to TaskQueue when empty; the remaining optional interfaces may be nil when their capability is disabled.

type SessionSkillSource

type SessionSkillSource interface {
	SessionSkillsForRuntime(context.Context, string) ([]domain.SkillVersion, error)
}

type SessionTerminator

type SessionTerminator interface {
	TerminateSession(ctx context.Context, sessionID string) error
}

type SessionThreadSkillSource

type SessionThreadSkillSource interface {
	SessionThreadSkillRuntime(
		context.Context,
		string,
		string,
	) (domain.SkillRuntime, error)
}

type SessionThreadSource

type SessionThreadSource interface {
	GetSessionThread(context.Context, string, string) (domain.SessionThread, error)
}

type SessionThreadWorkflowInput

type SessionThreadWorkflowInput struct {
	SessionID   string `json:"session_id"`
	ThreadID    string `json:"thread_id"`
	StartCursor int64  `json:"start_cursor"`
}

type SessionWorkflowInput

type SessionWorkflowInput struct {
	SessionID   string `json:"session_id"`
	StartCursor int64  `json:"start_cursor"`
}

SessionWorkflowInput starts (or restarts, via Continue-As-New) a SessionWorkflow. StartCursor is the durable last-observed event sequence; a fresh session starts at 0, and Continue-As-New carries the current cursor forward so a new history run does not reprocess consumed events.

type Signaler

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

Signaler delivers a wakeup to a SessionWorkflow using Signal-With-Start, so a wakeup for a session with no live workflow starts one idempotently (Workflow ID = session ID) and a wakeup for a running one simply signals it.

This is the single delivery primitive used by both the API's post-commit fast path and the outbox relay. Duplicate delivery is harmless: the workflow reads events by PostgreSQL receipt sequence after its durable cursor and ignores anything already observed.

func NewSignaler

func NewSignaler(c client.Client) *Signaler

func NewSignalerOnTaskQueue

func NewSignalerOnTaskQueue(c client.Client, taskQueue string) *Signaler

func (*Signaler) TerminateSession

func (s *Signaler) TerminateSession(ctx context.Context, sessionID string) error

TerminateSession stops the live Workflow execution for a session before its PostgreSQL projection is physically deleted. A session that never started a Workflow is already stopped, so Temporal NotFound is idempotent success.

func (*Signaler) TerminateThread

func (s *Signaler) TerminateThread(
	ctx context.Context,
	_ string,
	threadID string,
) error

TerminateThread stops one child Workflow after PostgreSQL has durably ended the owning Thread. Temporal NotFound is idempotent success: an idle child may never have started, or a concurrent relay may already have stopped it.

func (*Signaler) Wake

func (s *Signaler) Wake(ctx context.Context, sessionID string, maxEventSeq int64) error

Wake starts-or-signals the SessionWorkflow for a session, delivering wakeup metadata (the highest known receipt sequence) only. The workflow's start cursor is 0 on first start; the durable cursor then advances as it consumes PostgreSQL.

func (*Signaler) WakeThread

func (s *Signaler) WakeThread(
	ctx context.Context,
	sessionID string,
	threadID string,
	maxEventSeq int64,
) error

WakeThread starts-or-signals the independent Workflow for a child Session Thread. The stable Workflow ID makes relay redelivery and operation retries harmless; authoritative work is still loaded from PostgreSQL by sequence.

type SkillInstructionLoader

type SkillInstructionLoader interface {
	LoadSkillInstructions(context.Context, domain.SkillVersion) ([]byte, error)
}

SkillInstructionLoader reads the immutable instruction entry owned by the control plane. The Agent loop uses the same canonical source for every execution environment; self-hosted workers independently receive the pinned bundle for supporting-file access, without reverse filesystem access.

type StartModelRequestInput

type StartModelRequestInput struct {
	SessionID           string `json:"session_id"`
	TriggerEventID      string `json:"trigger_event_id"`
	ModelRequestStartID string `json:"model_request_start_id"`
}

StartModelRequestInput identifies one logical provider request span. The Workflow derives the ID from Session, trigger, and model-round ordinal.

type ThreadCompletionSource

type ThreadCompletionSource interface {
	CompleteThreadWorkflowTurn(
		ctx context.Context,
		sessionID string,
		threadID string,
		triggerEventID string,
		output []domain.EventDraft,
		status domain.Status,
		attemptID string,
		attemptState domain.RunAttemptState,
		attemptError *string,
		pendingActionEventIDs []string,
		resolutionEventIDs []string,
		transcriptDelta []domain.Message,
		toolUseMappings []domain.ProviderToolUseMapping,
		usage domain.TokenUsage,
	) (TurnCompletionResult, error)
}

type ThreadContextSnapshotSource

type ThreadContextSnapshotSource interface {
	GetThreadContextSnapshotForTrigger(
		ctx context.Context,
		sessionID string,
		threadID string,
		triggerEventID string,
	) (domain.ContextSnapshot, bool, error)
	PutThreadContextSnapshot(
		ctx context.Context,
		sessionID string,
		threadID string,
		triggerEventID string,
		transcriptTriggerEventIDs []string,
		messages []domain.Message,
		projection domain.ContextProjection,
	) (domain.ContextSnapshot, error)
}

ThreadContextSnapshotSource persists an immutable compacted projection for one Thread trigger. Production PostgreSQL implements it; lightweight test sources may omit it and continue exercising projection in memory.

type ThreadEventSource

type ThreadEventSource interface {
	ThreadEventsAfter(
		context.Context, string, string, int64, int,
	) ([]domain.Event, error)
}

type ThreadInterruptSource

type ThreadInterruptSource interface {
	FirstUnprocessedThreadInterruptAfter(
		ctx context.Context,
		sessionID string,
		threadID string,
		afterSeq int64,
	) (*domain.Event, error)
}

type ThreadOutboxStore

type ThreadOutboxStore interface {
	DeleteThreadWakeupIfUnchanged(
		ctx context.Context, sessionID string, threadID string, maxSeq int64,
		intent pg.OrchestrationIntent,
	) (bool, error)
	RecordThreadAttempt(
		ctx context.Context, sessionID string, threadID string, cause string,
	) error
}

type ThreadPendingActionSource

type ThreadPendingActionSource interface {
	UnresolvedThreadPendingActions(
		context.Context, string, string,
	) ([]domain.PendingAction, error)
}

type ThreadProviderTranscriptSource

type ThreadProviderTranscriptSource interface {
	LoadThreadProviderTranscript(
		ctx context.Context,
		sessionID string,
		threadID string,
	) (domain.ProviderTranscript, error)
}

type ThreadTerminationDeliverer

type ThreadTerminationDeliverer interface {
	TerminateThread(ctx context.Context, sessionID string, threadID string) error
}

type ThreadWakeupDeliverer

type ThreadWakeupDeliverer interface {
	WakeThread(
		ctx context.Context, sessionID string, threadID string, maxEventSeq int64,
	) error
}

type ThreadWorkflowRetrySource

type ThreadWorkflowRetrySource interface {
	RecordThreadWorkflowRetry(
		context.Context, string, string, string, string, string, map[string]any,
	) error
	ResumeThreadWorkflowRetry(
		context.Context, string, string, string, string,
	) error
}

type ToolOutputStore

type ToolOutputStore interface {
	StoreToolOutput(context.Context, string, string, string) error
}

ToolOutputStore publishes an immutable full textual result under a stable File ID.

type TurnCompletionResult

type TurnCompletionResult struct {
	Events  []domain.Event
	Applied bool
	Status  domain.Status
	// Parked is a pointer for Workflow-history compatibility. Activity results
	// recorded before durable interrupt support have no field; nil tells
	// CompleteWorkflowTurn to use the legacy input-derived disposition. New
	// results always carry an explicit true/false from PostgreSQL.
	Parked *bool
}

TurnCompletionResult mirrors pg.TurnCompletion without importing the pg package into the workflow-facing types, keeping the domain boundary intact. Status is the session's projected status after the completion committed, so the Activity can tell the workflow whether the session is now terminated.

type TurnDisposition

type TurnDisposition string

TurnDisposition tells the SessionWorkflow whether a completed turn finished, parked on a newly committed client-action barrier, or terminated the session.

const (
	TurnCompleted  TurnDisposition = "completed"
	TurnParked     TurnDisposition = "parked"
	TurnTerminated TurnDisposition = "terminated"
)

type TurnTool

type TurnTool struct {
	Name        string                  `json:"name"`
	Kind        TurnToolKind            `json:"kind"`
	Permission  domain.PermissionPolicy `json:"permission"`
	MCPServer   domain.MCPServer        `json:"mcp_server,omitempty"`
	MCPToolName string                  `json:"mcp_tool_name,omitempty"`
	Model       string                  `json:"model,omitempty"`
}

TurnTool is the immutable, Workflow-facing classification of an offered tool.

type TurnToolKind

type TurnToolKind string

TurnToolKind is the execution owner recorded by PrepareTurn. The Workflow consumes this durable Activity result rather than consulting a mutable process-global registry while replaying.

const (
	TurnToolBuiltin      TurnToolKind = "builtin"
	TurnToolCustom       TurnToolKind = "custom"
	TurnToolMCP          TurnToolKind = "mcp"
	TurnToolSelfHosted   TurnToolKind = "self_hosted"
	TurnToolRuntimeSkill TurnToolKind = "runtime_skill"
	TurnToolCoordinator  TurnToolKind = "coordinator"
	TurnToolAdvisor      TurnToolKind = "advisor"
)

type UsageCompletionSource

type UsageCompletionSource interface {
	CompleteWorkflowTurnWithUsage(
		ctx context.Context,
		sessionID string,
		triggerEventID string,
		output []domain.EventDraft,
		status domain.Status,
		attemptID string,
		attemptState domain.RunAttemptState,
		attemptError *string,
		pendingActionEventIDs []string,
		resolutionEventIDs []string,
		usage domain.TokenUsage,
	) (TurnCompletionResult, error)
}

UsageCompletionSource is implemented by stores that can atomically account model usage while completing a public turn. It is optional so lightweight EventSource implementations remain source-compatible.

type WakeupDeliverer

type WakeupDeliverer interface {
	Wake(ctx context.Context, sessionID string, maxEventSeq int64) error
}

WakeupDeliverer delivers one wakeup to Temporal. *Signaler implements it; a fake implements it in tests to simulate a temporarily-unavailable Worker or Temporal service.

type WakeupSignal

type WakeupSignal struct {
	MaxEventSeq int64 `json:"max_event_seq"`
}

WakeupSignal is the wakeup metadata delivered to a SessionWorkflow. It carries only the highest known public receipt sequence, never event payloads. The workflow loads authoritative events from PostgreSQL after its own durable cursor and ignores anything at or below it, so a duplicate or out-of-order signal cannot reorder public events or double-process a turn.

type WorkflowEventSource

type WorkflowEventSource interface {
	AppendWorkflowEvents(
		ctx context.Context,
		sessionID string,
		triggerEventID string,
		drafts []domain.EventDraft,
	) error
}

WorkflowEventSource appends already-completed public progress without processing the turn trigger or changing the Session projection. PostgreSQL implements it idempotently by explicit event ID; the Workflow uses it to make model-span starts and completed intermediate rounds visible in order.

type WorkflowRetrySource

type WorkflowRetrySource interface {
	RecordWorkflowRetry(
		ctx context.Context,
		sessionID string,
		triggerEventID string,
		errorEventID string,
		statusEventID string,
		errorPayload map[string]any,
	) error
	ResumeWorkflowRetry(
		ctx context.Context,
		sessionID string,
		triggerEventID string,
		statusEventID string,
	) error
}

WorkflowRetrySource atomically publishes retry status events with the corresponding Session projection transition.

Jump to

Keyboard shortcuts

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