Adapter Authoring

Runlane lanes are composed from one storage adapter and an explicit delivery mode. Storage owns durable truth. Polling workers discover runnable work through storage leases; transport delivery adds an external wakeup publisher. A lane package wires compatible pieces together, drives optional lifecycle hooks, and reports the composed capabilities.

interface Lane {
  readonly name: string
  readonly capabilities: LaneCapabilities
  readonly delivery: LaneDelivery
  readonly storage: StorageAdapter
  start?(): Promise<void>
  close?(): Promise<void>
}

type LaneDelivery =
  | { readonly mode: 'storage_polling' }
  | { readonly mode: 'transport'; readonly transport: TransportAdapter }

Storage

Implement StorageAdapter when you are building a reusable durable backend such as Postgres, MySQL, DynamoDB, SQLite, or Redis.

import type { StorageCapabilities } from '@runlane/contracts'

export const postgresStorageCapabilities = {
  claimsScheduleOccurrences: true,
  durableState: true,
  durableSteps: true,
  exportsObservations: true,
  enforcesIdempotency: true,
  enforcesQueueConcurrency: true,
  enforcesSingleton: true,
  leasesRuns: true,
  persistsOutbox: true,
  processLocalState: false,
  prunesRuns: true,
  readsRunHistory: true,
} satisfies StorageCapabilities

The important rule is the append boundary: core owns reducer semantics; storage owns atomic persistence.

durableState and processLocalState answer different questions. durableState: false means process loss can lose committed state. processLocalState: true means another process cannot construct an equivalent adapter and reach the same state. Redis without persistence can be non-durable but shared across processes; in-memory local storage is both non-durable and process-local.

Report capability flags as public promises, not implementation wishes. Core and operator APIs use these flags to decide whether runtime construction, maintenance, and operator commands are valid.

Storage capability flags:

appendRunEvents() receives append-only events and the core-projected RunRecord. It verifies environment, runId, and expectedSequence, persists the events and supplied projection, and creates outbox rows for any run.delivery_requested events in the same transaction when the command's persistDeliveryOutbox option is true or omitted. Storage-polling lanes pass false so delivery intent stays in run state without creating provider outbox rows.

Append commands include a required expectedSequence; callers use the current RunRecord.eventSequence or contractDefaults.run.newEventSequence for a new run.

Storage adapters may expose start() and close() hooks for pools, embedded engines, or external clients. Lane/runtime code starts storage before transport delivery and closes transport delivery before storage.

StorageAdapter.start() and StorageAdapter.close() are optional lifecycle hooks. Every other StorageAdapter method below is required by the contract and must exist on the adapter instance. When a required method is guarded by an unsupported capability, it should fail fast with RunlaneError and ErrorCode.CapabilityUnsupported; it must not silently no-op or return an empty success value.

Storage concurrency guarantees

Adapter correctness depends on the guarantees below, not on any particular database primitive. Postgres uses advisory locks, row locks, generated columns, and ON CONFLICT, but another backend can satisfy the same contract with conditional writes, compare-and-swap transactions, serializable partitions, leases, or single-writer streams. The important rule is the observable behavior under competing callers.

appendRunEvents() must compare the current persisted event sequence with expectedSequence before validating projected run invariants. If it does not match, reject with RunlaneError, ErrorCode.StorageConflict, and StorageConflictKind.EventSequence. This preserves optimistic-concurrency recovery: a stale append is a lost race, not an adapter contract violation.

When an append succeeds, these records commit atomically:

• event-history rows
• materialized projection, including RunRecord.output when run.succeeded carries JSON output, RunRecord.wait while released, and RunRecord.attemptDeadlineAt while a running attempt has a deadline
• storage-owned idempotency or singleton ownership rows
• derived outbox rows requested by the command
• derived observation records when exportsObservations is supported

The RunEventRecord values returned from appendRunEvents(), claimRunLease(), and heartbeatRunLease() must be the exact records persisted to history, including ids and sequence numbers. Do not generate one event id for storage and a second event id for the method result.

