feat(agents): replica-safe wake claims and dones — the backend is the claim authority

.changeset/agents-backend-issued-write-tokens.md

@@ -0,0 +1,6 @@
+---
+'@electric-ax/agents-runtime': patch
+'@electric-ax/agents-server': patch
+---
+
+Adopt backend-issued claim write tokens: when the Durable Streams backend implements the Write Fencing extension, the Agents Server adopts the `write_token` it delivers with wake notifications, pull claims, and heartbeat acks as the claim's write token, and the runtime refreshes its token from heartbeat responses. An opt-in `fencedSessionStreams` server option (env `ELECTRIC_AGENTS_FENCED_SESSION_STREAMS`) creates entity session streams with `Write-Fence: true` and forwards the token plus the fenced-class assertion on runtime appends, so the backend itself is the write authority and rejects deposed or lapsed writers — including across a server restart or from another server instance, where the in-memory token store would not know the claim. In this mode a session-stream create or fork fails unless the backend echoes `Write-Fence: true`, so the option never yields silently unfenced streams. Off by default, and behaviour is unchanged when the backend supplies no token.

.changeset/agents-manifest-wake-replacement.md

@@ -0,0 +1,5 @@
+---
+'@electric-ax/agents-server': patch
+---
+
+Fix a lost `runFinished` wake when a child finishes while its parent's manifest lands. Syncing a manifest entry's wake registration used to unregister the entry's rows and then register the replacement, and a source event evaluated between the two statements — the spawned child's run completing a few milliseconds after the parent's run wrote the child's manifest entry — matched nothing and was never re-evaluated. The registry now replaces a manifest entry's registration in place: the replacement is upserted first (a manifest that re-describes the registration spawn already made resolves to that same row, so nothing is deleted or re-created), and only then are the entry's other rows removed.

.changeset/agents-replica-safe-claims.md

@@ -0,0 +1,5 @@
+---
+'@electric-ax/agents-server': patch
+---
+
+Make wake claims and dones safe across Agents Server instances. A runtime's claim callback is now forwarded to the Durable Streams backend as a lease-renewing callback, like a heartbeat, so the claim adopts the write token the backend re-mints for it rather than one held in the memory of whichever instance happened to receive the delivery; with `fencedSessionStreams` on, a claim the backend issues no token for is refused (`502 WRITE_TOKEN_UNAVAILABLE`) instead of being answered with a token the backend would reject. The claim is materialised in `consumer_claims` once the backend has accepted it, which lets a done that lands on a different instance release it and settle the entity's status (`running` → `idle`, `stopping` → `stopped`) in one conditional update that cannot clobber a newer wake; a redelivered wake's second claim is refused with `409 WAKE_ALREADY_CLAIMED`. Without `fencedSessionStreams` this store remains the sole write authority, so a claim the backend refuses or never answers still falls back to a local mint and the wake runs as before. In-memory claim write tokens expire a grace period past the claim lease unless a heartbeat refreshes them, and a runtime that refuses a forwarded wake reverts the entity to `idle` only if it is still `running`, as a failed forward now does too.

.changeset/agents-runtime-unrecorded-write-failure.md

@@ -0,0 +1,5 @@
+---
+'@electric-ax/agents-runtime': patch
+---
+
+Stop a wake whose failure never reached its stream from consuming its trigger: when the wake's writes failed and the error event recording that failure was itself lost with the producer, the done is sent without acks so the backend re-wakes the entity instead of marking a message answered that no run answered. The runtime also ignores a redelivery of a wake already in flight for the same stream and generation (the backend's retry of a delivery whose 2xx it never saw), and adopts a `write_token` delivered with the wake notification when the claim callback returns none.

packages/agents-runtime/src/create-handler.ts

@@ -148,8 +148,11 @@ export interface RuntimeRouter {
     options?: Pick<ProcessWakeConfig, `claimHeaders` | `claimTokenHeader`>
   ) => void
 
-  /** True when a wake for the stream path is already in flight. */
-  isWakeActive: (streamPath: string) => boolean
+  /**
+   * True when a wake for the stream path is already in flight — with
+   * `epoch`, only one for that generation.
+   */
+  isWakeActive: (streamPath: string, epoch?: number) => boolean
 
   /** Dispatch an already-parsed webhook wake notification. */
   dispatchWebhookWake: (notification: WebhookNotification) => void
