automations

package
v1.799.2 Latest Latest
Warning

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

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

Documentation

Overview

Package automations mounts the Hanzo Cloud /v1/automations/* surface: the Connectors+Automations engine (HIP-0106, task #51). It composes THREE existing seams rather than reinventing them:

  • clients/integrations — per-org connector credentials (KMS-sealed). Connectors reach a token ONLY through integrations.TokenFor, never KMS directly.
  • cloud.EmbeddedTasks — the ONE shared in-process durable engine. A flow runs as a durable workflow in the OWNER's namespace (engine.go).
  • clients/principal — the ONE tenant gate. Every data handler resolves the org from principal.Org; a client-forged X-Org-Id with no bearer is refused.

Surface (all under /v1/automations/*, all org-gated except the compose-root generic GET /v1/automations/health):

GET    /v1/automations/connectors                 the connector catalogue (org-gated)
GET    /v1/automations/pieces                     back-compat alias of /connectors
GET    /v1/automations/flows                      list flows
POST   /v1/automations/flows                      create a flow (+ initial draft version)
GET    /v1/automations/flows/:id                  flow + latest version
PATCH  /v1/automations/flows/:id                  update flow metadata
DELETE /v1/automations/flows/:id                  delete a flow (+ versions + runs)
GET    /v1/automations/flows/:id/versions         list versions
POST   /v1/automations/flows/:id/versions         create a draft version
POST   /v1/automations/flows/:id/operations       apply a FlowOperation
POST   /v1/automations/flows/:id/run              start a durable run
POST   /v1/automations/flows/:id/enable           enable (POLLING → CreateSchedule)
POST   /v1/automations/flows/:id/disable          disable (POLLING → DeleteSchedule)
GET    /v1/automations/runs                        list runs
GET    /v1/automations/runs/:id                    run detail (refreshed from engine)
POST   /v1/automations/runs/:id/resume             resume a paused run (SignalWorkflow)
POST   /v1/automations/mcp                          MCP JSON-RPC tool surface

Package automations mounts the Hanzo Cloud /v1/automations/* surface: a native-Go Connectors+Automations engine (HIP-0106, task #51) that runs an org's flows durably on the ONE shared in-process hanzoai/tasks engine and invokes third-party connectors whose credentials are custodied by clients/integrations (KMS-sealed, per-org).

This file ports the ActivePieces shared contract (auto/packages/shared/src/lib/automation/) to plain Go structs + string-const enums. The JSON tags match the TypeScript field names verbatim because the reused web/ flow builder is the contract consumer — it authors the same trigger→action tree this engine walks.

TENANT ISOLATION is a physical property, not a policy: every stored row leads its indexes with `org`, and the durable engine's ONLY credential scope is FlowRunInput.Owner — the VALIDATED org resolved from principal.Org at flow-start, never a client-supplied field. See engine.go (ExecuteStepActivity).

Index

Constants

View Source
const (
	TriggerTypePiece = "PIECE_TRIGGER"
	TriggerTypeEmpty = "EMPTY"

	ActionTypePiece  = "PIECE"
	ActionTypeCode   = "CODE"
	ActionTypeBranch = "BRANCH"
	ActionTypeLoop   = "LOOP_ON_ITEMS"
	ActionTypeRouter = "ROUTER"
)

FlowTriggerType / FlowActionType mirror the discriminants of the step nodes.

View Source
const LatestFlowSchemaVersion = "21"

LatestFlowSchemaVersion is the shared schema stamp new versions carry, matching flow-version.ts LATEST_FLOW_SCHEMA_VERSION.

Variables

View Source
var ErrEngineNotReady = errors.New("automations: engine not ready")

ErrEngineNotReady is returned when cloud.EmbeddedTasks() is still nil (the engine is wired after MountAll). Handlers render it as 503 "automation engine not ready".

Functions

func Mount

func Mount(app *zip.App, deps cloud.Deps) error

Mount wires /v1/automations/* onto app per HIP-0106. Complex flavour: it keeps a package global (mounted) for Shutdown and the engine's run hooks, so it constructs the Service value directly.

func RecordRunEndActivity

func RecordRunEndActivity(ctx context.Context, in RunEndInput) error

RecordRunEndActivity records a run's terminal status so listRuns reflects it without a getRun refresh. Best-effort; mounted==nil is a no-op.

func RecordRunStartActivity

func RecordRunStartActivity(ctx context.Context, in RunStartInput) error

RecordRunStartActivity persists the run row + meters + audits EXACTLY once (idempotent by run id via the store's metered-flag claim). The single owner of run bookkeeping for every entrypoint. mounted==nil (an engine-only test without Mount) is a no-op — nothing to record against.

func Shutdown

func Shutdown(_ context.Context) error

Shutdown closes the store. Idempotent — safe when nothing is mounted.

Types

type Action

type Action struct {
	Name        string
	DisplayName string
	Description string
	Props       []PropSpec
	Run         func(ctx context.Context, rc RunContext) (any, error)
}

Action is one invocable capability of a connector. Props declares its inputs (drives the catalogue + the MCP tool input schema). Run is the side-effecting body; it MUST handle every error explicitly and MUST NOT leak another org's data.

type AddActionRequest

type AddActionRequest struct {
	ParentStep string      `json:"parentStep"`
	Action     *FlowAction `json:"action"`
}

AddActionRequest inserts Action after the step named ParentStep (empty ⇒ append to the end of the chain, or set as the trigger's first action).

type Catalog

type Catalog struct {
	ConnectorCount int                 `json:"connectorCount"`
	Connectors     []ConnectorMetadata `json:"connectors"`
}

Catalog is the browse catalogue served at GET /v1/automations/connectors. It is the go:embed'd catalog/catalog.json, seeded with the Tier-A connectors and later overwritten with the full 700+ connector set at this EXACT schema.

type ChangeNameRequest

type ChangeNameRequest struct {
	DisplayName string `json:"displayName"`
}

ChangeNameRequest renames the version.

type ChangeStatusRequest

type ChangeStatusRequest struct {
	Status FlowStatus `json:"status"`
}

ChangeStatusRequest flips ENABLED/DISABLED.

type Connector

type Connector struct {
	Name        string
	DisplayName string
	AuthType    string // "none" | "bot_token" | "oauth2" — for the catalogue card
	AuthReq     bool
	Actions     map[string]*Action
	Triggers    map[string]*Trigger
}

Connector is one connectable capability provider. Name == the integrations provider id where credentials apply.

type ConnectorAction added in v1.786.131

type ConnectorAction struct {
	Name        string     `json:"name"`
	DisplayName string     `json:"displayName"`
	Description string     `json:"description"`
	Props       []PropSpec `json:"props"`
}

ConnectorAction / ConnectorTrigger are the catalogue's action/trigger descriptors.

type ConnectorAuth added in v1.786.131

type ConnectorAuth struct {
	Type     string `json:"type"`
	Required bool   `json:"required"`
}

ConnectorAuth is the catalogue's auth descriptor: which credential a connector needs ("none" for core, "oauth2"/"bot_token" for a connected provider) and whether it is required.

type ConnectorMetadata added in v1.786.131

type ConnectorMetadata struct {
	Name        string             `json:"name"`
	DisplayName string             `json:"displayName"`
	Description string             `json:"description"`
	LogoURL     string             `json:"logoUrl"`
	Version     string             `json:"version"`
	Categories  []string           `json:"categories"`
	Auth        ConnectorAuth      `json:"auth"`
	Actions     []ConnectorAction  `json:"actions"`
	Triggers    []ConnectorTrigger `json:"triggers"`
}

ConnectorMetadata is one catalogue entry. The catalog WIRE schema models a connector's actions/triggers as arrays, so this Go shape uses arrays to match the wire exactly, as the contract requires.

type ConnectorTrigger added in v1.786.131

type ConnectorTrigger struct {
	Name        string     `json:"name"`
	DisplayName string     `json:"displayName"`
	Description string     `json:"description"`
	Strategy    string     `json:"strategy"`
	Props       []PropSpec `json:"props"`
}

type DeleteActionRequest

type DeleteActionRequest struct {
	Names []string `json:"names"`
}

DeleteActionRequest removes the named steps and relinks the chain.

type Flow

type Flow struct {
	ID                 string          `json:"id"`
	Org                string          `json:"projectId"` // projectId == org (server-derived)
	ExternalID         string          `json:"externalId"`
	FolderID           string          `json:"folderId"`
	Status             FlowStatus      `json:"status"`
	PublishedVersionID string          `json:"publishedVersionId"`
	Metadata           json.RawMessage `json:"metadata,omitempty"`
	Created            int64           `json:"created"`
	Updated            int64           `json:"updated"`
}

Flow is an org-scoped automation. projectId IS the org (the TS `projectId` field), always server-derived from principal.Org and NEVER trusted from a request body — a caller can never author a flow into another tenant.

type FlowAction

type FlowAction struct {
	Name        string       `json:"name"`
	Type        string       `json:"type"` // PIECE | CODE | ROUTER | LOOP_ON_ITEMS
	DisplayName string       `json:"displayName"`
	Valid       bool         `json:"valid"`
	Skip        bool         `json:"skip,omitempty"`
	Settings    StepSettings `json:"settings"`
	NextAction  *FlowAction  `json:"nextAction,omitempty"`
}

FlowAction is an action node (actions/action.ts). The engine walks the linear NextAction chain; the tree discriminants (ROUTER/LOOP) are modeled for contract fidelity but branch/loop execution is out of Phase-1 scope (see operations.go).

type FlowOperation

type FlowOperation struct {
	Type    FlowOperationType `json:"type"`
	Request json.RawMessage   `json:"request"`
}

FlowOperation is the discriminated-union envelope the builder POSTs to /v1/automations/flows/:id/operations: a type + an opaque request the apply switch decodes per type.

type FlowOperationType

type FlowOperationType string

FlowOperationType mirrors operations/index.ts FlowOperationType — the builder's edit-op discriminants. The engine applies the linear-chain subset (see operations.go); tree-restructuring ops are honestly rejected, never faked.

const (
	OpAddAction     FlowOperationType = "ADD_ACTION"
	OpUpdateAction  FlowOperationType = "UPDATE_ACTION"
	OpDeleteAction  FlowOperationType = "DELETE_ACTION"
	OpMoveAction    FlowOperationType = "MOVE_ACTION"
	OpUpdateTrigger FlowOperationType = "UPDATE_TRIGGER"
	OpChangeName    FlowOperationType = "CHANGE_NAME"
	OpChangeStatus  FlowOperationType = "CHANGE_STATUS"
)

type FlowRun

type FlowRun struct {
	ID            string        `json:"id"`
	Org           string        `json:"-"`
	FlowID        string        `json:"flowId"`
	FlowVersionID string        `json:"flowVersionId"`
	WorkflowID    string        `json:"-"`
	Status        FlowRunStatus `json:"status"`
	StartTime     int64         `json:"startTime"`
	FinishTime    int64         `json:"finishTime"`
	Created       int64         `json:"created"`
	Updated       int64         `json:"updated"`
}

FlowRun is an execution record (flow-run/flow-run.ts). WorkflowID (the tasks engine handle) equals ID and is internal — resume/describe address the engine through it, scoped to the org's namespace.

type FlowRunInput

type FlowRunInput struct {
	Owner         string     `json:"owner"`
	FlowID        string     `json:"flowId"`
	FlowVersionID string     `json:"flowVersionId"`
	RunID         string     `json:"runId"`
	Steps         []FlowStep `json:"steps"`
}

FlowRunInput is the durable workflow's typed input. Owner is the VALIDATED org set at flow-start from principal.Org — the SOLE credential scope and the cross-tenant isolation boundary; it is NEVER read from a request body. Steps is the flattened trigger→action chain (side-effecting steps in order).

type FlowRunResult

type FlowRunResult struct {
	RunID   string         `json:"runId"`
	Status  FlowRunStatus  `json:"status"`
	Steps   int            `json:"steps"`
	Outputs map[string]any `json:"outputs"`
}

FlowRunResult is the workflow's terminal result: the final status plus every step's output, keyed by step name (the threaded outputs).

func FlowRunWorkflow

func FlowRunWorkflow(ctx workflow.Context, in FlowRunInput) (FlowRunResult, error)

FlowRunWorkflow walks the flow's flattened step chain. Each side-effecting step runs as a retried activity (ExecuteStepActivity); a core.wait_for_approval step is a durable PAUSE that blocks on the resume signal. Prior step outputs are threaded so later steps can reference them ({{step.field}}). Deterministic: it executes steps strictly in order, Get-ing each before dispatching the next.

It also owns run bookkeeping (MED-1): the per-EXECUTION run id is the workflow id (workflow.GetInfo) — manual runs set it to the run id; a scheduled cron mints a fresh one per tick — so a run-start activity persists+meters+audits the run EXACTLY once, and a run-end activity records the terminal status.

type FlowRunStatus

type FlowRunStatus string

FlowRunStatus mirrors execution/flow-execution.ts FlowRunStatus (the subset the engine emits; the full enum has memory/log/quota terminal states we map onto FAILED).

const (
	RunRunning   FlowRunStatus = "RUNNING"
	RunSucceeded FlowRunStatus = "SUCCEEDED"
	RunFailed    FlowRunStatus = "FAILED"
	RunPaused    FlowRunStatus = "PAUSED"
	RunQueued    FlowRunStatus = "QUEUED"
	RunCanceled  FlowRunStatus = "CANCELED"
	RunTimeout   FlowRunStatus = "TIMEOUT"
)

type FlowStatus

type FlowStatus string

FlowStatus mirrors flows/flow.ts FlowStatus.

const (
	FlowEnabled  FlowStatus = "ENABLED"
	FlowDisabled FlowStatus = "DISABLED"
)

type FlowStep

type FlowStep struct {
	Name       string         `json:"name"`
	PieceName  string         `json:"pieceName"`
	ActionName string         `json:"actionName"`
	Input      map[string]any `json:"input"`
}

FlowStep is one resolved, executable step in the flattened chain.

type FlowTrigger

type FlowTrigger struct {
	Name        string          `json:"name"`
	Type        string          `json:"type"` // PIECE_TRIGGER | EMPTY
	DisplayName string          `json:"displayName"`
	Valid       bool            `json:"valid"`
	Strategy    TriggerStrategy `json:"strategy,omitempty"`
	Settings    StepSettings    `json:"settings"`
	NextAction  *FlowAction     `json:"nextAction,omitempty"`
}

FlowTrigger is the root of the step tree (triggers/trigger.ts). Strategy drives enable/disable (POLLING → CreateSchedule); NextAction chains into the action list.

type FlowVersion

type FlowVersion struct {
	ID            string           `json:"id"`
	Org           string           `json:"-"`
	FlowID        string           `json:"flowId"`
	DisplayName   string           `json:"displayName"`
	Trigger       *FlowTrigger     `json:"trigger"`
	Valid         bool             `json:"valid"`
	State         FlowVersionState `json:"state"`
	SchemaVersion string           `json:"schemaVersion"`
	Created       int64            `json:"created"`
	Updated       int64            `json:"updated"`
}

FlowVersion is one editable revision of a flow: a display name plus the root trigger of the step tree. Org is the isolation key (never serialized).

type FlowVersionState

type FlowVersionState string

FlowVersionState mirrors flows/flow-version.ts FlowVersionState.

const (
	VersionDraft  FlowVersionState = "DRAFT"
	VersionLocked FlowVersionState = "LOCKED"
)

type MoveActionRequest

type MoveActionRequest struct {
	Name          string `json:"name"`
	NewParentStep string `json:"newParentStep"`
}

MoveActionRequest moves Name to sit immediately after NewParentStep.

type PropSpec

type PropSpec struct {
	Name        string `json:"name"`
	DisplayName string `json:"displayName,omitempty"`
	Type        string `json:"type"` // string|number|boolean|object|array
	Required    bool   `json:"required,omitempty"`
	Description string `json:"description,omitempty"`
}

PropSpec describes one input property of an action/trigger. It is the ONE prop shape shared by the connector framework (connector.go), the catalogue, and the MCP tool input-schema derivation (mcp.go) — one definition, three consumers.

type RunContext

type RunContext struct {
	Org         string
	Input       map[string]any
	PrevOutputs map[string]any
	Token       func(secretName string) ([]byte, error)
}

RunContext is what an action's Run receives. It is the connector's ENTIRE view of the world: the org (for logging/attribution only — never a place to widen scope), the resolved input, the prior steps' outputs (threaded), and Token — the ONLY door to a credential. Token is bound at dispatch time to the VALIDATED org (StepInput.Owner) and the connector's own provider id, so a connector can reach no other tenant's and no other provider's secret.

type RunEndInput

type RunEndInput struct {
	RunID  string `json:"runId"`
	Owner  string `json:"owner"`
	Status string `json:"status"`
}

RunStartInput / RunEndInput are the durable run-bookkeeping activity payloads. The durable path is the SINGLE owner of run bookkeeping (MED-1): whichever entrypoint drives a workflow — manual /run, MCP, or a scheduled cron tick — records the run exactly once here, so metering/audit never double-count and every execution lands a FlowRun row visible to listRuns/getRun.

type RunStartInput

type RunStartInput struct {
	RunID         string `json:"runId"`
	Owner         string `json:"owner"`
	FlowID        string `json:"flowId"`
	FlowVersionID string `json:"flowVersionId"`
}

RunStartInput / RunEndInput are the durable run-bookkeeping activity payloads. The durable path is the SINGLE owner of run bookkeeping (MED-1): whichever entrypoint drives a workflow — manual /run, MCP, or a scheduled cron tick — records the run exactly once here, so metering/audit never double-count and every execution lands a FlowRun row visible to listRuns/getRun.

type StepInput

type StepInput struct {
	Owner       string         `json:"owner"`
	RunID       string         `json:"runId"`
	Name        string         `json:"name"`
	PieceName   string         `json:"pieceName"`
	ActionName  string         `json:"actionName"`
	Input       map[string]any `json:"input"`
	PrevOutputs map[string]any `json:"prevOutputs"`
}

StepInput is the activity's per-step input. Owner is copied verbatim from FlowRunInput.Owner by the workflow — an activity can never widen its own scope.

type StepOutput

type StepOutput struct {
	Name   string `json:"name"`
	Output any    `json:"output"`
}

StepOutput is the activity's result for one step.

func ExecuteStepActivity

func ExecuteStepActivity(ctx context.Context, in StepInput) (StepOutput, error)

ExecuteStepActivity is the side-effecting body of ONE step. It is THE isolation boundary: RunContext.Token is bound to in.Owner (the VALIDATED org set at flow-start) and the step's own connector id, so a flow authored by org A can never reach org B's connection nor another provider's secret. in.Owner is used for token custody and NOTHING from in.Input can change it.

type StepSettings

type StepSettings struct {
	PieceName    string         `json:"pieceName,omitempty"`
	PieceVersion string         `json:"pieceVersion,omitempty"`
	ActionName   string         `json:"actionName,omitempty"`
	TriggerName  string         `json:"triggerName,omitempty"`
	Input        map[string]any `json:"input,omitempty"`
}

StepSettings is the flattened union of PieceTriggerSettings / PieceActionSettings / CodeActionSettings — the fields the engine actually reads to dispatch a step: which piece, which action/trigger, and the input map. (The builder-only settings — propertySettings, sampleData, errorHandlingOptions — round-trip opaquely via the raw version JSON in the store; they are not modeled here.)

type Store

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

Store is the automations database. ONE SQLite file ({DataDir}/automations.db) holds every org's flows, versions, and runs; tenant isolation is the `org` column, physical on EVERY uniqueness + lookup index (each leads with org). MaxOpenConns(1) serializes writes against the single-writer WAL file.

func (*Store) ClaimMeter

func (s *Store) ClaimMeter(ctx context.Context, org, id string) (bool, error)

ClaimMeter atomically flips the run's metered flag 0→1 for (org,id) and reports whether THIS call won the flip. It is the exactly-once billing gate: only the winner meters + audits the run, so no entrypoint double-bills a run (MED-1). The (org,id) predicate keeps the claim tenant-scoped.

func (*Store) Close

func (s *Store) Close() error

Close closes the underlying database. Idempotent-safe via sql.DB.

func (*Store) CreateFlow

func (s *Store) CreateFlow(ctx context.Context, f Flow) (Flow, error)

func (*Store) CreateRun

func (s *Store) CreateRun(ctx context.Context, r FlowRun) (FlowRun, error)

func (*Store) CreateRunIfAbsent

func (s *Store) CreateRunIfAbsent(ctx context.Context, r FlowRun) (bool, error)

CreateRunIfAbsent inserts a run row keyed on its id, doing NOTHING if it already exists. It reports whether THIS call created the row. Idempotent by run id (=the workflow execution id), so a retried run-start bookkeeping step, or a manual handler + the durable path racing to record the same run, converge on one row.

func (*Store) CreateVersion

func (s *Store) CreateVersion(ctx context.Context, v FlowVersion) (FlowVersion, error)

CreateVersion inserts a version. The flow must exist in the SAME org (validated here, not by a SQL FK, so a cross-tenant flow_id can never anchor a version).

func (*Store) DeleteFlow

func (s *Store) DeleteFlow(ctx context.Context, org, id string) (bool, error)

DeleteFlow removes a flow and all its versions + runs within the org. One transaction so a partial delete never strands a version/run.

func (*Store) GetFlow

func (s *Store) GetFlow(ctx context.Context, org, id string) (Flow, error)

func (*Store) GetRun

func (s *Store) GetRun(ctx context.Context, org, id string) (FlowRun, error)

func (*Store) GetVersion

func (s *Store) GetVersion(ctx context.Context, org, id string) (FlowVersion, error)

func (*Store) LatestVersion

func (s *Store) LatestVersion(ctx context.Context, org, flowID string) (FlowVersion, error)

LatestVersion returns the most-recently-created version for a flow, or errNotFound if the flow has none.

func (*Store) ListFlows

func (s *Store) ListFlows(ctx context.Context, org string, limit int) ([]Flow, error)

func (*Store) ListRuns

func (s *Store) ListRuns(ctx context.Context, org, flowID string, limit int) ([]FlowRun, error)

func (*Store) ListVersions

func (s *Store) ListVersions(ctx context.Context, org, flowID string, limit int) ([]FlowVersion, error)

func (*Store) UpdateFlow

func (s *Store) UpdateFlow(ctx context.Context, f Flow) (Flow, error)

UpdateFlow persists the mutable flow fields (status, folder, published version, metadata) for (org,id). RowsAffected==0 ⇒ errNotFound (cross-tenant or missing).

func (*Store) UpdateRunStatus

func (s *Store) UpdateRunStatus(ctx context.Context, org, id string, status FlowRunStatus, finish, updated int64) error

UpdateRunStatus persists a terminal/observed status transition for (org,id).

func (*Store) UpdateVersion

func (s *Store) UpdateVersion(ctx context.Context, v FlowVersion) (FlowVersion, error)

UpdateVersion replaces a version's editable content (display name, trigger tree, valid, state). RowsAffected==0 ⇒ errNotFound.

type Trigger

type Trigger struct {
	Name        string
	DisplayName string
	Description string
	Strategy    TriggerStrategy
	Props       []PropSpec
}

Trigger is one entry point of a connector. Strategy selects POLLING (cron) vs WEBHOOK vs MANUAL. Phase-1 triggers are catalogue/metadata only (the entry wiring lives in the flow's root trigger + enable/disable); a Trigger carries no Run.

type TriggerStrategy

type TriggerStrategy string

TriggerStrategy mirrors trigger/index.ts TriggerStrategy. It selects HOW a flow starts: POLLING → a cron schedule on the tasks engine; WEBHOOK/APP_WEBHOOK → an inbound HTTP event; MANUAL → an explicit /run.

const (
	StrategyPolling    TriggerStrategy = "POLLING"
	StrategyWebhook    TriggerStrategy = "WEBHOOK"
	StrategyAppWebhook TriggerStrategy = "APP_WEBHOOK"
	StrategyManual     TriggerStrategy = "MANUAL"
)

type UpdateActionRequest

type UpdateActionRequest struct {
	Name     string       `json:"name"`
	Type     string       `json:"type"`
	Settings StepSettings `json:"settings"`
}

UpdateActionRequest replaces the settings of the step named Name.

type UpdateTriggerRequest

type UpdateTriggerRequest FlowTrigger

UpdateTriggerRequest replaces the root trigger node.

Jump to

Keyboard shortcuts

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