Idempotency owner creation is a serialized ownership decision for (environment, taskId, idempotencyKey).

A valid implementation may lock the partition, conditionally insert then read back the owner, or use a transaction primitive that guarantees one owner.

Idempotency retention uses the policy captured at owner creation:

• active owners never expire while active
• with a finite TTL, successful and cancelled terminal owners remain readable until their captured TTL expires or resetIdempotencyKey() clears them
• with ttl: "active", ownership is released when the run reaches any terminal state
• failed owners are released

getRunByIdempotencyKey() is side-effect free: an expired owner returns undefined; cleanup can happen on a later write, reset, or prune path.

Bounded queue capacity is partitioned by (environment, queue, concurrencyKey). Absent concurrency keys share one partition.

listRunnableRuns() and listRunsNeedingDispatch() should apply capacity predicates before returning candidates so a capacity-blocked early row does not starve runnable work in another partition.

claimRunLease() and reserveRunDispatch() must enforce the same capacity partition inside the write operation. A scan-time check alone is not enough, and row-locking only the candidate run is not enough when two different runs in the same partition can be claimed or reserved concurrently.

reserveRunDispatch() is the atomic bounded-queue delivery path. In one operation, it:

1. Checks the current sequence.
2. Confirms the run is still dispatchable.
3. Verifies partition capacity.
4. Appends the supplied run.delivery_requested event.
5. Writes the projected reservation.
6. Creates the outbox row when the command requests transport wakeup persistence.

A stale sequence, expired/ineligible dispatch state, or full capacity returns undefined; malformed command data remains an adapter contract violation.

claimRunLease() is a race-tolerant ownership operation. A stale sequence, expired dispatch reservation, or full bounded queue returns undefined so competing workers can keep polling without treating normal contention as a framework error. heartbeatRunLease() is stricter because the caller believes it already owns the run: stale sequence is StorageConflictKind.EventSequence, and lost lease ownership is StorageConflictKind.LeaseOwnership.

Outbox mutation methods are batch-only by contract.

claimOutboxMessages() atomically transitions only due rows whose status is pending, failed, or claimed with an expired claim into a new claim, and returns the exact rows now owned by that claim. When outboxMessageIds is provided, do not widen the claim into unrelated backlog rows.

markOutboxMessagesPublished(), markOutboxMessagesFailed(), and markOutboxMessagesDeadLettered() must only mutate messages whose current claim token still matches the command. A stale or missing claim rejects with StorageConflictKind.OutboxClaim.

Promise-returning adapter methods must report validation, contract, conflict, and backend failures as rejected promises. If an adapter implementation performs synchronous validation, wrap it so callers never observe a synchronous throw from a Promise-typed method.

Records returned from storage must be defensive copies. Mutable Date instances and nested objects must not share references with adapter-owned state or command input.

Successful task output is storage-owned durable run state, not transport state. Store it in canonical run_json, preserve run.succeeded.output in event history, and keep any indexed output column such as Postgres output_json synchronized from the same projection. Failed, retrying, released, and cancelled runs must not expose current success output.

Durable step output is storage-owned checkpoint state keyed by (environment, runId, stepKey). getRunStep() is a side-effect-free read. completeRunStep() must validate the active lease token, the active attempt number, and the run status before inserting the step. Lost ownership rejects with StorageConflictKind.LeaseOwnership. Replaying completion with the same JSON output is idempotent and returns the stored row with completed: false; replaying the same step key with different output is ErrorCode.InvariantViolation. Do not infer steps from run events, release metadata, or task output.

Observation export stream

The durable observation stream is a storage-owned export boundary over committed Runlane facts. Adapters that report exportsObservations: true must record sanitized observation records for persisted run events and newly completed durable steps. Adapters that report exportsObservations: false must still expose the required methods, but each method should reject with ErrorCode.CapabilityUnsupported.

