From 93bc0b879c18a7bf411a5d175b4898916e9e0190 Mon Sep 17 00:00:00 2001 From: Andres Berrios Date: Wed, 15 Jul 2026 18:34:11 +0200 Subject: [PATCH 1/3] fix(agents-server): drop of parent wake on parallel sub-agent spawn MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Each child's runFinished wake is registered from two paths (spawn + manifest-sync) keyed by the same manifestKey. The second insert hits the uq_wake_registration unique constraint and takes the conflict branch, which called loadRegistrations() — a full clear-and-rebuild of the in-memory registration cache from a snapshot read across an await. Under parallel spawn several of these reloads interleave. A stale snapshot whose continuation lands last clears the whole cache and rebuilds it without a sibling's newer registration, evicting it. When that sibling later finishes, evaluate() finds no cache entry and the wake is silently dropped (no error path). Sequential spawn never overlaps the reloads, so only the parallel fan-out reproduced it. Re-read only the single conflicting row and cache just that entry, leaving sibling registrations untouched. Add a regression test that registers two siblings for the same subscriber with the second hitting the conflict branch against a stale full-table snapshot, and asserts both still evaluate. --- .../wake-registry-parallel-spawn-clobber.md | 5 + packages/agents-server/src/wake-registry.ts | 47 +++++++++- .../agents-server/test/wake-registry.test.ts | 91 +++++++++++++++++++ 3 files changed, 139 insertions(+), 4 deletions(-) create mode 100644 .changeset/wake-registry-parallel-spawn-clobber.md diff --git a/.changeset/wake-registry-parallel-spawn-clobber.md b/.changeset/wake-registry-parallel-spawn-clobber.md new file mode 100644 index 0000000000..ecd16a309c --- /dev/null +++ b/.changeset/wake-registry-parallel-spawn-clobber.md @@ -0,0 +1,5 @@ +--- +"@electric-ax/agents-server": patch +--- + +Fix a dropped parent wake when a parent spawns sub-agents in parallel. Each child's `runFinished` wake is registered from two paths (spawn + manifest-sync) keyed by the same `manifestKey`; the second insert hits `uq_wake_registration` and takes the conflict branch. That branch called `loadRegistrations()`, a full clear-and-rebuild of the in-memory registration cache from a snapshot read across an `await`. Under parallel spawn several such reloads interleave, and a stale snapshot landing last evicts a sibling's newer registration from the cache — so when that sibling finishes, `evaluate()` finds no match and the wake is silently dropped (no error). Sequential spawn never overlaps the reloads, which is why only the parallel fan-out reproduced it. The conflict branch now re-reads only the single conflicting row and caches just that entry, leaving sibling registrations untouched. diff --git a/packages/agents-server/src/wake-registry.ts b/packages/agents-server/src/wake-registry.ts index ff8ef6e75b..d0b3ed7785 100644 --- a/packages/agents-server/src/wake-registry.ts +++ b/packages/agents-server/src/wake-registry.ts @@ -3,7 +3,7 @@ import { isChangeMessage, isControlMessage, } from '@electric-sql/client' -import { and, eq } from 'drizzle-orm' +import { and, eq, isNull } from 'drizzle-orm' import { wakeRegistrations } from './db/schema.js' import { serverLog } from './utils/log.js' import { electricUrlWithPath } from './utils/electric-url.js' @@ -344,9 +344,48 @@ export class WakeRegistry { .returning({ id: wakeRegistrations.id }) if (result.length === 0) { - // Another path (e.g. manifest-sync) may have created the row first. - // Refresh the cache so this process still sees the effective registration. - await this.loadRegistrations() + // Another path (e.g. manifest-sync) created the row first. Re-read only + // the conflicting row and cache just that entry. A full loadRegistrations() + // here clears and rebuilds the entire cache from a snapshot taken across an + // await; when several register() calls conflict concurrently (parallel + // sub-agent spawn), a stale snapshot landing last can evict a sibling's + // newer registration, silently dropping its wake. See uq_wake_registration. + const existing = await this.db + .select() + .from(wakeRegistrations) + .where( + and( + eq(wakeRegistrations.tenantId, tenantId), + eq(wakeRegistrations.subscriberUrl, reg.subscriberUrl), + eq(wakeRegistrations.sourceUrl, reg.sourceUrl), + eq(wakeRegistrations.oneShot, reg.oneShot), + eq(wakeRegistrations.debounceMs, reg.debounceMs ?? 0), + eq(wakeRegistrations.timeoutMs, reg.timeoutMs ?? 0), + eq(wakeRegistrations.condition, reg.condition), + reg.manifestKey == null + ? isNull(wakeRegistrations.manifestKey) + : eq(wakeRegistrations.manifestKey, reg.manifestKey) + ) + ) + .limit(1) + + const row = existing[0] + if (row) { + this.upsertCachedRegistration({ + tenantId: row.tenantId, + subscriberUrl: row.subscriberUrl, + sourceUrl: row.sourceUrl, + condition: row.condition as WakeRegistration[`condition`], + debounceMs: row.debounceMs || undefined, + timeoutMs: row.timeoutMs || undefined, + oneShot: row.oneShot, + includeResponse: row.includeResponse === false ? false : undefined, + manifestKey: row.manifestKey ?? undefined, + dbId: row.id, + createdAt: row.createdAt, + timeoutConsumed: row.timeoutConsumed, + }) + } return } diff --git a/packages/agents-server/test/wake-registry.test.ts b/packages/agents-server/test/wake-registry.test.ts index c879a124ca..5b9460dd3f 100644 --- a/packages/agents-server/test/wake-registry.test.ts +++ b/packages/agents-server/test/wake-registry.test.ts @@ -525,6 +525,97 @@ describe(`Wake Registry`, () => { ).rejects.toThrow(`connection refused`) }) + it(`register() conflict re-reads only the conflicting row without evicting siblings`, async () => { + // Regression for the parallel sub-agent spawn dropped-wake race: when two + // register() calls for the same subscriber land at nearly the same instant + // (spawn path + manifest-sync path), the one that hits the unique constraint + // must NOT clear-and-rebuild the whole cache from a possibly-stale snapshot. + // Doing so evicts a sibling's already-cached registration, so its runFinished + // wake is silently dropped (evaluate returns []). The fix re-reads only the + // single conflicting row and caches just that, leaving siblings untouched. + const conflictRow = { + id: 2, + tenantId: `default`, + subscriberUrl: `/parent/p1`, + sourceUrl: `/child/c2`, + condition: `runFinished`, + debounceMs: 0, + timeoutMs: 0, + oneShot: false, + timeoutConsumed: false, + includeResponse: true, + manifestKey: null, + createdAt: new Date(), + } + // A full clear-and-rebuild would read this STALE snapshot — which is missing + // the already-cached sibling (row 1) — and evict it. Models the interleaved + // reload whose late-landing continuation clobbers a newer registration. + const staleSnapshot = [conflictRow] + let conflictNext = false + const db: any = { + insert: () => ({ + values: () => ({ + onConflictDoNothing: () => ({ + returning: () => + conflictNext ? Promise.resolve([]) : Promise.resolve([{ id: 1 }]), + }), + }), + }), + delete: () => ({ where: () => Promise.resolve() }), + update: () => ({ set: () => ({ where: () => Promise.resolve() }) }), + select: () => ({ + from: () => + Object.assign(Promise.resolve(staleSnapshot), { + // loadRegistrations (the pre-fix reload) awaits this directly. + where: () => + Object.assign(Promise.resolve(staleSnapshot), { + // Targeted single-row re-read (the fix) resolves the conflicting row. + limit: () => Promise.resolve([conflictRow]), + orderBy: () => Promise.resolve([]), + }), + }), + }), + } + + const registry = new WakeRegistry(db) + + // First registration succeeds and is cached as row 1. + await registry.register({ + subscriberUrl: `/parent/p1`, + sourceUrl: `/child/c1`, + condition: `runFinished`, + oneShot: false, + }) + + // Second registration hits the unique constraint (the other path created it). + conflictNext = true + await registry.register({ + subscriberUrl: `/parent/p1`, + sourceUrl: `/child/c2`, + condition: `runFinished`, + oneShot: false, + }) + + // Both siblings must still evaluate — the conflict must not have clobbered row 1. + const first = registry.evaluate(`/child/c1`, { + type: `run`, + key: `run-1`, + value: { status: `completed` }, + headers: { operation: `update` }, + }) + const second = registry.evaluate(`/child/c2`, { + type: `run`, + key: `run-2`, + value: { status: `completed` }, + headers: { operation: `update` }, + }) + + expect(first).toHaveLength(1) + expect(first[0]!.subscriberUrl).toBe(`/parent/p1`) + expect(second).toHaveLength(1) + expect(second[0]!.subscriberUrl).toBe(`/parent/p1`) + }) + it(`rebuilds registry from register calls`, async () => { const registry = new WakeRegistry(createMockDb()) await registry.register({ From deb8c53d6cf8aec6c23e10ca2a9bbe99d6fffd68 Mon Sep 17 00:00:00 2001 From: Andres Berrios Date: Thu, 16 Jul 2026 03:08:26 +0200 Subject: [PATCH 2/3] fix(agents-server): drop of sibling sub-agent wakes on parallel spawn When a parent spawns 2+ sub-agents in one turn, only some children's runFinished wakes reached the parent; the rest were silently dropped. Root cause is in ElectricAgentsTenantRuntime.syncManifestWakes: for each manifest `upsert` it ran `unregisterByManifestKey()` then `await register()`. That non-atomically removes the wake registration from the in-memory cache and only re-adds it after a DB round-trip. Every spawned child re-syncs a registration identical to the one the spawn already created, so each child's registration churns through a remove -> re-add window; a sibling whose run finishes inside that window evaluates against a cache with no matching registration and its wake is lost. The earlier parallel-spawn-clobber fix guarded register()'s on-conflict path, which distinct-source children never exercise, so it did not cover this case. Fix: WakeRegistry.reconcileManifestRegistration registers the desired registration first (register() now returns the dbId it left in the cache: the fresh insert, or the pre-existing row re-read on conflict), then deletes only the other rows anchored to that manifest key. The registration is continuously present, so no in-flight wake falls through a gap. Pruning by the exact kept id (never a re-derived field key) guarantees the row just kept is never removed. Verified with a new pull-wake e2e (parent spawns 6 workers in parallel via the real runner + mock LLM): every child's wake reaches the parent's stream across 15/15 runs; it failed ~1-in-1..1-in-3 before. Adds two deterministic unit tests for the reconcile contract. Existing wake-registry suites unchanged (36/36). Note: the pull-wake runner legitimately coalesces multiple wake events into one handler invocation, so the assertion is at the production layer (one wake event per child on the parent stream), not handler-invocation count. Co-Authored-By: Claude Opus 4.8 (1M context) --- packages/agents-server/src/runtime.ts | 27 +- packages/agents-server/src/wake-registry.ts | 86 ++++- .../test/parallel-spawn-wake-repro.test.ts | 349 ++++++++++++++++++ .../agents-server/test/wake-registry.test.ts | 125 +++++++ 4 files changed, 572 insertions(+), 15 deletions(-) create mode 100644 packages/agents-server/test/parallel-spawn-wake-repro.test.ts diff --git a/packages/agents-server/src/runtime.ts b/packages/agents-server/src/runtime.ts index c545fa714c..d44131419e 100644 --- a/packages/agents-server/src/runtime.ts +++ b/packages/agents-server/src/runtime.ts @@ -242,31 +242,34 @@ export class ElectricAgentsTenantRuntime { if (!manifestKey) continue if (operation === `delete`) { - await this.manager.wakeRegistry.unregisterByManifestKey( + await this.manager.wakeRegistry.reconcileManifestRegistration( subscriberUrl, manifestKey, + null, this.serviceId ) continue } - await this.manager.wakeRegistry.unregisterByManifestKey( + // Reconcile idempotently and WITHOUT a delivery gap. The old + // unregister-then-register sequence briefly removed the registration + // from the cache; a source (e.g. a sibling sub-agent) that finished in + // that window had its wake dropped. reconcileManifestRegistration + // registers the desired reg first, then prunes only stale rows. + const reg = value + ? buildManifestWakeRegistration(subscriberUrl, value, manifestKey) + : null + if (reg) { + reg.tenantId = this.serviceId + } + await this.manager.wakeRegistry.reconcileManifestRegistration( subscriberUrl, manifestKey, + reg, this.serviceId ) if (value) { - const reg = buildManifestWakeRegistration( - subscriberUrl, - value, - manifestKey - ) - if (reg) { - reg.tenantId = this.serviceId - await this.manager.wakeRegistry.register(reg) - } - const cronSpec = extractManifestCronSpec(value) if (cronSpec) { void this.manager diff --git a/packages/agents-server/src/wake-registry.ts b/packages/agents-server/src/wake-registry.ts index d0b3ed7785..bc085b7f44 100644 --- a/packages/agents-server/src/wake-registry.ts +++ b/packages/agents-server/src/wake-registry.ts @@ -3,7 +3,7 @@ import { isChangeMessage, isControlMessage, } from '@electric-sql/client' -import { and, eq, isNull } from 'drizzle-orm' +import { and, eq, isNull, ne } from 'drizzle-orm' import { wakeRegistrations } from './db/schema.js' import { serverLog } from './utils/log.js' import { electricUrlWithPath } from './utils/electric-url.js' @@ -325,7 +325,15 @@ export class WakeRegistry { return this.syncRecoveryPromise } - async register(reg: WakeRegistration): Promise { + /** + * Register a wake subscription. Returns the dbId of the row now backing the + * cached registration — the freshly-inserted row, or (on unique-constraint + * conflict) the pre-existing row that was re-read and re-cached. Returns -1 + * only in the pathological case where the conflicting row could not be + * re-read. Callers reconciling a set of registrations use the returned id to + * prune siblings unambiguously (see reconcileManifestRegistration). + */ + async register(reg: WakeRegistration): Promise { const tenantId = this.resolveTenantId(reg.tenantId) const result = await this.db .insert(wakeRegistrations) @@ -385,8 +393,9 @@ export class WakeRegistry { createdAt: row.createdAt, timeoutConsumed: row.timeoutConsumed, }) + return row.id } - return + return -1 } const dbId = result[0]!.id @@ -397,6 +406,7 @@ export class WakeRegistry { createdAt: new Date(), timeoutConsumed: false, }) + return dbId } private startTimeoutTimer(reg: CachedWakeRegistration, dbId: number): void { @@ -452,6 +462,76 @@ export class WakeRegistry { } } + /** + * Idempotently reconcile the single wake registration anchored to a manifest + * entry, without ever leaving a delivery gap. + * + * The obvious sequence — `unregisterByManifestKey()` then `register()` — drops + * the registration from the cache and only re-adds it after an async DB + * round-trip. A source whose run finishes inside that window evaluates against + * an empty cache and its wake is silently lost. Parallel sub-agent spawn hits + * this constantly: every child's manifest entry re-syncs a registration that + * is *identical* to the one the spawn already created, so each child gets a + * remove→re-add churn, and any sibling that finishes mid-churn is dropped. + * + * Instead we register the desired registration FIRST — register() returns the + * id of the row it left in the cache (a fresh insert, or the pre-existing row + * re-read on conflict), so an equivalent registration is continuously present + * — THEN delete only the rows for this manifest key that differ from it. + * Pruning by that exact id (never by a re-derived field key, which can diverge + * from what register() actually cached) guarantees we never delete the row we + * just kept. `desired == null` (a manifest delete, or an entry carrying no + * wake) prunes them all, matching the previous unregister-only behaviour. + */ + async reconcileManifestRegistration( + subscriberUrl: string, + manifestKey: string, + desired: WakeRegistration | null, + tenantId?: string + ): Promise { + const resolvedTenantId = this.resolveTenantId(tenantId) + + let keptDbId = -1 + if (desired) { + keptDbId = await this.register({ + ...desired, + tenantId: resolvedTenantId, + manifestKey, + }) + } + + // Delete every other row anchored to this manifest key in one predicate — + // covering rows that are not currently cached, so a later loadRegistrations() + // can't resurrect them. `keptDbId === -1` (a delete/no-wake reconcile) + // matches all rows for the key, preserving the old unregister-only delete. + await this.db + .delete(wakeRegistrations) + .where( + and( + eq(wakeRegistrations.tenantId, resolvedTenantId), + eq(wakeRegistrations.subscriberUrl, subscriberUrl), + eq(wakeRegistrations.manifestKey, manifestKey), + ne(wakeRegistrations.id, keptDbId) + ) + ) + + const staleDbIds = Array.from(this.registrationCache.values()).flatMap( + (regs) => + regs + .filter( + (r) => + r.tenantId === resolvedTenantId && + r.subscriberUrl === subscriberUrl && + r.manifestKey === manifestKey && + r.dbId !== keptDbId + ) + .map((r) => r.dbId) + ) + for (const dbId of staleDbIds) { + this.removeCachedRegistrationByDbId(dbId) + } + } + async unregisterBySubscriber( subscriberUrl: string, tenantId?: string diff --git a/packages/agents-server/test/parallel-spawn-wake-repro.test.ts b/packages/agents-server/test/parallel-spawn-wake-repro.test.ts new file mode 100644 index 0000000000..3a84acaa0a --- /dev/null +++ b/packages/agents-server/test/parallel-spawn-wake-repro.test.ts @@ -0,0 +1,349 @@ +/** + * Reproduction: parent spawns N sub-agents in one turn (parallel), and must be + * woken once per child when each child's run finishes. + * + * Mirrors the domo app: a single pull-wake runner hosts a `parent` agent (our + * custom deterministic handler) plus the built-in `worker`. The parent, on its + * first inbox wake, spawns N workers concurrently with a `runFinished` wake on + * each, then ends its turn. Each worker runs (mock LLM) and finishes; the server + * should deliver a wake to the parent for EVERY child. + * + * Bug (AGENTS.md gotcha 4): only one child's runFinished wakes the parent; the + * others are silently dropped. This test asserts the parent observes ALL N. + */ +import { afterAll, beforeAll, describe, expect, it, vi } from 'vitest' +import { DurableStreamTestServer } from '@durable-streams/server' +import { createBuiltinAgentHandler } from '../../agents/src/bootstrap' +import { createPullWakeRunner } from '@electric-ax/agents-runtime' +import { ElectricAgentsServer } from '../src/server' +import { parsePrincipalKey } from '../src/principal' +import { + durableStreamTestServerUrl, + readStreamEvents, + waitFor, +} from './test-utils' +import { + TEST_POSTGRES_URL, + resetElectricAgentsTestBackend, +} from './test-backend' +import type { HandlerContext, WakeEvent } from '@electric-ax/agents-runtime' +import type { StreamFn } from '@mariozechner/pi-agent-core' + +const CHILD_COUNT = 6 + +// Per-parent record of which child sources have woken it. +const parentWakes = new Map>() +// Per-parent record of the full wake invocation log (for diagnostics). +const parentWakeLog = new Map>() +const parentSpawned = new Set() +const spawnedChildren = new Map>() + +function record( + map: Map>, + parent: string, + child: string +): void { + const set = map.get(parent) ?? new Set() + set.add(child) + map.set(parent, set) +} + +function createMockStreamFn(responseText: string): StreamFn { + return vi.fn(((model) => { + const message = { + role: `assistant`, + content: [{ type: `text`, text: responseText }], + api: model.api, + provider: model.provider, + model: model.id, + usage: { + input: 0, + output: 0, + cacheRead: 0, + cacheWrite: 0, + totalTokens: 0, + cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 }, + }, + stopReason: `stop`, + timestamp: Date.now(), + } as any + const events = [ + { type: `start`, partial: { ...message, content: [] } }, + { + type: `text_start`, + contentIndex: 0, + partial: { ...message, content: [{ type: `text`, text: `` }] }, + }, + { + type: `text_delta`, + contentIndex: 0, + delta: responseText, + partial: message, + }, + { + type: `text_end`, + contentIndex: 0, + content: responseText, + partial: message, + }, + { type: `done`, reason: `stop`, message }, + ] as any[] + return { + async *[Symbol.asyncIterator]() { + for (const event of events) yield event + }, + result: async () => message, + } as any + }) as StreamFn) +} + +describe(`parallel sub-agent spawn wake delivery`, () => { + let dsServer: DurableStreamTestServer + let electricAgentsServer: ElectricAgentsServer + let bootstrap: NonNullable< + Awaited> + > + let puller: ReturnType + let baseUrl = `` + let streamBaseUrl = `` + const runnerId = `parallel-spawn-repro-runner` + const authHeaders = { authorization: `Bearer test-token` } + const testPrincipal = parsePrincipalKey(`user:test-user`) + const mockStreamFn = createMockStreamFn(`mock child response`) + + beforeAll(async () => { + dsServer = new DurableStreamTestServer({ + port: 0, + longPollTimeout: 500, + webhooks: true, + }) + await Promise.all([resetElectricAgentsTestBackend(), dsServer.start()]) + + electricAgentsServer = new ElectricAgentsServer({ + durableStreamsUrl: durableStreamTestServerUrl(dsServer.url), + port: 0, + postgresUrl: TEST_POSTGRES_URL, + electricUrl: undefined, + authenticateRequest: (req) => + req.headers.get(`authorization`) === authHeaders.authorization + ? testPrincipal + : null, + }) + baseUrl = await electricAgentsServer.start() + streamBaseUrl = electricAgentsServer.streamClient.baseUrl + + const created = await createBuiltinAgentHandler({ + agentServerUrl: baseUrl, + workingDirectory: process.cwd(), + streamFn: mockStreamFn, + serverHeaders: authHeaders, + defaultDispatchPolicyForType: () => ({ + targets: [{ type: `runner`, runnerId }], + }), + }) + if (!created) throw new Error(`bootstrap failed (no model catalog)`) + bootstrap = created + + // Custom deterministic parent: on first inbox wake, spawn CHILD_COUNT workers + // in parallel each with a runFinished wake back to us; on later wakes, record + // which child fired. No LLM run in the parent — keeps spawn timing exact. + bootstrap.registry.define(`parent`, { + description: `Repro parent that spawns ${CHILD_COUNT} workers in parallel`, + permissionGrants: [ + { + subject_kind: `principal_kind`, + subject_value: `user`, + permission: `spawn`, + }, + { + subject_kind: `principal_kind`, + subject_value: `user`, + permission: `manage`, + }, + ], + async handler(ctx: HandlerContext, wake: WakeEvent) { + const self = ctx.entityUrl + const log = parentWakeLog.get(self) ?? [] + log.push({ type: wake.type, source: wake.source }) + parentWakeLog.set(self, log) + + const isChildWake = + wake.source !== self && wake.source.startsWith(`/worker/`) + if (isChildWake) { + record(parentWakes, self, wake.source) + return + } + + if (parentSpawned.has(self)) return + parentSpawned.add(self) + + const ids = Array.from( + { length: CHILD_COUNT }, + (_, i) => `child-${i}-${Math.random().toString(36).slice(2, 8)}` + ) + spawnedChildren.set( + self, + ids.map((id) => `/worker/${id}`) + ) + await Promise.all( + ids.map((id) => + ctx.spawn( + `worker`, + id, + { + systemPrompt: `You are a worker. Reply "done".`, + tools: [`send`], + }, + { + initialMessage: `Do your task and finish.`, + wake: { on: `runFinished`, includeResponse: true }, + } + ) + ) + ) + }, + }) + + await bootstrap.runtime.registerTypes() + + // Register the runner row (advertise sandbox profiles), then start the puller. + const regRes = await fetch(`${baseUrl}/_electric/runners`, { + method: `POST`, + headers: { 'content-type': `application/json`, ...authHeaders }, + body: JSON.stringify({ + id: runnerId, + owner_principal: testPrincipal.url, + label: `Parallel spawn repro`, + kind: `local`, + admin_status: `enabled`, + sandbox_profiles: bootstrap.runtime.sandboxProfileDescriptors, + }), + }) + if (!regRes.ok) + throw new Error( + `runner registration failed: ${regRes.status} ${await regRes.text()}` + ) + const registration = (await regRes.json()) as { + wake_stream_offset?: string + } + + puller = createPullWakeRunner({ + baseUrl, + runnerId, + runtime: bootstrap.runtime, + headers: authHeaders, + claimHeaders: authHeaders, + claimTokenHeader: `electric-claim-token`, + offset: registration.wake_stream_offset, + onError: (error) => + console.error(`[repro] pull-wake runner error:`, error), + }) + puller.start() + }, 120_000) + + afterAll(async () => { + await puller?.stop().catch(() => {}) + bootstrap?.runtime.abortWakes() + await bootstrap?.runtime.drainWakes().catch(() => {}) + await bootstrap?.shutdownSandboxes?.().catch(() => {}) + await Promise.allSettled([electricAgentsServer?.stop(), dsServer?.stop()]) + }, 120_000) + + it(`wakes the parent once per parallel child (${CHILD_COUNT} children)`, async () => { + const id = `p-${Date.now()}` + const parentUrl = `/parent/${id}` + const entityApiUrl = `${baseUrl}/_electric/entities/parent/${id}` + + const spawnRes = await fetch(entityApiUrl, { + method: `PUT`, + headers: { 'content-type': `application/json`, ...authHeaders }, + body: JSON.stringify({}), + }) + expect(spawnRes.status).toBe(201) + + const sendRes = await fetch(`${entityApiUrl}/send`, { + method: `POST`, + headers: { 'content-type': `application/json`, ...authHeaders }, + body: JSON.stringify({ + from: testPrincipal.url, + payload: `Kick off the workers.`, + }), + }) + expect(sendRes.status).toBeLessThan(300) + + // Wait for the parent to have spawned its children. + await waitFor( + async () => (spawnedChildren.get(parentUrl)?.length ?? 0) === CHILD_COUNT, + 20_000, + 100 + ) + + const children = spawnedChildren.get(parentUrl) ?? [] + + // The wake-delivery GUARANTEE we assert is at the production layer: every + // child's runFinished must append a distinct `wake` event to the parent's + // stream (source = child url). Bug #1 (the registration-churn gap) drops + // some of these events entirely. We deliberately do NOT assert on how many + // times the parent HANDLER ran: the pull-wake runner legitimately coalesces + // several wake events into one handler invocation (a range wake), and a + // real parent reads its child-status state rather than the single triggering + // source — so handler-level coalescing loses no information and is expected. + const wakeSources = async (): Promise> => { + const events = await readStreamEvents( + streamBaseUrl, + `${parentUrl}/main` + ).catch(() => []) + const sources = new Set() + for (const e of events) { + const isWake = + (e as any).type === `wake` || (e as any).value?.type === `wake` + if (!isWake) continue + const src = ((e as any).value?.value ?? (e as any).value)?.source + if (typeof src === `string`) sources.add(src) + } + return sources + } + + let timedOut = false + try { + await waitFor( + async () => { + const s = await wakeSources() + return children.every((c) => s.has(c)) + }, + 30_000, + 200 + ) + } catch { + timedOut = true + } + + const delivered = await wakeSources() + const missing = children.filter((c) => !delivered.has(c)) + + if (missing.length > 0) { + console.error( + `\n[repro] MISSING WAKES for ${missing.length}/${CHILD_COUNT} children (no wake event on parent stream):`, + missing + ) + console.error( + `[repro] parent handler wake log (coalescing is OK):`, + parentWakeLog.get(parentUrl) + ) + for (const child of children) { + const events = await readStreamEvents( + streamBaseUrl, + `${child}/main` + ).catch(() => []) + const runs = events.filter( + (e) => (e as any).type === `run` || (e as any).value?.type === `run` + ) + console.error( + `[repro] child ${child}: ${events.length} events, ${runs.length} run events, delivered=${delivered.has(child)}` + ) + } + } + + expect({ timedOut, missing }).toEqual({ timedOut: false, missing: [] }) + }, 90_000) +}) diff --git a/packages/agents-server/test/wake-registry.test.ts b/packages/agents-server/test/wake-registry.test.ts index 5b9460dd3f..f97483c65f 100644 --- a/packages/agents-server/test/wake-registry.test.ts +++ b/packages/agents-server/test/wake-registry.test.ts @@ -947,6 +947,131 @@ describe(`Wake Registry`, () => { result.wakeMessage.changes[result.wakeMessage.changes.length - 1]! expect(lastChange.key).toBe(`run-2`) }) + + it(`reconcileManifestRegistration keeps an equivalent registration without a churn gap`, async () => { + // Regression for the parallel sub-agent spawn dropped-wake race. A child's + // spawn creates a runFinished registration; its manifest entry then re-syncs + // the SAME registration. The old syncManifestWakes did unregister→register, + // leaving the cache empty across an await — a sibling finishing there lost + // its wake. reconcileManifestRegistration must register-first and prune only + // OTHER rows, so the equivalent registration is never removed. + const seededRow = { + id: 7, + tenantId: `default`, + subscriberUrl: `/parent/p1`, + sourceUrl: `/worker/c1`, + condition: `runFinished`, + debounceMs: 0, + timeoutMs: 0, + oneShot: false, + timeoutConsumed: false, + includeResponse: true, + manifestKey: `child:worker:c1`, + createdAt: new Date(), + } + let insertConflicts = false + const db: any = { + insert: () => ({ + values: () => ({ + onConflictDoNothing: () => ({ + returning: () => + insertConflicts + ? Promise.resolve([]) + : Promise.resolve([{ id: seededRow.id }]), + }), + }), + }), + delete: () => ({ where: () => Promise.resolve() }), + update: () => ({ set: () => ({ where: () => Promise.resolve() }) }), + select: () => ({ + from: () => + Object.assign(Promise.resolve([seededRow]), { + where: () => + Object.assign(Promise.resolve([seededRow]), { + limit: () => Promise.resolve([seededRow]), + orderBy: () => Promise.resolve([]), + }), + }), + }), + } + + const registry = new WakeRegistry(db) + const removeSpy = vi.spyOn( + registry as any, + `removeCachedRegistrationByDbId` + ) + + // Spawn-path registration (fresh insert → dbId 7 cached). + await registry.register({ + subscriberUrl: `/parent/p1`, + sourceUrl: `/worker/c1`, + condition: `runFinished`, + oneShot: false, + includeResponse: true, + manifestKey: `child:worker:c1`, + }) + + // Manifest re-sync of the identical registration now conflicts on insert. + insertConflicts = true + await registry.reconcileManifestRegistration( + `/parent/p1`, + `child:worker:c1`, + { + subscriberUrl: `/parent/p1`, + sourceUrl: `/worker/c1`, + condition: `runFinished`, + oneShot: false, + includeResponse: true, + manifestKey: `child:worker:c1`, + } + ) + + // The registration survived the reconcile — evaluate still delivers, and the + // kept row (dbId 7) was never removed. + const results = registry.evaluate(`/worker/c1`, { + type: `run`, + key: `run-1`, + value: { status: `completed` }, + headers: { operation: `update` }, + }) + expect(results).toHaveLength(1) + expect(results[0]!.registrationDbId).toBe(7) + expect(removeSpy).not.toHaveBeenCalledWith(7) + }) + + it(`reconcileManifestRegistration with null desired removes the anchored registration`, async () => { + const registry = new WakeRegistry(createMockDb()) + await registry.register({ + subscriberUrl: `/parent/p1`, + sourceUrl: `/worker/c1`, + condition: `runFinished`, + oneShot: false, + manifestKey: `child:worker:c1`, + }) + expect( + registry.evaluate(`/worker/c1`, { + type: `run`, + key: `run-1`, + value: { status: `completed` }, + headers: { operation: `update` }, + }) + ).toHaveLength(1) + + await registry.reconcileManifestRegistration( + `/parent/p1`, + `child:worker:c1`, + null + ) + + expect( + registry.evaluate(`/worker/c1`, { + type: `run`, + key: `run-2`, + value: { status: `completed` }, + headers: { operation: `update` }, + }) + ).toHaveLength(0) + }) }) // ============================================================================ From 781c9eb9777bc418e0b9c78e8e76a68bee1a2f7e Mon Sep 17 00:00:00 2001 From: Andres Berrios Date: Thu, 16 Jul 2026 03:11:41 +0200 Subject: [PATCH 3/3] fix(agents-server): close the same wake-registration gap at cron & webhook-source sites MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit upsertCronSchedule and upsertWebhookSourceSubscription used the same unregister-then-register sequence that dropped wakes on parallel sub-agent spawn: the manifest-anchored wake registration is briefly absent from the in-memory cache across the register()'s DB round-trip, so a cron tick or webhook event arriving in that window would be missed. These are single API calls rather than concurrent bursts, so the race is far narrower than the spawn path, but the gap is the same class of bug — and both also race against the manifest-sync reconcile triggered by their own writeManifestEntry. Route both through WakeRegistry.reconcileManifestRegistration, which registers the desired reg first (never leaving the cache empty) and then prunes only stale rows for the manifest key. Delete-only sites (deleteSchedule, deleteWebhookSourceSubscription, deletePgSyncObservation, upsertFutureSendSchedule) have no re-register and so no gap; left as-is. Cron/webhook/manifest suites pass (30/30). Co-Authored-By: Claude Opus 4.8 (1M context) --- packages/agents-server/src/entity-manager.ts | 59 +++++++++++--------- 1 file changed, 32 insertions(+), 27 deletions(-) diff --git a/packages/agents-server/src/entity-manager.ts b/packages/agents-server/src/entity-manager.ts index 200f518661..57872e1743 100644 --- a/packages/agents-server/src/entity-manager.ts +++ b/packages/agents-server/src/entity-manager.ts @@ -2977,23 +2977,26 @@ export class EntityManager { const spec = resolveCronScheduleSpec(req.expression, req.timezone) const manifestKey = `schedule:${req.id}` - await this.wakeRegistry.unregisterByManifestKey( + // Gap-free reconcile (register-first, then prune stale). The old + // unregister-then-register briefly left this manifest key's wake absent from + // the cache; a cron tick landing in that window would be missed. See + // WakeRegistry.reconcileManifestRegistration. + await this.wakeRegistry.reconcileManifestRegistration( entityUrl, manifestKey, + { + subscriberUrl: entityUrl, + sourceUrl: getCronStreamPath(spec.expression, spec.timezone), + condition: { + on: `change`, + }, + debounceMs: req.debounceMs, + timeoutMs: req.timeoutMs, + oneShot: false, + manifestKey, + }, this.tenantId ) - await this.wakeRegistry.register({ - tenantId: this.tenantId, - subscriberUrl: entityUrl, - sourceUrl: getCronStreamPath(spec.expression, spec.timezone), - condition: { - on: `change`, - }, - debounceMs: req.debounceMs, - timeoutMs: req.timeoutMs, - oneShot: false, - manifestKey, - }) await this.getOrCreateCronStream(spec.expression, spec.timezone) const txid = randomUUID() @@ -3162,24 +3165,26 @@ export class EntityManager { ) // The manifest is the durable source of truth. Register side effects after - // it is appended so failures can be repaired by manifest replay. - await this.wakeRegistry.unregisterByManifestKey( + // it is appended so failures can be repaired by manifest replay. Gap-free + // reconcile (register-first, then prune stale) so a webhook event landing + // during a re-subscribe isn't missed — see + // WakeRegistry.reconcileManifestRegistration. + await this.wakeRegistry.reconcileManifestRegistration( entityUrl, manifestKey, + { + subscriberUrl: entityUrl, + sourceUrl: req.subscription.sourceUrl, + condition: { + on: `change`, + collections: [`webhook_event`], + ops: [`insert`], + }, + oneShot: false, + manifestKey, + }, this.tenantId ) - await this.wakeRegistry.register({ - tenantId: this.tenantId, - subscriberUrl: entityUrl, - sourceUrl: req.subscription.sourceUrl, - condition: { - on: `change`, - collections: [`webhook_event`], - ops: [`insert`], - }, - oneShot: false, - manifestKey, - }) return { txid, subscription: req.subscription } }