registry

package
v1.18.2 Latest Latest
Warning

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

 Go to latest Published: Oct 13, 2026 License: MIT Imports: 68 Imported by: 0

Documentation

Index

Constants

View Source 
const MaxLookupCacheSize = 10000

maxLookupCacheSize bounds the number of entries kept in the in-memory lookup cache so it cannot grow without limit and leak memory.

Variables

This section is empty.

Functions

This section is empty.

Types

type DatabaseLogger 

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

func NewDatabaseLogger 

func NewDatabaseLogger(parentCtx context.Context, s LogCreator, workflowID string, fallback hermod.Logger) *DatabaseLogger

func (*DatabaseLogger) Close 

func (l *DatabaseLogger) Close()

Close stops the background flusher and drains what is buffered. It is safe to call more than once.

func (*DatabaseLogger) Debug 

func (l *DatabaseLogger) Debug(msg string, keysAndValues ...any)

func (*DatabaseLogger) DebugEnabled added in v1.8.1 

func (l *DatabaseLogger) DebugEnabled() bool

DebugEnabled reports whether Debug output is kept.

It exists so a caller can skip building an argument that is about to be discarded. The engine's per-write line measures the message's payload, which for a data-map message means marshalling it to JSON; a logger that cannot answer this is assumed to want the line, so without it that marshal ran for every message written.

func (*DatabaseLogger) Error 

func (l *DatabaseLogger) Error(msg string, keysAndValues ...any)

func (*DatabaseLogger) Flush 

func (l *DatabaseLogger) Flush()

func (*DatabaseLogger) Info 

func (l *DatabaseLogger) Info(msg string, keysAndValues ...any)

func (*DatabaseLogger) UpdateStorage 

func (l *DatabaseLogger) UpdateStorage(s LogCreator)

func (*DatabaseLogger) Warn 

func (l *DatabaseLogger) Warn(msg string, keysAndValues ...any)

type DebuggerEvent 

type DebuggerEvent struct {
	WorkflowID string         `json:"workflow_id"`
	NodeID     string         `json:"node_id"`
	MsgID      string         `json:"msg_id"`
	Data       map[string]any `json:"data"`
	State      string         `json:"state"` // "paused", "resumed", "aborted"
}

type LiveMessage 

type LiveMessage struct {
	WorkflowID string         `json:"workflow_id"`
	NodeID     string         `json:"node_id"`
	Timestamp  time.Time      `json:"timestamp"`
	Data       map[string]any `json:"data"`
	IsError    bool           `json:"is_error"`
	Error      string         `json:"error,omitempty"`
}

type LogCreator 

type LogCreator interface {
	CreateLog(ctx context.Context, log storage.Log) error
	CreateLogs(ctx context.Context, logs []storage.Log) error
}

type PIIStats 

type PIIStats struct {
	Discoveries map[string]uint64 `json:"discoveries"`
	LastUpdated time.Time         `json:"last_updated"`
}

type Registry 

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

func NewRegistry 

func NewRegistry(s storage.Storage, ls ...storage.Storage) *Registry

func (*Registry) AIBudget added in v1.18.0 

func (r *Registry) AIBudget() *aibudget.Service

AIBudget is the vhosts' AI spending policy over whatever storage the registry holds right now: its store when it has a database, the control plane's API when it is a worker. It is built once, since it caches each vhost's budget, and cmd/hermod installs it with genai.SetBudget so that every AI node and the agent are held to it.

func (*Registry) ApplyTransformation 

func (r *Registry) ApplyTransformation(ctx context.Context, msg hermod.Message, transType string, config map[string]any) (hermod.Message, error)

func (*Registry) BroadcastLiveMessage 

func (r *Registry) BroadcastLiveMessage(workflowID, nodeID string, msg hermod.Message, isError bool, errMsg string)

func (*Registry) BroadcastLog 

func (r *Registry) BroadcastLog(engineID, level, msg, data string)

func (*Registry) BroadcastStatus 

func (r *Registry) BroadcastStatus(update telemetry.StatusUpdate)

