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/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 } } 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 ff8ef6e75b..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 } 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) @@ -344,10 +352,50 @@ 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() - return + // 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 row.id + } + return -1 } const dbId = result[0]!.id @@ -358,6 +406,7 @@ export class WakeRegistry { createdAt: new Date(), timeoutConsumed: false, }) + return dbId } private startTimeoutTimer(reg: CachedWakeRegistration, dbId: number): void { @@ -413,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 c879a124ca..f97483c65f 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({ @@ -856,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) + }) }) // ============================================================================