@@ -372,6 +375,7 @@ export function createRuntimeRouter(
     process.env.ELECTRIC_AGENTS_DEBUG_REGISTRATION_TIMING === `1`
   const pendingWakes = new Set<Promise<void>>()
   const pendingWakeLabels = new Map<Promise<void>, string>()
+  const pendingWakeEpochs = new Map<Promise<void>, number>()
   const pendingWakeControllers = new Map<Promise<void>, AbortController>()
   const wakeErrors: Array<Error> = []
   const debugCleanup = process.env.ELECTRIC_AGENTS_DEBUG_CLEANUP === `1`
@@ -434,15 +438,26 @@ export function createRuntimeRouter(
       .finally(() => {
         pendingWakes.delete(wake)
         pendingWakeLabels.delete(wake)
+        pendingWakeEpochs.delete(wake)
         pendingWakeControllers.delete(wake)
       })
     pendingWakes.add(wake)
     pendingWakeLabels.set(wake, wakeLabel)
+    pendingWakeEpochs.set(wake, notification.epoch)
     pendingWakeControllers.set(wake, controller)
   }
 
-  const isWakeActive: RuntimeRouter[`isWakeActive`] = (streamPath) =>
-    [...pendingWakeLabels.values()].includes(streamPath)
+  const isWakeActive: RuntimeRouter[`isWakeActive`] = (streamPath, epoch) => {
+    for (const [wake, label] of pendingWakeLabels) {
+      if (
+        label === streamPath &&
+        (epoch === undefined || pendingWakeEpochs.get(wake) === epoch)
+      ) {
+        return true
+      }
+    }
+    return false
+  }
 
   const dispatchWebhookWake: RuntimeRouter[`dispatchWebhookWake`] = dispatchWake
 
@@ -524,6 +539,19 @@ export function createRuntimeRouter(
       )
     }
 
+    if (isWakeActive(notification.streamPath, notification.epoch)) {
+      // The backend redelivers a wake whose 2xx it never saw; a second wake
+      // for the same generation would claim it again (the backend accepts
+      // that — the lease is live) and run the handler twice. The in-flight
+      // wake's done settles the delivery, so acknowledge without starting
+      // another.
+      runtimeLog.warn(
+        `[agent-runtime]`,
+        `wake for ${notification.streamPath} (epoch=${notification.epoch}) is already in flight; ignoring redelivery`
+      )
+      return json({ ok: true }, 200)
+    }
+
     dispatchWebhookWake(notification)
     return json({ ok: true }, 200)
   }

packages/agents-runtime/src/process-wake.ts

@@ -548,6 +548,11 @@ export async function processWake(
       return globalThis.fetch(input, { ...init, headers })
     },
     onError: (error) => {
+      // A batch that fails once this wake's failure is on record may be the
+      // one carrying the error event failBackgroundWake appended; whether it
+      // was is settled at done time against the producer's last written
+      // offset. See errorEventUnrecorded.
+      if (liveProcessError) writeFailedAfterErrorEvent = true
       failBackgroundWake(error, `WRITE_FAILED`)
     },
   })
@@ -634,6 +639,16 @@ export async function processWake(
     close: () => void
   }> = []
   let liveProcessError: Error | null = null
+  // Set when a producer batch fails after failBackgroundWake appended this
+  // wake's error event — possibly the batch carrying it, since batches are
+  // sent concurrently. `producer.lastSuccessfulOffset` as it stood when the
+  // event was appended decides which: an append that landed after it proves
+  // the error event landed too. When it did not, nothing durable records that
+  // the wake failed, so the done carries no acks — the trigger stays pending
+  // and the backend re-wakes the entity instead of consuming a message no run
+  // answered, the same shape as a crash mid-wake.
+  let writeFailedAfterErrorEvent = false
+  let offsetBeforeErrorEvent: string | undefined
   let acceptLiveInputs = false
   const handledSignalKeys = new Set<string>()
 