func (*Registry) BrowseSinkTable 

func (r *Registry) BrowseSinkTable(ctx context.Context, cfg factory.SinkConfig, table string, limit int) ([]hermod.Message, error)

func (*Registry) CanTransform added in v1.8.1 

func (r *Registry) CanTransform(transType string) bool

CanTransform reports whether transType names a registered transformer.

Some names are both: "validator" is a transformer and a node executor, and the transformer is the one a preview wants. Callers deciding how to preview a node need to ask this first, or a node that transforms perfectly well gets turned away for being a node.

func (*Registry) Close 

func (r *Registry) Close()

func (*Registry) ContextWithPipelineSnapshot 

func (r *Registry) ContextWithPipelineSnapshot(ctx context.Context) context.Context

ContextWithPipelineSnapshot returns a new context with a shared snapshot pointer for optimizing transformation pipelines by reducing redundant ToMap() calls.

func (*Registry) CreateLog 

func (r *Registry) CreateLog(ctx context.Context, l storage.Log) error

func (*Registry) CreateLogs 

func (r *Registry) CreateLogs(ctx context.Context, logs []storage.Log) error

func (*Registry) CreateSink 

func (r *Registry) CreateSink(ctx context.Context, cfg factory.SinkConfig) (hermod.Sink, error)

func (*Registry) CreateSource 

func (r *Registry) CreateSource(ctx context.Context, cfg factory.SourceConfig) (hermod.Source, error)

func (*Registry) DebuggerCommand 

func (r *Registry) DebuggerCommand(workflowID, msgID, action string)

func (*Registry) DeleteLogs 

func (r *Registry) DeleteLogs(ctx context.Context, filter storage.LogFilter) error

func (*Registry) DeleteWorkflowTraces added in v1.10.0 

func (r *Registry) DeleteWorkflowTraces(ctx context.Context, workflowID string)

DeleteWorkflowTraces removes the traces of a workflow that has been deleted.

Here rather than in storage.DeleteWorkflow because traces live in the log store, and SetLogStorage may point that at a different database from the one holding workflows — a cascade inside the workflow store would delete from the wrong place and silently leave the real rows behind. The Registry is the object that knows both.

Best effort by design: the hourly sweep reclaims anything this misses, so a failure here costs space until the next tick rather than failing the delete the operator actually asked for.

func (*Registry) DiscoverDatabases 

func (r *Registry) DiscoverDatabases(ctx context.Context, cfg factory.SourceConfig) ([]string, error)

func (*Registry) DiscoverPublications 

func (r *Registry) DiscoverPublications(ctx context.Context, cfg factory.SourceConfig) ([]hermod.PublicationInfo, error)

func (*Registry) DiscoverReplicationSlots 

func (r *Registry) DiscoverReplicationSlots(ctx context.Context, cfg factory.SourceConfig) ([]hermod.ReplicationSlotInfo, error)

func (*Registry) DiscoverSinkColumns 

func (r *Registry) DiscoverSinkColumns(ctx context.Context, cfg factory.SinkConfig, table string) ([]hermod.ColumnInfo, error)

func (*Registry) DiscoverSinkDatabases 

func (r *Registry) DiscoverSinkDatabases(ctx context.Context, cfg factory.SinkConfig) ([]string, error)

func (*Registry) DiscoverSinkTables 

func (r *Registry) DiscoverSinkTables(ctx context.Context, cfg factory.SinkConfig) ([]string, error)

func (*Registry) DiscoverSourceColumns 

func (r *Registry) DiscoverSourceColumns(ctx context.Context, cfg factory.SourceConfig, table string) ([]hermod.ColumnInfo, error)

func (*Registry) DiscoverTables 

func (r *Registry) DiscoverTables(ctx context.Context, cfg factory.SourceConfig) ([]string, error)

func (*Registry) DrainWorkflowDLQ 

func (r *Registry) DrainWorkflowDLQ(ctx context.Context, id string) error

func (*Registry) EvaluateConditions 

