From 48725d31d72c270394f715edd81ce0e214aa8931 Mon Sep 17 00:00:00 2001 From: eshurakov <54751+eshurakov@users.noreply.github.com> Date: Thu, 6 Aug 2026 08:37:40 +0000 Subject: [PATCH 1/2] fix(cloud-agent-sdk): allow same-id event rebroadcasts Change the duplicate event filtering logic to allow events with an eventId equal to the current cursor. Previously, events with `eventId <= lastEventId` were dropped. This caused issues with entity upserts (e.g., tool part updates) where the system re-emits the same eventId with a new payload (e.g., transitioning from `pending` to `completed`). By changing the check to `raw.eventId < lastEventId`, we ensure that live tool updates and message updates are correctly delivered while still dropping strictly older replayed events. Co-authored-by: kiloconnect[bot] <240665456+kiloconnect[bot]@users.noreply.github.com> --- .../src/cloud-agent-transport.test.ts | 68 +++++++++++++++---- .../src/cloud-agent-transport.ts | 9 ++- 2 files changed, 61 insertions(+), 16 deletions(-) diff --git a/packages/cloud-agent-sdk/src/cloud-agent-transport.test.ts b/packages/cloud-agent-sdk/src/cloud-agent-transport.test.ts index e4df5261118..41009ede8b9 100644 --- a/packages/cloud-agent-sdk/src/cloud-agent-transport.test.ts +++ b/packages/cloud-agent-sdk/src/cloud-agent-transport.test.ts @@ -1315,22 +1315,63 @@ describe('CloudAgentTransport exactly-once event replay', () => { }; } - it('drops a duplicate event ID at the cursor', async () => { - const { transport, serviceEvents } = createTransportWithSinks(); + it('delivers same event ID rebroadcasts (entity upsert updates)', async () => { + const { transport, serviceEvents, chatEvents } = createTransportWithSinks(); transport.connect(); await flushPromises(); - // snapshot replay puts session.created at serviceEvents[0] const serviceCountBefore = serviceEvents.length; + const chatCountBefore = chatEvents.length; - // First delivery sets the cursor. + // First delivery sets the cursor. A second frame with the same eventId is + // how the DO broadcasts message.part.updated after an entity upsert + // (pending → running → completed keep the same stored row id). sendRaw(eventWithId(5)); - // Duplicate — same eventId as the cursor — must be dropped. sendRaw(eventWithId(5)); - // Only one new service event (the first delivery) reached the sink. - expect(serviceEvents.length).toBe(serviceCountBefore + 1); + expect(serviceEvents.length).toBe(serviceCountBefore + 2); + + sendRaw({ + eventId: 5, + executionId: null, + sessionId: 'ses-1', + streamEventType: 'kilocode', + timestamp: new Date().toISOString(), + data: { + type: 'message.part.updated', + properties: { + part: { + id: 'part-tool-1', + sessionID: 'ses-1', + messageID: 'msg-1', + type: 'tool', + callID: 'call-1', + tool: 'bash', + state: { + status: 'completed', + input: { command: 'pwd' }, + output: '/tmp', + title: 'pwd', + metadata: {}, + time: { start: 1, end: 2 }, + }, + }, + }, + }, + }); + + const toolUpdates = chatEvents + .slice(chatCountBefore) + .filter(e => e.type === 'message.part.updated'); + expect(toolUpdates).toHaveLength(1); + expect(toolUpdates[0]).toMatchObject({ + type: 'message.part.updated', + part: { + tool: 'bash', + state: { status: 'completed', input: { command: 'pwd' } }, + }, + }); transport.destroy(); }); @@ -1441,9 +1482,10 @@ describe('CloudAgentTransport exactly-once event replay', () => { jest.advanceTimersByTime(2000); await flushMicrotasks(); - // Phase 2: the DO replays 3 through 8 on the new socket. - // Events 3, 4, 5 must be dropped (≤ cursor at 5). - // Events 6, 7, 8 must be delivered. + // Phase 2: reconnect uses exclusive fromId (id > cursor). The DO may still + // overlap below the cursor during catch-up; those must be dropped. Same-id + // as the cursor is kept (entity upsert rebroadcasts), but exclusive replay + // normally starts at cursor+1. const newMockWs = webSocketConstructor.mock.results.at(-1)?.value as MockWebSocket; newMockWs.onopen?.(new Event('open')); for (let id = 3; id <= 8; id++) { @@ -1468,9 +1510,9 @@ describe('CloudAgentTransport exactly-once event replay', () => { .map(e => e.status.attempt) .sort((a, b) => a - b); - // Each ID from 1 through 8 must appear exactly once. - // Duplicates, missing IDs, or extra deliveries will fail the assertion. - expect(deliveredIds).toEqual([1, 2, 3, 4, 5, 6, 7, 8]); + // 3–4 dropped (below cursor). 5 re-delivered if present (same-id upsert). + // 6–8 are new. Live tool updates rely on same-id delivery. + expect(deliveredIds).toEqual([1, 2, 3, 4, 5, 5, 6, 7, 8]); transport.destroy(); newMockWs.onclose?.({ diff --git a/packages/cloud-agent-sdk/src/cloud-agent-transport.ts b/packages/cloud-agent-sdk/src/cloud-agent-transport.ts index 8dce2fea12c..015fd64ae45 100644 --- a/packages/cloud-agent-sdk/src/cloud-agent-transport.ts +++ b/packages/cloud-agent-sdk/src/cloud-agent-transport.ts @@ -201,10 +201,13 @@ function createCloudAgentTransport(config: CloudAgentTransportConfig): Transport lifecycleHooks: config.lifecycleHooks, websocketHeaders: config.websocketHeaders, onEvent: raw => { - // Drop replayed duplicates: positive IDs at or below the cursor - // are redundant — the sink saw them on the first delivery. + // Drop strictly older positive IDs (reconnect/replay catch-up). + // Same-id rebroadcasts must be kept: the DO entity-upserts tool + // parts (and messages) under a stable row id and re-emits that + // eventId with a newer payload (pending → running → completed). + // Treating `<=` as duplicate left live tools stuck on empty input. // eventId 0 (synthetic sentinel) is always delivered. - if (raw.eventId > 0 && lastEventId !== null && raw.eventId <= lastEventId) { + if (raw.eventId > 0 && lastEventId !== null && raw.eventId < lastEventId) { return; } From 2738dbd1f84182e185af99c5fe14a2aae68c9448 Mon Sep 17 00:00:00 2001 From: eshurakov <54751+eshurakov@users.noreply.github.com> Date: Thu, 6 Aug 2026 08:51:19 +0000 Subject: [PATCH 2/2] fix(cloud-agent-sdk): allow same-id event rebroadcasts Stop dropping events with an `eventId` less than or equal to the `lastEventId`. The Durable Object (DO) performs entity upserts for tool parts and messages using a stable row ID, which can result in rebroadcasting the same `eventId` with updated payloads (e.g., transitioning from `pending` to `completed`). Previously, the transport dropped these events, causing live tools to remain stuck with empty or stale input. The `lastEventId` now only tracks the high-water mark for reconnection purposes. Co-authored-by: kiloconnect[bot] <240665456+kiloconnect[bot]@users.noreply.github.com> --- .../src/cloud-agent-transport.test.ts | 69 ++++++++++++++----- .../src/cloud-agent-transport.ts | 17 ++--- 2 files changed, 58 insertions(+), 28 deletions(-) diff --git a/packages/cloud-agent-sdk/src/cloud-agent-transport.test.ts b/packages/cloud-agent-sdk/src/cloud-agent-transport.test.ts index 41009ede8b9..23facf06fb4 100644 --- a/packages/cloud-agent-sdk/src/cloud-agent-transport.test.ts +++ b/packages/cloud-agent-sdk/src/cloud-agent-transport.test.ts @@ -1296,7 +1296,7 @@ describe('CloudAgentTransport page-seam', () => { // Exactly-once event replay // --------------------------------------------------------------------------- -describe('CloudAgentTransport exactly-once event replay', () => { +describe('CloudAgentTransport event delivery and replay cursor', () => { /** * Create a session.status event with an explicit eventId and a valid * session status shape. Count-based tests verify which IDs were delivered. @@ -1376,21 +1376,59 @@ describe('CloudAgentTransport exactly-once event replay', () => { transport.destroy(); }); - it('drops a positive event ID below the cursor', async () => { - const { transport, serviceEvents } = createTransportWithSinks(); + it('delivers entity upserts with an event ID below the high-water mark', async () => { + const { transport, serviceEvents, chatEvents } = createTransportWithSinks(); transport.connect(); await flushPromises(); const serviceCountBefore = serviceEvents.length; + const chatCountBefore = chatEvents.length; - // Advance the cursor to 10. + // A later append-only event advances the cursor past an earlier tool part + // row; the DO still rebroadcasts that older id when the part completes. sendRaw(eventWithId(10)); - // eventId 5 is below the cursor — must be dropped. - sendRaw(eventWithId(5)); + sendRaw({ + eventId: 5, + executionId: null, + sessionId: 'ses-1', + streamEventType: 'kilocode', + timestamp: new Date().toISOString(), + data: { + type: 'message.part.updated', + properties: { + part: { + id: 'part-tool-1', + sessionID: 'ses-1', + messageID: 'msg-1', + type: 'tool', + callID: 'call-1', + tool: 'bash', + state: { + status: 'completed', + input: { command: 'pwd' }, + output: '/tmp', + title: 'pwd', + metadata: {}, + time: { start: 1, end: 2 }, + }, + }, + }, + }, + }); - // Only the eventId=10 delivery reached the sink. expect(serviceEvents.length).toBe(serviceCountBefore + 1); + const toolUpdates = chatEvents + .slice(chatCountBefore) + .filter(e => e.type === 'message.part.updated'); + expect(toolUpdates).toHaveLength(1); + expect(toolUpdates[0]).toMatchObject({ + type: 'message.part.updated', + part: { + tool: 'bash', + state: { status: 'completed', input: { command: 'pwd' } }, + }, + }); transport.destroy(); }); @@ -1430,7 +1468,7 @@ describe('CloudAgentTransport exactly-once event replay', () => { transport.destroy(); }); - it('replays 3 through 8 after 1 through 5 on resume, delivering each once', async () => { + it('delivers overlapping reconnect frames without filtering by event ID', async () => { jest.useFakeTimers(); try { async function flushMicrotasks(): Promise { @@ -1482,10 +1520,9 @@ describe('CloudAgentTransport exactly-once event replay', () => { jest.advanceTimersByTime(2000); await flushMicrotasks(); - // Phase 2: reconnect uses exclusive fromId (id > cursor). The DO may still - // overlap below the cursor during catch-up; those must be dropped. Same-id - // as the cursor is kept (entity upsert rebroadcasts), but exclusive replay - // normally starts at cursor+1. + // Phase 2: DO exclusive fromId should start after the cursor, but if + // overlapping frames arrive the client must still deliver them — entity + // upserts can reuse older ids with newer payloads. const newMockWs = webSocketConstructor.mock.results.at(-1)?.value as MockWebSocket; newMockWs.onopen?.(new Event('open')); for (let id = 3; id <= 8; id++) { @@ -1510,9 +1547,7 @@ describe('CloudAgentTransport exactly-once event replay', () => { .map(e => e.status.attempt) .sort((a, b) => a - b); - // 3–4 dropped (below cursor). 5 re-delivered if present (same-id upsert). - // 6–8 are new. Live tool updates rely on same-id delivery. - expect(deliveredIds).toEqual([1, 2, 3, 4, 5, 5, 6, 7, 8]); + expect(deliveredIds).toEqual([1, 2, 3, 3, 4, 4, 5, 5, 6, 7, 8]); transport.destroy(); newMockWs.onclose?.({ @@ -1545,9 +1580,9 @@ describe('CloudAgentTransport exactly-once event replay', () => { sendRaw(eventWithId(10)); expect(serviceEvents.length).toBe(serviceCountBefore + 1); - // Send eventId 5 — dropped, cursor must stay at 10. + // Older id is still delivered (entity upsert), but cursor stays at 10. sendRaw(eventWithId(5)); - expect(serviceEvents.length).toBe(serviceCountBefore + 1); + expect(serviceEvents.length).toBe(serviceCountBefore + 2); // Disconnect and reconnect to inspect the replay cursor. mockWs.onclose?.({ diff --git a/packages/cloud-agent-sdk/src/cloud-agent-transport.ts b/packages/cloud-agent-sdk/src/cloud-agent-transport.ts index 015fd64ae45..604441a805d 100644 --- a/packages/cloud-agent-sdk/src/cloud-agent-transport.ts +++ b/packages/cloud-agent-sdk/src/cloud-agent-transport.ts @@ -201,17 +201,12 @@ function createCloudAgentTransport(config: CloudAgentTransportConfig): Transport lifecycleHooks: config.lifecycleHooks, websocketHeaders: config.websocketHeaders, onEvent: raw => { - // Drop strictly older positive IDs (reconnect/replay catch-up). - // Same-id rebroadcasts must be kept: the DO entity-upserts tool - // parts (and messages) under a stable row id and re-emits that - // eventId with a newer payload (pending → running → completed). - // Treating `<=` as duplicate left live tools stuck on empty input. - // eventId 0 (synthetic sentinel) is always delivered. - if (raw.eventId > 0 && lastEventId !== null && raw.eventId < lastEventId) { - return; - } - - if (raw.eventId > 0) { + // Track high-water mark for reconnect fromId only. Do not filter + // by eventId: the DO entity-upserts tool/message parts under a + // stable row id and rebroadcasts that same (or older) id with a + // newer payload. Dropping those left live tools stuck on empty input. + // eventId 0 is a synthetic sentinel and never advances the cursor. + if (raw.eventId > 0 && (lastEventId === null || raw.eventId > lastEventId)) { lastEventId = raw.eventId; }