Class WorkflowService

Checks if an error is a HotMesh reserved error type that indicates a HotMesh interruption rather than a true error condition.

When this returns true, you can safely return rethrow the error. The workflow engine will handle the interruption automatically.

Example

import { MemFlow } from '@hotmeshio/hotmesh';

try {
  await someWorkflowOperation();
} catch (error) {
  // Check if this is a HotMesh interruption
  if (MemFlow.workflow.didInterrupt(error)) {
    // Rethrow the error
    throw error;
  }
  // Handle actual error
  console.error('Workflow failed:', error);
}
Copy

Emits events to the event bus provider. Topics are prefixed with the quorum namespace.

Adds custom user data to the backend workflow record (writes to HASH). Runs exactly once during workflow execution.

Returns an entity session handle for interacting with the workflow's JSONB entity storage.

Example

const entity = await workflow.entity();
await entity.set({ user: { id: 123 } });
await entity.merge({ user: { name: "John" } });
const user = await entity.get("user");
Copy

Spawns a child workflow and awaits the result, or if await is false, returns immediately.

Executes a hook function and awaits the signal response. This is a convenience method that combines hook() and waitFor() operations.

Signal Injection: The signalId is automatically injected as the LAST argument to the hooked function. The hooked function should check for this signal parameter and emit the signal when processing is complete.

This behaves like execChild but targets the existing workflow instead of spawning a new workflow.

Example

// Execute a hook and await its signal response
const result = await MemFlow.workflow.execHook({
  taskQueue: 'processing',
  workflowName: 'processData',
  args: ['user123', 'batch-process'],
  signalId: 'processing-complete'
});

// The hooked function receives the signal as the last argument:
export async function processData(userId: string, processType: string, signalInfo?: { signal: string }) {
  // ... do processing work ...
  const result = { userId, processType, status: 'completed' };

  // Check if called via execHook (signalInfo will be present)
  if (signalInfo?.signal) {
    await MemFlow.workflow.signal(signalInfo.signal, result);
  }

  return result;
}
Copy

Example

// Alternative pattern - check if last arg is signal object
export async function myHookFunction(arg1: string, arg2: number, ...rest: any[]) {
  // ... process arg1 and arg2 ...
  const result = { processed: true, data: [arg1, arg2] };

  // Check if last argument is a signal object
  const lastArg = rest[rest.length - 1];
  if (lastArg && typeof lastArg === 'object' && lastArg.signal) {
    await MemFlow.workflow.signal(lastArg.signal, result);
  }

  return result;
}
Copy

Executes multiple hooks in parallel and awaits all their signal responses. This solves the race condition where Promise.all() with execHook() would prevent all waitFor() registrations from completing.

The method ensures all waitFor() registrations happen before any hooks execute, preventing signals from being sent before the framework is ready to receive them.

Example

// Execute multiple research perspectives in parallel
const results = await MemFlow.workflow.execHookBatch<{
  optimistic: OptimisticResult;
  skeptical: SkepticalResult;
}>([
  {
    key: 'optimistic',
    options: {
      taskQueue: 'agents',
      workflowName: 'optimisticPerspective',
      args: [query],
      signalId: 'optimistic-complete'
    }
  },
  {
    key: 'skeptical', 
    options: {
      taskQueue: 'agents',
      workflowName: 'skepticalPerspective',
      args: [query],
      signalId: 'skeptical-complete'
    }
  }
]);

// results.optimistic contains the OptimisticResult
// results.skeptical contains the SkepticalResult
Copy

Spawns a child workflow and awaits the result, or if await is false, returns immediately.

Returns the current workflow context, restored from storage.

Spawns a hook from the main thread or a hook thread. If entity/workflowName are not provided, defaults to the current workflow.

Interrupts a running job by sending an interruption request.

Provides a proxy for defined activities, ensuring deterministic replay and retry.

Returns a deterministic random number between 0 and 1. The number is derived from the current execution index, ensuring deterministic replay.

Returns a search session handle for interacting with the workflow's HASH storage.

Sends a signal payload to any paused workflow thread awaiting this signal. This method is commonly used to coordinate between workflows, hook functions, and external events.

Example

// Basic usage - send a simple signal with data
await MemFlow.workflow.signal('signal-id', { name: 'WarmMash' });
Copy

Example

// Hook function signaling completion
export async function exampleHook(name: string): Promise<void> {
  const result = await processData(name);
  await MemFlow.workflow.signal('hook-complete', { data: result });
}
Copy

Example

// Signal with complex data structure
await MemFlow.workflow.signal('process-complete', {
  status: 'success',
  data: { id: 123, name: 'test' },
  timestamp: new Date().toISOString()
});
Copy

Sleeps the workflow for a specified duration, deterministically. On replay, it will not actually sleep again, but resume after sleep.

Example

// Basic usage - sleep for a specific duration
await MemFlow.workflow.sleepFor('2 seconds');
Copy

Example

// Using with Promise.all for parallel operations
const [greeting, timeInSeconds] = await Promise.all([
  someActivity(name),
  MemFlow.workflow.sleepFor('1 second')
]);
Copy

Example

// Multiple sequential sleeps
await MemFlow.workflow.sleepFor('1 seconds');  // First pause
await MemFlow.workflow.sleepFor('2 seconds');  // Second pause
Copy

Spawns a child workflow and returns the child Job ID without awaiting its completion.

Executes a distributed trace, outputting the provided attributes to the telemetry sink (e.g. OpenTelemetry).

This trace will only run once per workflow execution by default.

Pauses the workflow until a signal with the given signalId is received. This method is commonly used to coordinate between the main workflow and hook functions, or to wait for external events.

Example

// Basic usage - wait for a single signal
const payload = await MemFlow.workflow.waitFor<PayloadType>('abcdefg');
Copy

Example

// Wait for multiple signals in parallel
const [signal1, signal2] = await Promise.all([
  MemFlow.workflow.waitFor<Record<string, any>>('signal1'),
  MemFlow.workflow.waitFor<Record<string, any>>('signal2')
]);
Copy

Example

// Typical pattern with hook functions
// In main workflow:
await MemFlow.workflow.waitFor<ResponseType>('hook-complete');

// In hook function:
await MemFlow.workflow.signal('hook-complete', { data: result });
Copy