fix(agents-server): stop orphaned cron wakes (#4559)

Fixes orphaned cron wakes in Agents Server after a manifest-backed
schedule is deleted. Users should no longer see agents repeatedly woken
by a cron schedule that `list_schedules` correctly reports as absent.

## Root Cause

Cron schedules are represented by two linked pieces of state:

1. a manifest-backed schedule row on the agent stream, used by tools
like `list_schedules`
2. wake-registry + scheduler side effects, used to deliver cron ticks

For `/horton/RbYyjlX3ey`, the manifest schedule had been deleted, so
`list_schedules` returned `[]`. However, cron wake delivery kept going
because stale wake/scheduler side effects could outlive the manifest
deletion:

- `WakeRegistry` removed cached registrations on Electric shape deletes
by deriving the DB id from the shape message key. Shape keys are
protocol-level identifiers and are not guaranteed to be the table
primary key, so the DB row could be gone while the in-memory
registration remained.
- Cron ticks rescheduled themselves indefinitely, even when no
`wake_registrations` rows still subscribed to that cron stream.

## Approach

This PR makes the cron side effects self-healing in both places.

### Wake registry delete handling

`WakeRegistry.applyShapeMessage` now treats `old_value.id` as
authoritative for delete messages. The shape uses `replica: full`, so
deletes should carry the deleted row; if the id is missing, the registry
resets its cache to fail closed instead of keeping a potentially stale
wake registration alive.

```ts
const oldValue = (message as unknown as {
  old_value?: { id?: unknown }
}).old_value
const oldId = Number(oldValue?.id)
if (Number.isFinite(oldId)) {
  this.removeCachedRegistrationByDbId(oldId)
} else {
  this.resetCachedRegistrations()
}
```

### Cron chain rescheduling

Before inserting the next cron tick, the scheduler checks whether any
wake registration still points at that cron stream:

```ts
select 1 as exists
from wake_registrations
where tenant_id = ${tenantId}
  and source_url = ${streamPath}
limit 1
```

If there are no subscribers, it completes the current task and stops the
chain. A future schedule registration will seed a fresh tick through
`getOrCreateCronStream` / rehydration.

## Key Invariants

- The manifest remains the user-visible source of truth for schedule
tools.
- A cron stream should only keep producing future ticks while at least
one wake registration subscribes to it.
- Deleting a manifest-backed cron schedule must remove both durable wake
registrations and any effective in-memory wake registrations.
- Shared cron streams remain reusable: deleting the last subscriber
stops the chain; adding a subscriber later recreates it.

## Non-goals

- This does not change the public schedule tool API.
- This does not delete historical cron stream events.
- This does not introduce per-schedule cron streams; cron streams remain
shared by expression/timezone.
- This does not attempt to clean up already-delivered wake events from
agent timelines.

## Trade-offs

An alternative would be to always reschedule cron ticks forever for
every cron stream ever created. That is simpler, but it allows orphaned
cron streams to keep waking stale in-memory subscribers and creates
unnecessary scheduler churn. Checking for subscribers before
rescheduling keeps cron streams lazy and aligns their lifecycle with
active wake registrations.

Another option would be to rely only on wake-registry cache
invalidation. The scheduler guard is still useful as a second line of
defense and prevents orphaned cron task chains even if a process
restarts or a wake cache bug reappears.

## Verification

Commands run:

```bash
pnpm install
pnpm --filter @electric-ax/agents-runtime build
pnpm --filter @electric-ax/agents-mcp build
pnpm --filter @electric-sql/client build
pnpm --filter @electric-ax/agents-server typecheck
cd packages/agents-server && pnpm test test/scheduler.test.ts
cd packages/agents-server && pnpm test test/wake-registry.test.ts -t "removes cached registrations"
GITHUB_BASE_REF=main node scripts/check-changeset.mjs
git diff --check
```

Notes:

- `pnpm install` completed dependency linking but reported warnings,
including `examples/.shared prepare` failing due SST.
- The focused wake-registry unit test passes. Running all of
`wake-registry.test.ts` locally reached the integration section and
timed out with a webhook delivery `fetch failed` warning, so the new
unit path was verified with `-t`.
- `check-changeset` passes.

## Files changed

- `packages/agents-server/src/wake-registry.ts` — remove cached
registrations using `old_value.id` from shape delete messages; reset the
cache if a delete id is unavailable.
- `packages/agents-server/src/scheduler.ts` — stop rescheduling cron
ticks when no registrations subscribe to the cron stream; clarify cron
payload stream-path extraction and subscriber lookup.
- `packages/agents-server/test/scheduler.test.ts` — cover
subscriber-present and no-subscriber cron reschedule paths.
- `packages/agents-server/test/wake-registry.test.ts` — cover non-id
shape keys with delete ids coming from `old_value.id`.
- `.changeset/fix-orphaned-cron-wakes.md` — patch changeset for
`@electric-ax/agents-server`.

Author: KyleAMathews

.changeset/fix-orphaned-cron-wakes.md

@@ -0,0 +1,5 @@
+---
+'@electric-ax/agents-server': patch
+---
+
+Stop orphaned cron wakes after schedule deletion by clearing stale wake-registry entries and ending cron tick chains with no subscribers.

packages/agents-server/src/scheduler.ts

@@ -228,6 +228,14 @@ export function isPermanentElectricAgentsError(err: unknown): boolean {
   )
 }
 
