From 7d8acc3b82de3a7eb1a3a6acc243ca19a4264b19 Mon Sep 17 00:00:00 2001 From: kjgbot Date: Mon, 24 Aug 2026 14:54:09 +0200 Subject: [PATCH 1/2] fix(engine): bound never-dispatched pending invocations by absolute age MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The 0032 handler-unreachable TTL fails an open invocation only after a DISPATCHED handler connection is observed unreachable for the TTL. A `pending` row that was never dispatched to any node (`dispatch_attempts = 0`) has no handler to observe, so the clock never starts and it can sit pending indefinitely — a 7,527-row backlog on the `relaycast-cloud` D1 (oldest 2026-06-26) is the direct measurement. The failure mode is not inert. When a matching node eventually returns and the queue drains, week-old spawn briefs come back to life as fresh agents. One landed as `chief` on a second node, and because agent rows carry one `token_hash` per name it evicted the live resident's token — every DM to `chief` routed to the impostor while the resident sat silently deaf. `sweepTimedOutInvocations` now runs a second pass that fails `status='pending' AND dispatch_attempts=0 AND created_at older than PENDING_INVOCATION_MAX_AGE_MS (default 72h)` with the distinguishing error `never_dispatched_expired`. 72h rides out a weekend-scale node outage with a day of buffer, and is far shorter than the observed multi-month backlog. The existing handler-unreachable TTL runs first and is unchanged; the new bound only touches the shape that guard cannot see. Tests: - must-fire: an old, never-dispatched pending row IS failed with `never_dispatched_expired`. - must-not-fire: a recent pending row is untouched. - must-not-fire: a dispatched-then-unreachable row inside the 0032 TTL is untouched (has `dispatch_attempts > 0`). - control: reset the same row to the never-dispatched shape and the age bound DOES fail it, proving the must-not-fire is guarded by the query filter rather than by accident. Both must-not-fire cases were verified to turn red when their guarded condition was removed from the sweep query. Fixes #357 Co-Authored-By: Claude Opus 4.7 Session-Id: 10847086-04c3-455f-9fba-c975ab69ab62 Session-Id: 10847086-04c3-455f-9fba-c975ab69ab62 Session-Id: 10847086-04c3-455f-9fba-c975ab69ab62 --- CHANGELOG.md | 6 +- packages/engine/CHANGELOG.md | 6 +- .../pendingInvocationAgeBound.test.ts | 203 ++++++++++++++++++ packages/engine/src/engine/action.ts | 58 +++++ packages/engine/src/node-invocations.ts | 1 + 5 files changed, 272 insertions(+), 2 deletions(-) create mode 100644 packages/engine/src/__tests__/conformance/pendingInvocationAgeBound.test.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index 416e293b..dfba2eaf 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -16,7 +16,11 @@ This project follows [Semantic Versioning](https://semver.org/spec/v2.0.0.html). Packages without a separate changelog are covered by the cross-package notes below. -## [Unreleased] +## [Unreleased - Patch] + +### Fixed + +- Never-dispatched pending action invocations are now bounded by absolute age (default 72h), so stale spawn queues can no longer drain into live agents that evict an existing resident under the same name. ## [8.2.1] - 2026-08-24 diff --git a/packages/engine/CHANGELOG.md b/packages/engine/CHANGELOG.md index 87954cc4..57fb059a 100644 --- a/packages/engine/CHANGELOG.md +++ b/packages/engine/CHANGELOG.md @@ -7,7 +7,11 @@ See the [root changelog](../../CHANGELOG.md) for cross-package release highlight The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and this project follows [Semantic Versioning](https://semver.org/spec/v2.0.0.html). -## [Unreleased] +## [Unreleased - Patch] + +### Fixed + +- `sweepTimedOutInvocations` now fails `pending` invocations that never dispatched (`dispatch_attempts = 0`) and are older than `PENDING_INVOCATION_MAX_AGE_MS` (default 72h) with the distinguishing error `never_dispatched_expired`. The existing `handler_unavailable` TTL guard, which requires a dispatched handler connection to observe unreachable, still runs first; the age bound covers the never-dispatched shape it cannot see. ## [8.2.1] - 2026-08-24 diff --git a/packages/engine/src/__tests__/conformance/pendingInvocationAgeBound.test.ts b/packages/engine/src/__tests__/conformance/pendingInvocationAgeBound.test.ts new file mode 100644 index 00000000..139e535b --- /dev/null +++ b/packages/engine/src/__tests__/conformance/pendingInvocationAgeBound.test.ts @@ -0,0 +1,203 @@ +import { afterEach, beforeEach, describe, expect, it } from 'vitest'; +import { eq } from 'drizzle-orm'; +import { + attachDirectNodeSocket, + createWorkspace, + makeNodeStack, + registerAgent, + type TestStack, +} from './harness.js'; +import { + PENDING_INVOCATION_MAX_AGE_MS, + sweepTimedOutInvocations, +} from '../../engine/action.js'; +import { actionInvocations } from '../../db/schema.js'; + +// Issue #357: a `pending` invocation that was never dispatched to any node +// (`dispatch_attempts = 0`) has no handler connection for the 0032 TTL sweep +// to observe unreachable, so its clock never starts and the row can sit +// pending indefinitely. When a matching node eventually returns, week-old +// spawn briefs drain into live agents that can evict the resident chief's +// token — a delayed-action identity hazard. The fix bounds such rows by +// absolute age with a distinguishing error (`never_dispatched_expired`). + +function registerBody(handler: string, overrides: Record = {}) { + return JSON.stringify({ + name: 'crm.get_person_batch', + description: 'Fetch a batch of people', + handler_agent: handler, + ...overrides, + }); +} + +async function insertPendingInvocation( + stack: TestStack, + workspaceId: string, + opts: { id: string; createdAt: Date; dispatchAttempts?: number }, +): Promise { + await stack.runtime.handle.db.insert(actionInvocations).values({ + id: opts.id, + workspaceId, + actionId: null, + actionName: 'spawn', + callerId: null, + callerName: null, + input: { capability: 'claude' }, + status: 'pending', + dispatchAttempts: opts.dispatchAttempts ?? 0, + createdAt: opts.createdAt, + }); +} + +async function readInvocation(stack: TestStack, id: string) { + const [row] = await stack.runtime.handle.db + .select({ + status: actionInvocations.status, + error: actionInvocations.error, + dispatchAttempts: actionInvocations.dispatchAttempts, + }) + .from(actionInvocations) + .where(eq(actionInvocations.id, id)); + return row; +} + +describe('pending invocation age bound (issue #357)', () => { + let stack: TestStack; + beforeEach(() => { stack = makeNodeStack(); }); + afterEach(() => stack.close()); + + // must-fire + it('fails a never-dispatched pending invocation older than the age bound with never_dispatched_expired', async () => { + const ws = await createWorkspace(stack.app, 'ndx-must-fire'); + + const staleId = 'inv_stale_never_dispatched'; + await insertPendingInvocation(stack, ws.workspaceId, { + id: staleId, + createdAt: new Date(Date.now() - 10_000), + dispatchAttempts: 0, + }); + + await sweepTimedOutInvocations(stack.runtime.handle.db, stack.runtime.realtime, { + pendingInvocationMaxAgeMs: 1_000, + }); + + const row = await readInvocation(stack, staleId); + expect(row).toMatchObject({ status: 'failed', error: 'never_dispatched_expired' }); + }); + + // must-not-fire + it('leaves a recently created pending invocation alone', async () => { + const ws = await createWorkspace(stack.app, 'ndx-recent-untouched'); + + const freshId = 'inv_fresh_never_dispatched'; + await insertPendingInvocation(stack, ws.workspaceId, { + id: freshId, + createdAt: new Date(), + dispatchAttempts: 0, + }); + + await sweepTimedOutInvocations(stack.runtime.handle.db, stack.runtime.realtime, { + pendingInvocationMaxAgeMs: PENDING_INVOCATION_MAX_AGE_MS, + }); + + const row = await readInvocation(stack, freshId); + expect(row.status).toBe('pending'); + expect(row.error).toBeNull(); + }); + + // must-not-fire: existing 0032 TTL guard is not regressed. A dispatched-then- + // unreachable invocation still inside the handler-unreachable TTL is left + // alone by the new age bound (it has dispatch_attempts > 0) AND by the + // existing sweep (the TTL has not elapsed). + it('leaves a dispatched-then-unreachable invocation inside the handler TTL alone', async () => { + const ws = await createWorkspace(stack.app, 'ndx-inside-ttl'); + const caller = await registerAgent(stack.app, ws.workspaceKey, 'worker'); + const handler = await registerAgent(stack.app, ws.workspaceKey, 'orchestrator'); + + const register = await stack.app.request('/v1/actions', { + method: 'POST', + headers: { 'content-type': 'application/json', authorization: `Bearer ${handler.token}` }, + body: registerBody('orchestrator'), + }); + expect(register.status).toBe(201); + + const handlerNode = await attachDirectNodeSocket(stack, ws.workspaceId, handler); + const invoke = await stack.app.request('/v1/actions/crm.get_person_batch/invoke', { + method: 'POST', + headers: { 'content-type': 'application/json', authorization: `Bearer ${caller.token}` }, + body: JSON.stringify({ input: { batchSize: 5 } }), + }); + expect(invoke.status).toBe(201); + const invocationId = (await invoke.json() as { data: { invocation_id: string } }).data.invocation_id; + + // Backdate past the new age bound. The row was dispatched + // (dispatch_attempts > 0), so the age bound must NOT touch it. + await stack.runtime.handle.db + .update(actionInvocations) + .set({ createdAt: new Date(Date.now() - PENDING_INVOCATION_MAX_AGE_MS - 60_000) }) + .where(eq(actionInvocations.id, invocationId)); + + await handlerNode.handle.handleClose(); + + // First sweep observes the disconnect: stamps handler_unreachable_since, + // no failure yet. Age bound must not fire (dispatch_attempts > 0). + await sweepTimedOutInvocations(stack.runtime.handle.db, stack.runtime.realtime, { + handlerUnreachableTtlMs: 60_000, + pendingInvocationMaxAgeMs: 1_000, + }); + + const row = await readInvocation(stack, invocationId); + expect(['pending', 'dispatched']).toContain(row.status); + expect(row.error).toBeNull(); + expect(row.dispatchAttempts).toBeGreaterThan(0); + }); + + // Prove the must-not-fire guard can actually fail: remove the dispatch_attempts + // qualifier on the row and the age bound will (correctly) fire on it. If this + // control test doesn't turn red, the must-not-fire above is not really testing + // the age bound — the row was safe for some other reason. + it('control: the same row IS failed if its dispatch_attempts qualifier is removed', async () => { + const ws = await createWorkspace(stack.app, 'ndx-control'); + const caller = await registerAgent(stack.app, ws.workspaceKey, 'worker'); + const handler = await registerAgent(stack.app, ws.workspaceKey, 'orchestrator'); + + await stack.app.request('/v1/actions', { + method: 'POST', + headers: { 'content-type': 'application/json', authorization: `Bearer ${handler.token}` }, + body: registerBody('orchestrator'), + }); + const handlerNode = await attachDirectNodeSocket(stack, ws.workspaceId, handler); + const invoke = await stack.app.request('/v1/actions/crm.get_person_batch/invoke', { + method: 'POST', + headers: { 'content-type': 'application/json', authorization: `Bearer ${caller.token}` }, + body: JSON.stringify({ input: { batchSize: 5 } }), + }); + const invocationId = (await invoke.json() as { data: { invocation_id: string } }).data.invocation_id; + + // Simulate the never-dispatched shape on an old row: restore status to + // 'pending', zero dispatch_attempts, and backdate. The age bound MUST fire, + // proving the must-not-fire above is guarded by the (status, dispatch_attempts) + // qualifier rather than by accident. + await stack.runtime.handle.db + .update(actionInvocations) + .set({ + status: 'pending', + dispatchAttempts: 0, + dispatchedAt: null, + dispatchedNodeId: null, + dispatchedProvider: null, + createdAt: new Date(Date.now() - 10_000), + }) + .where(eq(actionInvocations.id, invocationId)); + + await handlerNode.handle.handleClose(); + + await sweepTimedOutInvocations(stack.runtime.handle.db, stack.runtime.realtime, { + handlerUnreachableTtlMs: 60_000, + pendingInvocationMaxAgeMs: 1_000, + }); + + const row = await readInvocation(stack, invocationId); + expect(row).toMatchObject({ status: 'failed', error: 'never_dispatched_expired' }); + }); +}); diff --git a/packages/engine/src/engine/action.ts b/packages/engine/src/engine/action.ts index e5fb264c..170afdb6 100644 --- a/packages/engine/src/engine/action.ts +++ b/packages/engine/src/engine/action.ts @@ -36,6 +36,19 @@ export const ACTION_DISPATCH_TIMEOUT_MS = 30_000; * a signal instead of an unbounded hang. */ export const ACTION_HANDLER_UNREACHABLE_TTL_MS = 120_000; +/** + * Absolute age bound for a `pending` invocation that was NEVER dispatched to a + * node (`dispatch_attempts = 0`). The handler-unreachable TTL only fires once a + * dispatched handler connection has been observed offline; an invocation that + * never left the queue has no handler to observe, so the clock never starts + * and the row can sit `pending` forever. When the queue eventually drains + * (e.g. a node returns after a long outage), week-old spawn briefs come back + * to life as fresh agents that can evict the live resident under the same + * name — a delayed-action identity hazard, not inert backlog. Sized deliberately: + * long enough to ride out a weekend-scale node outage plus a day of buffer, + * far shorter than the observed multi-month backlog that motivated it. + */ +export const PENDING_INVOCATION_MAX_AGE_MS = 72 * 60 * 60 * 1000; const ACTION_RETRY_BACKOFF_MS = 5_000; const NODE_DRAIN_REQUEUE_RETRY_MS = 5_000; @@ -43,6 +56,8 @@ export interface SweepTimedOutInvocationsOptions { timeoutMs?: number; /** Override for {@link ACTION_HANDLER_UNREACHABLE_TTL_MS}. */ handlerUnreachableTtlMs?: number; + /** Override for {@link PENDING_INVOCATION_MAX_AGE_MS}. */ + pendingInvocationMaxAgeMs?: number; /** When provided, TTL failures emit `action.failed` back to the caller. */ completionDeps?: InvocationCompletionDeps; } @@ -2177,6 +2192,44 @@ async function failUnreachableAgentInvocations( } } +/** + * Absolute-age bound for never-dispatched pending invocations. The + * handler-unreachable TTL keys off a dispatched connection going quiet; an + * invocation that never left the queue has no handler to observe, so it needs + * its own bound. Fails rows that are `pending`, have `dispatch_attempts = 0`, + * and are older than `maxAgeMs` with a distinguishing error so operators can + * tell a never-dispatched expiry from a handler-that-went-quiet expiry. + */ +async function failNeverDispatchedExpiredInvocations( + db: Db, + maxAgeMs: number, + completionDeps?: InvocationCompletionDeps, +): Promise { + const cutoff = new Date(Date.now() - maxAgeMs); + const rows = await db + .select({ id: actionInvocations.id, workspaceId: actionInvocations.workspaceId }) + .from(actionInvocations) + .where(and( + eq(actionInvocations.status, 'pending'), + eq(actionInvocations.dispatchAttempts, 0), + lte(actionInvocations.createdAt, cutoff), + )); + + const byWorkspace = new Map(); + for (const row of rows) { + const list = byWorkspace.get(row.workspaceId) ?? []; + list.push(row.id); + byWorkspace.set(row.workspaceId, list); + } + for (const [workspaceId, ids] of byWorkspace) { + try { + await failOpenInvocations(db, workspaceId, ids, 'never_dispatched_expired', completionDeps); + } catch { + // Leave for the next sweep. + } + } +} + export async function sweepTimedOutInvocations( db: Db, registry: NodeConnectionRegistry, @@ -2190,6 +2243,11 @@ export async function sweepTimedOutInvocations( sweepOpts.handlerUnreachableTtlMs ?? ACTION_HANDLER_UNREACHABLE_TTL_MS, sweepOpts.completionDeps, ); + await failNeverDispatchedExpiredInvocations( + db, + sweepOpts.pendingInvocationMaxAgeMs ?? PENDING_INVOCATION_MAX_AGE_MS, + sweepOpts.completionDeps, + ); const now = new Date(); const cutoff = new Date(Date.now() - timeoutMs); const rows = await db diff --git a/packages/engine/src/node-invocations.ts b/packages/engine/src/node-invocations.ts index 947e91c8..28d5fcaa 100644 --- a/packages/engine/src/node-invocations.ts +++ b/packages/engine/src/node-invocations.ts @@ -1,5 +1,6 @@ export { ACTION_HANDLER_UNREACHABLE_TTL_MS, + PENDING_INVOCATION_MAX_AGE_MS, drainNodeInvocations, sweepTimedOutInvocations, type SweepTimedOutInvocationsOptions, From ea114f4499dcb8187f0ac22dc89876fa9507688c Mon Sep 17 00:00:00 2001 From: kjgbot Date: Tue, 25 Aug 2026 21:09:16 +0200 Subject: [PATCH 2/2] fix(engine): chunk id list and atomic re-check for never-dispatched sweep MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two review findings on the age-bound sweep: 1) D1 100-bound-parameter cap: failNeverDispatchedExpiredInvocations was passing every stale id for a workspace into one inArray, so any workspace with >~100 aged rows threw "too many bound parameters" that the per- workspace catch swallowed — leaving the exact backlog the fix targets unable to drain. Chunk by D1_SAFE_IN_QUERY_CHUNK_SIZE with a per-chunk try/catch and per-chunk completion emission. 2) SELECT/UPDATE race: a row selected as (pending, dispatch_attempts=0, past cutoff) could be concurrently dispatched before failOpenInvocationRows ran; that UPDATE only re-checks status IN OPEN_INVOCATION_STATUSES, so it would fail a row that had just been dispatched. Introduce failNeverDispatchedInvocationRows whose UPDATE re-checks status='pending', dispatch_attempts=0, and created_at<=cutoff atomically, so a concurrently dispatched row is skipped. Never-dispatched rows have no spawn reservation to release, so the held-rows lookup is unnecessary here. Test: pendingInvocationAgeBound now includes an isolation case for a pending row with dispatch_attempts>0 past the cutoff — must NOT be failed with never_dispatched_expired. Proves the dispatch_attempts=0 filter is load- bearing on its own (the existing must-not-fire uses a dispatched row, so it alone couldn't isolate the two filters). Co-Authored-By: Claude Opus 4.7 --- .../pendingInvocationAgeBound.test.ts | 26 +++++++++ packages/engine/src/engine/action.ts | 53 +++++++++++++++++-- 2 files changed, 75 insertions(+), 4 deletions(-) diff --git a/packages/engine/src/__tests__/conformance/pendingInvocationAgeBound.test.ts b/packages/engine/src/__tests__/conformance/pendingInvocationAgeBound.test.ts index 139e535b..9e40c9c3 100644 --- a/packages/engine/src/__tests__/conformance/pendingInvocationAgeBound.test.ts +++ b/packages/engine/src/__tests__/conformance/pendingInvocationAgeBound.test.ts @@ -152,6 +152,32 @@ describe('pending invocation age bound (issue #357)', () => { expect(row.dispatchAttempts).toBeGreaterThan(0); }); + // must-not-fire: isolate the `dispatch_attempts = 0` filter. A `pending` row + // whose `dispatchAttempts > 0` (e.g. rescheduled back to pending after a prior + // dispatch) is exactly the handler-unreachable TTL's territory and the new age + // bound must NOT touch it — even when the row is past the cutoff. The earlier + // must-not-fire uses a `dispatched` row, so this case isolates the attempts + // filter from the status filter and proves each is load-bearing on its own. + it('leaves a pending row with dispatch_attempts > 0 past the cutoff alone', async () => { + const ws = await createWorkspace(stack.app, 'ndx-attempts-nonzero'); + + const requeuedId = 'inv_requeued_pending'; + await insertPendingInvocation(stack, ws.workspaceId, { + id: requeuedId, + createdAt: new Date(Date.now() - 10_000), + dispatchAttempts: 1, + }); + + await sweepTimedOutInvocations(stack.runtime.handle.db, stack.runtime.realtime, { + pendingInvocationMaxAgeMs: 1_000, + }); + + const row = await readInvocation(stack, requeuedId); + expect(row.status).toBe('pending'); + expect(row.error).toBeNull(); + expect(row.dispatchAttempts).toBe(1); + }); + // Prove the must-not-fire guard can actually fail: remove the dispatch_attempts // qualifier on the row and the age bound will (correctly) fire on it. If this // control test doesn't turn red, the must-not-fire above is not really testing diff --git a/packages/engine/src/engine/action.ts b/packages/engine/src/engine/action.ts index 170afdb6..cb62fc6a 100644 --- a/packages/engine/src/engine/action.ts +++ b/packages/engine/src/engine/action.ts @@ -5,6 +5,7 @@ import { generateId } from './snowflake.js'; import { RELEASED_AGENT_STATUS, releasedAgentName } from './agent.js'; import { randomHex, sha256Hex } from '../lib/crypto.js'; import { codedError } from '../lib/httpError.js'; +import { D1_SAFE_IN_QUERY_CHUNK_SIZE } from '../lib/queryChunks.js'; import { toFleetWireJson } from './deliveryWire.js'; import { emitAgentExitedEffects, @@ -2222,14 +2223,58 @@ async function failNeverDispatchedExpiredInvocations( byWorkspace.set(row.workspaceId, list); } for (const [workspaceId, ids] of byWorkspace) { - try { - await failOpenInvocations(db, workspaceId, ids, 'never_dispatched_expired', completionDeps); - } catch { - // Leave for the next sweep. + // Chunk ids so the IN clause stays under D1's 100-bound-parameter cap. A per- + // chunk try/catch keeps one bad chunk from stalling the whole workspace, so a + // large stale backlog can actually drain across sweeps. + for (let i = 0; i < ids.length; i += D1_SAFE_IN_QUERY_CHUNK_SIZE) { + const chunk = ids.slice(i, i + D1_SAFE_IN_QUERY_CHUNK_SIZE); + try { + const failed = await failNeverDispatchedInvocationRows( + db, + workspaceId, + chunk, + cutoff, + 'never_dispatched_expired', + ); + if (completionDeps && failed.length > 0) { + await emitFailedInvocationEffects(completionDeps, workspaceId, failed); + } + } catch { + // Leave this chunk for the next sweep; other chunks still get their shot. + } } } } +/** + * Terminally fail never-dispatched rows atomically: the UPDATE re-checks the + * SELECT predicates (`status = 'pending'`, `dispatch_attempts = 0`, `created_at + * <= cutoff`), so a row that was concurrently dispatched between the sweep's + * SELECT and this UPDATE is skipped instead of being killed mid-flight. Never- + * dispatched rows carry no spawn reservation (dispatch_attempts = 0 means no + * node was ever chosen), so no capacity release is needed. + */ +async function failNeverDispatchedInvocationRows( + db: Db, + workspaceId: string, + invocationIds: string[], + cutoff: Date, + error: string, +): Promise { + if (invocationIds.length === 0) return []; + return await db + .update(actionInvocations) + .set({ status: 'failed', error, completedAt: new Date(), spawnReservedAt: null }) + .where(and( + eq(actionInvocations.workspaceId, workspaceId), + inArray(actionInvocations.id, invocationIds), + eq(actionInvocations.status, 'pending'), + eq(actionInvocations.dispatchAttempts, 0), + lte(actionInvocations.createdAt, cutoff), + )) + .returning(); +} + export async function sweepTimedOutInvocations( db: Db, registry: NodeConnectionRegistry,