func (r *Registry) EvaluateConditions(msg hermod.Message, conditions []map[string]any) bool

func (*Registry) ExecSinkStatement 

func (r *Registry) ExecSinkStatement(ctx context.Context, cfg factory.SinkConfig, stmt string) error

func (*Registry) ExecuteSQL 

func (r *Registry) ExecuteSQL(ctx context.Context, cfg factory.SourceConfig, query string, userSample map[string]any) ([]map[string]any, error)

func (*Registry) ExecuteSinkSQL 

func (r *Registry) ExecuteSinkSQL(ctx context.Context, cfg factory.SinkConfig, query string, userSample map[string]any) ([]map[string]any, error)

func (*Registry) GetAllStatuses 

func (r *Registry) GetAllStatuses() []telemetry.StatusUpdate

func (*Registry) GetDB 

func (r *Registry) GetDB(ctx context.Context, typeName string, config map[string]string) (*sql.DB, error)

func (*Registry) GetDQScorer 

func (r *Registry) GetDQScorer() *governance.Scorer

func (*Registry) GetDashboardHistory added in v1.8.1 

func (r *Registry) GetDashboardHistory(ctx context.Context, vhost string, since time.Time, limit int) ([]storage.DashboardSample, error)

GetDashboardHistory reads back the persisted series for a vhost.

func (*Registry) GetDashboardStats 

func (r *Registry) GetDashboardStats(ctx context.Context, vhost string) (storage.DashboardStats, error)

func (*Registry) GetEngine 

func (r *Registry) GetEngine(id string) (*pkgengine.Engine, bool)

func (*Registry) GetLogStorage 

func (r *Registry) GetLogStorage() storage.Storage

func (*Registry) GetLogger 

func (r *Registry) GetLogger() hermod.Logger

func (*Registry) GetLookupCache 

func (r *Registry) GetLookupCache(key string) (any, bool)

func (*Registry) GetLookupCacheSize 

func (r *Registry) GetLookupCacheSize() (int, int)

func (*Registry) GetMeshManager 

func (r *Registry) GetMeshManager() *mesh.Manager

func (*Registry) GetNodeState 

func (r *Registry) GetNodeState(key string) (any, bool)

func (*Registry) GetNodeStates 

func (r *Registry) GetNodeStates(ctx context.Context, workflowID string) (map[string]any, error)

func (*Registry) GetOrOpenDB 

func (r *Registry) GetOrOpenDB(src storage.Source) (*sql.DB, error)

func (*Registry) GetOrOpenDBByID 

func (r *Registry) GetOrOpenDBByID(ctx context.Context, id string) (*sql.DB, string, error)

func (*Registry) GetPIIStats 

func (r *Registry) GetPIIStats() map[string]*PIIStats

func (*Registry) GetSink 

func (r *Registry) GetSink(workflowID, nodeID string) (hermod.Sink, bool)

func (*Registry) GetSinkConfig 

func (r *Registry) GetSinkConfig(ctx context.Context, id string) (storage.Sink, error)

func (*Registry) GetSinkConfigs 

func (r *Registry) GetSinkConfigs(id string) ([]factory.SinkConfig, bool)

func (*Registry) GetSource 

func (r *Registry) GetSource(ctx context.Context, id string) (storage.Source, error)

func (*Registry) GetSourceConfig 

func (r *Registry) GetSourceConfig(ctx context.Context, id string) (storage.Source, error)

func (*Registry) GetSourceConfigs 

func (r *Registry) GetSourceConfigs(id string) ([]factory.SourceConfig, bool)

func (*Registry) GetSourceFactoryConfig 

func (r *Registry) GetSourceFactoryConfig(ctx context.Context, id string) (factory.SourceConfig, error)

func (*Registry) GetSourceFormSamples 

func (r *Registry) GetSourceFormSamples(ctx context.Context, path string, limit int) ([]hermod.Message, error)

func (*Registry) GetStorage 

func (r *Registry) GetStorage() storage.Storage

func (*Registry) GetWorkflowConfig 

func (r *Registry) GetWorkflowConfig(id string) (storage.Workflow, bool)

