Effect Integration

The Effect integration layer is Smithers’ third and lowest API tier. The TOON builder API and JSX mirror it, and most workflows never need to reach this far. You reach for it when you need direct control over execution boundaries, custom bridging logic, or the full expressiveness of the Effect type system against Smithers internals. All modules live under src/effect/ and are imported from smithers-orchestrator/effect/*. They assume familiarity with the Effect library.

Runners

Three functions run Effect programs inside the shared Smithers managed runtime. The runtime is initialized once, annotates all logs with "service": "smithers", and normalizes failures to SmithersError.

EFFECT_RUN_PROMISE

Runs an Effect and returns a Promise. The promise rejects with a SmithersError on failure.

import { runPromise } from "smithers-orchestrator/effect/runtime";
import { Effect } from "effect";

const result = await runPromise(
  Effect.gen(function* () {
    yield* Effect.log("starting");
    return 42;
  }),
  { signal: abortController.signal },
);

function runPromise<A, E, R>(
  effect: Effect.Effect<A, E, R>,
  options?: { signal?: AbortSignal },
): Promise<A>

EFFECT_RUN_SYNC

Runs a synchronous Effect immediately. Throws SmithersError on failure. Use this only when you are certain the Effect performs no async work.

import { runSync } from "smithers-orchestrator/effect/runtime";

const value = runSync(Effect.succeed("hello"));

function runSync<A, E, R>(
  effect: Effect.Effect<A, E, R>,
): A

EFFECT_RUN_FORK

Forks an Effect as a background fiber. Returns the fiber immediately without awaiting the result. Suitable for fire-and-forget side effects like metric updates.

import { runFork } from "smithers-orchestrator/effect/runtime";
import { Metric } from "effect";
import { runsTotal } from "smithers-orchestrator/effect/metrics";

runFork(Metric.increment(runsTotal));

function runFork<A, E, R>(
  effect: Effect.Effect<A, E, R>,
): Fiber<A, E>

EFFECT_SINGLE_RUNNER

The EFFECT_SINGLE_RUNNER pattern provides a singleton @effect/cluster SingleRunner backed by an in-memory SQLite database. It manages the task worker entity lifecycle, serializes dispatches by bridge key, and survives multiple concurrent callers sharing the same runtime. The singleton is initialized lazily on first dispatch and reused for the lifetime of the process.

import {
  dispatchWorkerTask,
  subscribeTaskWorkerDispatches,
} from "smithers-orchestrator/effect/single-runner";

// Dispatch a registered worker task
const result = await dispatchWorkerTask(task, async () => {
  await doWork();
  return { terminal: true };
});

// Observe dispatches (useful for testing or TUI)
const unsubscribe = subscribeTaskWorkerDispatches((task) => {
  console.log("dispatched", task.executionId);
});

function dispatchWorkerTask(
  task: WorkerTask,
  execute: () => Promise<{ terminal: boolean }>,
): Promise<{ terminal: boolean }>

function subscribeTaskWorkerDispatches(
  subscriber: (task: WorkerTask) => void,
): () => void

EFFECT_TASK_RUNTIME

Task-scoped context propagated via AsyncLocalStorage. Tools and compute callbacks read this to find the current runId, stepId, heartbeat handle, and abort signal without threading the values through call stacks.

import {
  getTaskRuntime,
  requireTaskRuntime,
  withTaskRuntime,
  SmithersTaskRuntime,
} from "smithers-orchestrator/effect/task-runtime";

// Inside a compute callback — access the current runtime
const rt = requireTaskRuntime();
rt.heartbeat({ progress: 0.5 });

// Establish a new runtime scope (used internally by the engine)
const result = withTaskRuntime(
  { runId, stepId, attempt, iteration, signal, db, heartbeat, lastHeartbeat: null },
  () => desc.computeFn!(),
);

type SmithersTaskRuntime = {
  runId: string;
  stepId: string;
  attempt: number;
  iteration: number;
  signal: AbortSignal;
  db: any;
  heartbeat: (data?: unknown) => void;
  lastHeartbeat: unknown | null;
};

function withTaskRuntime<T>(runtime: SmithersTaskRuntime, execute: () => T): T
function getTaskRuntime(): SmithersTaskRuntime | undefined
function requireTaskRuntime(): SmithersTaskRuntime

requireTaskRuntime throws SmithersError("TASK_RUNTIME_UNAVAILABLE") when called outside a task execution scope.

Bridges

Bridges connect the Smithers engine to Effect programs. The engine dispatches execution to a bridge; the bridge translates that into the correct Effect constructs and reports results back.

EFFECT_ACTIVITY_BRIDGE

The activity bridge wraps legacy task execution in an @effect/workflow Activity with idempotency and retry semantics. Each task maps to a SmithersTaskBridge workflow instance keyed by adapter namespace, workflow name, run ID, node ID, and iteration.

import {
  executeTaskActivity,
  makeTaskActivity,
  makeTaskBridgeKey,
  RetriableTaskFailure,
} from "smithers-orchestrator/effect/activity-bridge";

const result = await executeTaskActivity(
  adapter,
  "my-workflow",
  runId,
  desc,
  async (context) => {
    // context.attempt — current attempt number (1-based)
    // context.idempotencyKey — stable key for this attempt
    return computeResult();
  },
  {
    initialAttempt: 1,
    retry: { times: 3, while: (e) => e instanceof RetriableTaskFailure },
  },
);

type TaskActivityContext = {
  attempt: number;
  idempotencyKey: string;
};

type ExecuteTaskActivityOptions = {
  initialAttempt?: number;
  retry?: false | { times: number; while?: (error: unknown) => boolean };
  includeAttemptInIdempotencyKey?: boolean;
};

function executeTaskActivity<A>(
  adapter: SmithersDb,
  workflowName: string,
  runId: string,
  desc: TaskDescriptor,
  executeFn: (context: TaskActivityContext) => Promise<A> | A,
  options?: ExecuteTaskActivityOptions,
): Promise<A>

RetriableTaskFailure is a sentinel error class. Throw it from an activity to trigger a retry within the bridge’s retry loop.

EFFECT_WORKFLOW_BRIDGE

The workflow bridge is the top-level seam that routes a task to the appropriate execution path: compute, static, or legacy. It manages inflight and completed execution maps to prevent duplicate dispatches across concurrent engine invocations.

import { executeTaskBridge } from "smithers-orchestrator/effect/workflow-bridge";

await executeTaskBridge(
  adapter, db, runId, desc, descriptorMap,
  inputTable, eventBus, toolConfig, "my-workflow",
  cacheEnabled, signal, disabledAgents,
  runAbortController, hijackState,
  legacyExecuteTaskFn,
);

The bridge classifies each task before dispatch:

Classification	Condition
`"compute"`	`desc.computeFn` set, no agent, no cache, no worktree, no scorers
`"static"`	`desc.staticPayload` set, no agent, no cache, no worktree, no scorers
`"legacy"`	Everything else — forwarded to `legacyExecuteTaskFn`

EFFECT_WORKFLOW_MAKE_BRIDGE

Wraps an entire workflow body execution in an @effect/workflow Workflow using AsyncLocalStorage to thread the bridge runtime through the call stack. Used for child workflow execution and continue-as-new semantics.

import {
  runWorkflowWithMakeBridge,
  withWorkflowMakeBridgeRuntime,
  getWorkflowMakeBridgeRuntime,
} from "smithers-orchestrator/effect/workflow-make-bridge";

const result = await runWorkflowWithMakeBridge(
  workflow,
  { runId, input: { repo: "acme/core" }, resume: false },
  (wf, opts) => engine.run(wf, opts),
);

The bridge handles continue-as-new by looping internally until the run settles at a terminal or suspending status. Child workflows registered within the same scope share the parent’s Scope and engine context.

function runWorkflowWithMakeBridge<Schema>(
  workflow: SmithersWorkflow<Schema>,
  opts: RunOptions & { runId: string },
  executeBody: (workflow: SmithersWorkflow<Schema>, opts: RunOptions) => Promise<RunResult>,
): Promise<RunResult>

function withWorkflowMakeBridgeRuntime<T>(
  runtime: WorkflowMakeBridgeRuntime,
  execute: () => T,
): T

function getWorkflowMakeBridgeRuntime(): WorkflowMakeBridgeRuntime | undefined

EFFECT_COMPUTE_TASK_BRIDGE

Executes a computeFn task directly within the bridge — no agent involved. Manages the full attempt lifecycle: DB insert, node state transitions, heartbeat flushing, timeout enforcement, and event emission. Integrates with the heartbeat watchdog when desc.heartbeatTimeoutMs is set.

import { executeComputeTaskBridge } from "smithers-orchestrator/effect/compute-task-bridge";

await executeComputeTaskBridge(
  adapter, db, runId, desc, eventBus,
  { rootDir: "/workspace" },
  "my-workflow",
  signal,
);

Eligibility check:

import { canExecuteBridgeManagedComputeTask } from "smithers-orchestrator/effect/compute-task-bridge";

const eligible = canExecuteBridgeManagedComputeTask(desc, cacheEnabled);
// true when: desc.computeFn set, no agent, no cache, no worktree, no scorers

EFFECT_STATIC_TASK_BRIDGE

Executes a staticPayload task without invoking any agent or compute function. The payload is validated against the output schema and persisted immediately.

import { executeStaticTaskBridge } from "smithers-orchestrator/effect/static-task-bridge";

await executeStaticTaskBridge(
  adapter, runId, desc, eventBus,
  { rootDir: "/workspace" },
  "my-workflow",
  signal,
);

Eligibility check:

import { canExecuteBridgeManagedStaticTask } from "smithers-orchestrator/effect/static-task-bridge";

const eligible = canExecuteBridgeManagedStaticTask(desc, cacheEnabled);
// true when: desc.staticPayload set, no agent, no cache, no worktree, no scorers

EFFECT_DEFERRED_BRIDGE

Lightweight in-memory deferred resolution map. Stores Exit values keyed by (runId, nodeId, iteration) and reads them back during replay or resume. Used for simple approval and timer synchronization that does not need durable persistence.

import {
  makeApprovalDeferred,
  makeTimerDeferred,
  makeDeferredBridgeKey,
  bridgeApprovalResolve,
  bridgeTimerResolve,
  getDeferredResolution,
} from "smithers-orchestrator/effect/deferred-bridge";

// Resolve an approval decision
bridgeApprovalResolve(runId, nodeId, iteration, { approved: true });

// Retrieve the stored resolution
const exit = getDeferredResolution(runId, nodeId, iteration);

function makeDeferredBridgeKey(runId: string, nodeId: string, iteration: number): string
function bridgeApprovalResolve(runId: string, nodeId: string, iteration: number, decision: { approved: boolean }): void
function bridgeTimerResolve(runId: string, nodeId: string, iteration: number): void
function getDeferredResolution(runId: string, nodeId: string, iteration: number): Exit.Exit<...> | undefined

EFFECT_DURABLE_DEFERRED_BRIDGE

Durable version of the deferred bridge built on @effect/workflow DurableDeferred. Approval and WaitForEvent nodes use this so that the resolution is durable across restarts. Each deferral is keyed by an execution ID derived from the adapter namespace, run ID, node ID, and iteration.

import {
  awaitApprovalDurableDeferred,
  awaitWaitForEventDurableDeferred,
  bridgeApprovalResolve,
  bridgeWaitForEventResolve,
  bridgeSignalResolve,
  makeDurableDeferredBridgeExecutionId,
} from "smithers-orchestrator/effect/durable-deferred-bridge";

// Wait for an approval decision (called from within a task execution)
const resolution = await awaitApprovalDurableDeferred(adapter, runId, nodeId, iteration);
// resolution.approved, .note, .decidedBy, .decisionJson, .autoApproved

// Resolve it from outside (e.g. from the gateway or HTTP handler)
await bridgeApprovalResolve(adapter, runId, nodeId, iteration, {
  approved: true,
  note: "looks good",
  decidedBy: "alice",
});

// Resolve a WaitForEvent node when a signal arrives
await bridgeWaitForEventResolve(adapter, runId, nodeId, iteration, {
  signalName: "payment.received",
  correlationId: "order-42",
  payloadJson: JSON.stringify({ amount: 100 }),
  seq: 1,
  receivedAtMs: Date.now(),
});

// Resolve all matching WaitForEvent nodes in a run
await bridgeSignalResolve(adapter, runId, {
  signalName: "payment.received",
  correlationId: "order-42",
  payloadJson: JSON.stringify({ amount: 100 }),
  seq: 1,
  receivedAtMs: Date.now(),
});

Resolution schemas:

type ApprovalDurableDeferredResolution = {
  approved: boolean;
  note: string | null;
  decidedBy: string | null;
  decisionJson: string | null;
  autoApproved: boolean;
};

type WaitForEventDurableDeferredResolution = {
  signalName: string;
  correlationId: string | null;
  payloadJson: string;
  seq: number;
  receivedAtMs: number;
};

EFFECT_DEFERRED_STATE_BRIDGE

Manages the state machine for timer, approval, and WaitForEvent nodes. Reads attempt metadata from the database, determines whether a deferred task is still pending or has already been resolved, and drives the appropriate resolution path. Key exports:

import {
  resolveDeferredTaskStateBridge,
  isBridgeManagedTimerTask,
  isBridgeManagedWaitForEventTask,
  cancelPendingTimersBridge,
} from "smithers-orchestrator/effect/deferred-state-bridge";

// Check if a task is bridge-managed
if (isBridgeManagedTimerTask(desc)) { ... }
if (isBridgeManagedWaitForEventTask(desc)) { ... }

// Cancel all pending timers for a run
await cancelPendingTimersBridge(adapter, runId, eventBus);

The resolveDeferredTaskStateBridge function drives the resolution loop. It reads the current attempt snapshot, handles timer expiry, matches incoming signals, and emits the appropriate SmithersEvent when the node settles.

EFFECT_CHILD_WORKFLOW_EXECUTION

Child workflow execution is provided by the WorkflowMakeBridgeRuntime context, accessible from within an active workflow body via getWorkflowMakeBridgeRuntime().

import { getWorkflowMakeBridgeRuntime } from "smithers-orchestrator/effect/workflow-make-bridge";

const bridgeRuntime = getWorkflowMakeBridgeRuntime();
if (bridgeRuntime) {
  const childResult = await bridgeRuntime.executeChildWorkflow(childWorkflow, {
    runId: generateRunId(),
    input: { ...childInput },
  });
}

The child workflow is registered as its own Workflow in the shared engine context and executed under the parent’s Scope. Its continue-as-new loop runs independently but shares the parent engine scope’s lifecycle.

EFFECT_WORKER_ENTITY_DISPATCH

Task dispatches pass through an @effect/cluster Entity (the TaskWorkerEntity). The entity is defined using @effect/rpc and sharded via the SingleRunner. Each invocation is addressed by its bridgeKey — a composite of adapter namespace, workflow name, run ID, node ID, and iteration.

import {
  TaskWorkerEntity,
  WorkerTask,
  WorkerDispatchKind,
  makeWorkerTask,
} from "smithers-orchestrator/effect/entity-worker";

// Schema types for tasks
type WorkerTask = {
  executionId: string;
  bridgeKey: string;
  workflowName: string;
  runId: string;
  nodeId: string;
  iteration: number;
  retries: number;
  taskKind: "agent" | "compute" | "static";
  dispatchKind: "compute" | "static" | "legacy";
};

EFFECT_SANDBOX_ENTITY_TRANSPORT

Sandbox execution (Bubblewrap, Docker, Codeplane) is routed through an Entity transport. The SandboxEntity and SandboxEntityExecutor bridge the SandboxTransportService interface into the cluster entity model.

import {
  SandboxEntity,
  SandboxEntityExecutor,
  makeSandboxEntityId,
  makeSandboxTransportServiceEffect,
} from "smithers-orchestrator/effect/sandbox-entity";

// Build the Effect layer that provides SandboxTransportService
const transportLayer = makeSandboxTransportServiceEffect(executorLayer);

HTTP-backed executors:

import {
  CodeplaneSandboxExecutorLive,
  DockerSandboxExecutorLive,
  SandboxHttpRunner,
} from "smithers-orchestrator/effect/http-runner";

Socket-backed executor:

import {
  BubblewrapSandboxExecutorLive,
  SandboxSocketRunner,
} from "smithers-orchestrator/effect/socket-runner";

Infrastructure

EFFECT_CHILD_PROCESS

Wraps Node.js child_process.spawn in an Effect with full lifecycle management: output capture, truncation, idle timeout, total timeout, AbortSignal forwarding, and detached process group cleanup.

import { spawnCaptureEffect } from "smithers-orchestrator/effect/child-process";
import { runPromise } from "smithers-orchestrator/effect/runtime";

const result = await runPromise(
  spawnCaptureEffect("git", ["diff", "--stat"], {
    cwd: "/workspace",
    timeoutMs: 30_000,
    idleTimeoutMs: 10_000,
    maxOutputBytes: 200_000,
    onStdout: (chunk) => process.stdout.write(chunk),
  }),
);

// result.stdout, result.stderr, result.exitCode

type SpawnCaptureOptions = {
  cwd: string;
  env?: Record<string, string | undefined>;
  input?: string;
  signal?: AbortSignal;
  timeoutMs?: number;
  idleTimeoutMs?: number;
  maxOutputBytes?: number;       // default: 200_000 bytes
  detached?: boolean;
  onStdout?: (chunk: string) => void;
  onStderr?: (chunk: string) => void;
};

type SpawnCaptureResult = {
  stdout: string;
  stderr: string;
  exitCode: number | null;
};

function spawnCaptureEffect(
  command: string,
  args: string[],
  options: SpawnCaptureOptions,
): Effect.Effect<SpawnCaptureResult, SmithersError>

Output exceeding maxOutputBytes is truncated and a smithers.tool.output_truncated_total metric is incremented. The process group is killed with SIGKILL on abort or timeout.

EFFECT_INTEROP

Utilities for wrapping non-Effect code so it fits cleanly into Effect pipelines.

import {
  fromPromise,
  fromSync,
  ignoreSyncError,
  toError,
} from "smithers-orchestrator/effect/interop";

// Wrap a promise-returning function
const effect = fromPromise("fetch user", () => fetch("/api/user").then(r => r.json()));

// Wrap a synchronous function that may throw
const syncEffect = fromSync("parse JSON", () => JSON.parse(raw));

// Best-effort cleanup — swallows thrown errors
const cleanup = ignoreSyncError("close db", () => db.close());

type ErrorWrapOptions = {
  code?: SmithersErrorCode;
  details?: Record<string, unknown>;
};

function fromPromise<A>(
  label: string,
  evaluate: () => PromiseLike<A>,
  options?: ErrorWrapOptions,
): Effect.Effect<A, SmithersError>

function fromSync<A>(
  label: string,
  evaluate: () => A,
  options?: ErrorWrapOptions,
): Effect.Effect<A, SmithersError>

function ignoreSyncError(label: string, fn: () => void): Effect.Effect<void>

function toError(cause: unknown, label?: string, options?: ErrorWrapOptions): SmithersError

All failures are normalized to SmithersError. Pass code to set a specific SmithersErrorCode; without it the code defaults to "INTERNAL_ERROR".

EFFECT_SQL_MESSAGE_STORAGE

Provides a SQLite-backed implementation of @effect/workflow’s message storage interface. Used by the workflow engine to persist workflow state across process restarts.

import {
  SqlMessageStorage,
  ensureSqlMessageStorage,
  ensureSqlMessageStorageEffect,
  getSqlMessageStorage,
} from "smithers-orchestrator/effect/sql-message-storage";

// Ensure message storage is initialized for a given database
const storage = await ensureSqlMessageStorage(db);

// Effect version
const storageEffect = ensureSqlMessageStorageEffect(db);

The storage creates and manages the _smithers_runs, _smithers_nodes, and _smithers_attempts tables plus supporting indices. The getSqlMessageStorage(db) function returns an existing instance without creating one.

EFFECT_LOGGING

Thin wrappers around Effect.logDebug/Info/Warning/Error that fire-and-forget via runFork. Each accepts an optional annotations map and an optional log span name.

import {
  logDebug,
  logInfo,
  logWarning,
  logError,
} from "smithers-orchestrator/effect/logging";

logInfo("workflow started", { runId, workflowName: "deploy" }, "engine:run");

logError("task failed", { runId, nodeId, error: err.message }, "engine:task");

type LogAnnotations = Record<string, unknown> | undefined;

function logDebug(message: string, annotations?: LogAnnotations, span?: string): void
function logInfo(message: string, annotations?: LogAnnotations, span?: string): void
function logWarning(message: string, annotations?: LogAnnotations, span?: string): void
function logError(message: string, annotations?: LogAnnotations, span?: string): void

EFFECT_LOG_FORMATS_JSON_PRETTY_LOGFMT

Log format selection is controlled via SmithersObservabilityOptions.logFormat. The observability layer applies the chosen format to the shared runtime layer:

Format	Description
`"json"`	Structured JSON lines — suitable for log aggregation pipelines
`"pretty"`	Human-readable colored output for local development
`"logfmt"`	Key=value logfmt — compatible with Loki and similar systems

Configure via the observability API:

import { createSmithers } from "smithers-orchestrator";

const { smithers } = createSmithers({ ... });
// then in your run call:
await smithers.run(workflow, {
  input: { ... },
  observability: { logFormat: "json" },
});

EFFECT_METRICS

Pre-defined Effect Metric instances for every significant system boundary in Smithers. Import individual metrics and compose them with Metric.increment, Metric.update, Metric.set, or Metric.tagged. Counters (selection)

Name	Metric
`runsTotal`	`smithers.runs.total`
`nodesStarted` / `nodesFinished` / `nodesFailed`	`smithers.nodes.*`
`toolCallsTotal` / `toolCallErrorsTotal`	`smithers.tool_calls.*`
`approvalsRequested` / `approvalsGranted` / `approvalsDenied`	`smithers.approvals.*`
`tokensInputTotal` / `tokensOutputTotal`	`smithers.tokens.*`
`tokensContextWindowBucketTotal`	`smithers.tokens.context_window_bucket_total`
`runsFinishedTotal` / `runsFailedTotal` / `runsCancelledTotal`	`smithers.runs.*_total`
`errorsTotal`	`smithers.errors.total`
`sandboxCreatedTotal` / `sandboxCompletedTotal`	`smithers.sandbox.*`

Gauges (selection)

Name	Metric
`activeRuns` / `activeNodes`	`smithers.runs.active`, `smithers.nodes.active`
`schedulerQueueDepth`	`smithers.scheduler.queue_depth`
`approvalPending`	`smithers.approval.pending`
`timersPending`	`smithers.timers.pending`
`processMemoryRssBytes` / `processHeapUsedBytes`	`smithers.process.*`

Histograms (selection)

Name	Metric
`nodeDuration` / `attemptDuration`	`smithers.node.duration_ms`, `smithers.attempt.duration_ms`
`toolDuration`	`smithers.tool.duration_ms`
`runDuration`	`smithers.run.duration_ms`
`tokensInputPerCall` / `tokensOutputPerCall`	`smithers.tokens.*_per_call`
`tokensContextWindowPerCall`	`smithers.tokens.context_window_per_call`
`sandboxDurationMs` / `sandboxBundleSizeBytes`	`smithers.sandbox.*`
`heartbeatDataSizeBytes` / `heartbeatIntervalMs`	`smithers.heartbeats.*`

import { Effect, Metric } from "effect";
import {
  toolCallsTotal,
  toolDuration,
} from "smithers-orchestrator/effect/metrics";

// Record a tool call with tags
const record = Effect.all([
  Metric.increment(Metric.tagged(toolCallsTotal, "tool", "bash")),
  Metric.update(Metric.tagged(toolDuration, "tool", "bash"), durationMs),
], { discard: true });

trackEvent(event: SmithersEvent) is the high-level entry point — it maps every event type to the correct set of metric updates:

import { trackEvent } from "smithers-orchestrator/effect/metrics";

runFork(trackEvent({ type: "NodeStarted", runId, nodeId, ... }));

updateProcessMetrics() snapshots process.memoryUsage() and process uptime into gauges. Call it on a recurring interval from a background fiber.

EFFECT_DIFF_BUNDLE_COMPUTE

Produces a serializable diff bundle by comparing a git working tree against a base ref. Captures tracked file changes, binary files, and untracked files.

import { computeDiffBundle, DiffBundle, FilePatch } from "smithers-orchestrator/effect/diff-bundle";

const bundle = await computeDiffBundle("HEAD", "/workspace");
// bundle.seq       — sequence number (default: 1)
// bundle.baseRef   — the ref passed in ("HEAD")
// bundle.patches   — FilePatch[]

type FilePatch = {
  path: string;
  operation: "add" | "modify" | "delete";
  diff: string;
  binaryContent?: string;  // base64-encoded for binary files
};

type DiffBundle = {
  seq: number;
  baseRef: string;
  patches: FilePatch[];
};

function computeDiffBundle(
  baseRef: string,
  currentDir: string,
  seq?: number,
): Promise<DiffBundle>

EFFECT_DIFF_BUNDLE_APPLY

Applies a previously computed DiffBundle to a target directory. Attempts git apply first; falls back to per-patch application using the diff library for text patches and direct file writes for binary patches.

import { applyDiffBundle } from "smithers-orchestrator/effect/diff-bundle";

await applyDiffBundle(bundle, "/sandbox/workspace");

function applyDiffBundle(
  bundle: DiffBundle,
  targetDir: string,
): Promise<void>

The target directory is created recursively if it does not exist. Delete operations remove the target file silently.

EFFECT_SCHEDULER_WAKE_QUEUE

A lightweight notify/wait queue used to wake the scheduler when new work becomes available, avoiding busy polling.

import { createSchedulerWakeQueue } from "smithers-orchestrator/effect/workflow-make-bridge";

const queue = createSchedulerWakeQueue();

// Background: notify when new tasks are ready
queue.notify();

// Scheduler loop: wait for the next notification
await queue.wait();

type SchedulerWakeQueue = {
  notify(): void;
  wait(): Promise<void>;
};

function createSchedulerWakeQueue(): SchedulerWakeQueue

notify() resolves a pending wait() immediately, or increments an internal pending counter so the next wait() call returns without suspending. Multiple notify() calls before a wait() are coalesced into a single wakeup.

EFFECT_WORKFLOW_VERSIONING_RUNTIME

Manages workflow patch decisions for safe migration of long-running workflows. A patch decision is a boolean recorded against a string patchId. New runs see true; runs that pre-date the patch see false (as stored in the run config).

import {
  createWorkflowVersioningRuntime,
  withWorkflowVersioningRuntime,
  getWorkflowVersioningRuntime,
  usePatched,
  getWorkflowPatchDecisions,
} from "smithers-orchestrator/effect/versioning";

// Create a runtime for the current run
const versioningRuntime = createWorkflowVersioningRuntime({
  baseConfig: run.configJson ?? {},
  initialDecisions: getWorkflowPatchDecisions(run.configJson),
  isNewRun: !opts.resume,
  persist: async (config) => { await adapter.updateRunConfig(runId, config); },
});

// Wrap the workflow body execution
const result = withWorkflowVersioningRuntime(versioningRuntime, () =>
  engine.executeBody(workflow, opts),
);

// Flush decisions to the database after each render frame
await versioningRuntime.flush();

Inside a workflow (JSX or TOON), use the usePatched hook to branch on a patch:

import { usePatched } from "smithers-orchestrator/effect/versioning";

function MyWorkflow({ ctx }) {
  const hasNewRetryLogic = usePatched("2026-04-retry-overhaul");

  return (
    <Workflow name="deploy">
      <Task id="run" retries={hasNewRetryLogic ? 3 : 0}>
        Deploy
      </Task>
    </Workflow>
  );
}

type WorkflowVersioningRuntime = {
  resolve(patchId: string): boolean;
  flush(): Promise<void>;
  snapshot(): WorkflowPatchDecisions;
};

function createWorkflowVersioningRuntime(options: WorkflowVersioningRuntimeOptions): WorkflowVersioningRuntime
function withWorkflowVersioningRuntime<T>(runtime: WorkflowVersioningRuntime, execute: () => T): T
function getWorkflowVersioningRuntime(): WorkflowVersioningRuntime | undefined
function getWorkflowPatchDecisions(config: Record<string, unknown> | null | undefined): WorkflowPatchDecisions
function usePatched(patchId: string): boolean

usePatched calls resolve() on the ambient WorkflowVersioningRuntime. Outside a versioning scope it always returns false.

Getting Started

Build Workflows

Run and Operate

Core Concepts

Guides

Components

Integrations

Runtime API

Examples

Reference

Effect Integration

Runners

EFFECT_RUN_PROMISE

EFFECT_RUN_SYNC

EFFECT_RUN_FORK

EFFECT_SINGLE_RUNNER

EFFECT_TASK_RUNTIME

Bridges

EFFECT_ACTIVITY_BRIDGE

EFFECT_WORKFLOW_BRIDGE

EFFECT_WORKFLOW_MAKE_BRIDGE

EFFECT_COMPUTE_TASK_BRIDGE

EFFECT_STATIC_TASK_BRIDGE

EFFECT_DEFERRED_BRIDGE

EFFECT_DURABLE_DEFERRED_BRIDGE

EFFECT_DEFERRED_STATE_BRIDGE

EFFECT_CHILD_WORKFLOW_EXECUTION

EFFECT_WORKER_ENTITY_DISPATCH

EFFECT_SANDBOX_ENTITY_TRANSPORT

Infrastructure

EFFECT_CHILD_PROCESS

EFFECT_INTEROP

EFFECT_SQL_MESSAGE_STORAGE

EFFECT_LOGGING

EFFECT_LOG_FORMATS_JSON_PRETTY_LOGFMT

EFFECT_METRICS

EFFECT_DIFF_BUNDLE_COMPUTE

EFFECT_DIFF_BUNDLE_APPLY

EFFECT_SCHEDULER_WAKE_QUEUE

EFFECT_WORKFLOW_VERSIONING_RUNTIME

Getting Started

Build Workflows

Run and Operate

Core Concepts

Guides

Components

Integrations

Runtime API

Examples

Reference

​Runners

​EFFECT_RUN_PROMISE

​EFFECT_RUN_SYNC

​EFFECT_RUN_FORK

​EFFECT_SINGLE_RUNNER

​EFFECT_TASK_RUNTIME

​Bridges

​EFFECT_ACTIVITY_BRIDGE

​EFFECT_WORKFLOW_BRIDGE

​EFFECT_WORKFLOW_MAKE_BRIDGE

​EFFECT_COMPUTE_TASK_BRIDGE

​EFFECT_STATIC_TASK_BRIDGE

​EFFECT_DEFERRED_BRIDGE

​EFFECT_DURABLE_DEFERRED_BRIDGE

​EFFECT_DEFERRED_STATE_BRIDGE

​EFFECT_CHILD_WORKFLOW_EXECUTION

​EFFECT_WORKER_ENTITY_DISPATCH

​EFFECT_SANDBOX_ENTITY_TRANSPORT

​Infrastructure

​EFFECT_CHILD_PROCESS

​EFFECT_INTEROP

​EFFECT_SQL_MESSAGE_STORAGE

​EFFECT_LOGGING

​EFFECT_LOG_FORMATS_JSON_PRETTY_LOGFMT

​EFFECT_METRICS

​EFFECT_DIFF_BUNDLE_COMPUTE

​EFFECT_DIFF_BUNDLE_APPLY

​EFFECT_SCHEDULER_WAKE_QUEUE

​EFFECT_WORKFLOW_VERSIONING_RUNTIME

Runners

EFFECT_RUN_PROMISE

EFFECT_RUN_SYNC

EFFECT_RUN_FORK

EFFECT_SINGLE_RUNNER

EFFECT_TASK_RUNTIME

Bridges

EFFECT_ACTIVITY_BRIDGE

EFFECT_WORKFLOW_BRIDGE

EFFECT_WORKFLOW_MAKE_BRIDGE

EFFECT_COMPUTE_TASK_BRIDGE

EFFECT_STATIC_TASK_BRIDGE

EFFECT_DEFERRED_BRIDGE

EFFECT_DURABLE_DEFERRED_BRIDGE

EFFECT_DEFERRED_STATE_BRIDGE

EFFECT_CHILD_WORKFLOW_EXECUTION

EFFECT_WORKER_ENTITY_DISPATCH

EFFECT_SANDBOX_ENTITY_TRANSPORT

Infrastructure

EFFECT_CHILD_PROCESS

EFFECT_INTEROP

EFFECT_SQL_MESSAGE_STORAGE

EFFECT_LOGGING

EFFECT_LOG_FORMATS_JSON_PRETTY_LOGFMT

EFFECT_METRICS

EFFECT_DIFF_BUNDLE_COMPUTE

EFFECT_DIFF_BUNDLE_APPLY

EFFECT_SCHEDULER_WAKE_QUEUE

EFFECT_WORKFLOW_VERSIONING_RUNTIME