Observation records must be committed with the source fact. A successful appendRunEvents(), claimRunLease(), heartbeatRunLease(), or reserveRunDispatch() write that returns run event records must also make those events visible to scanObservationRecords(). A successful first-time completeRunStep() must make the step visible to the stream; an idempotent replay returning completed: false must not create a second observation record.

Observation record ids are duplicate-safe. The same durable run event id maps to the same run-event observation record id, and the same durable step token maps to the same run-step observation record id. The stream payload must stay sanitized: no task payload, successful output, durable step output, failure detail, release metadata, or operator-expanded run rows by default.

scanObservationRecords({ environment, cursor, limit }) returns records after the supplied cursor in stable forward order. Storage owns the cursor encoding and should clamp limits with contractDefaults.pagination. The returned cursor is the checkpoint value after the final returned record; exporters store it only after their sink accepts the batch. Observation streamPosition values are opaque strings at the contract boundary so adapters can use database-native cursor precision internally.

getObservationCheckpoint() and advanceObservationCheckpoint() are scoped by (environment, consumer). advanceObservationCheckpoint() must compare the stored checkpoint with expectedCursor and reject stale writers with StorageConflictKind.ObservationCheckpoint. It must also reject non-monotonic advances and cursors that do not reference an existing stream record in the same environment, so a crashed, slow, or malformed exporter cannot move the checkpoint backward or skip unexported records.

Released run waits are storage-owned durable run state too. Preserve RunRecord.wait from the supplied projection, index the durable condition needed for signal and run-completion scans, and clear it only when the projected lifecycle event clears it. Do not infer signal or source-run waits from reason strings or metadata.

Attempt deadlines are fixed at lease claim. Preserve RunLeaseClaimedEvent.attemptDeadlineAt and the matching RunRecord.attemptDeadlineAt; heartbeat writes must not move it. Terminal, retry, release, cancellation, and delivery-recovery projections clear the field.

listRunnableRuns() returns compact RunnableRunRef candidates for worker acquisition. It must not return payload-bearing RunRecord values.

Workers use the refs to pick candidates, read the full run, and then supply core-projected lease events to claimRunLease(). Use getRunRunnableAvailableAt() for the due-work predicate and ordering timestamp, and use runStatusValues.isRunnableCandidate() when adapter code needs status narrowing.

listRunsNeedingDelivery() is the maintenance query for fresh delivery requests.

Use runStatusValues.isDeliveryRecoveryCandidate() when adapter code needs to narrow the compact ref status. Maintenance then calls getRuns() once for the selected ids, appends run.delivery_requested, and asks storage to create a new outbox row only for transport delivery lanes.

listRunsNeedingDispatch() and reserveRunDispatch() are the bounded-queue dispatch path. Queue capacity must be enforced inside the storage transaction because multiple schedulers and workers can race.

Capacity counts currently running runs plus unexpired dispatch reservations in the same environment, queue, and concurrencyKey partition. When no concurrencyKey is present, the whole queue is one partition.

Dispatch reservations use dispatchExpiresAt; if no worker claims the run before that time, maintenance may reserve and publish it again.

listRunsNeedingCancellationFinalization() is the cancellation maintenance query. It returns only cancellation_requested runs whose current lease has expired.

Use getRunCancellationFinalizationAvailableAt() for the predicate and runStatusValues.isCancellationFinalizationCandidate() for status-only narrowing. Maintenance then reads current projections with getRuns() and appends run.cancelled; it does not wake workers for cancellation finalization.

listRunsNeedingTimeoutFinalization() is the attempt-deadline maintenance query. It returns only running runs whose attemptDeadlineAt is at or before the requested time.

Use getRunAttemptTimeoutFinalizationAvailableAt() for the predicate. Maintenance then reads current projections with getRuns() and appends either run.retry_scheduled or run.failed with ErrorCode.TaskTimedOut according to the registered task retry policy.

listRunsWaitingForSignal() and listRunsWaitingForRunCompletion() are event-driven resume scans. They return compact refs for released runs whose current wait condition exactly matches the requested signal key or source run id. They must not return runs that are no longer released, no longer have the matching wait condition, or are merely time-released.

