Class Activity

Base class for all HotMesh activity types. Activities are the execution units within a YAML-defined workflow graph. Each activity represents a node in a Directed Acyclic Graph (DAG) that the engine orchestrates.

Activity Categories

Activities fall into three execution categories:

  • Category A (Duplex): Two-phase activities with Leg 1 (dispatch) and Leg 2 (response). Used by Worker and Await. Leg 1 publishes a message and waits; Leg 2 handles the response via processEvent and transitions to adjacent activities.

  • Category B (Leg1-only with children): Single-phase activities that execute work and transition to children using the crash-safe executeLeg1StepProtocol. Used by Hook (passthrough mode), Signal, and Interrupt (target mode).

  • Category C (Leg1-only, no children): Terminal activities that execute without spawning children. Used by Interrupt (self mode).

Shared YAML Configuration

All activity types support these base properties in the YAML descriptor:

Property Type Description
type string Activity type: trigger, worker, await, hook, signal, interrupt, cycle
title string Human-readable label for the activity
input.schema object JSON Schema for input validation
input.maps object Maps data from other activities into this activity's input
output.schema object JSON Schema for output validation
output.maps object Maps/transforms the activity's own output data
job.maps object Maps activity data to the shared job state
emit boolean If true, emits a message to the graph's publishes topic
persist boolean If true, emits the job-completed event while keeping the job active
expire number Seconds until the job expires after completion (-1 = forever)
statusThreshold number Custom semaphore threshold for Dynamic Activation Control
cycle boolean If true, leaves Leg 2 open so the activity can be re-entered

Data Mapping Syntax

Mapping expressions use curly-brace references to bind data between activities and the shared job state:

input:
  maps:
    x: '{t1.output.data.fieldName}'      # reference another activity's output
    y: '{$self.output.data.fieldName}'    # reference own output
    z: '{$job.data.fieldName}'            # reference shared job state
    s: '{$app.settings.configKey}'        # reference app-level settings

See

Hierarchy (view full)

Constructors

constructor

Properties

adjacencyList adjacentIndex code config context data engine guidLedger hook leg logger metadata status

Methods

assertGenerationalId authorizeEntry bindActivityData bindActivityError bindActivityMetadata bindActivityMetadataPaths bindActivityState bindDimensionalAddress bindJobError bindJobMetadata bindJobMetadataPaths bindJobState executeLeg1StepProtocol executeStepProtocol filterAdjacent getJobStatus getState getTriggerConfig initDimensionalAddress initPolicies initSelf isJobComplete jobWasInterrupted mapInputData mapJobData mapOutputData mapStatusThreshold processEvent registerTimeout resolveAdjacentDad resolveDad resolveStatus resolveThresholdHit setLeg setState setStatus shouldEmit verifyEntry verifyLeg1Entry verifyReentry

Constructors

constructor

Properties

adjacencyList

adjacencyList: StreamData[]

adjacentIndex

adjacentIndex: number = 0

code

code: number = 200

config

config: ActivityType

context

context: JobState

data

data: ActivityData

engine

engine: EngineService

guidLedger

guidLedger: number = 0

hook

hook: ActivityData

leg

leg: ActivityLeg

logger

logger: ILogger

metadata

metadata: ActivityMetadata

status

status: StreamStatus = StreamStatus.SUCCESS

Methods

assertGenerationalId

  • assertGenerationalId(jobGID, msgGID?): void

  • if the job is created/deleted/created with the same key, the 'gid' ensures no stale messages (such as sleep delays) enter the workstream. Any message with a mismatched gid belongs to a prior job and can safely be ignored/dropped.

    Parameters

    • jobGID: string
    • OptionalmsgGID: string

    Returns void

authorizeEntry

bindActivityData

bindActivityError

  • bindActivityError(data): void

  • Any StreamMessage with a status of ERROR is bound to the activity

    Parameters

    • data: Record<string, unknown>

    Returns void

bindActivityMetadata

bindActivityMetadataPaths

bindActivityState

bindDimensionalAddress

bindJobError

  • bindJobError(data): void

  • unhandled activity errors (activities that return an ERROR StreamMessage status and have no adjacent children to transition to) are bound to the job

    Parameters

    • data: Record<string, unknown>

    Returns void

bindJobMetadata

bindJobMetadataPaths

bindJobState

executeLeg1StepProtocol

  • executeLeg1StepProtocol(delta): Promise<boolean>

  • Executes the 3-step Leg1 protocol for Category B activities (Leg1-only with children, e.g., Hook passthrough, Signal, Interrupt-another). Uses the incoming Leg1 message GUID as the GUID ledger key.

    Step A: setState + notarizeLeg1Completion + step1 markers (transaction 1) Step B: publish children + step2 markers + setStatusAndCollateGuid (transaction 2) Step C: if edge → runJobCompletionTasks + step3 markers + finalize (transaction 3)

    Parameters

    • delta: number

    Returns Promise<boolean>

    true if this transition caused the job to complete

executeStepProtocol

  • executeStepProtocol(delta, shouldFinalize): Promise<boolean>

  • Executes the 3-step Leg2 protocol using GUID ledger for crash-safe resume. Each step bundles durable writes with its concluding digit update in a single transaction.

    Parameters

    • delta: number
    • shouldFinalize: boolean

    Returns Promise<boolean>

    true if this transition caused the job to complete

filterAdjacent

getJobStatus

getState

getTriggerConfig

initDimensionalAddress

initPolicies

initSelf

isJobComplete

jobWasInterrupted

  • jobWasInterrupted(jobStatus): boolean

  • A job with a vale < -100_000_000 is considered interrupted, as the interruption event decrements the job status by 1billion.

    Parameters

    • jobStatus: number

    Returns boolean

mapInputData

mapJobData

mapOutputData

mapStatusThreshold

  • mapStatusThreshold(): number

  • A job is assumed to be complete when its status (a semaphore) reaches 0. A different threshold can be set in the activity YAML, in support of Dynamic Activation Control.

    Returns number

processEvent

registerTimeout

resolveAdjacentDad

resolveDad

resolveStatus

resolveThresholdHit

setLeg

setState

setStatus

shouldEmit

verifyEntry

verifyLeg1Entry

  • verifyLeg1Entry(): Promise<boolean>

  • Leg1 entry verification for Category B activities (Leg1-only with children). Returns true if this is a resume (Leg1 already completed on a prior attempt). On resume, loads the GUID ledger for step-level resume decisions.

    Returns Promise<boolean>

verifyReentry

  • verifyReentry(): Promise<number>

  • Upon entering leg 2 of a duplexed activity. Increments both the activity ledger (+1) and GUID ledger (+1). Stores the GUID ledger value for step-level resume decisions.

    Returns Promise<number>

Settings

Member Visibility

On This Page

Activity CategoriesShared YAML ConfigurationData Mapping Syntax
Constructors
Properties
Methods