feat(runtime): keep run attributes public and secrets beside the log

.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.

.changeset/run-context.md

@@ -0,0 +1,5 @@
+---
+'@swirls/looms': minor
+---
+
+Runs accept immutable `attributes` and a `secrets` context. Attributes are recorded on `runtime.run.started`, the run state, and the run summary, and are passed to `authorize` when present. Secrets stay in a side store: they are available to effect handlers and in-process worker dispatch, and they are not written to events, snapshots, or run responses. A worker whose transport persists tasks reads them with `POST /runs/:runId/effects/:effectId/:attempt/secrets` (`client.workerSecrets`), which only answers the attempt that is currently running.

apps/docs/src/docs-manifest.json

@@ -830,7 +830,7 @@
         "level": 2
       }
     ],
-    "text": "Reliability and recovery Durability depends on the host and storage you choose. In-memory examples lose history on exit. SQLite persists locally; production actor hosts must preserve one writer per run and provide wake scheduling. Replay and recovery are different operations Historical replay folds recorded events to reconstruct state and does not dispatch effects. Recovery loads a snapshot and later events, then resumes unfinished work. An effect that lacks a recorded outcome may execute again. Looms does not promise exactly-once actions across arbitrary external services. A crash after an external action Looms requests an external action. The service performs it. The host stops before recording success. Recovery cannot infer success from the local log alone. Pass ctx.effectId as an idempotency key where the external API supports it. Check that the service retains the key for as long as you may need to retry the action. For unsupported APIs, store a durable operation ID, query the provider for the outcome, or reconcile before retrying. Never assume adding an arbitrary header makes an API idempotent. Retries and errors Effect retry policies count total attempts: maxAttempts: 3 means one initial attempt and up to two retries. Attempt starts and retry deadlines are recorded before the runtime waits, so 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 runtime.effect.failed . Signals and webhook delivery Use the embed start/signal idempotencyKey option with a stable runId for retries of the same logical request. Verify webhook signatures and transform allowed webhook payloads into typed events on the server. Deduplicate by the provider delivery ID and correlate by run, thread, and domain identifiers. Retrying with a fresh key defeats deduplication. Interrupted model streams A model call can produce partial deltas before failing. Those deltas may already be recorded; resuming work may incur another model call and different output. Render partial content as provisional, inspect completion events, and decide how your application handles abandoned partial turns. The adapter does not guarantee exactly-once billing or identical regenerated text. Snapshots, retention, and projections A compatible snapshot bounds replay work; it is not a backup or a schema migration. Trimming events removes older audit and time-travel history and can prevent rebuilding new projections from the beginning. Define a retention and archive policy before enabling trimAfterSnapshot. Projectors run after local commit. Durable delivery stores a cursor per projector name and version, retries without advancing that cursor, and dead-letters exhausted batches. A run can finish before an external index or webhook catches up, so monitor lag and keep projector writes idempotent. Test recovery Test a restart while parked, an interruption during an external action, duplicate webhook delivery, and a deployment with existing runs. The testing guide provides the local restart test. Review versioning before changing definitions. ) }"
+    "text": "Reliability and recovery Durability depends on the host and storage you choose. In-memory examples lose history on exit. SQLite persists locally; production actor hosts must preserve one writer per run and provide wake scheduling. Replay and recovery are different operations Historical replay folds recorded events to reconstruct state and does not dispatch effects. Recovery loads a snapshot and later events, then resumes unfinished work. An effect that lacks a recorded outcome may execute again. Looms does not promise exactly-once actions across arbitrary external services. A crash after an external action Looms requests an external action. The service performs it. The host stops before recording success. Recovery cannot infer success from the local log alone. Pass ctx.effectId as an idempotency key where the external API supports it. Check that the service retains the key for as long as you may need to retry the action. For unsupported APIs, store a durable operation ID, query the provider for the outcome, or reconcile before retrying. Never assume adding an arbitrary header makes an API idempotent. Retries and errors Effect retry policies count total attempts: maxAttempts: 3 means one initial attempt and up to two retries. Attempt starts and retry deadlines are recorded before the runtime waits, so 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 runtime.effect.failed . A handler can stop that budget early with failEffect and nonRetryable: true , or ask the next attempt to wait at least retryAfterMs without shortening the policy backoff. A worker heartbeat can carry details ; the next dispatch of that effect includes the latest details so the attempt can resume. The callback response is {'{ status, cancelRequested }'} , not the run state. cancelRequested tells an attempt that is already running to stop. Signals and webhook delivery Use the embed start/signal idempotencyKey option with a stable runId for retries of the same logical request. Verify webhook signatures and transform allowed webhook payloads into typed events on the server. Deduplicate by the provider delivery ID and correlate by run, thread, and domain identifiers. Retrying with a fresh key defeats deduplication. Interrupted model streams A model call can produce partial deltas before failing. Those deltas may already be recorded; resuming work may incur another model call and different output. Render partial content as provisional, inspect completion events, and decide how your application handles abandoned partial turns. The adapter does not guarantee exactly-once billing or identical regenerated text. Snapshots, retention, and projections A compatible snapshot bounds replay work; it is not a backup or a schema migration. Trimming events removes older audit and time-travel history and can prevent rebuilding new projections from the beginning. Define a retention and archive policy before enabling trimAfterSnapshot. Projectors run after local commit. Durable delivery stores a cursor per projector name and version, retries without advancing that cursor, and dead-letters exhausted batches. A run can finish before an external index or webhook catches up, so monitor lag and keep projector writes idempotent. Test recovery Test a restart while parked, an interruption during an external action, duplicate webhook delivery, and a deployment with existing runs. The testing guide provides the local restart test. Review versioning before changing definitions. ) }"
   },
   {
     "path": "/docs/schedules",

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

@@ -17,15 +17,51 @@ All built-in stores reject an event over 1 MiB and a snapshot over 8 MiB,
 measured as serialized UTF-8, before storage is mutated. The stable error names
 are `EventPayloadTooLargeError` and `SnapshotPayloadTooLargeError`.
 
+## Run attributes and secret context
+
+`attributes` on `start` are immutable labels (tenant, project, org). They are
+written on `runtime.run.started`, folded into run state, and copied onto the
+run summary. Authorization receives them on `route.attributes` when a run has
+any. They must be safe for every reader of the run.
+
+`secrets` are bearer material for the run. The memory and SQLite stores keep
+them in a side table, `looms_run_secrets`, which is not part of the event log
+or a snapshot. Effect handlers read them on `ctx.secrets`. An in-process worker
+receives them on the dispatch task. A transport that persists that task must
+omit `secrets`. Its worker reads them with
+`POST /runs/:runId/effects/:effectId/:attempt/secrets` (`client.workerSecrets`)
+after its `started` callback is accepted. The route uses `worker` access, so it
+returns 503 until the host configures worker auth. It answers
+`{ status: 'accepted', secrets }` only while that attempt is started and
+unsettled, and `{ status: 'stale' }` or `{ status: 'cancelled' }` otherwise.
+Responses are sent with `cache-control: no-store`. A handler must not copy
+secrets into an emitted event.
+
+A retry of the same start may omit `secrets` and keeps the stored context. Sending
+a different context conflicts, and the error does not include the secret values.
+Forks and effect-launched runs inherit the parent's attributes and secrets.
+A store without `runSecrets` rejects a start that supplies secrets.
+
 ## 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.test.ts

@@ -66,6 +66,30 @@ describe('createLoomsClient', () => {
     ])
   })
 