func (*Registry) GetWorkflowHealth 

func (r *Registry) GetWorkflowHealth(id string) (storage.WorkflowHealth, bool)

GetWorkflowHealth returns a real-time health summary for a running workflow.

func (*Registry) GetWorkflowStatus 

func (r *Registry) GetWorkflowStatus(id string) (telemetry.StatusUpdate, bool)

func (*Registry) InvalidateVHostSecret added in v1.17.0 

func (r *Registry) InvalidateVHostSecret(vhost, name string)

InvalidateVHostSecret forgets what expressions cached for one vhost's secret. The API calls it after a secret is saved or deleted, so the next message reads the new value instead of waiting out the cache.

func (*Registry) IsBranchPreviewable added in v1.8.1 

func (r *Registry) IsBranchPreviewable(nodeType string) bool

IsBranchPreviewable reports whether nodeType names a routing node the editor can preview, so a caller can pick the right path before doing any work rather than reacting to an error afterwards.

func (*Registry) IsDebuggerAttached 

func (r *Registry) IsDebuggerAttached(workflowID string) bool

func (*Registry) IsEngineRunning 

func (r *Registry) IsEngineRunning(id string) bool

func (*Registry) IsResourceInUse 

func (r *Registry) IsResourceInUse(ctx context.Context, resourceID string, excludeID string, isSource bool) bool

func (*Registry) IsWorkflowNode added in v1.8.1 

func (r *Registry) IsWorkflowNode(nodeType string) bool

IsWorkflowNode reports whether nodeType names a registered node executor.

Used to tell two failures apart. "stateful" is a real node that a preview cannot run, because running it would mutate state a live workflow depends on; "statefull" is a typo. Both used to produce 

unknown transformation type: no transformer is registered under that name

which describes the typo and misdescribes the node.

func (*Registry) LogWritesDropped added in v1.18.0 

func (r *Registry) LogWritesDropped() uint64

LogWritesDropped reports log lines discarded because log storage could not keep up.

func (*Registry) Logger 

func (r *Registry) Logger() hermod.Logger

func (*Registry) MLDatasetFromQuery added in v1.18.0 

func (r *Registry) MLDatasetFromQuery(ctx context.Context, vhost, dataset, sourceID, query string, maxRows int) (int, error)

MLDatasetFromQuery replaces a vhost's dataset with what a query on one of the vhost's database sources returns. A source of another vhost is refused: its credentials are not this vhost's to use.

func (*Registry) MLPredict added in v1.18.0 

func (r *Registry) MLPredict(ctx context.Context, vhost, model string, rows []inference.Row) ([]inference.Row, error)

MLPredict is what the Predict transformer calls: the vhost's model, with the rows it was given. The workflow is the caller the prediction log names.

func (*Registry) MLRecommend added in v1.18.0 

func (r *Registry) MLRecommend(ctx context.Context, vhost, model string, req map[string]any) ([]map[string]any, error)

MLRecommend is what the Predict node calls in Recommend mode: the vhost's recommendation model, for req's "user" or "item", with "k" and "exclude_seen" when set. The workflow is the caller the log names.

func (*Registry) MLRecommendBuild added in v1.18.0 

func (r *Registry) MLRecommendBuild(ctx context.Context, vhost, model string, config map[string]any) ([]map[string]any, int, error)

MLRecommendBuild is what the Build Recommendations node calls: the model's answers for every salesman, customer or product as rows, written to the target table when the config names one. config is the node's: for, periodDays, k, targetSourceId, targetTable, targetMode, and maxRows, over which the build is refused before anything is written.

func (*Registry) MLService added in v1.18.0 

func (r *Registry) MLService() *ml.Service

MLService is the model registry and inference path over whatever storage and secrets the registry holds right now. It is cheap to build, so it is built per use rather than cached against a store that may be replaced. The monitor is not: it is the one that holds the drift windows. Its ML worker is the one in force, with the one saved in storage re-read when that is due, so a worker connected from the console, here or in another process sharing the database, is used without a restart.