+function cronTaskStreamPath(
+  payload: DelayedSendPayload | CronTickPayload
+): string | null {
+  return typeof (payload as { streamPath?: unknown }).streamPath === `string`
+    ? (payload as { streamPath: string }).streamPath
+    : null
+}
+
 function normalizeTask(row: ScheduledTaskRow): {
   id: number
   tenantId: string
@@ -680,6 +688,24 @@ export class Scheduler implements SchedulerClient {
         task.fireAt
       )
 
+      const streamPath = cronTaskStreamPath(task.payload)
+      const subscriberRows = streamPath
+        ? await sql<Array<{ exists: number }>>`
+            select 1 as exists
+            from wake_registrations
+            where tenant_id = ${tenantId}
+              and source_url = ${streamPath}
+            limit 1
+          `
+        : []
+
+      // Cron streams are virtual shared sources. If no wake registrations
+      // still point at this cron stream (e.g. the owning manifest schedule was
+      // deleted), stop the chain here instead of keeping a forever-global tick
+      // alive. Rehydration/getOrCreateCronStream will seed a fresh tick when a
+      // subscription is recreated.
+      if (subscriberRows.length === 0) return
+
       await sql`
         insert into scheduled_tasks (
           tenant_id,

packages/agents-server/src/wake-registry.ts

@@ -739,7 +739,23 @@ export class WakeRegistry {
     }
 
     if (message.headers.operation === `delete`) {
-      this.removeCachedRegistrationByDbId(Number(message.key))
+      // Shape keys are protocol-level identifiers and are not guaranteed to be
+      // the table primary key. The wake_registrations shape uses
+      // `replica: full`, so deletes should carry the deleted row in old_value;
+      // use that row id to remove the matching in-memory registration. If the
+      // id is unavailable, reset the cache so we fail closed rather than
+      // keeping a stale wake registration alive.
+      const oldValue = (
+        message as unknown as {
+          old_value?: { id?: unknown }
+        }
+      ).old_value
+      const oldId = Number(oldValue?.id)
+      if (Number.isFinite(oldId)) {
+        this.removeCachedRegistrationByDbId(oldId)
+      } else {
+        this.resetCachedRegistrations()
+      }
       return
     }
 

packages/agents-server/test/scheduler.test.ts

@@ -314,7 +314,7 @@ describe(`Scheduler`, () => {
 
   it(`reschedules cron from the stored fireAt instead of now`, async () => {
     const mock = createMockPgClient({
-      txResponses: [[{ id: 11 }], []],
+      txResponses: [[{ id: 11 }], [{ exists: 1 }], []],
     })
     const scheduler = new Scheduler({
       pgClient: mock.pgClient,
@@ -346,6 +346,42 @@ describe(`Scheduler`, () => {
     expect(insertCall!.values[2]).toBe(`2026-04-09T10:30:00.000Z`)
   })
 
+  it(`stops rescheduling cron ticks when the stream has no subscribers`, async () => {
+    const mock = createMockPgClient({
+      txResponses: [[{ id: 11 }], []],
+    })
+    const scheduler = new Scheduler({
+      pgClient: mock.pgClient,
+      instanceId: `instance-1`,
+      executors: {
+        delayed_send: vi.fn(),
+        cron_tick: vi.fn(),
+      },
+    })
+
+    await (scheduler as any).completeAndRescheduleCron({
+      id: 11,
+      kind: `cron_tick`,
+      payload: { streamPath: `/_cron/test` },
+      fireAt: new Date(`2026-04-09T10:00:00.000Z`),
+      cronExpression: `*/30 * * * *`,
+      cronTimezone: `UTC`,
+      cronTickNumber: 4,
+    })
+
+    expect(
+      mock.txCalls.some((call) => call.sql.includes(`update scheduled_tasks`))
+    ).toBe(true)
+    expect(
+      mock.txCalls.some((call) => call.sql.includes(`from wake_registrations`))
+    ).toBe(true)
+    expect(
+      mock.txCalls.some((call) =>
+        call.sql.includes(`insert into scheduled_tasks`)
+      )
+    ).toBe(false)
+  })
+
   it(`recovers from transient run loop errors instead of stopping permanently`, async () => {
     const scheduler = new Scheduler({
       pgClient: createMockPgClient().pgClient,

packages/agents-server/test/wake-registry.test.ts

@@ -201,6 +201,38 @@ describe(`Wake Registry`, () => {
     expect(results[0]!.sourceEventKey).toBe(`insert:tick-7`)
   })
 
+  it(`removes cached registrations from shape delete old_value ids`, async () => {
+    const registry = new WakeRegistry(createMockDb())
+    await registry.register({
+      subscriberUrl: `/watcher/w1`,
+      sourceUrl: `/_cron/abc`,
+      condition: { on: `change` },
+      oneShot: false,
+    })
+
+    const before = registry.evaluate(`/_cron/abc`, {
+      type: `cron_tick`,
+      key: `tick-7`,
+      value: {},
+      headers: { operation: `insert` },
+    })
+    expect(before).toHaveLength(1)
+
+    await (registry as any).applyShapeMessage({
+      key: `shape-key-is-not-the-registration-id`,
+      old_value: { id: before[0]!.registrationDbId },
+      headers: { operation: `delete` },
+    })
+
+    const after = registry.evaluate(`/_cron/abc`, {
+      type: `cron_tick`,
+      key: `tick-8`,
+      value: {},
+      headers: { operation: `insert` },
+    })
+    expect(after).toHaveLength(0)
+  })
+
   it(`keeps distinct registrations distinct for the same source event`, async () => {
     const registry = new WakeRegistry(createMockDb())
     await registry.register({