workflows

package
v0.37.0 Latest Latest
Warning

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

 Go to latest Published: Oct 11, 2026 License: AGPL-3.0 Imports: 19 Imported by: 0

Documentation

Overview

Package workflows contains the storage layer for Epic 64 (triggers & workflows).

Pure data-access on the seven Epic 64 tables. Crypto (encrypt/decrypt of webhook HMAC secrets) is NOT done here — handlers encrypt before calling CreateWebhook; the webhook receiver decrypts after calling GetWebhookByTriggerID (the public hook URL carries the trigger id). This mirrors the pkg/secrets/mcp_store.go split exactly.

Index

Constants

View Source 
const SchemaViolationPayloadCap = 4 << 10 // 4 KiB

SchemaViolationPayloadCap is the hard cap on a marshaled schema-mismatch action_result document. The payload size is an invariant, never a best-effort.

Variables

View Source 
var ErrConcurrentRun = errors.New("workflow: a run is already in flight for this workflow")

ErrConcurrentRun is returned by CreateWorkflowRun when the single-in-flight partial unique index (uq_workflow_run_single_inflight) rejects the insert because another queued/running run exists for the same workflow_id. Handlers map this to 409 "already running".

View Source 
var ErrDedupConflict = errors.New("workflow: duplicate webhook delivery")

ErrDedupConflict is returned by RecordWebhookDelivery when the (webhook_id, dedup_key) UNIQUE constraint rejects a duplicate delivery. Handlers map this to 200 "duplicate" (idempotent re-delivery).

View Source 
var ErrInputSchemaNotObject = errors.New(`inputSchema must be object-rooted (type "object", a type list containing "object", or no root type)`)

ErrInputSchemaNotObject rejects a declared inputSchema whose root type no run input can satisfy: runs always carry a JSON object (an absent run input is validated as the JSON null document), so e.g. {"type":"string"} would fail validation on every run.

View Source 
var ErrInvalidInputSchema = errors.New("invalid inputSchema")

ErrInvalidInputSchema is the sentinel for "the stored schema itself is broken" — malformed JSON or no-longer-compiling. A workflow defect, not an input defect: the fire path records a failed fire with {"code":"invalid_input_schema"} instead of a schema-mismatch payload (design 0059 D3).

View Source 
var ErrNotFound = errors.New("workflow: not found")