func (*Registry) MLTrain added in v1.18.0 

func (r *Registry) MLTrain(ctx context.Context, vhost, model string, req map[string]any) (map[string]any, error)

MLTrain is what the Train Model transformer calls: refill the dataset from its source when the node names one, then train and register the model.

func (*Registry) NotifyWorkerShutdown added in v1.11.0 

func (r *Registry) NotifyWorkerShutdown(ctx context.Context, activeWorkflows int)

NotifyWorkerShutdown reports that this worker is going down and taking its running workflows with it.

One alert for the worker rather than one per workflow: a worker running forty workflows would otherwise send forty messages at the moment its operator is least able to read them.

func (*Registry) OpenMLTarget added in v1.18.0 

func (r *Registry) OpenMLTarget(ctx context.Context, vhost, sourceID string) (*sql.DB, string, error)

OpenMLTarget opens a database source of the vhost for a recommendation build's writes: the source must belong to the vhost, as a dataset query's must. It returns the pool and the source's type.

func (*Registry) PauseForDebugger 

func (r *Registry) PauseForDebugger(workflowID, nodeID string, msg hermod.Message)

PauseForDebugger is PauseForDebuggerContext on context.Background().

func (*Registry) PauseForDebuggerContext added in v1.18.0 

func (r *Registry) PauseForDebuggerContext(ctx context.Context, workflowID, nodeID string, msg hermod.Message)

PauseForDebuggerContext holds a message at a breakpoint until the debugger answers, debuggerPauseLimit passes, or ctx ends. The last is what lets a workflow stopped while paused stop now rather than after the limit.

func (*Registry) PreviewBranch added in v1.8.1 

func (r *Registry) PreviewBranch(ctx context.Context, nodeType string, config map[string]any, msg hermod.Message) ([]hermod.Message, string, error)

PreviewBranch runs a routing node against a sample message and reports the branch it would take.

The editor's "Test" button used to have one path: the transformer registry. That covers every node whose job is to change a message, and none of the nodes whose job is to choose where it goes. switch, condition and router are registered as node executors and dispatched at runtime by executeNodeInternal, so a workflow containing them ran correctly while the preview of the same node answered 

unknown transformation type "switch": no transformer is registered under that name

which reads as a broken configuration rather than a preview that never looked in the right registry.

Only executors that declare themselves interfaces.PreviewSafe run here. A node that exists but has not declared it is refused by name, because a preview is not a reason to write to a sink or record an approval.

func (*Registry) PurgeLogs 

func (r *Registry) PurgeLogs(ctx context.Context, before time.Time) error

func (*Registry) RebuildWorkflow 

func (r *Registry) RebuildWorkflow(ctx context.Context, workflowID string, fromOffset int64) error

func (*Registry) RecordCircuitBreakerFailure 

func (r *Registry) RecordCircuitBreakerFailure(workflowID, breakerNodeID string)

RecordCircuitBreakerFailure counts a downstream failure against a circuit breaker node.

The breaker reads this count and opens once it passes its threshold. Nothing incremented it before, so a node offered in the editor as "Stop flow on failure threshold" always reported success however broken the downstream was — a control that cannot fire, which is worse than no control because someone believes it is there.

The registry is the natural place for this: it already holds node state and is what the traversal has a handle on.

func (*Registry) RecordStep 

func (r *Registry) RecordStep(ctx context.Context, workflowID, messageID string, step hermod.TraceStep)

func (*Registry) ResolveSecrets added in v1.18.0 

func (r *Registry) ResolveSecrets(ctx context.Context, vhost string, config map[string]string) map[string]string

ResolveSecrets is resolveSecrets for a transport that checks a source's credentials itself, outside a running workflow (a chat source's endpoint). The config passed in is not modified.

func (*Registry) ResumeApproval 

func (r *Registry) ResumeApproval(ctx context.Context, app storage.Approval, branch string) error

ResumeApproval resumes a halted workflow at an approval node with the specified decision branch ("approved" or "rejected").

func (*Registry) RunWorkflowNode 

