From 8148877b4a798494cd58bfd772559b4365eacd08 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=9B=B7=E7=94=B5=E8=8A=BD=E8=A1=A3?= Date: Wed, 22 Apr 2026 18:43:05 -0400 Subject: [PATCH] fix(runtime): preserve inbound correlations for active runs --- .../inbound-correlation-registry.ts | 13 ----- .../unit/inbound-correlation-registry.test.ts | 50 +++++++++++++++++++ 2 files changed, 50 insertions(+), 13 deletions(-) create mode 100644 tests/unit/inbound-correlation-registry.test.ts diff --git a/src/main/agent-runtime/inbound-correlation-registry.ts b/src/main/agent-runtime/inbound-correlation-registry.ts index 929272ca..923eee16 100644 --- a/src/main/agent-runtime/inbound-correlation-registry.ts +++ b/src/main/agent-runtime/inbound-correlation-registry.ts @@ -17,8 +17,6 @@ interface InboundCorrelationRegistry { lastAllocatedTimestamp: number; } -const MAX_INBOUND_CORRELATIONS_PER_RUN = 512; - export class InboundCorrelationRegistryStore { private readonly registries = new Map(); @@ -34,7 +32,6 @@ export class InboundCorrelationRegistryStore { }; registry.byMessageId.set(record.id, entry); registry.byPiTimestamp.set(allocated, record.id); - this.evictOldestIfNeeded(registry); return allocated; } @@ -62,14 +59,4 @@ export class InboundCorrelationRegistryStore { this.registries.set(runId, registry); return registry; } - - private evictOldestIfNeeded(registry: InboundCorrelationRegistry): void { - while (registry.byMessageId.size > MAX_INBOUND_CORRELATIONS_PER_RUN) { - const oldestMessageId = registry.byMessageId.keys().next().value; - if (!oldestMessageId) break; - const entry = registry.byMessageId.get(oldestMessageId); - registry.byMessageId.delete(oldestMessageId); - if (entry) registry.byPiTimestamp.delete(entry.piTimestamp); - } - } } diff --git a/tests/unit/inbound-correlation-registry.test.ts b/tests/unit/inbound-correlation-registry.test.ts new file mode 100644 index 00000000..ff60e4f6 --- /dev/null +++ b/tests/unit/inbound-correlation-registry.test.ts @@ -0,0 +1,50 @@ +import { describe, expect, it } from "vitest"; +import { InboundCorrelationRegistryStore } from "../../src/main/agent-runtime/inbound-correlation-registry"; +import type { AgentMessageRecord } from "../../src/shared/types"; + +function makeMessage(index: number): AgentMessageRecord { + return { + id: `message-${index}`, + companyId: "company-1", + fromAgentId: `agent-${index}`, + toAgentId: "agent-recipient", + senderKind: "agent", + channel: "direct", + channelTargetId: null, + subject: `message ${index}`, + body: `body ${index}`, + priority: "normal", + readAt: null, + parentMessageId: null, + attachmentsJson: "[]", + createdAt: "2026-04-22T00:00:00.000Z", + }; +} + +describe("InboundCorrelationRegistryStore", () => { + it("preserves every inbound correlation for the active run", () => { + const store = new InboundCorrelationRegistryStore(); + const timestamps: number[] = []; + + for (let i = 0; i < 700; i += 1) { + timestamps.push(store.register("run-1", makeMessage(i), 1_000)); + } + + expect(new Set(timestamps).size).toBe(700); + for (let i = 0; i < timestamps.length; i += 1) { + expect(store.lookup("run-1", timestamps[i])).toMatchObject({ + fromAgentId: `agent-${i}`, + messageId: `message-${i}`, + }); + } + }); + + it("drops correlations only when the run is released", () => { + const store = new InboundCorrelationRegistryStore(); + const timestamp = store.register("run-1", makeMessage(1), 1_000); + + expect(store.lookup("run-1", timestamp)?.messageId).toBe("message-1"); + store.deleteRun("run-1"); + expect(store.lookup("run-1", timestamp)).toBeNull(); + }); +});