diff --git a/container/agent-runner/src/db/session-state.test.ts b/container/agent-runner/src/db/session-state.test.ts index af95b93047f..16d46e4f5c2 100644 --- a/container/agent-runner/src/db/session-state.test.ts +++ b/container/agent-runner/src/db/session-state.test.ts @@ -2,11 +2,14 @@ import { beforeEach, describe, expect, test } from 'bun:test'; import { getOutboundDb, initTestSessionDb } from './connection.js'; import { + clearTurnSends, clearContinuation, getContinuation, migrateLegacyContinuation, + recordTurnSend, setContinuation, setCurrentRequestMessageId, + wasSentThisTurn, } from './session-state.js'; beforeEach(() => { @@ -29,6 +32,43 @@ describe('session-state — active request attribution', () => { }); }); +describe('session-state — turn send deduplication', () => { + test('persists the active request send in shared session state', () => { + setCurrentRequestMessageId('request-1'); + recordTurnSend('slack', 'slack:C1', 'slack:C1:thread-1', '같은 답변'); + + expect(wasSentThisTurn('slack', 'slack:C1', 'slack:C1:thread-1', '같은 답변')).toBe(true); + const row = getOutboundDb().prepare("SELECT value FROM session_state WHERE key GLOB 'turn_send:*'").get() as { + value: string; + }; + expect(row.value).toBe('request-1'); + }); + + test('does not suppress the same text in a copied thread', () => { + setCurrentRequestMessageId('request-1'); + recordTurnSend('slack', 'slack:C1', 'slack:C1:thread-original', '같은 답변'); + + expect(wasSentThisTurn('slack', 'slack:C1', 'slack:C1:thread-copy', '같은 답변')).toBe(false); + }); + + test('does not suppress the same text in a later request', () => { + setCurrentRequestMessageId('request-1'); + recordTurnSend('slack', 'slack:C1', 'slack:C1:thread-1', '같은 답변'); + + setCurrentRequestMessageId('request-2'); + expect(wasSentThisTurn('slack', 'slack:C1', 'slack:C1:thread-1', '같은 답변')).toBe(false); + }); + + test('clears all send records after the turn completes', () => { + setCurrentRequestMessageId('request-1'); + recordTurnSend('slack', 'slack:C1', 'slack:C1:thread-1', '같은 답변'); + + clearTurnSends(); + + expect(wasSentThisTurn('slack', 'slack:C1', 'slack:C1:thread-1', '같은 답변')).toBe(false); + }); +}); + function seedLegacy(value: string): void { getOutboundDb() .prepare('INSERT INTO session_state (key, value, updated_at) VALUES (?, ?, ?)') diff --git a/container/agent-runner/src/db/session-state.ts b/container/agent-runner/src/db/session-state.ts index 92501d49b08..1a63cf1157e 100644 --- a/container/agent-runner/src/db/session-state.ts +++ b/container/agent-runner/src/db/session-state.ts @@ -9,6 +9,8 @@ * providers is therefore lossless: each provider's last thread stays * on file and resumes cleanly if the user flips back. */ +import crypto from 'crypto'; + import { getOutboundDb } from './connection.js'; const LEGACY_KEY = 'sdk_session_id'; @@ -92,6 +94,8 @@ export function clearContinuation(providerName: string): void { */ const IN_REPLY_TO_KEY = 'current_in_reply_to'; const REQUEST_MESSAGE_KEY = 'current_request_message_id'; +const TURN_SEND_KEY_PREFIX = 'turn_send:'; +const TURN_SEND_SEPARATOR = '\0'; /** * Ignore a stamp older than this. The poll loop clears the stamp in a @@ -99,7 +103,16 @@ const REQUEST_MESSAGE_KEY = 'current_request_message_id'; * the guard stops a later out-of-batch read from picking up a dead stamp. * Generous so a long-running batch's late sends still stamp correctly. */ -const IN_REPLY_TO_MAX_AGE_MS = 30 * 60 * 1000; +const ACTIVE_STAMP_MAX_AGE_MS = 30 * 60 * 1000; + +function getFreshValue(key: string): string | null { + const row = getOutboundDb().prepare('SELECT value, updated_at FROM session_state WHERE key = ?').get(key) as + | { value: string; updated_at: string } + | undefined; + if (!row) return null; + const age = Date.now() - new Date(row.updated_at).getTime(); + return Number.isFinite(age) && age <= ACTIVE_STAMP_MAX_AGE_MS ? row.value : null; +} export function setCurrentInReplyTo(id: string | null): void { if (id === null) { @@ -114,13 +127,7 @@ export function clearCurrentInReplyTo(): void { } export function getCurrentInReplyTo(): string | null { - const row = getOutboundDb() - .prepare('SELECT value, updated_at FROM session_state WHERE key = ?') - .get(IN_REPLY_TO_KEY) as { value: string; updated_at: string } | undefined; - if (!row) return null; - const age = Date.now() - new Date(row.updated_at).getTime(); - if (!Number.isFinite(age) || age > IN_REPLY_TO_MAX_AGE_MS) return null; - return row.value; + return getFreshValue(IN_REPLY_TO_KEY); } /** @@ -132,3 +139,40 @@ export function setCurrentRequestMessageId(id: string | null): void { if (id === null) deleteValue(REQUEST_MESSAGE_KEY); else setValue(REQUEST_MESSAGE_KEY, id); } + +function turnSendKey(channelType: string | null, platformId: string | null, threadId: string | null, text: string) { + const requestMessageId = getFreshValue(REQUEST_MESSAGE_KEY); + if (!requestMessageId) return null; + const digest = crypto + .createHash('sha256') + .update([requestMessageId, channelType ?? '', platformId ?? '', threadId ?? '', text].join(TURN_SEND_SEPARATOR)) + .digest('hex'); + return [`${TURN_SEND_KEY_PREFIX}${digest}`, requestMessageId] as const; +} + +/** Record a user-facing message delivered this turn via send_message. */ +export function recordTurnSend( + channelType: string | null, + platformId: string | null, + threadId: string | null, + text: string, +): void { + const record = turnSendKey(channelType, platformId, threadId, text); + if (record) setValue(...record); +} + +/** True if this exact text was already sent to this destination this turn. */ +export function wasSentThisTurn( + channelType: string | null, + platformId: string | null, + threadId: string | null, + text: string, +): boolean { + const record = turnSendKey(channelType, platformId, threadId, text); + return record !== null && getValue(record[0]) !== undefined; +} + +/** Clear the record — called when a turn completes. */ +export function clearTurnSends(): void { + getOutboundDb().prepare("DELETE FROM session_state WHERE key GLOB 'turn_send:*'").run(); +} diff --git a/container/agent-runner/src/mcp-tools/core.ts b/container/agent-runner/src/mcp-tools/core.ts index 29f1899ca93..5758d12194e 100644 --- a/container/agent-runner/src/mcp-tools/core.ts +++ b/container/agent-runner/src/mcp-tools/core.ts @@ -13,9 +13,8 @@ import path from 'path'; import { findByName, getAllDestinations } from '../destinations.js'; import { getDestinationReplyRouting } from '../db/messages-in.js'; import { getMessageIdBySeq, getRoutingBySeq, writeMessageOut } from '../db/messages-out.js'; -import { getCurrentInReplyTo } from '../db/session-state.js'; +import { getCurrentInReplyTo, recordTurnSend } from '../db/session-state.js'; import { getSessionRouting } from '../db/session-routing.js'; -import { recordTurnSend } from '../turn-sends.js'; import { registerTools } from './server.js'; import type { McpToolDefinition } from './types.js'; diff --git a/container/agent-runner/src/poll-loop.test.ts b/container/agent-runner/src/poll-loop.test.ts index a4bf33b87da..c0d9549d774 100644 --- a/container/agent-runner/src/poll-loop.test.ts +++ b/container/agent-runner/src/poll-loop.test.ts @@ -3,16 +3,16 @@ import { describe, it, expect, beforeEach, afterEach } from 'bun:test'; import { initTestSessionDb, closeSessionDb, getInboundDb, getOutboundDb } from './db/connection.js'; import { getPendingMessages, markCompleted } from './db/messages-in.js'; import { getUndeliveredMessages, writeMessageOut } from './db/messages-out.js'; +import { clearTurnSends, recordTurnSend, setCurrentRequestMessageId } from './db/session-state.js'; import { formatMessages, extractRouting } from './formatter.js'; import { isCorruptionError, processQuery } from './poll-loop.js'; import { MockProvider } from './providers/mock.js'; import type { AgentQuery, ProviderEvent } from './providers/types.js'; -import { clearTurnSends, recordTurnSend } from './turn-sends.js'; beforeEach(() => { initTestSessionDb(); - // turn-sends is process-global; reset it so a prior test's record can't - // suppress a message here. + // Turn-send records live in session_state; reset them so a prior test + // cannot suppress a message here. clearTurnSends(); }); @@ -224,7 +224,13 @@ describe('origin metadata (from= attribute)', () => { .run(name, name, channelType, platformId); } - function insertWithRouting(id: string, kind: string, content: object, channelType: string | null, platformId: string | null): void { + function insertWithRouting( + id: string, + kind: string, + content: object, + channelType: string | null, + platformId: string | null, + ): void { getInboundDb() .prepare( `INSERT INTO messages_in (id, kind, timestamp, status, platform_id, channel_type, content) @@ -503,6 +509,7 @@ describe('interactive turn deduplication', () => { }; function seedMidTurnSend(text: string): void { + setCurrentRequestMessageId(routing.inReplyTo); getInboundDb() .prepare( `INSERT INTO destinations (name, display_name, type, channel_type, platform_id, agent_group_id) @@ -514,8 +521,8 @@ describe('interactive turn deduplication', () => { // the mid-turn send recorded under. getInboundDb() .prepare( - `INSERT INTO messages_in (id, kind, timestamp, status, trigger, channel_type, platform_id, thread_id, content) - VALUES ('m1', 'chat-sdk', datetime('now'), 'processing', 1, 'slack', 'slack:D1', ?, ?)`, + `INSERT INTO messages_in (id, seq, kind, timestamp, status, trigger, channel_type, platform_id, thread_id, content) + VALUES ('m1', 2, 'chat-sdk', datetime('now'), 'processing', 1, 'slack', 'slack:D1', ?, ?)`, ) .run(routing.threadId, JSON.stringify({ text: 'hi' })); // Mirror what the send_message MCP tool does: write the standalone chat row @@ -565,6 +572,39 @@ describe('interactive turn deduplication', () => { expect(finalRow).toBeDefined(); expect(JSON.parse(finalRow!.content).text).toBe(finalText); }); + + it('keeps identical text when the question comes from a copied thread', async () => { + const text = '복사한 스레드에서도 다시 답해 주세요.'; + seedMidTurnSend(text); + const copiedThreadId = 'slack:D1:1712345678.000200'; + setCurrentRequestMessageId('m-copy'); + getInboundDb() + .prepare( + `INSERT INTO messages_in (id, seq, kind, timestamp, status, trigger, channel_type, platform_id, thread_id, content) + VALUES ('m-copy', 4, 'chat-sdk', datetime('now'), 'processing', 1, 'slack', 'slack:D1', ?, ?)`, + ) + .run(copiedThreadId, JSON.stringify({ text: 'same question' })); + const { query } = makeResultQuery({ + type: 'result', + text: `${text}`, + }); + + await processQuery( + query, + { ...routing, threadId: copiedThreadId, inReplyTo: 'm-copy' }, + ['m-copy'], + 'claude', + undefined, + 'prompt', + undefined, + ); + + const out = getUndeliveredMessages(); + expect(out).toHaveLength(2); + const finalRow = out.find((message) => JSON.parse(message.content)._nanoclawFinal === true); + expect(finalRow?.thread_id).toBe(copiedThreadId); + expect(JSON.parse(finalRow!.content).text).toBe(text); + }); }); describe('Slack native stream events', () => { @@ -636,10 +676,7 @@ describe('Slack native stream events', () => { ); const out = getUndeliveredMessages(); - expect(out.map((row) => JSON.parse(row.content).action)).toEqual([ - 'stream_task_update', - 'stream_end', - ]); + expect(out.map((row) => JSON.parse(row.content).action)).toEqual(['stream_task_update', 'stream_end']); }); }); diff --git a/container/agent-runner/src/poll-loop.ts b/container/agent-runner/src/poll-loop.ts index 0af7c2c3e79..218ac1b19fb 100644 --- a/container/agent-runner/src/poll-loop.ts +++ b/container/agent-runner/src/poll-loop.ts @@ -8,15 +8,16 @@ import { type MessageInRow, } from './db/messages-in.js'; import { hasTaskSend, writeMessageOut } from './db/messages-out.js'; -import { clearTurnSends, wasSentThisTurn } from './turn-sends.js'; import { touchHeartbeat, clearStaleProcessingAcks } from './db/connection.js'; import { + clearTurnSends, clearContinuation, clearCurrentInReplyTo, migrateLegacyContinuation, setContinuation, setCurrentInReplyTo, setCurrentRequestMessageId, + wasSentThisTurn, } from './db/session-state.js'; import { formatMessages, diff --git a/container/agent-runner/src/turn-sends.ts b/container/agent-runner/src/turn-sends.ts deleted file mode 100644 index 01f9d7c344a..00000000000 Binary files a/container/agent-runner/src/turn-sends.ts and /dev/null differ