--- title: Effect Integration description: Low-level Effect-ts integration layer for power users who need direct access to Smithers internals. --- 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](https://effect.website) 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. ```ts 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 }, ); ``` ```ts function runPromise( effect: Effect.Effect, options?: { signal?: AbortSignal }, ): Promise ``` ### 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. ```ts import { runSync } from "smithers-orchestrator/effect/runtime"; const value = runSync(Effect.succeed("hello")); ``` ```ts function runSync( effect: Effect.Effect, ): 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. ```ts import { runFork } from "smithers-orchestrator/effect/runtime"; import { Metric } from "effect"; import { runsTotal } from "smithers-orchestrator/effect/metrics"; runFork(Metric.increment(runsTotal)); ``` ```ts function runFork( effect: Effect.Effect, ): Fiber ``` --- ### 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. ```ts 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); }); ``` ```ts 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. ```ts 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!(), ); ``` ```ts type SmithersTaskRuntime = { runId: string; stepId: string; attempt: number; iteration: number; signal: AbortSignal; db: any; heartbeat: (data?: unknown) => void; lastHeartbeat: unknown | null; }; function withTaskRuntime(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. ```ts 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 }, }, ); ``` ```ts type TaskActivityContext = { attempt: number; idempotencyKey: string; }; type ExecuteTaskActivityOptions = { initialAttempt?: number; retry?: false | { times: number; while?: (error: unknown) => boolean }; includeAttemptInIdempotencyKey?: boolean; }; function executeTaskActivity( adapter: SmithersDb, workflowName: string, runId: string, desc: TaskDescriptor, executeFn: (context: TaskActivityContext) => Promise | A, options?: ExecuteTaskActivityOptions, ): Promise ``` `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. ```ts 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. ```ts 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. ```ts function runWorkflowWithMakeBridge( workflow: SmithersWorkflow, opts: RunOptions & { runId: string }, executeBody: (workflow: SmithersWorkflow, opts: RunOptions) => Promise, ): Promise function withWorkflowMakeBridgeRuntime( 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. ```ts import { executeComputeTaskBridge } from "smithers-orchestrator/effect/compute-task-bridge"; await executeComputeTaskBridge( adapter, db, runId, desc, eventBus, { rootDir: "/workspace" }, "my-workflow", signal, ); ``` Eligibility check: ```ts 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. ```ts import { executeStaticTaskBridge } from "smithers-orchestrator/effect/static-task-bridge"; await executeStaticTaskBridge( adapter, runId, desc, eventBus, { rootDir: "/workspace" }, "my-workflow", signal, ); ``` Eligibility check: ```ts 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. ```ts 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); ``` ```ts 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. ```ts 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: ```ts 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: ```ts 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()`. ```ts 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. ```ts 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. ```ts 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: ```ts import { CodeplaneSandboxExecutorLive, DockerSandboxExecutorLive, SandboxHttpRunner, } from "smithers-orchestrator/effect/http-runner"; ``` Socket-backed executor: ```ts 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. ```ts 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 ``` ```ts type SpawnCaptureOptions = { cwd: string; env?: Record; 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 ``` 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. ```ts 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()); ``` ```ts type ErrorWrapOptions = { code?: SmithersErrorCode; details?: Record; }; function fromPromise( label: string, evaluate: () => PromiseLike, options?: ErrorWrapOptions, ): Effect.Effect function fromSync( label: string, evaluate: () => A, options?: ErrorWrapOptions, ): Effect.Effect function ignoreSyncError(label: string, fn: () => void): Effect.Effect 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. ```ts 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. ```ts 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"); ``` ```ts type LogAnnotations = Record | 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: ```ts 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.*` | ```ts 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: ```ts 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. ```ts 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[] ``` ```ts 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 ``` --- ### 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. ```ts import { applyDiffBundle } from "smithers-orchestrator/effect/diff-bundle"; await applyDiffBundle(bundle, "/sandbox/workspace"); ``` ```ts function applyDiffBundle( bundle: DiffBundle, targetDir: string, ): Promise ``` 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. ```ts 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(); ``` ```ts type SchedulerWakeQueue = { notify(): void; wait(): Promise; }; 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). ```ts 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: ```ts import { usePatched } from "smithers-orchestrator/effect/versioning"; function MyWorkflow({ ctx }) { const hasNewRetryLogic = usePatched("2026-04-retry-overhaul"); return ( Deploy ); } ``` ```ts type WorkflowVersioningRuntime = { resolve(patchId: string): boolean; flush(): Promise; snapshot(): WorkflowPatchDecisions; }; function createWorkflowVersioningRuntime(options: WorkflowVersioningRuntimeOptions): WorkflowVersioningRuntime function withWorkflowVersioningRuntime(runtime: WorkflowVersioningRuntime, execute: () => T): T function getWorkflowVersioningRuntime(): WorkflowVersioningRuntime | undefined function getWorkflowPatchDecisions(config: Record | null | undefined): WorkflowPatchDecisions function usePatched(patchId: string): boolean ``` `usePatched` calls `resolve()` on the ambient `WorkflowVersioningRuntime`. Outside a versioning scope it always returns `false`.