func (r *Registry) RunWorkflowNode(workflowID string, node *storage.WorkflowNode, msg hermod.Message) ([]hermod.Message, string, error)

RunWorkflowNode runs one node on context.Background(). Nothing can cancel it, so a running workflow uses RunWorkflowNodeContext instead.

func (*Registry) RunWorkflowNodeContext added in v1.18.0 

func (r *Registry) RunWorkflowNodeContext(ctx context.Context, workflowID string, node *storage.WorkflowNode, msg hermod.Message) ([]hermod.Message, string, error)

RunWorkflowNodeContext runs one node bound to ctx. The traversal passes the workflow's lifetime context, so stopping or draining a workflow cancels a lookup or API call in flight instead of waiting out the node's own timeout.

func (*Registry) RunWorkflowOnce added in v1.18.0 

func (r *Registry) RunWorkflowOnce(ctx context.Context, wf storage.Workflow, msg hermod.Message, sourceNodeID string) (RunResult, error)

RunWorkflowOnce sends msg through wf once, starting at sourceNodeID (the workflow's first source node when empty), and returns the run's outcome.

It is "run now with this input": the nodes really run and the sinks really write, through the same walk a resumed approval takes, whether or not the workflow is running. Every step is written to the workflow's traces under the run id regardless of the workflow's trace sample rate, so the run is in the workflow's run history. msg is not modified; the run works on a copy with a fresh id.

func (*Registry) SampleSinkTable 

func (r *Registry) SampleSinkTable(ctx context.Context, cfg factory.SinkConfig, table string) (hermod.Message, error)

func (*Registry) SampleTable 

func (r *Registry) SampleTable(ctx context.Context, cfg factory.SourceConfig, table string) (hermod.Message, error)

func (*Registry) SetConfig 

func (r *Registry) SetConfig(cfg config.Config)

func (*Registry) SetFactories 

func (r *Registry) SetFactories(sourceFactory SourceFactory, sinkFactory SinkFactory)

func (*Registry) SetLogStorage 

func (r *Registry) SetLogStorage(s storage.Storage)

func (*Registry) SetLogger 

func (r *Registry) SetLogger(logger hermod.Logger)

func (*Registry) SetLookupCache 

func (r *Registry) SetLookupCache(key string, value any, ttl time.Duration)

SetLookupCache stores a value with an optional TTL. Expired entries are reclaimed lazily on read and during a bounded sweep on write, so no per-key goroutine is spawned (which previously leaked goroutines and memory under high lookup throughput).

func (*Registry) SetNodeState 

func (r *Registry) SetNodeState(key string, val any)

func (*Registry) SetSecretManager 

func (r *Registry) SetSecretManager(mgr secrets.Manager)

func (*Registry) SetSinkFactory 

func (r *Registry) SetSinkFactory(f SinkFactory)

func (*Registry) SetSourceFactory 

func (r *Registry) SetSourceFactory(f SourceFactory)

func (*Registry) SetStateStore 

func (r *Registry) SetStateStore(ss hermod.StateStore)

func (*Registry) SetStorage 

func (r *Registry) SetStorage(s storage.Storage)

func (*Registry) SetWorkerID added in v1.11.0 

func (r *Registry) SetWorkerID(id string)

SetWorkerID records which worker this registry belongs to, so a worker-level alert can say which one went down. A deployment runs several and "a worker is shutting down" is not actionable without the name.

func (*Registry) SimulateWorkflow added in v1.13.0 

func (r *Registry) SimulateWorkflow(ctx context.Context, wf storage.Workflow, in SimulationInput) ([]WorkflowStepResult, error)

SimulateWorkflow runs sample messages through a workflow without starting an engine, and reports what every node emitted.

Partial previews a workflow that is still being built. It keeps the checks that make the workflow a graph — a source, no cycle, no edge to nowhere — and drops the ones about whether it could run: a reachable sink, configured references, the dead-letter settings. None of those change what a node emits, and a workflow is missing its sink exactly while its nodes are being set up, which is when the editor needs the preview.

func (*Registry) StartWorkflow 

func (r *Registry) StartWorkflow(id string, wf storage.Workflow) error

StartWorkflow creates and starts a workflow engine for the given workflow configuration.

func (*Registry) StateStore 

func (r *Registry) StateStore() hermod.StateStore

func (*Registry) StatusSubscriberCount 

func (r *Registry) StatusSubscriberCount() int

StatusSubscriberCount returns the number of active status subscribers. It is primarily useful for observability and tests that assert subscribers are released when a client disconnects.

func (*Registry) StopAll 

func (r *Registry) StopAll()

func (*Registry) StopEngine 

func (r *Registry) StopEngine(ctx context.Context, id string) error

StopEngine stops a workflow on an operator's instruction. Unlike StopEngineWithoutUpdate — which the supervisor and the worker's own reconciliation use — this marks the end of a stall episode, so the workflow's automatic-restart budget starts fresh the next time it runs.

func (*Registry) StopEngineWithoutUpdate 

func (r *Registry) StopEngineWithoutUpdate(ctx context.Context, id string) error

func (*Registry) Storage 

func (r *Registry) Storage() interfaces.RegistryStorage

func (*Registry) SubscribeDashboardStats 

func (r *Registry) SubscribeDashboardStats(vhost string) chan storage.DashboardStats

func (*Registry) SubscribeDebugger 

func (r *Registry) SubscribeDebugger(workflowID string) chan DebuggerEvent

func (*Registry) SubscribeLiveMessages 

func (r *Registry) SubscribeLiveMessages() chan LiveMessage

func (*Registry) SubscribeLogs 

func (r *Registry) SubscribeLogs() chan storage.Log

func (*Registry) SubscribeStatus 

func (r *Registry) SubscribeStatus() chan telemetry.StatusUpdate

func (*Registry) SubscribeWorkflowLiveMessages 

func (r *Registry) SubscribeWorkflowLiveMessages(workflowID string) chan LiveMessage

func (*Registry) SubscribeWorkflowLogs 

func (r *Registry) SubscribeWorkflowLogs(workflowID string) chan storage.Log

func (*Registry) SubscribeWorkflowStatus 

func (r *Registry) SubscribeWorkflowStatus(workflowID string) chan telemetry.StatusUpdate

func (*Registry) TestSink 

func (r *Registry) TestSink(ctx context.Context, cfg factory.SinkConfig) error

func (*Registry) TestSource 

func (r *Registry) TestSource(ctx context.Context, cfg factory.SourceConfig) error

func (*Registry) TestTransformationPipeline 

func (r *Registry) TestTransformationPipeline(ctx context.Context, transformations []storage.Transformation, msg hermod.Message) ([]hermod.Message, error)

func (*Registry) TestWorkflow 

func (r *Registry) TestWorkflow(ctx context.Context, wf storage.Workflow, msg hermod.Message) ([]WorkflowStepResult, error)

TestWorkflow runs msg through a workflow the engine could start, seeding every source node with it. It is what the editor's Test button calls.

func (*Registry) TriggerSnapshot 

func (r *Registry) TriggerSnapshot(ctx context.Context, sourceID string, tables ...string) error

func (*Registry) UnsubscribeDashboardStats 

func (r *Registry) UnsubscribeDashboardStats(ch chan storage.DashboardStats)

func (*Registry) UnsubscribeDebugger 

func (r *Registry) UnsubscribeDebugger(workflowID string, ch chan DebuggerEvent)

func (*Registry) UnsubscribeLiveMessages 

func (r *Registry) UnsubscribeLiveMessages(ch chan LiveMessage)

func (*Registry) UnsubscribeLogs 

func (r *Registry) UnsubscribeLogs(ch chan storage.Log)

func (*Registry) UnsubscribeStatus 

func (r *Registry) UnsubscribeStatus(ch chan telemetry.StatusUpdate)

func (*Registry) UpdateNodeState 

func (r *Registry) UpdateNodeState(ctx context.Context, workflowID, nodeID string, state any) error

func (*Registry) UpdateSink 

func (r *Registry) UpdateSink(ctx context.Context, snk storage.Sink) error

func (*Registry) UpdateSinkStatus 

func (r *Registry) UpdateSinkStatus(ctx context.Context, id string, status string) error

func (*Registry) UpdateSource 

func (r *Registry) UpdateSource(ctx context.Context, src storage.Source) error

func (*Registry) UpdateSourceStatus 

func (r *Registry) UpdateSourceStatus(ctx context.Context, id string, status string) error

func (*Registry) ValidateWorkflow 

func (r *Registry) ValidateWorkflow(ctx context.Context, wf storage.Workflow) error

type RunResult added in v1.18.0 

type RunResult struct {
	RunID  string             `json:"run_id"`
	Status RunStatus          `json:"status"`
	Steps  []hermod.TraceStep `json:"steps"`
}

RunResult is what RunWorkflowOnce reports. RunID is the id of the message the run sent, which is also the id its trace is stored under.

type RunStatus added in v1.18.0 

type RunStatus string

RunStatus is how a manual run ended. 

const (
	// RunCompleted: every branch ran to its end without a node failing.
	RunCompleted RunStatus = "completed"
	// RunFailed: at least one node or sink write failed.
	RunFailed RunStatus = "failed"
	// RunWaiting: nothing failed and the message is held at an approval (or
	// another node that holds it) until someone acts.
	RunWaiting RunStatus = "waiting"
)

type SimulationInput added in v1.13.0 

type SimulationInput struct {
	// Message seeds every source node PerSource does not name. Nil leaves such
	// a source unseeded, and everything downstream of it unreached.
	Message hermod.Message
	// PerSource seeds source nodes, by node ID, with their own sample.
	PerSource map[string]hermod.Message
	// Partial previews a workflow that is still being built; see
	// SimulateWorkflow.
	Partial bool
}

SimulationInput is what a simulation feeds its source nodes.

type SinkFactory 

type SinkFactory func(factory.SinkConfig) (hermod.Sink, error)

type SourceFactory 

type SourceFactory func(factory.SourceConfig) (hermod.Source, error)

type WorkflowStepResult 

type WorkflowStepResult struct {
	NodeID   string            `json:"node_id"`
	NodeType string            `json:"node_type"`
	Payload  map[string]any    `json:"payload,omitempty"`
	Metadata map[string]string `json:"metadata,omitempty"`
	Error    string            `json:"error,omitempty"`
	Filtered bool              `json:"filtered,omitempty"`
	// Skipped marks a node no message reached: every edge into it was on a
	// branch that was not taken. Filtered is set as well, which is how such a
	// node was reported before Skipped existed; Skipped is what tells it apart
	// from a node that was reached and emitted nothing.
	Skipped bool   `json:"skipped,omitempty"`
	Branch  string `json:"branch,omitempty"`
	// TakenEdges are the IDs of the edges this node's output travelled along:
	// the path the editor highlights. Empty for a node that emitted nothing.
	TakenEdges []string `json:"taken_edges,omitempty"`
}

Source Files

View all Source files

Directories

Path Synopsis
Package noderetry is the per-step retry policy any workflow node can carry.
 Package noderetry is the per-step retry policy any workflow node can carry.
Package nodes is the one import that makes every workflow node type available.
 Package nodes is the one import that makes every workflow node type available.
ai
Package ai holds the workflow node executors that call a language model and route on its answer.
 Package ai holds the workflow node executors that call a language model and route on its answer.
ai/agent
Package agent is the ai_agent node: a bounded tool-use loop in which a language model works towards a goal by calling the tools the node allows.
 Package agent is the ai_agent node: a bounded tool-use loop in which a language model works towards a goal by calling the tools the node allows.
ai/agent/mcptool
Package mcptool lets the ai_agent node use one tool of a remote MCP server over the Streamable HTTP transport.
 Package mcptool lets the ai_agent node use one tool of a remote MCP server over the Streamable HTTP transport.
control
core
flow
reliability
util

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.