getRunByIdempotencyKey() is the recovery read behind retained idempotent trigger() semantics. Idempotency ownership is scoped by environment, task id, and key.

Retention rules:

• active owners never expire while active
• successful and cancelled terminal owners with a finite captured idempotencyKeyTTL remain until that TTL expires or resetIdempotencyKey() clears them
• owners captured with idempotencyKeyTTL: "active" are released on terminal state
• failed owners are cleared automatically

Storage must use its authoritative owner table for this lookup, not operator pagination.

pruneRuns() receives terminal statuses and a concrete olderThan: Date from core and must never delete active runs. Core resolves public duration strings before calling storage and freezes that cutoff into public continuation cursors.

When the command omits limit, storage uses contractDefaults.pruning.batchLimit. When storage cannot finish in one bounded call, it returns nextCursor; core passes the adapter cursor back on the next prune command with the same retention filter.

Ownership tokens are part of the contract. Lease tokens are supplied in the core-projected lease events passed to claimRunLease() and heartbeatRunLease() so storage does not compute run projections. Schedule occurrence and outbox claim tokens are storage-generated, and callers return those tokens for schedule completion, publish, failure, or dead-letter updates.

Adapter indexes should treat Runlane ids as opaque strings. Public Runlane ids are non-empty and reserve : for backend-internal key composition; adapters should still avoid parsing prefixes or separators for behavior.

Schedule occurrence ids and generated schedule run ids use deterministic hashed scope data and may be longer than random run ids. Do not parse them to recover environment, schedule, or fire time.

Outbox mutations are batch-only so networked adapters can persist publish results without one round-trip per row.

OutboxFailureRecord.code must be an ErrorCode. Provider-specific response codes belong in meta so operator tooling can branch on Runlane's stable vocabulary.

Run events are durable replay records. Existing event types stay backward compatible: add optional fields with reducer defaults, or introduce a new RunEventType when new required data is needed.

Lease heartbeats are real events in v1. Adapters must preserve observable sequence and history semantics for listRunEvents(). Physical compaction is only valid if callers still see the same ordered event records, or if a future compaction contract explicitly changes that behavior.

Use environmentKey(environment) for durable scoping and uniqueness indexes. Do not build parallel environment key logic from environment.name; future environment identity fields must widen the same contract everywhere.

Operator list APIs use opaque cursors. Storage should apply contractDefaults.pagination, and include filter and sort state in cursor semantics so a cursor cannot silently resume a different ordering. Event sequence sorting is only valid inside one run, so adapters should require runId when callers request RunEventSortField.Sequence.

Transport

Implement TransportAdapter when you are building a reusable wakeup backend such as SQS, Redis, RabbitMQ, Pub/Sub, Kafka, or HTTP push.

Transport should carry the minimum needed to wake a worker. It does not own payloads, successful output, failure detail, or materialized run state.

import { asId, type DeliveryMessage } from '@runlane/contracts'

const wakeup = {
  environment: { name: 'production' },
  queue: asId<'queue'>('emails'),
  requestedAt: new Date(),
  runId: asId<'run'>('run_123'),
} satisfies DeliveryMessage

Workers use the wakeup environment and runId to read current durable state from storage before executing. Duplicate or delayed transport messages are safe only when storage remains the durable truth boundary.

Transport capability flags:

If a transport exposes queues, runtime construction verifies that every runtime queue has a matching provider binding and policy. Omit queues only for transports whose binding cannot be described as static Runlane queue definitions.

Transport-driven execution uses runlane.executeDelivery(message), not a drain worker. Drain asks storage what work is due; delivery execution uses the transport-acquired runId, verifies the stored run is still runnable for that queue, and returns an ignored result for ack-safe stale wakeups such as terminal, not-due, wrong-queue, already-leased, claim-lost, or abandoned attempts.

