diff --git a/src/gateway/server-maintenance.test.ts b/src/gateway/server-maintenance.test.ts index c337e80a0fc4..74bc50afb58d 100644 --- a/src/gateway/server-maintenance.test.ts +++ b/src/gateway/server-maintenance.test.ts @@ -688,40 +688,6 @@ describe("startGatewayMaintenanceTimers", () => { stopMaintenanceTimers(timers); }); - it("keeps retained in-flight entries through ttl and overflow until settlement", async () => { - const { startGatewayMaintenanceTimers, deps, now } = await createTimedMaintenanceScenario(); - seedStableDedupeEntries(deps, now); - deps.dedupe.set("message.action:route-binding:active", { - ts: now - DEDUPE_TTL_MS - 1, - ok: true, - requestIdentity: '["slack","primary"]', - retainUntilSettled: true, - }); - deps.dedupe.set("overflow-newest", { ts: now, ok: true }); - const timers = startGatewayMaintenanceTimers(deps); - - await vi.advanceTimersByTimeAsync(60_000); - - expect(deps.dedupe.size).toBe(DEDUPE_MAX); - expect(deps.dedupe.has("message.action:route-binding:active")).toBe(true); - expect(deps.dedupe.has("stable-0")).toBe(false); - expect(deps.dedupe.has("stable-1")).toBe(false); - - const retained = deps.dedupe.get("message.action:route-binding:active"); - if (!retained) { - throw new Error("Expected retained route binding"); - } - deps.dedupe.set("message.action:route-binding:active", { - ...retained, - ts: Date.now(), - retainUntilSettled: false, - }); - await vi.advanceTimersByTimeAsync(DEDUPE_TTL_MS + 60_000); - - expect(deps.dedupe.has("message.action:route-binding:active")).toBe(false); - stopMaintenanceTimers(timers); - }); - it("evicts dedupe overflow by oldest timestamp even after reinsertion", async () => { const { startGatewayMaintenanceTimers, deps, now } = await createTimedMaintenanceScenario(); diff --git a/src/gateway/server-maintenance.ts b/src/gateway/server-maintenance.ts index e05b1a941c67..b255357a096f 100644 --- a/src/gateway/server-maintenance.ts +++ b/src/gateway/server-maintenance.ts @@ -218,13 +218,8 @@ export function startGatewayMaintenanceTimers(params: { } return Boolean(isChatKey && runId && params.chatQueuedTurns.has(runId)); }; - const isRetainedDedupeEntry = (entry: DedupeEntry) => entry.retainUntilSettled === true; for (const [k, v] of params.dedupe) { - if ( - isRetainedDedupeEntry(v) || - isActiveRunDedupeKey(k, v) || - isPendingAcceptedRunDedupeKey(k, v) - ) { + if (isActiveRunDedupeKey(k, v) || isPendingAcceptedRunDedupeKey(k, v)) { continue; } if (now - v.ts > DEDUPE_TTL_MS) { @@ -236,9 +231,7 @@ export function startGatewayMaintenanceTimers(params: { const oldestKeys = [...params.dedupe.entries()] .filter( ([key, entry]) => - !isRetainedDedupeEntry(entry) && - !isActiveRunDedupeKey(key, entry) && - !isPendingAcceptedRunDedupeKey(key, entry), + !isActiveRunDedupeKey(key, entry) && !isPendingAcceptedRunDedupeKey(key, entry), ) .toSorted(([, left], [, right]) => left.ts - right.ts) .slice(0, excess) diff --git a/src/gateway/server-methods/send.test.ts b/src/gateway/server-methods/send.test.ts index cce29b53a3c3..c1b8a350bb5a 100644 --- a/src/gateway/server-methods/send.test.ts +++ b/src/gateway/server-methods/send.test.ts @@ -20,7 +20,7 @@ import { createTestRegistry, } from "../../test-utils/channel-plugins.js"; import { captureEnv, setTestEnvValue } from "../../test-utils/env.js"; -import { DEDUPE_TTL_MS } from "../server-constants.js"; +import { DEDUPE_MAX, DEDUPE_TTL_MS } from "../server-constants.js"; import { startGatewayMaintenanceTimers } from "../server-maintenance.js"; import { createGatewayMaintenanceStateForTest } from "../test-helpers.maintenance-state.js"; import type { GatewayRequestContext } from "./types.js"; @@ -1272,10 +1272,7 @@ describe("gateway send mirroring", () => { await vi.waitFor(() => { expect(mocks.dispatchChannelMessageAction).toHaveBeenCalledTimes(1); }); - const activeBinding = [...context.dedupe.entries()].find(([key]) => - key.includes(":route-binding:"), - ); - expect(activeBinding?.[1].retainUntilSettled).toBe(true); + expect([...context.dedupe.keys()].some((key) => key.includes(":route-binding:"))).toBe(false); await vi.advanceTimersByTimeAsync(DEDUPE_TTL_MS + 60_000); defaultAccountId = "secondary"; @@ -1292,7 +1289,61 @@ describe("gateway send mirroring", () => { expect(firstRespondCall(retryRespond)?.[0]).toBe(true); expect(firstRespondCall(retryRespond)?.[3]?.cached).toBe(true); expect(mocks.dispatchChannelMessageAction).toHaveBeenCalledTimes(1); - expect(context.dedupe.get(activeBinding?.[0] ?? "")?.retainUntilSettled).toBe(false); + } finally { + clearInterval(maintenance.tickInterval); + clearInterval(maintenance.healthInterval); + clearInterval(maintenance.dedupeCleanup); + clearInterval(maintenance.worktreeCleanup); + if (maintenance.mediaCleanup) { + clearInterval(maintenance.mediaCleanup); + } + maintenance.skillCuratorCleanup(); + vi.useRealTimers(); + } + }); + + it("does not let settled route aliases evict canonical results before ttl", async () => { + vi.useFakeTimers(); + vi.setSystemTime(new Date("2026-07-26T00:00:00Z")); + let defaultAccountId = "primary"; + mockMutableMessageRouteAccounts(() => defaultAccountId); + const context = makeContext(); + const maintenance = startGatewayMaintenanceTimers({ + ...createGatewayMaintenanceStateForTest(), + dedupe: context.dedupe, + runWorktreeGc: vi.fn(async () => undefined), + }); + const operationCount = Math.floor(DEDUPE_MAX / 2) + 1; + const invoke = (idempotencyKey: string, respond: ReturnType) => + invokeGatewayMessageMethod({ + method: "message.action", + request: { + channel: "slack", + action: "send", + params: { target: "channel:current", message: "hi" }, + idempotencyKey, + }, + respond, + context, + }); + + try { + for (let index = 0; index < operationCount; index += 1) { + await invoke(`idem-action-capacity-${index}`, vi.fn()); + } + + expect(mocks.dispatchChannelMessageAction).toHaveBeenCalledTimes(operationCount); + expect(context.dedupe.size).toBe(operationCount); + expect([...context.dedupe.keys()].some((key) => key.includes(":route-binding:"))).toBe(false); + + await vi.advanceTimersByTimeAsync(60_000); + defaultAccountId = "secondary"; + const retryRespond = vi.fn(); + await invoke("idem-action-capacity-0", retryRespond); + + expect(mocks.dispatchChannelMessageAction).toHaveBeenCalledTimes(operationCount); + expect(firstRespondCall(retryRespond)?.[0]).toBe(true); + expect(firstRespondCall(retryRespond)?.[3]?.cached).toBe(true); } finally { clearInterval(maintenance.tickInterval); clearInterval(maintenance.healthInterval); diff --git a/src/gateway/server-methods/send.ts b/src/gateway/server-methods/send.ts index 2df88a5aec54..d698a16f7fe4 100644 --- a/src/gateway/server-methods/send.ts +++ b/src/gateway/server-methods/send.ts @@ -76,6 +76,7 @@ import { INTERNAL_MESSAGE_CHANNEL, normalizeMessageChannel } from "../../utils/m import { resolveGatewayConversationReadOrigin } from "../conversation-read-origin.js"; import { ADMIN_SCOPE } from "../operator-scopes.js"; import { resolveGatewayPluginConfig } from "../runtime-plugin-config.js"; +import { DEDUPE_MAX, DEDUPE_TTL_MS } from "../server-constants.js"; import { loadSessionEntry } from "../session-utils.js"; import { formatForLog } from "../ws-log.js"; import { @@ -100,8 +101,56 @@ type MessageOperationRouteBinding = { reservedRoute?: MessageOperationRoute; }; +type MessageOperationRouteBindingEntry = { + requestScope: string; + retainUntilSettled: boolean; + ts: number; +}; + +// Send and poll callers can spell one canonical route four ways by omitting or +// supplying channel/account defaults. Preserve every alias for the full result budget. +const MESSAGE_OPERATION_ROUTE_BINDING_MAX = DEDUPE_MAX * 4; +const messageOperationRouteBindings = new WeakMap< + GatewayRequestContext, + Map +>(); const messageOperationRouteBindingQueues = new WeakMap(); +function pruneMessageOperationRouteBindings( + bindings: Map, + now: number, +): void { + for (const [key, entry] of bindings) { + if (!entry.retainUntilSettled && now - entry.ts > DEDUPE_TTL_MS) { + bindings.delete(key); + } + } + const excess = bindings.size - MESSAGE_OPERATION_ROUTE_BINDING_MAX; + if (excess <= 0) { + return; + } + const oldestSettledKeys = [...bindings.entries()] + .filter(([, entry]) => !entry.retainUntilSettled) + .toSorted(([, left], [, right]) => left.ts - right.ts) + .slice(0, excess) + .map(([key]) => key); + for (const key of oldestSettledKeys) { + bindings.delete(key); + } +} + +function getMessageOperationRouteBindings( + context: GatewayRequestContext, +): Map { + let bindings = messageOperationRouteBindings.get(context); + if (!bindings) { + bindings = new Map(); + messageOperationRouteBindings.set(context, bindings); + } + pruneMessageOperationRouteBindings(bindings, Date.now()); + return bindings; +} + function getMessageOperationRouteBindingQueue(context: GatewayRequestContext): KeyedAsyncQueue { let queue = messageOperationRouteBindingQueues.get(context); if (!queue) { @@ -320,7 +369,9 @@ function resolveMessageOperationRouteBinding(params: { const key = `${params.prefix}${authorityScope}:route-binding:${explicitRouteScope}:${params.idempotencyKey}`; return { key, - reservedRoute: parseMessageOperationRoute(params.context.dedupe.get(key)?.requestIdentity), + reservedRoute: parseMessageOperationRoute( + getMessageOperationRouteBindings(params.context).get(key)?.requestScope, + ), }; } @@ -332,21 +383,23 @@ function bindMessageOperationRoute(params: { if (!params.binding) { return true; } - const existing = params.context.dedupe.get(params.binding.key); + const bindings = getMessageOperationRouteBindings(params.context); + const existing = bindings.get(params.binding.key); if (existing) { - if (existing.requestIdentity !== params.requestScope) { + if (existing.requestScope !== params.requestScope) { return false; } - params.context.dedupe.set(params.binding.key, { ...existing, ts: Date.now() }); + bindings.set(params.binding.key, { ...existing, ts: Date.now() }); return true; } // Bind the canonical route before dispatch so retries can replay without // consulting mutable defaults or plugin/account configuration. - params.context.dedupe.set(params.binding.key, { + bindings.set(params.binding.key, { ts: Date.now(), - ok: true, - requestIdentity: params.requestScope, + requestScope: params.requestScope, + retainUntilSettled: false, }); + pruneMessageOperationRouteBindings(bindings, Date.now()); return true; } @@ -358,13 +411,15 @@ function refreshMessageOperationRouteBinding(params: { if (!params.binding) { return; } - const existing = params.context.dedupe.get(params.binding.key); - if (existing?.requestIdentity === params.requestScope) { - params.context.dedupe.set(params.binding.key, { + const bindings = getMessageOperationRouteBindings(params.context); + const existing = bindings.get(params.binding.key); + if (existing?.requestScope === params.requestScope) { + bindings.set(params.binding.key, { ...existing, ts: Date.now(), retainUntilSettled: false, }); + pruneMessageOperationRouteBindings(bindings, Date.now()); } } @@ -376,11 +431,12 @@ function retainMessageOperationRouteBinding(params: { if (!params.binding) { return; } - const existing = params.context.dedupe.get(params.binding.key); - if (existing?.requestIdentity === params.requestScope) { - // Maintenance must not sever the mutable-default binding from its canonical - // in-flight operation; settlement below starts the ordinary bounded TTL. - params.context.dedupe.set(params.binding.key, { + const bindings = getMessageOperationRouteBindings(params.context); + const existing = bindings.get(params.binding.key); + if (existing?.requestScope === params.requestScope) { + // Active provider work owns this alias even past TTL or capacity pressure; + // settlement below restarts ordinary expiry. + bindings.set(params.binding.key, { ...existing, retainUntilSettled: true, }); diff --git a/src/gateway/server-shared.ts b/src/gateway/server-shared.ts index 433066077e2f..cce7ef9a9187 100644 --- a/src/gateway/server-shared.ts +++ b/src/gateway/server-shared.ts @@ -13,8 +13,6 @@ export function pendingChatSendDedupeKey(runId: string): string { export type DedupeEntry = { ts: number; ok: boolean; - /** Active effectful work owns this entry until settlement starts its normal TTL. */ - retainUntilSettled?: boolean; /** Optional effectful-request fingerprint for methods with caller-supplied operation ids. */ requestIdentity?: string; payload?: unknown;