diff --git a/CHANGELOG.md b/CHANGELOG.md index 37814dc9..0f65ef3e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -20,6 +20,7 @@ Packages without a separate changelog are covered by the cross-package notes bel ### 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. - Retried action invocations reuse the original invocation instead of executing a provider action twice, wait for a durable dispatch outcome, and preserve locally completed release routing identity. ## [8.2.1] - 2026-08-24 diff --git a/packages/engine/CHANGELOG.md b/packages/engine/CHANGELOG.md index 3a750423..aa771752 100644 --- a/packages/engine/CHANGELOG.md +++ b/packages/engine/CHANGELOG.md @@ -11,6 +11,7 @@ and this project follows [Semantic Versioning](https://semver.org/spec/v2.0.0.ht ### 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. - `POST /v1/actions/:name/invoke` atomically claims caller-scoped idempotency keys before provider dispatch, waits for a durable dispatch outcome, and replays the original invocation with its immutable handler and node identity, including locally completed releases. ## [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..2cb93f55 --- /dev/null +++ b/packages/engine/src/__tests__/conformance/pendingInvocationAgeBound.test.ts @@ -0,0 +1,234 @@ +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, + neverDispatchedSweepGraceMs: 0, + }); + + 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, + neverDispatchedSweepGraceMs: 0, + }); + + 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, + neverDispatchedSweepGraceMs: 0, + }); + + const row = await readInvocation(stack, invocationId); + expect(['pending', 'dispatched']).toContain(row.status); + expect(row.error).toBeNull(); + 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, + neverDispatchedSweepGraceMs: 0, + }); + + 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 + // 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, + neverDispatchedSweepGraceMs: 0, + }); + + 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 58c8944a..0608a841 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, @@ -37,6 +38,31 @@ 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; +/** + * Sleep between the never-dispatched sweep's candidate SELECT and its atomic + * UPDATE. `dispatchNodeInvocation` sends the frame before recording + * `dispatchAttempts += 1`, so a sweep landing between those two dispatcher + * calls could otherwise mark an in-flight dispatch `never_dispatched_expired` + * while the handler already has the frame. The atomic UPDATE re-checks + * `dispatch_attempts = 0`, so once the dispatcher's UPDATE lands, the sweep's + * WHERE excludes the row. This grace window (default 5s) covers the send→record + * gap by a wide margin — the dispatcher's UPDATE is one D1 round-trip — while + * still bounding sweep latency. Set to 0 in tests to keep them fast. + */ +export const NEVER_DISPATCHED_SWEEP_GRACE_MS = 5_000; const ACTION_RETRY_BACKOFF_MS = 5_000; const NODE_DRAIN_REQUEUE_RETRY_MS = 5_000; @@ -44,6 +70,10 @@ 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; + /** Override for {@link NEVER_DISPATCHED_SWEEP_GRACE_MS}. */ + neverDispatchedSweepGraceMs?: number; /** When provided, TTL failures emit `action.failed` back to the caller. */ completionDeps?: InvocationCompletionDeps; } @@ -2526,6 +2556,100 @@ 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, + graceMs: 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), + )); + + if (rows.length === 0) return; + + // Give any concurrent dispatcher's send→record window (`dispatchNodeAttempt` + // UPDATE, one D1 round-trip) time to close before the atomic UPDATE fires. + // Combined with the UPDATE's `dispatch_attempts = 0` re-check, this makes an + // in-flight dispatch reliably invisible to the age sweep instead of racing + // with it. See `NEVER_DISPATCHED_SWEEP_GRACE_MS`. + if (graceMs > 0) { + await new Promise((resolve) => setTimeout(resolve, graceMs)); + } + + 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) { + // 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, @@ -2539,6 +2663,12 @@ export async function sweepTimedOutInvocations( sweepOpts.handlerUnreachableTtlMs ?? ACTION_HANDLER_UNREACHABLE_TTL_MS, sweepOpts.completionDeps, ); + await failNeverDispatchedExpiredInvocations( + db, + sweepOpts.pendingInvocationMaxAgeMs ?? PENDING_INVOCATION_MAX_AGE_MS, + sweepOpts.neverDispatchedSweepGraceMs ?? NEVER_DISPATCHED_SWEEP_GRACE_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,