feat(runtime)!: report worker callback outcome instead of run state

.changeset/effect-reliability.md

@@ -0,0 +1,9 @@
+---
+'@swirls/looms': minor
+---
+
+Worker callbacks now answer `{ status: 'accepted' | 'stale' | 'cancelled', cancelRequested }` instead of the run state. `createLooms().workerCallback` and `createLoomsClient().workerCallback` return that result. `cancelRequested` tells an attempt that is already running to stop, because a dispatch that has left the host cannot be recalled. A stale report records nothing.
+
+`failEffect({ error, nonRetryable, retryAfterMs })` fails the current attempt. `nonRetryable` ends the effect even when the retry policy has attempts left. `retryAfterMs` waits at least that long before the next attempt and never shortens the policy backoff. Worker `fail` callbacks take the same two fields.
+
+A heartbeat may include `details`. The latest details are kept across retries and included on the next `EffectWorkerTask`, so a resumed attempt can continue from the recorded progress.

apps/docs/src/routes/docs.reliability.tsx

@@ -67,7 +67,13 @@ function Page() {
         a restart preserves the retry budget and backoff. Configure start, heartbeat, and execution
         timeouts for work delivered to external workers. Worker callbacks include the effect ID and
         attempt number; stale callbacks cannot complete a newer attempt. Exhaustion records{' '}
-        <code>runtime.effect.failed</code>.
+        <code>runtime.effect.failed</code>. A handler can stop that budget early with{' '}
+        <code>failEffect</code> and <code>nonRetryable: true</code>, or ask the next attempt to wait
+        at least <code>retryAfterMs</code> without shortening the policy backoff. A worker heartbeat
+        can carry <code>details</code>; the next dispatch of that effect includes the latest details
+        so the attempt can resume. The callback response is{' '}
+        <code>{'{ status, cancelRequested }'}</code>, not the run state.{' '}
+        <code>cancelRequested</code> tells an attempt that is already running to stop.
       </p>
       <h2 id="signals-and-webhook-delivery">
         Signals and webhook delivery

docs/production.md

@@ -19,13 +19,24 @@ are `EventPayloadTooLargeError` and `SnapshotPayloadTooLargeError`.
 
 ## Worker effect protocol
 
-1. The runtime dispatches `{ runId, effectId, attempt, type, input }`.
+1. The runtime dispatches `{ runId, effectId, attempt, type, input, details? }`.
+   `details` is the progress carried by the previous attempt's last heartbeat.
 2. The worker calls `started` before the schedule-to-start deadline.
-3. Long work calls `heartbeat`; each accepted heartbeat extends the heartbeat
-   deadline without exceeding start-to-close.
-4. The worker calls exactly one of `complete` or `fail`.
+3. Long work calls `heartbeat`. Each accepted heartbeat extends the heartbeat
+   deadline without exceeding start-to-close. A heartbeat may include `details`;
+   the latest details are kept across retries and sent on the next dispatch.
+4. The worker calls exactly one of `complete` or `fail`. `fail` may set
+   `nonRetryable` to stop the retry policy, or `retryAfterMs` to wait at least
+   that long before the next attempt (the policy backoff is never shortened).
+   A local handler does the same by failing with `failEffect`.
 5. On cancellation, stop side effects and call `cancelled`.
 
+Each callback response is `{ status: 'accepted' | 'stale' | 'cancelled', cancelRequested }`.
+The host no longer returns the run state. `cancelRequested` is how a worker learns
+that an attempt already in progress should stop: a queued dispatch cannot be recalled.
+`stale` means the report was for another attempt, or the attempt had already settled,
+and the host recorded nothing.
+
 Callbacks include the same `effectId` and `attempt`. Stale callbacks are ignored.
 A dispatch transport failure is recorded as `ambiguous`: an operator must first
 verify the destination's idempotency record, then use `retryEffect` or

packages/client/src/client.ts

@@ -15,6 +15,7 @@ import {
   type JsonValue,
   type RunState,
   type WorkerCallbackInput,
+  type WorkerCallbackResult,
 } from '@looms/core'
 
 import { consumeSseStreamEffect, eventsFromSseData, sseFrames } from './sse'
@@ -41,6 +42,53 @@ const StartResultSchema = Schema.Struct({
 
 const RunResultSchema = Schema.Struct({ runId: Schema.String, state: RunStateSchema })
 
+interface WorkerFailRequestBody {
+  error: string
+  nonRetryable?: boolean
+  retryAfterMs?: number
+}
+
+function workerCallbackBody(input: WorkerCallbackInput) {
+  switch (input.kind) {
+    case 'complete':
+      return { events: input.events }
+
+    case 'fail': {
+      const body: WorkerFailRequestBody = { error: input.error }
+
+      if (input.nonRetryable !== undefined) {
+        body.nonRetryable = input.nonRetryable
+      }
+
+      if (input.retryAfterMs !== undefined) {
+        body.retryAfterMs = input.retryAfterMs
+      }
+
+      return body
+    }
+
+    case 'heartbeat':
+      return input.details === undefined ? {} : { details: input.details }
+    case 'started':
+    case 'cancelled':
+      return {}
+
+    default: {
+      const exhaustive: never = input
+      return exhaustive
+    }
+  }
+}
+
+const WorkerCallbackResultSchema = Schema.Struct({
+  status: Schema.Union([
+    Schema.Literal('accepted'),
+    Schema.Literal('stale'),
+    Schema.Literal('cancelled'),
+  ]),
+  cancelRequested: Schema.Boolean,
+})
+
 /** `appended` counts the events the run gained while the wake ran. */
 const WakeResultSchema = Schema.Struct({
   runId: Schema.String,
@@ -181,17 +229,15 @@ export function createLoomsClient(options: LoomsClientOptions = {}) {
     runId?: string
   }) => request('/runs', StartResultSchema, { method: 'POST', body: stringifyJson(args) })
 
-  const workerCallback = (runId: string, input: WorkerCallbackInput) => {
-    const body =
-      input.kind === 'complete'
-        ? { events: input.events }
-        : input.kind === 'fail'
-          ? { error: input.error }
-          : {}
+  const workerCallback = (
+    runId: string,
+    input: WorkerCallbackInput,
+  ): Promise<WorkerCallbackResult> => {
+    const body = workerCallbackBody(input)
 
     return request(
       `/runs/${encodeURIComponent(runId)}/effects/${encodeURIComponent(input.effectId)}/${input.attempt}/${input.kind}`,
-      RunResultSchema,
+      WorkerCallbackResultSchema,
       {
         method: 'POST',
         headers: options.workerCallbackToken

packages/core/src/effects.ts

@@ -215,6 +215,52 @@ export class EffectHandlerError extends Data.TaggedError('EffectHandlerError')<{
   }
 }
 
+export interface EffectFailureOptions {
+  readonly error: string
+  /** Fail the attempt even when the retry policy has attempts remaining. */
+  readonly nonRetryable?: boolean
+  /** Wait at least this long before the next attempt. Does not shorten the policy backoff. */
+  readonly retryAfterMs?: number
+}
+
+/** A handler's deliberate attempt failure, distinct from an unexpected thrown error. */
+export class EffectFailure extends Data.TaggedError('EffectFailure')<{
+  readonly error: string
+  readonly nonRetryable: boolean
+  readonly retryAfterMs?: number
+}> {
+  constructor(options: EffectFailureOptions) {
+    const error = options.error
+    const nonRetryable = options.nonRetryable === true
+
+    if (options.retryAfterMs === undefined) {
+      super({ error, nonRetryable })
+    } else {
+      super({ error, nonRetryable, retryAfterMs: options.retryAfterMs })
+    }
+
+    this.name = 'EffectFailure'
+  }
+}
+
+/** Fail the current attempt. The retry policy decides whether another attempt is scheduled. */
+export function failEffect(options: EffectFailureOptions): Effect.Effect<never, EffectFailure> {
+  return Effect.fail(new EffectFailure(options))
+}
+
+/** Unwrap a handler failure, including one wrapped by `EffectHandlerError`. */
+export function readEffectFailure(cause: unknown): EffectFailure | null {
+  if (cause instanceof EffectFailure) {
+    return cause
+  }
+
+  if (cause instanceof EffectHandlerError) {
+    return readEffectFailure(cause.cause)
+  }
+
+  return null
+}
+
 export type EffectHandlerResult<
   R = never,
   E extends EventInput = EventInput,
@@ -266,6 +312,18 @@ function liftHandlerResult<R, E extends EventInput, HandlerError>(
   return Effect.succeed(result)
 }
 
+function invokeHandler<R, E extends EventInput, HandlerError>(
+  handler: (input: JsonValue, ctx: EffectContext) => EffectHandlerResult<R, E, HandlerError>,
+  input: JsonValue,
+  ctx: EffectContext,
+): Effect.Effect<ReadonlyArray<E>, EffectHandlerError, R> {
+  try {
+    return liftHandlerResult(handler(input, ctx))
+  } catch (cause) {
+    return Effect.fail(new EffectHandlerError(cause))
+  }
+}
+
 /**
  * Define an effect. With `execute`, a run's actor cell runs the handler at
  * most once while it is alive; if the cell dies mid-execution, the replacement
@@ -327,7 +385,12 @@ export function defineEffect<R, E extends EventInput, HandlerError>(def: {
       Effect.gen(function* () {
         const input = schema ? yield* validateInputEffect(schema, raw) : raw
 
-        return handler ? yield* liftHandlerResult(handler(input, ctx)) : noOutput
+        if (!handler) {
+          return noOutput
+        }
+
+        // invokeHandler turns a synchronous throw into EffectHandlerError.
+        return yield* invokeHandler(handler, input, ctx)
       }),
   }
 }

packages/core/src/fold.ts

@@ -35,6 +35,28 @@ function readString(obj: { [key: string]: JsonValue }, key: string): string | un
   return Predicate.isString(value) ? value : undefined
 }
 
+/** Heartbeat progress already recorded for this effect, if any. */
+function priorHeartbeatDetails(state: RunState, effectId: string): JsonValue | null {
+  return state.effectExecutions[effectId]?.heartbeatDetails ?? null
+}
+
+/**
+ * A heartbeat that carries `details` replaces the saved progress. Every other execution event
+ * keeps it, including the retry that starts the next attempt.
+ */
+function nextHeartbeatDetails(
+  state: RunState,
+  effectId: string,
+  event: EventEnvelope,
+  payload: { [key: string]: JsonValue },
+): JsonValue | null {
+  if (event.type === 'runtime.effect.heartbeat' && Object.hasOwn(payload, 'details')) {
+    return payload.details ?? null
+  }
+
+  return priorHeartbeatDetails(state, effectId)
+}
+
 function cloneState(state: RunState): RunState {
   return {
     runId: state.runId,
@@ -290,6 +312,7 @@ function applyProtocol(state: RunState, event: EventEnvelope, registry: FoldRegi
             deadlineAt: null,
             startedAt: event.ts,
             lastHeartbeatAt: null,
+            heartbeatDetails: priorHeartbeatDetails(state, effectId),
             lastError: null,
           },
         },
@@ -352,6 +375,7 @@ function applyProtocol(state: RunState, event: EventEnvelope, registry: FoldRegi
               : sameAttempt
                 ? previous.lastHeartbeatAt
                 : null,
+            heartbeatDetails: nextHeartbeatDetails(state, effectId, event, payload),
             lastError: readString(payload, 'error') ?? null,
           },
         },
@@ -384,6 +408,7 @@ function applyProtocol(state: RunState, event: EventEnvelope, registry: FoldRegi
             deadlineAt: nextAttemptAt,
             startedAt: null,
             lastHeartbeatAt: null,
+            heartbeatDetails: priorHeartbeatDetails(state, effectId),
             lastError: readString(payload, 'error') ?? 'effect failed',
           },
         },

packages/core/src/protocol.ts

@@ -94,6 +94,8 @@ export const protocolCatalog = defineEventCatalog('runtime', {
   'effect.failed': Schema.Struct({
     effectId: Schema.String,
     error: Schema.String,
+    nonRetryable: Schema.optional(Schema.Boolean),
+    retryAfterMs: Schema.optional(Schema.Finite),
   }),
   'effect.attempt.started': Schema.Struct({
     effectId: Schema.String,
@@ -119,6 +121,8 @@ export const protocolCatalog = defineEventCatalog('runtime', {
     attempt: Schema.Finite,
     heartbeatAt: Schema.Finite,
     deadlineAt: Schema.NullOr(Schema.Finite),
+    /** Progress the next attempt should resume from. Absent on heartbeats that only extend the deadline. */
+    details: Schema.optional(JsonValueSchema),
   }),
   'effect.cancel.requested': Schema.Struct({
     effectId: Schema.String,

packages/core/src/state.ts

@@ -58,6 +58,8 @@ export interface EffectExecutionRecord {
   /** When the current attempt started executing, locally or on a worker. */
   startedAt: number | null
   lastHeartbeatAt: number | null
+  /** Progress reported by the latest heartbeat. Survives retries so the next attempt can resume. */
+  heartbeatDetails: JsonValue | null
   lastError: string | null
 }
 
@@ -189,6 +191,9 @@ export const RunStateSchema = Schema.Struct({
       lastHeartbeatAt: Schema.NullOr(Schema.Finite).pipe(
         Schema.withDecodingDefaultKey(Effect.succeed(null)),
       ),
+      heartbeatDetails: Schema.NullOr(JsonValueSchema).pipe(
+        Schema.withDecodingDefaultKey(Effect.succeed(null)),
+      ),
       lastError: Schema.NullOr(Schema.String),
     }),
   ).pipe(Schema.withDecodingDefaultKey(Effect.succeed({}))),

packages/core/src/worker-callback.ts

@@ -1,4 +1,5 @@
 import type { EventInput } from './envelope'
+import type { JsonValue } from './types'
 
 /** Identifies one attempt of an effect executed by an external worker. */
 export interface WorkerCallback {
@@ -8,8 +9,25 @@ export interface WorkerCallback {
 
 /** What an external worker reports about an effect attempt. */
 export type WorkerCallbackInput =
-  | (WorkerCallback & { readonly kind: 'started' | 'heartbeat' | 'cancelled' })
+  | (WorkerCallback & { readonly kind: 'started' | 'cancelled' })
+  | (WorkerCallback & { readonly kind: 'heartbeat'; readonly details?: JsonValue })
   | (WorkerCallback & { readonly kind: 'complete'; readonly events: ReadonlyArray<EventInput> })
-  | (WorkerCallback & { readonly kind: 'fail'; readonly error: string })
+  | (WorkerCallback & {
+      readonly kind: 'fail'
+      readonly error: string
+      /** When set, the attempt is terminal even if the retry policy has attempts left. */
+      readonly nonRetryable?: boolean
+      /** Extra delay before the next attempt. The policy backoff is never shortened. */
+      readonly retryAfterMs?: number
+    })
 
 export type WorkerCallbackKind = WorkerCallbackInput['kind']
+
+/**
+ * What the host tells the worker about the callback it just sent.
+ * A queued message cannot be recalled, so `cancelRequested` is how an in-flight attempt learns to stop.
+ */
+export interface WorkerCallbackResult {
+  readonly status: 'accepted' | 'stale' | 'cancelled'
+  readonly cancelRequested: boolean
+}