Reusable transport adapters should stay contract-only and should not import core. Provider-specific runtime helpers, such as an SQS/Lambda bridge, can parse provider records into DeliveryMessage values and then call executeDelivery().

Persisted user task outcomes are not transport failures. Storage, projection, registration, and persistence failures should propagate so the provider integration can retry or report batch item failure.

Transport publish commands receive claimed outbox attempts keyed by outboxMessageId and claimToken. A successful publishWakeups() return must include outcomes[index] for command.attempts[index], with each outcome tagged by WakeupPublishOutcomeType.Published or WakeupPublishOutcomeType.Failed.

Throw TransportUnavailable or operation-level TransportPublishFailed when the adapter cannot produce trustworthy per-attempt outcomes; core records that operation-level failure against every claimed attempt. Do not echo ids or split successful and failed rows into parallel arrays.

Use publishWakeupsCommandSchema when an adapter or adapter test intentionally validates a publish command. Do not use provider serialization or clone heuristics as a substitute for the Runlane command contract.

Transport adapters may expose start() and close() hooks for clients, sockets, or subscriptions. They must not own durable run state even when the transport backend is itself durable.

Lane Packages

A lane package composes storage and an explicit delivery mode with createLane() from @runlane/contracts. The helper validates the supplied adapters, reports composed capabilities, and uses the standard lifecycle order: start storage before transport delivery, close transport delivery before storage.

createLane() is a composition boundary, not a capability shim. It keeps the original adapter instances, exposes storage.capabilities through lane.capabilities.storage, and adds only lane-level metadata and lifecycle ordering. It does not make an adapter durable, add operator reads, emulate unsupported storage behavior, or invent a fake transport for polling lanes.

Malformed options or incomplete adapters fail fast with RunlaneError and ErrorCode.ConfigurationInvalid. Unsupported option names are rejected instead of ignored so lane packages do not grow accidental shadow configuration.

import { createLane, LaneDeliveryMode, type Lane, type StorageAdapter, type TransportAdapter } from '@runlane/contracts'

export interface MyLaneOptions {
  readonly name?: string
  readonly operatorReads?: boolean
  readonly productionDurable?: boolean
  readonly storage: StorageAdapter
  readonly transport?: TransportAdapter
}

export function myLane(options: MyLaneOptions): Lane {
  return createLane({
    delivery:
      options.transport === undefined
        ? { mode: LaneDeliveryMode.StoragePolling }
        : { mode: LaneDeliveryMode.Transport, transport: options.transport },
    name: options.name ?? 'my-lane',
    operatorReads: options.operatorReads ?? options.storage.capabilities.readsRunHistory,
    productionDurable: options.productionDurable ?? false,
    storage: options.storage,
  })
}

Errors

Adapters throw RunlaneError with stable ErrorCode values. Raw driver errors can be attached as cause for logs, but public callers should branch on error.code.

Use the narrowest code that describes the boundary that failed:

Create storage conflicts with createStorageConflictError({ kind: StorageConflictKind.* }). Do not hand-roll bare RunlaneError storage conflicts; that drops the storageConflictKind metadata core and tooling use to distinguish expected races from real storage failures.

Do not expose raw driver messages as the public error contract. Callers should be able to handle adapter failures by code, retryability, and structured metadata without parsing provider text.

Conformance

Adapter packages should import the shared conformance suites from @runlane/testing and provide a fresh adapter or lane resource per test. Vitest packages can use @runlane/testing/vitest; other runners can import defineStorageConformanceSuite(), defineTransportConformanceSuite(), or defineLaneCompositionConformanceSuite() from @runlane/testing and pass the resulting suite to runConformanceSuite() with a runner object that supplies describe and test.

Conformance covers one primitive boundary at a time: storage conformance for durable run truth, transport conformance for wakeup publishing, and lane composition conformance for wiring compatible delivery mode into a Lane. Keep backend-specific tests beside the adapter for migrations, SQLSTATE or SDK error mapping, connection lifecycle, and operational behavior the generic contract cannot observe.