@@ -701,6 +716,7 @@ export async function processWake(
 
     liveProcessError = toError(err)
     log.error(`wake background task failed for ${entityUrl}:`, liveProcessError)
+    offsetBeforeErrorEvent = producer.lastSuccessfulOffset
     writeEvent(
       entityStateSchema.errors.insert({
         key: `error-${epoch}-${crypto.randomUUID()}`,
@@ -1222,7 +1238,7 @@ export async function processWake(
       return null
     }
     claimedWake = true
-    writeToken = claimed.writeToken ?? ``
+    writeToken = claimed.writeToken ?? notification.write_token ?? ``
 
     handleRuntimeSideEffectEvents(catchUpEvents)
 
@@ -1248,6 +1264,7 @@ export async function processWake(
             ok: boolean
             claimToken?: string
             token?: string
+            writeToken?: string
           }
           if (!data.ok) {
             failBackgroundWake(
@@ -1258,6 +1275,10 @@ export async function processWake(
           }
           if (data.claimToken) activeClaimToken = data.claimToken
           if (data.token) activeClaimToken = data.token
+          // The server may re-mint the write token on each heartbeat (its
+          // backend's token TTL tracks the claim lease); adopt the refresh
+          // so appends from a long-running activation keep a live token.
+          if (data.writeToken) writeToken = data.writeToken
         })
         .catch((err: unknown) => {
           failBackgroundWake(err, `HEARTBEAT_FAILED`)
@@ -2556,7 +2577,12 @@ export async function processWake(
         cleanupErrors.push(toError(err))
       }
     }
-    const doneOffset = safeAckOffset
+    // Every batch has been flushed by now, so an offset past the one the
+    // error event was queued behind is an append that succeeded after it.
+    const errorEventUnrecorded =
+      writeFailedAfterErrorEvent &&
+      producer.lastSuccessfulOffset === offsetBeforeErrorEvent
+    const doneOffset = errorEventUnrecorded ? `-1` : safeAckOffset
     for (const sdb of secondaryDbs) {
       try {
         await sdb.flushWrites?.()
@@ -2595,11 +2621,22 @@ export async function processWake(
       }
     }
     if (claimedWake) {
-      log.info(
-        doneOffset === `-1`
-          ? `done without ack (no consumed offset)`
-          : `done acking ${streamPath} at ${doneOffset}`
-      )
+      if (errorEventUnrecorded) {
+        // Warn, not info: the trigger stays pending, so the backend re-wakes
+        // this entity immediately and the whole handler — model and tool
+        // calls included — runs again. One line is a duplicated run; a stream
+        // of them for the same entity is a wake loop over a write failure
+        // that is not clearing, and the writes are what to fix.
+        log.warn(
+          `done without ack for ${streamPath} (epoch=${epoch}): the write failure was never recorded on the stream, leaving the wake pending for redelivery`
+        )
+      } else {
+        log.info(
+          doneOffset === `-1`
+            ? `done without ack (no consumed offset)`
+            : `done acking ${streamPath} at ${doneOffset}`
+        )
+      }
       if (shutdownRequested) {
         log.info(`shutdown requested, sending done callback at checkpoint`)
       }

packages/agents-runtime/src/types.ts

@@ -704,6 +704,13 @@ export interface WebhookNotification {
   triggeredBy?: Array<string>
   callback: string
   claimToken: string
+  /**
+   * The wake's write token as the Durable Streams backend delivered it
+   * (Write Fencing extension), passed through by the server. The claim
+   * callback's `writeToken` is authoritative; this is the fallback for a
+   * server that adopts none.
+   */
+  write_token?: string
   triggerEvent?: string
   wakeEvent?: WakeEvent
   entity?: {

packages/agents-runtime/test/create-handler.test.ts

@@ -244,6 +244,67 @@ describe(`createRuntimeHandler`, () => {
     })
   })
 
+  it(`acknowledges a redelivered wake for an in-flight (stream, epoch) without starting it again`, async () => {
+    defineEntity(`test-agent`, { handler: async () => {} })
+
+    const resolvers: Array<() => void> = []
+    processWakeMock.mockImplementation(
+      () =>
+        new Promise<void>((resolve) => {
+          resolvers.push(resolve)
+        })
+    )
+
+    const notification = {
+      consumerId: `wake-1`,
+      epoch: 1,
+      wakeId: `wake-1`,
+      streamPath: `/streams/entity:test-1`,
+      streams: [{ path: `/streams/entity:test-1`, offset: `0_0` }],
+      callback: `http://localhost:3000/_electric/wakes/wake-1`,
+      claimToken: `tok-1`,
+      entity: {
+        type: `test-agent`,
+        status: `active`,
+        url: `http://localhost:3000/test-agent/test-1`,
+        streams: {
+          main: `/streams/entity:test-1`,
+        },
+      },
+    }
+
+    const handler = createRuntimeHandler({
+      baseUrl: `http://localhost:3000`,
+      handlerUrl: `http://localhost:4000/electric-agents`,
+      webhookSignature: false,
+    })
+    const deliver = (body: unknown): Promise<Response> =>
+      handler.handleWebhookRequest(
+        new Request(`http://localhost/electric-agents`, {
+          method: `POST`,
+          headers: { 'content-type': `application/json` },
+          body: JSON.stringify(body),
+        })
+      )
+
+    expect((await deliver(notification)).status).toBe(200)
+    // The backend's retry of the same wake (its 2xx was lost) must be
+    // acknowledged — the in-flight wake's done settles it — not run twice.
+    expect((await deliver(notification)).status).toBe(200)
+    expect(processWakeMock).toHaveBeenCalledTimes(1)
+
+    // A later generation for the same stream is a new wake, not a redelivery.
+    expect(
+      (await deliver({ ...notification, epoch: 2, wakeId: `wake-2` })).status
+    ).toBe(200)
+    expect(processWakeMock).toHaveBeenCalledTimes(2)
+    expect(handler.debugState()).toMatchObject({ pendingWakeCount: 2 })
+
+    for (const resolve of resolvers) resolve()
+    await handler.waitForSettled()
+    expect(handler.isWakeActive(`/streams/entity:test-1`, 1)).toBe(false)
+  })
+
   it(`records wake errors in debugState() until drained`, async () => {
     defineEntity(`test-agent`, { handler: async () => {} })
     processWakeMock.mockRejectedValueOnce(new Error(`wake failed`))