fix(agent): 복사 스레드의 중복 응답 판정을 분리

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 (?, ?, ?)')

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,14 +94,25 @@ 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
  * finally, but a container killed mid-batch (SIGKILL) can leave one behind;
  * 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();
+}

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';
 

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: `<message to="dm">${text}</message>`,
+    });
+
+    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']);
   });
 });
 

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,