ErrNotFound is returned by Get/Update/Delete when the row does not exist (or the caller's owner scope does not match). Handlers map this to 404.

Functions

func NextCronFire added in v0.33.0 

func NextCronFire(cfg *types.CronSourceConfig, now time.Time) (time.Time, error)

NextCronFire returns the first fire time strictly after now for a validated cron config. The config is re-validated here so the engine keeps its defensive behavior for rows written before validation existed (callers translate the error into a fallback).

func NextCronFireFromConfig added in v0.33.0 

func NextCronFireFromConfig(sourceConfig json.RawMessage, now time.Time) (*types.CronSourceConfig, time.Time, error)

NextCronFireFromConfig validates source_config and computes the next fire after now in one step — the write-path convenience.

func NormalizeInputSchema added in v0.33.1 

func NormalizeInputSchema(inputSchema json.RawMessage) json.RawMessage

NormalizeInputSchema collapses an explicit JSON null declaration to a nil (absent) schema so the stored row never carries a literal jsonb null: "declared null" and "not declared" are the same contract at the storage boundary — schema-less (#1433). Everything else passes through untouched.

func NormalizeTriggerInputFrom added in v0.34.0 

func NormalizeTriggerInputFrom(inputFrom string) string

NormalizeTriggerInputFrom maps an unset/empty mode to the legacy default "envelope" (rows scanned after migration 000031 always carry a value; in-memory rows may not).

func RecordNodeDuration added in v0.8.12 

func RecordNodeDuration(nodeType string, durationSeconds float64)

RecordNodeDuration records per-node execution timing.

func RecordRunFinished added in v0.8.12 

func RecordRunFinished(status, errorCode, ownerType string, durationSeconds float64)

RecordRunFinished records a run's terminal status + duration.

func RecordRunStarted added in v0.8.12 

func RecordRunStarted()

RecordRunStarted increments the concurrent gauge.

func RecordSchedulerTick added in v0.8.12 

func RecordSchedulerTick(durationSeconds float64)

RecordSchedulerTick records scheduler tick timing.

func RecordTriggerFire added in v0.8.12 

func RecordTriggerFire(source, status string)

RecordTriggerFire records a trigger fire event.

func RecordWebhookDelivery added in v0.8.12 

func RecordWebhookDelivery(webhookID, status string)

RecordWebhookDelivery records a webhook delivery.

func RequiredNonEnvelopeProperties added in v0.34.0 

func RequiredNonEnvelopeProperties(schema json.RawMessage, sourceType string) []string

RequiredNonEnvelopeProperties returns the top-level `required` property names of schema that the given source's envelope key set provides no value for — the V6 wiring-guard predicate (design 0059 D4). Syntactic only: no deep schema walk (fire-time validation is the backstop for opted-in triggers). A schema that cannot be parsed requires nothing beyond the envelope — the guard skips it (a garbage stored schema is a workflow defect, caught at fire time or on the next workflow update).

func ResolveTriggerInput added in v0.34.0 

func ResolveTriggerInput(spec TriggerInputSpec, envelope json.RawMessage) (json.RawMessage, error)

ResolveTriggerInput computes the input a fired run should carry per the trigger's mapping (design 0059 §3.4): 

base := body mode → envelope.body; mapped → nil; else the envelope itself
no static input   → base (the envelope path returns the envelope bytes
                    VERBATIM — legacy triggers are byte-identical)
base is an object → shallow top-level merge, static keys win
otherwise         → the static document verbatim

Shallow merge is deliberate (deep-merge semantics are unspecified territory; the first 90% use case is "provide `topic` at the top level"). The envelope stays reachable under its own keys in envelope mode; mapped mode exists for additionalProperties:false schemas a merged envelope would violate.

func SchemaMismatchPayload added in v0.34.0 

func SchemaMismatchPayload(inputFrom string, err error) json.RawMessage

SchemaMismatchPayload builds the capped action_result document for a fire-time input/schema mismatch (status validation_error): typed, location-only violations extracted from verr, hard-capped at SchemaViolationPayloadCap with the design's recursive drop semantics:

  • records are appended only while the marshaled payload stays under the cap; a record that does not fit first drops its message, then the record — pointer included — is dropped entirely and counted in a final {"keyword":"truncated","message":"N more violations omitted"} marker;
  • if even the first record cannot fit without its message, the payload degrades to the bare {"code","inputFrom"} skeleton plus marker.

An err that is not a *jsonschema.ValidationError (or extraction that yields no records) still produces a well-formed payload with a single static, instance-free record — the invariant is "always ≤ cap, never instance values", including for unexpected error shapes.

func StaticInputPresent added in v0.34.0 

func StaticInputPresent(raw json.RawMessage) bool

StaticInputPresent reports whether raw carries a static input document: a non-nil, non-JSON-null document. The JSON literal null means "no static input" — same as an absent key (mirrors the #1433 null-means- absent convention on inputSchema).

func TriggerOptedIn added in v0.34.0 

func TriggerOptedIn(inputFrom string, input json.RawMessage) bool

TriggerOptedIn reports whether a trigger opted in to fire-time input validation (D3): any non-default inputFrom, or a static input document present. Legacy un-opted triggers keep byte-identical envelope behavior and no validation (§3.6).

func ValidateCronSourceConfig added in v0.33.0 

func ValidateCronSourceConfig(sourceConfig json.RawMessage) (*types.CronSourceConfig, error)

ValidateCronSourceConfig decodes and validates a cron trigger's source_config JSON: expr must parse under the five-field syntax and tz (when set) must be a loadable IANA location. It returns the typed config so callers don't decode twice.

func ValidateInputSchema added in v0.33.0 

func ValidateInputSchema(inputSchema json.RawMessage) error

ValidateInputSchema checks that a workflow's declared inputSchema is declarable: well-formed JSON that compiles as a JSON Schema AND roots at type "object" (or leaves the root type unset). An absent schema — or the explicit JSON literal null, which means the same thing (#1433) — accepts everything. Used at workflow create/update so a broken or unsatisfiable schema fails fast instead of at run time.

func ValidateRunInput added in v0.33.0 

func ValidateRunInput(inputSchema, input json.RawMessage) error

ValidateRunInput validates a run's input document against a workflow's inputSchema. An absent schema — or a stored JSON literal null row (#1433) — accepts everything; an absent input is validated as the JSON null document (so required properties fail).

Types

type AgentNodeData 

type AgentNodeData struct {
	Agent                   string          `json:"agent,omitempty"`
	Prompt                  string          `json:"prompt"`
	System                  string          `json:"system,omitempty"`
	OutputSchema            json.RawMessage `json:"outputSchema,omitempty"`
	EnforceStructuredOutput bool            `json:"enforceStructuredOutput,omitempty"`
	Session                 string          `json:"session,omitempty"`
	SessionID               string          `json:"sessionId,omitempty"`
}

AgentNodeData is the typed shape of SpecNode.Data for agent nodes.

type ConditionCase 

type ConditionCase struct {
	ID         string `json:"id"`
	Expression string `json:"expression"`
}

ConditionCase is one branch of a condition node.

type ConditionNodeData 

type ConditionNodeData struct {
	Conditions []ConditionCase `json:"conditions"`
}

ConditionNodeData is the typed shape of SpecNode.Data for condition nodes.

type DefaultsBlock 

type DefaultsBlock struct {
	MaxAttempts *int   `json:"maxAttempts,omitempty"`
	Timeout     string `json:"timeout,omitempty"`
}

DefaultsBlock carries workflow-level defaults that are merged into each node's config (node-level wins). Only maxAttempts and timeout may be defaulted; behavioral fields must be per-node.

type HTTPNodeData 

type HTTPNodeData struct {
	Method  string            `json:"method,omitempty"`
	URL     string            `json:"url"`
	Headers map[string]string `json:"headers,omitempty"`
	Body    string            `json:"body,omitempty"`
	Timeout string            `json:"timeout,omitempty"`
}

HTTPNodeData is the typed shape of SpecNode.Data for http nodes.

type SchemaViolation added in v0.34.0 

type SchemaViolation struct {
	JSONPointer string `json:"jsonPointer"`
	Keyword     string `json:"keyword"`
	Message     string `json:"message,omitempty"`
}

SchemaViolation is one typed, location-only record of a schema violation. JSONPointer locates the violation in the run input (RFC 6901); Keyword is the failing schema keyword; Message is generated from keyword + schema-side fields only (e.g. a name from the schema's own `required` list — never an instance value).

func ExtractSchemaViolations added in v0.34.0 

func ExtractSchemaViolations(err error) []SchemaViolation

ExtractSchemaViolations walks a validation error's typed trail and returns location-only violation records. Never consults ValidationError.Error(): pointers come from InstanceLocation, keywords from ErrorKind.KeywordPath(), and messages from the concrete kind's SCHEMA-side fields only.

type ScriptNodeData 

type ScriptNodeData struct {
	Language string `json:"language"`
	Handler  string `json:"handler"`
}

ScriptNodeData is the typed shape of SpecNode.Data for script nodes.

type SessionOriginRow added in v0.11.0 

type SessionOriginRow struct {
	SessionID   string
	WorkspaceID string
	Origin      string
	TriggerID   *string
	FireID      *string
	Title       string
	CreatedAt   time.Time
}

type Spec 

type Spec struct {
	Nodes []SpecNode `json:"nodes"`
	Edges []SpecEdge `json:"edges"`
}

Spec is the parsed and validated DAG stored in workflows.spec_json. It is the execution-ready form: defaults merged, node types validated, edges checked for cycles/dangling/condition-coverage.

func ParseSpec 

func ParseSpec(raw json.RawMessage) (*Spec, error)

ParseSpec parses raw JSON bytes into a Spec. Does NOT validate — call ValidateSpec for that. Returns an error only on JSON parse failure.

type SpecEdge 

type SpecEdge struct {
	Source       string `json:"source"`
	Target       string `json:"target"`
	SourceHandle string `json:"sourceHandle,omitempty"`
}

SpecEdge is a directed edge in the DAG. SourceHandle carries the condition-branch id (for condition nodes) or is empty (for all other types).

type SpecNode 

type SpecNode struct {
	ID          string          `json:"id"`
	Type        string          `json:"type"`
	Data        json.RawMessage `json:"data"`
	MaxAttempts int             `json:"maxAttempts,omitempty"`
	Timeout     string          `json:"timeout,omitempty"`
}

SpecNode is a single node in the workflow DAG.

type Store 

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

Store is the Postgres-backed data-access layer for Epic 64. Construct once at app boot via NewStore; share across handlers. Methods are safe for concurrent use (pgxpool serializes connections internally).

func NewStore 

func NewStore(pool *pgxpool.Pool) *Store

NewStore constructs a Store backed by the given pgxpool.

func (*Store) ClaimDueCronTriggers added in v0.12.2 

func (s *Store) ClaimDueCronTriggers(ctx context.Context, now time.Time, limit int, nextFireFn func(*TriggerRow) time.Time) ([]*TriggerRow, error)

ClaimDueCronTriggers atomically claims due cron triggers and advances each trigger's last_fired_at and next_fire_at within the same transaction. This prevents sibling API replicas from claiming the same trigger concurrently: FOR UPDATE SKIP LOCKED ensures that if replica A locks a row, replica B's SELECT skips it rather than blocking.

nextFireFn computes the next fire time for each claimed trigger from its cron expression (cron parsing lives in the engine layer, not the store). The returned triggers are in their pre-advance state so the caller can use the original next_fire_at for the missed-tick skip check.

func (*Store) ClaimQueuedRuns 

func (s *Store) ClaimQueuedRuns(ctx context.Context, limit int) ([]*WorkflowRunRow, error)

ClaimQueuedRuns selects up to limit queued runs and atomically marks them running, returning the claimed rows. Uses FOR UPDATE SKIP LOCKED so multiple API replicas (or multiple goroutines) can claim concurrently without contention. The reconciler polls this on each tick.

func (*Store) CountTriggersByOwner 

func (s *Store) CountTriggersByOwner(ctx context.Context, ownerType, ownerID string) (int, error)

CountTriggersByOwner returns the number of triggers owned by (ownerType, ownerID).

func (*Store) CountWorkflowsByOwner 

func (s *Store) CountWorkflowsByOwner(ctx context.Context, ownerType, ownerID string) (int, error)

CountWorkflowsByOwner returns the number of workflows owned by (ownerType, ownerID). Used for quota enforcement (workflows.maxPerUser / workflows.maxPerOrg).

func (*Store) CreateNodeRun 

func (s *Store) CreateNodeRun(ctx context.Context, row *WorkflowNodeRunRow) error

CreateNodeRun inserts a workflow_node_runs row.

func (*Store) CreateTrigger 

func (s *Store) CreateTrigger(ctx context.Context, row *TriggerRow) error

func (*Store) CreateTriggerFire 

func (s *Store) CreateTriggerFire(ctx context.Context, row *TriggerFireRow) error

CreateTriggerFire inserts a trigger_fires row.

func (*Store) CreateWebhook 

func (s *Store) CreateWebhook(ctx context.Context, row *WebhookRow) error

CreateWebhook inserts a row into webhooks. The caller supplies a pre-encrypted SecretCipher (KEK for org scope, session-DEK for user scope — exact mcp_servers crypto envelope pattern). trigger_id has a 1:1 UNIQUE constraint.

func (*Store) CreateWorkflow 

func (s *Store) CreateWorkflow(ctx context.Context, row *WorkflowRow) error

CreateWorkflow inserts a row into workflows. The caller supplies a pre-generated UUID. spec_json must already be parsed + validated (US-64.4).

func (*Store) CreateWorkflowRun 

func (s *Store) CreateWorkflowRun(ctx context.Context, row *WorkflowRunRow) error

CreateWorkflowRun inserts a new run row. The single-in-flight partial unique index (uq_workflow_run_single_inflight) may reject the insert with a unique violation — mapped to ErrConcurrentRun so the handler can return 409 + Retry-After. Callers should use CreateWorkflowRunTx (with the fire row) for webhook-triggered runs; this method is for manual runs.

func (*Store) CreateWorkflowRunWithFire 

func (s *Store) CreateWorkflowRunWithFire(ctx context.Context, fire *TriggerFireRow, run *WorkflowRunRow) error

CreateWorkflowRunWithFire atomically inserts a trigger_fires row AND a workflow_runs row in a single transaction. If the run insert hits the single-in-flight unique violation, the whole tx rolls back (no orphan fire row claiming "fired" with no run). The caller then commits a separate trigger_fires row with status='skipped' + reason='already_running' and returns 409 + Retry-After.

This is the v1 correctness pattern for webhook fire-row + run-create atomicity (design Edge Case 4 / Webhook Security atomicity bullet).

func (*Store) DeleteTrigger 

func (s *Store) DeleteTrigger(ctx context.Context, ownerType, ownerID, triggerID string) error

DeleteTrigger deletes a trigger by ID scoped to (ownerType, ownerID). FK cascades handle webhooks + trigger_fires. Returns ErrNotFound.

func (*Store) DeleteWorkflow 

func (s *Store) DeleteWorkflow(ctx context.Context, ownerType, ownerID, workflowID string) error

DeleteWorkflow deletes a workflow by ID scoped to (ownerType, ownerID). FK cascades handle workflow_runs (ON DELETE CASCADE). Returns ErrNotFound if the workflow doesn't exist or the caller's scope doesn't match.

func (*Store) DisableTrigger 

func (s *Store) DisableTrigger(ctx context.Context, triggerID string) error

DisableTrigger sets enabled=false (called when the circuit breaker trips).

func (*Store) GetLastRoutineResult added in v0.10.0 

func (s *Store) GetLastRoutineResult(ctx context.Context, triggerID string) (json.RawMessage, error)

GetLastRoutineResult returns the result from the most recent successful routine fire for a trigger. Used for memory_mode: last_result.

func (*Store) GetRecentRoutineResults added in v0.10.0 

func (s *Store) GetRecentRoutineResults(ctx context.Context, triggerID string, limit int) ([]json.RawMessage, error)

GetRecentRoutineResults returns the N most recent successful routine results. Used for memory_max_runs > 1.

func (*Store) GetTrigger 

func (s *Store) GetTrigger(ctx context.Context, ownerType, ownerID, triggerID string) (*TriggerRow, error)

GetTrigger returns a single trigger by ID scoped to (ownerType, ownerID), or ErrNotFound.

func (*Store) GetTriggerByID added in v0.8.12 

func (s *Store) GetTriggerByID(ctx context.Context, triggerID string) (*TriggerRow, error)

func (*Store) GetWebhookByTriggerID 

func (s *Store) GetWebhookByTriggerID(ctx context.Context, triggerID string) (*WebhookRow, error)

GetWebhookByTriggerID returns the webhook config for a trigger, or ErrNotFound. Used by the webhook receiver to fetch the HMAC secret + IP allowlist.

func (*Store) GetWorkflow 

func (s *Store) GetWorkflow(ctx context.Context, ownerType, ownerID, workflowID string) (*WorkflowRow, error)

GetWorkflow returns a single workflow by ID scoped to (ownerType, ownerID), or ErrNotFound if not found / wrong scope.

func (*Store) GetWorkflowPolicy added in v0.9.1 

func (s *Store) GetWorkflowPolicy(ctx context.Context, workflowID string) (onMissing, ownerType, ownerID string, err error)

GetWorkflowPolicy returns the on_missing_workspace policy + owner for a workflow. Used by the engine to decide whether to create a workspace on missing (Epic 64).

func (*Store) GetWorkflowRun 

func (s *Store) GetWorkflowRun(ctx context.Context, runID string) (*WorkflowRunRow, error)

GetWorkflowRun returns a single run by ID, or ErrNotFound. Not scoped — run IDs are unguessable UUIDs and the caller has already authorized via the workflow's owner scope.

func (*Store) HasInFlightRun 

func (s *Store) HasInFlightRun(ctx context.Context, workflowID string) (bool, error)

HasInFlightRun reports whether a non-terminal run exists for the workflow. Used for fast-path rejection at the API layer (the partial unique index is the authoritative gate; this is an early-reject optimization).

func (*Store) IncrementTriggerFailures 

func (s *Store) IncrementTriggerFailures(ctx context.Context, triggerID string) (int, error)

IncrementTriggerFailures increments consecutive_failures and returns the new value. If the new value crosses auto_disable_after, the caller sets enabled=false.

func (*Store) ListNodeRuns 

func (s *Store) ListNodeRuns(ctx context.Context, workflowRunID string) ([]*WorkflowNodeRunRow, error)

ListNodeRuns returns all node runs for a workflow run, ordered by started_at ASC.

func (*Store) ListPendingRoutineFires added in v0.10.0 

func (s *Store) ListPendingRoutineFires(ctx context.Context, limit int) ([]*TriggerFireRow, error)

ListPendingRoutineFires returns fire rows with action_type='routine' and status='fired' that have not yet been processed (no result). Used by the scheduler to pick up webhook-triggered routines on its next tick.

func (*Store) ListSessionOrigins added in v0.11.0 

func (s *Store) ListSessionOrigins(ctx context.Context, workspaceID string) ([]*SessionOriginRow, error)

func (*Store) ListTriggerFires 

func (s *Store) ListTriggerFires(ctx context.Context, triggerID string, limit, offset int) ([]*TriggerFireRow, error)

ListTriggerFires returns recent fires for a trigger, paginated by fired_at DESC.

func (*Store) ListTriggers 

func (s *Store) ListTriggers(ctx context.Context, ownerType, ownerID string) ([]*TriggerRow, error)

func (*Store) ListWorkflowRuns 

func (s *Store) ListWorkflowRuns(ctx context.Context, workflowID string, limit, offset int) ([]*WorkflowRunRow, error)

ListWorkflowRuns returns runs for a workflow, paginated by created_at DESC. limit must be > 0; offset is the number of rows to skip.

func (*Store) ListWorkflowRunsByWorkspace added in v0.11.0 

func (s *Store) ListWorkflowRunsByWorkspace(ctx context.Context, workspaceID string) ([]*WorkflowRunRow, error)

ListWorkflowRunsByWorkspace returns non-terminal runs for a workspace.

func (*Store) ListWorkflows 

func (s *Store) ListWorkflows(ctx context.Context, ownerType, ownerID string) ([]*WorkflowRow, error)

ListWorkflows returns all workflows owned by (ownerType, ownerID), ordered by created_at ASC. Never decrypts — display fields only.

func (*Store) RecordSessionOrigin added in v0.11.0 

func (s *Store) RecordSessionOrigin(ctx context.Context, row *SessionOriginRow) error

func (*Store) RecordWebhookDelivery 

func (s *Store) RecordWebhookDelivery(ctx context.Context, webhookID, dedupKey string) error

RecordWebhookDelivery inserts a dedup row. Returns ErrDedupConflict if a row with the same (webhook_id, dedup_key) already exists — the caller returns 200 "duplicate" in that case. The unique constraint name is webhook_deliveries_webhook_dedup_uniq.

func (*Store) ResetTriggerFailures 

func (s *Store) ResetTriggerFailures(ctx context.Context, triggerID string) error

ResetTriggerFailures resets consecutive_failures to 0 (called on first success).

func (*Store) UpdateNodeRunStatus 

func (s *Store) UpdateNodeRunStatus(ctx context.Context, nodeRunID, status string, output json.RawMessage, branch *string, errorCode *string, errMsg json.RawMessage) error

UpdateNodeRunStatus sets status + terminal fields. On a terminal transition, finished_at is set. branch is set only on condition nodes (matched edge handle).

func (*Store) UpdateRunWorkspace added in v0.9.1 

func (s *Store) UpdateRunWorkspace(ctx context.Context, runID, workspaceID string) error

UpdateRunWorkspace updates the workspace_id on a workflow run.

func (*Store) UpdateTrigger 

func (s *Store) UpdateTrigger(ctx context.Context, ownerType, ownerID, triggerID string, upd *TriggerUpdate) (*TriggerRow, error)

UpdateTrigger updates an existing trigger scoped to (ownerType, ownerID). source_type is NOT mutable (not in TriggerUpdate). Partial update via pointer fields: nil preserves existing values. Returns the updated row or ErrNotFound.

func (*Store) UpdateTriggerFireResult added in v0.10.0 

func (s *Store) UpdateTriggerFireResult(ctx context.Context, fireID string, result json.RawMessage, status string) error

UpdateTriggerFireResult sets the result + final status on a trigger fire row. Used by the routine executor after agent execution completes.

func (*Store) UpdateWebhookSecret added in v0.9.1 

func (s *Store) UpdateWebhookSecret(ctx context.Context, triggerID string, secretCipher []byte, keyVersion int) error

UpdateWebhookSecret replaces the encrypted HMAC secret + key version for a webhook. Used by the rotate-secret endpoint.

func (*Store) UpdateWorkflow 

func (s *Store) UpdateWorkflow(ctx context.Context, ownerType, ownerID, workflowID string, upd *WorkflowUpdate) (*WorkflowRow, error)

UpdateWorkflow updates an existing workflow scoped to (ownerType, ownerID). Partial update via WorkflowUpdate pointer fields: nil preserves existing values. Returns the updated row or ErrNotFound.

func (*Store) UpdateWorkflowRunStatus 

func (s *Store) UpdateWorkflowRunStatus(ctx context.Context, runID, status string, errorCode *string, errMsg json.RawMessage, output json.RawMessage) error

UpdateWorkflowRunStatus sets status + terminal fields. On a terminal transition (succeeded/failed/canceled/timed_out), finished_at is set. error_code + error are nullable (null on success).

func (*Store) WorkspaceExistsByID added in v0.34.7 

func (s *Store) WorkspaceExistsByID(ctx context.Context, workspaceID string) (bool, error)

WorkspaceExistsByID reports whether a workspace row with the id exists, UNSCOPED — the existence semantics of the workspaces(id) foreign keys (triggers.workspace_id, workflows.target_workspace_id). The handler parent-id contract checks ride this; ownership policy is NOT judged here (cross-owner targets remain the fire-time loud class, #1440). Same pool as the workflows tables — workspaces lives in the same database.

type TriggerFireRow 

type TriggerFireRow struct {
	ID            string
	TriggerID     string
	SourceType    string
	InputEnvelope json.RawMessage
	ActionType    string
	ActionResult  json.RawMessage
	Status        string
	FiredAt       time.Time
	CompletedAt   *time.Time
	// Result is the routine execution outcome (UpdateTriggerFireResult):
	// the captured agent output on success (captureMode full) or the
	// failure cause on error. Distinct from ActionResult (the
	// receiver/engine's pre-execution action payload). Exposing it closes
	// the routine-failure observability gap (#1441): failure causes were
	// written to a column no read path selected.
	Result json.RawMessage
}

TriggerFireRow is the DB row shape for trigger_fires.

type TriggerInputSpec added in v0.34.0 

type TriggerInputSpec struct {
	InputFrom string
	Input     json.RawMessage
}

TriggerInputSpec is the trigger-carried half of the resolution input: the mapping mode (InputFrom) and the optional static input document (Input). Zero values mean legacy envelope mode with no static input.

type TriggerRow 

type TriggerRow struct {
	ID                  string
	OwnerType           string
	OwnerID             string
	Name                string
	Description         string
	Enabled             bool
	SourceType          string
	SourceConfig        json.RawMessage
	WorkspaceID         *string
	WorkflowID          *string
	InputFrom           string // 0059: envelope (legacy default) | body | mapped
	Input               json.RawMessage
	Prompt              string
	Agent               string
	ScriptPath          string
	ScriptArgs          []string
	ScriptEnv           json.RawMessage
	MemoryMode          string
	MemoryMaxRuns       int
	CaptureMode         string
	PreserveSession     string
	ConsecutiveFailures int
	AutoDisableAfter    int
	LastFiredAt         *time.Time
	NextFireAt          *time.Time
	CreatedAt           time.Time
	UpdatedAt           time.Time
}

TriggerRow is the DB row shape for triggers.

type TriggerUpdate 

type TriggerUpdate struct {
	Name         *string
	Description  *string
	Enabled      *bool
	SourceConfig json.RawMessage
	WorkspaceID  *string
	WorkflowID   *string
	// InputFrom (0059): nil keeps the stored mode. Input follows the
	// jsonb CASE shape — a non-nil RawMessage (including the JSON
	// literal null, which lands as jsonb null = "no static input")
	// replaces the stored document; nil keeps it.
	InputFrom        *string
	Input            json.RawMessage
	Prompt           *string
	Agent            *string
	ScriptPath       *string
	ScriptArgs       []string
	ScriptEnv        json.RawMessage
	MemoryMode       *string
	MemoryMaxRuns    *int
	CaptureMode      *string
	PreserveSession  *string
	AutoDisableAfter *int
	// NextFireAt is set by the handler when a schedule change must take
	// effect immediately (#1410) — nil keeps the stored slot.
	NextFireAt *time.Time
}

TriggerUpdate carries only the fields a partial update may change. Pointer fields: nil means "keep existing". source_type is NOT in this struct — it's immutable after create (the source defines the trigger's identity).

type ValidationError 

type ValidationError struct {
	Code   string `json:"code"`
	NodeID string `json:"nodeId,omitempty"`
	Detail string `json:"detail,omitempty"`
}

ValidationError describes a single problem found during DAG validation.

func ValidateSpec 

func ValidateSpec(spec *Spec, predecessorSchemas map[string]json.RawMessage, defaults DefaultsBlock) []ValidationError

ValidateSpec validates a workflow DAG spec and applies defaults merging. It mutates the spec in place (defaults applied to nodes that omit maxAttempts/timeout) ONLY when validation passes without structural errors (dangling edges, duplicate IDs, cycles). Non-structural errors (unreachable nodes, missing branch edges) do NOT prevent defaults merging — the spec is still mutated. Callers should discard the spec if any errors are returned. predecessorSchemas maps node ID → that node's outputSchema (JSON Schema bytes), used for expr-lang type-checking condition expressions against the upstream node's output shape. nil/empty schemas skip type-checking for that predecessor.

Returns a slice of ValidationErrors (empty if valid).

func (ValidationError) Error 

func (e ValidationError) Error() string

type WebhookRow 

type WebhookRow struct {
	ID                string
	TriggerID         string
	SecretCipher      []byte
	KeyVersion        int
	AllowedIPs        []string
	IdempotencyMode   string
	IdempotencyHeader string
	CreatedAt         time.Time
}

WebhookRow is the DB row shape for webhooks. SecretCipher is the encrypted HMAC secret (crypto envelope); callers must decrypt before verifying signatures.

type WorkflowNodeRunRow 

type WorkflowNodeRunRow struct {
	ID            string
	WorkflowRunID string
	NodeID        string
	NodeType      string
	Status        string
	Attempt       int
	Input         json.RawMessage
	Output        json.RawMessage
	Branch        *string
	ErrorCode     *string
	Error         json.RawMessage
	StartedAt     time.Time
	FinishedAt    *time.Time
}

WorkflowNodeRunRow is the DB row shape for workflow_node_runs.

type WorkflowRow 

type WorkflowRow struct {
	ID                 string
	OwnerType          string
	OwnerID            string
	Name               string
	Slug               string
	Description        string
	SpecYAML           string
	SpecJSON           json.RawMessage
	InputSchema        json.RawMessage
	TargetWorkspaceID  *string
	OnMissingWorkspace string
	Status             string
	Defaults           json.RawMessage
	CreatedAt          time.Time
	UpdatedAt          time.Time
}

WorkflowRow is the DB row shape for workflows.

type WorkflowRunRow 

type WorkflowRunRow struct {
	ID            string
	WorkflowID    string
	SpecSnapshot  json.RawMessage
	Input         json.RawMessage
	Output        json.RawMessage
	Status        string
	ErrorCode     *string
	Error         json.RawMessage
	TriggerID     *string
	TriggerFireID *string
	WorkspaceID   string
	StartedAt     *time.Time
	FinishedAt    *time.Time
	CreatedAt     time.Time
	UpdatedAt     time.Time
}

WorkflowRunRow is the DB row shape for workflow_runs.

type WorkflowUpdate 

type WorkflowUpdate struct {
	Name               *string
	Slug               *string
	Description        *string
	SpecYAML           *string
	SpecJSON           json.RawMessage
	InputSchema        json.RawMessage
	TargetWorkspaceID  *string
	OnMissingWorkspace *string
	Status             *string
	Defaults           json.RawMessage
}

WorkflowUpdate carries only the fields a partial update may change. Pointer fields: nil means "keep existing". This mirrors the API DTO pattern (UpdateWorkflowRequest) so the handler can pass fields through directly. The zero-value WorkflowUpdate preserves every field (no changes).

Source Files

View all Source files

Directories

Path Synopsis
Package exprlang compiles condition expressions for workflow DAG validation.
 Package exprlang compiles condition expressions for workflow DAG validation.
Package scriptwrap executes user-authored inline script handlers (Python, Node) inside the workspace sandbox.
 Package scriptwrap executes user-authored inline script handlers (Python, Node) inside the workspace sandbox.

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.