+  test('workerSecrets posts to the attempt with the worker token', async () => {
+    const calls: { href: string; authorization?: string | null }[] = []
+
+    const client = createLoomsClient({
+      workerCallbackToken: 'worker-token',
+      fetch: async (input, init) => {
+        calls.push({
+          href: hrefOf(input),
+          authorization: new Headers(init?.headers).get('authorization'),
+        })
+
+        return Response.json({ status: 'accepted', secrets: { token: 'abc' } })
+      },
+    })
+
+    const result = await client.workerSecrets('run 1', { effectId: 'eff/1', attempt: 2 })
+
+    expect(result).toEqual({ status: 'accepted', secrets: { token: 'abc' } })
+
+    expect(calls).toEqual([
+      { href: '/runs/run%201/effects/eff%2F1/2/secrets', authorization: 'Bearer worker-token' },
+    ])
+  })
+
   test('listDefinitions reads the definitions route', async () => {
     const calls: string[] = []
 

packages/client/src/client.ts

@@ -7,14 +7,20 @@ import {
   ReplayStepSchema,
   RunStateSchema,
   RunSummariesSchema,
+  SecretContextSchema,
   stringifyJson,
   type DefinitionInput,
   type DefinitionRef,
   type EventEnvelope,
   type EventInput,
   type JsonValue,
+  type RunAttributes,
   type RunState,
+  type SecretContext,
+  type WorkerCallback,
   type WorkerCallbackInput,
+  type WorkerCallbackResult,
+  type WorkerSecretsResult,
 } from '@looms/core'
 
 import { consumeSseStreamEffect, eventsFromSseData, sseFrames } from './sse'
@@ -41,6 +47,62 @@ const StartResultSchema = Schema.Struct({
 
 const RunResultSchema = Schema.Struct({ runId: Schema.String, state: RunStateSchema })
 
+interface WorkerFailRequestBody {
+  error: string
+  nonRetryable?: boolean
+  retryAfterMs?: number
+}
+
+function attemptPath(runId: string, attempt: WorkerCallback): string {
+  return `/runs/${encodeURIComponent(runId)}/effects/${encodeURIComponent(attempt.effectId)}/${attempt.attempt}`
+}
+
+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,
+})
+
+const WorkerSecretsResultSchema = Schema.Union([
+  Schema.Struct({ status: Schema.Literal('accepted'), secrets: SecretContextSchema }),
+  Schema.Struct({ status: Schema.Union([Schema.Literal('stale'), Schema.Literal('cancelled')]) }),
+])
+
 /** `appended` counts the events the run gained while the wake ran. */
 const WakeResultSchema = Schema.Struct({
   runId: Schema.String,
@@ -179,29 +241,35 @@ export function createLoomsClient(options: LoomsClientOptions = {}) {
     definitionVersion?: string
     input?: JsonValue
     runId?: string
+    attributes?: RunAttributes
+    secrets?: SecretContext
   }) => 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 }
-          : {}
-
-    return request(
-      `/runs/${encodeURIComponent(runId)}/effects/${encodeURIComponent(input.effectId)}/${input.attempt}/${input.kind}`,
-      RunResultSchema,
-      {
-        method: 'POST',
-        headers: options.workerCallbackToken
-          ? { authorization: `Bearer ${options.workerCallbackToken}` }
-          : undefined,
-        body: stringifyJson(body),
-      },
-    )
+  const workerHeaders = options.workerCallbackToken
+    ? { authorization: `Bearer ${options.workerCallbackToken}` }
+    : undefined
+
+  const workerCallback = (
+    runId: string,
+    input: WorkerCallbackInput,
+  ): Promise<WorkerCallbackResult> => {
+    const body = workerCallbackBody(input)
+
+    return request(`${attemptPath(runId, input)}/${input.kind}`, WorkerCallbackResultSchema, {
+      method: 'POST',
+      headers: workerHeaders,
+      body: stringifyJson(body),
+    })
   }
 
+  /** Read the run's secret context. Only the attempt that is currently running gets it. */
+  const workerSecrets = (runId: string, attempt: WorkerCallback): Promise<WorkerSecretsResult> =>
+    request(`${attemptPath(runId, attempt)}/secrets`, WorkerSecretsResultSchema, {
+      method: 'POST',
+      headers: workerHeaders,
+      body: '{}',
+    })
+
   const streamRun = (args: {
     kind: string
     definitionName: string
@@ -300,6 +368,7 @@ export function createLoomsClient(options: LoomsClientOptions = {}) {
         body: stringifyJson(forkOptions?.runId === undefined ? {} : { runId: forkOptions.runId }),
       }),
     workerCallback,
+    workerSecrets,
     listRunSummaries: () => request('/runs/summaries', RunSummariesSchema),
     listDefinitions: () => request('/definitions', DefinitionCatalogSchema),
     replayTo: (runId: string, seq: number) =>

packages/core/src/effects.test.ts

@@ -10,6 +10,8 @@ const ctx: EffectContext = {
   runId: 'run_1',
   threadId: 'thr_1',
   causingEventId: 'evt_1',
+  attributes: {},
+  secrets: {},
   signal: new AbortController().signal,
   emit: async () => undefined,
 }