fix(agent): 스레드 조회를 torn read에 견디게 하고 선조회 지침을 복원한다

container/agent-runner/src/db/sqlite-errors.ts

@@ -0,0 +1,23 @@
+/**
+ * True for SQLite errors that indicate a corrupt READ view — almost always a
+ * cross-mount page-cache coherency issue on Docker Desktop macOS rather than
+ * actual file damage (host-side integrity_check passes).
+ *
+ * Two observed modes, with different recovery:
+ * - A one-off torn read: a single fresh read-only connection catches a page
+ *   mid-host-commit. The next open sees a consistent view again — callers
+ *   polling on an interval should treat this as "data not visible yet" and
+ *   retry (2026-07-15 read_current_thread incident: the tool's poller threw
+ *   once while the main poll loop read the same file cleanly).
+ * - A poisoned kernel page cache: every fresh open in the container keeps
+ *   seeing the same broken view. Reopening does NOT recover; only a fresh
+ *   container mount does. The poll loop exits after a streak of these so the
+ *   host respawns the container (see CORRUPTION_STREAK_EXIT in poll-loop.ts).
+ */
+export function isCorruptionError(msg: string): boolean {
+  return (
+    msg.includes('database disk image is malformed') ||
+    msg.includes('SQLITE_CORRUPT') ||
+    msg.includes('file is not a database')
+  );
+}

container/agent-runner/src/mcp-tools/current-thread.test.ts

@@ -0,0 +1,126 @@
+/**
+ * Tests for the read_current_thread container-side poller.
+ *
+ * Pins the 2026-07-15 incident behavior: the host wrote the thread response
+ * into inbound.db, but the tool's poller caught a torn read (`database disk
+ * image is malformed`) at exactly that moment, had no retry, surfaced the
+ * SQLite error to the user, and left the response row pending forever —
+ * nothing else consumes kind='system' rows.
+ */
+import { describe, it, expect, beforeEach, afterEach } from 'bun:test';
+
+import { initTestSessionDb, closeSessionDb, getInboundDb, getOutboundDb } from '../db/connection.js';
+import type { MessageInRow } from '../db/messages-in.js';
+import { requestCurrentThread } from './current-thread.js';
+
+function insertResponseRow(id: string, requestId: string, ageMs = 0): void {
+  const timestamp = new Date(Date.now() - ageMs).toISOString();
+  getInboundDb()
+    .prepare(
+      `INSERT INTO messages_in (id, kind, timestamp, status, trigger, content)
+       VALUES (?, 'system', ?, 'pending', 0, ?)`,
+    )
+    .run(
+      id,
+      timestamp,
+      JSON.stringify({ type: 'current_thread_response', requestId, messages: [{ text: 'hi' }] }),
+    );
+}
+
+function ackStatus(id: string): string | undefined {
+  const row = getOutboundDb()
+    .prepare('SELECT status FROM processing_ack WHERE message_id = ?')
+    .get(id) as { status: string } | undefined;
+  return row?.status;
+}
+
+/** The tool writes its host request synchronously before the first await. */
+function lastRequestId(): string {
+  const row = getOutboundDb()
+    .prepare("SELECT content FROM messages_out WHERE kind = 'system' ORDER BY rowid DESC LIMIT 1")
+    .get() as { content: string };
+  return (JSON.parse(row.content) as { requestId: string }).requestId;
+}
+
+beforeEach(() => {
+  initTestSessionDb();
+});
+
+afterEach(() => {
+  closeSessionDb();
+});
+
+describe('requestCurrentThread', () => {
+  it('returns messages once the host response lands', async () => {
+    const pending = requestCurrentThread(50, { timeoutMs: 2_000, pollIntervalMs: 10 });
+    insertResponseRow('resp-1', lastRequestId());
+
+    const result = await pending;
+
+    expect(result.error).toBeUndefined();
+    expect(result.messages).toEqual([{ text: 'hi' }]);
+    // Consumed responses are acked so they never ride along a later batch.
+    expect(ackStatus('resp-1')).toBe('completed');
+  });
+
+  it('keeps polling through torn reads while the host writes (SQLITE_CORRUPT)', async () => {
+    let calls = 0;
+    const find = (): MessageInRow | undefined => {
+      calls++;
+      if (calls <= 3) throw new Error('database disk image is malformed');
+      return { id: 'resp-2', content: JSON.stringify({ messages: [1] }) } as MessageInRow;
+    };
+
+    const result = await requestCurrentThread(50, { timeoutMs: 2_000, pollIntervalMs: 5, find });
+
+    expect(result.error).toBeUndefined();
+    expect(result.messages).toEqual([1]);
+    expect(calls).toBe(4);
+    expect(ackStatus('resp-2')).toBe('completed');
+  });
+
+  it('times out with a transient-read note when corruption persists', async () => {
+    const find = (): MessageInRow | undefined => {
+      throw new Error('SQLITE_CORRUPT: database disk image is malformed');
+    };
+
+    const result = await requestCurrentThread(50, { timeoutMs: 60, pollIntervalMs: 5, find });
+
+    expect(result.error).toContain('timed out');
+    expect(result.error).toContain('malformed');
+  });
+
+  it('does not blame a stale corruption once a later read succeeds', async () => {
+    let calls = 0;
+    const find = (): MessageInRow | undefined => {
+      calls++;
+      if (calls === 1) throw new Error('database disk image is malformed');
+      return undefined; // healthy reads, response just never arrives
+    };
+
+    const result = await requestCurrentThread(50, { timeoutMs: 60, pollIntervalMs: 5, find });
+
+    expect(result.error).toBe('Current thread request timed out.');
+  });
+
+  it('propagates non-corruption read errors unchanged', async () => {
+    const find = (): MessageInRow | undefined => {
+      throw new Error('no such table: messages_in');
+    };
+
+    await expect(requestCurrentThread(50, { timeoutMs: 60, pollIntervalMs: 5, find })).rejects.toThrow(
+      'no such table',
+    );
+  });
+
+  it('acks stray responses from a dead previous call, but never fresh ones', async () => {
+    insertResponseRow('stray', 'req-old', 10 * 60_000);
+    insertResponseRow('fresh', 'req-live', 1_000);
+
+    const result = await requestCurrentThread(50, { timeoutMs: 30, pollIntervalMs: 5 });
+
+    expect(result.error).toContain('timed out');
+    expect(ackStatus('stray')).toBe('completed');
+    expect(ackStatus('fresh')).toBeUndefined();
+  });
+});

container/agent-runner/src/mcp-tools/current-thread.ts

@@ -2,10 +2,21 @@
 import { openInboundDb, getOutboundDb } from '../db/connection.js';
 import { markCompleted, type MessageInRow } from '../db/messages-in.js';
 import { writeMessageOut } from '../db/messages-out.js';
+import { isCorruptionError } from '../db/sqlite-errors.js';
 import { registerTools } from './server.js';
 import type { McpToolDefinition } from './types.js';
 
 const RESPONSE_TYPE = 'current_thread_response';
+const REQUEST_TIMEOUT_MS = 30_000;
+const POLL_INTERVAL_MS = 250;
+/**
+ * Pending responses older than this are strays from a call that died
+ * mid-poll (e.g. a torn read killed the poller before it could ack).
+ * Nothing else ever consumes them — both poll-loop paths skip
+ * kind='system' rows — so they would sit pending forever. Gated well past
+ * REQUEST_TIMEOUT_MS so a concurrent call's live response is never swept.
+ */
+const STALE_RESPONSE_MS = 2 * 60_000;
 
 function ok(text: string) {
   return { content: [{ type: 'text' as const, text }] };
@@ -40,19 +51,67 @@ function findResponse(requestId: string): MessageInRow | undefined {
   }
 }
 
+/** Ack stray responses left behind by a previous call that never read them. */
+function expireStaleResponses(): void {
+  const inbound = openInboundDb();
+  try {
+    const cutoff = new Date(Date.now() - STALE_RESPONSE_MS).toISOString();
+    const stray = inbound
+      .prepare(
+        `SELECT id FROM messages_in
+          WHERE status = 'pending'
+            AND json_extract(content, '$.type') = ?
+            AND datetime(timestamp) < datetime(?)`,
+      )
+      .all(RESPONSE_TYPE, cutoff) as Array<{ id: string }>;
+    if (stray.length > 0) markCompleted(stray.map((r) => r.id));
+  } catch {
+    // Hygiene only — never let the sweep block the actual request.
+  } finally {
+    inbound.close();
+  }
+}
+
 const sleep = (ms: number) => new Promise<void>((resolve) => setTimeout(resolve, ms));
 
-async function requestCurrentThread(limit: number): Promise<{ messages?: unknown; error?: string }> {
+export interface RequestCurrentThreadOptions {
+  timeoutMs?: number;
+  pollIntervalMs?: number;
+  /** Test seam — production always polls inbound.db via findResponse. */
+  find?: (requestId: string) => MessageInRow | undefined;
+}
+
+export async function requestCurrentThread(
+  limit: number,
+  options: RequestCurrentThreadOptions = {},
+): Promise<{ messages?: unknown; error?: string }> {
+  const { timeoutMs = REQUEST_TIMEOUT_MS, pollIntervalMs = POLL_INTERVAL_MS, find = findResponse } = options;
+  expireStaleResponses();
+
   const requestId = `current-thread-${Date.now()}-${Math.random().toString(36).slice(2, 8)}`;
   writeMessageOut({
     id: requestId,
     kind: 'system',
     content: JSON.stringify({ action: 'read_current_thread', requestId, limit }),
   });
 
-  const deadline = Date.now() + 30_000;
+  const deadline = Date.now() + timeoutMs;
+  let lastCorruptRead: string | undefined;
   while (Date.now() < deadline) {
-    const response = findResponse(requestId);
+    let response: MessageInRow | undefined;
+    try {
+      response = find(requestId);
+      lastCorruptRead = undefined;
+    } catch (cause) {
+      const message = cause instanceof Error ? cause.message : String(cause);
+      if (!isCorruptionError(message)) throw cause;
+      // This poller reads inbound.db right up to the moment the host commits
+      // the response row, so it is the likeliest reader to catch a torn page
+      // (SQLITE_CORRUPT while the file itself is intact — see
+      // db/sqlite-errors.ts). Treat it as "response not visible yet"; the
+      // next tick opens a fresh connection and usually sees a clean view.
+      lastCorruptRead = message;
+    }
     if (response) {
       markCompleted([response.id]);
       try {
@@ -61,16 +120,21 @@ async function requestCurrentThread(limit: number): Promise<{ messages?: unknown
         return { error: 'Invalid response from the host.' };
       }
     }
-    await sleep(250);
+    await sleep(pollIntervalMs);
   }
-  return { error: 'Current thread request timed out.' };
+  return {
+    error: lastCorruptRead
+      ? `Current thread request timed out; reads kept failing with a transient error (${lastCorruptRead}). ` +
+        'Call the tool once more before reporting a system problem to the user.'
+      : 'Current thread request timed out.',
+  };
 }
 
 export const readCurrentThread: McpToolDefinition = {
   tool: {
     name: 'read_current_thread',
     description:
-      '현재 Slack 입력만으로 필요한 앞선 맥락을 알 수 없을 때 현재 thread를 한 번 읽습니다. 본문 없는 mention에서는 같은 작성자의 아직 답변되지 않은 명시적 요청을 하나로 특정할 수 있고 새 권한이나 위험한 작업이 아닐 때만 이어서 처리하세요. 다른 사용자의 요청, 이미 답변된 요청, 민감한 작업을 추측하지 말고 불명확하면 짧게 물어보세요. 요청 자체에 충분한 맥락이 있으면 호출하지 마세요. 다른 채널이나 thread는 조회할 수 없습니다.',
+      '현재 Slack 입력만으로 필요한 앞선 맥락을 알 수 없을 때 현재 thread를 읽습니다. 사용자가 “위 내용”, “이 이슈”, “스레드”처럼 현재 입력에 없는 앞선 내용을 가리키거나 본문 없는 mention만 보내면 답하거나 되묻기 전에 먼저 호출하세요. 본문 없는 mention은 조회 후에도 같은 작성자의 아직 답변되지 않은 명시적 요청을 하나로 특정할 수 있고 새 권한이나 위험한 작업이 아닐 때만 이어서 처리하고, 다른 사용자의 요청, 이미 답변된 요청, 민감한 작업은 추측하지 말고 짧게 물어보세요. 요청 자체에 충분한 맥락이 있으면 호출하지 마세요. 다른 채널이나 thread는 조회할 수 없습니다.',
     inputSchema: {
       type: 'object' as const,
       properties: {

container/agent-runner/src/poll-loop.test.ts

@@ -4,8 +4,9 @@ import { initTestSessionDb, closeSessionDb, getInboundDb, getOutboundDb } from '
 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 { isCorruptionError } from './db/sqlite-errors.js';
 import { formatMessages, extractRouting } from './formatter.js';
-import { isCorruptionError, processQuery } from './poll-loop.js';
+import { processQuery } from './poll-loop.js';
 import { MockProvider } from './providers/mock.js';
 import type { AgentQuery, ProviderEvent } from './providers/types.js';
 

container/agent-runner/src/poll-loop.ts

@@ -9,6 +9,7 @@ import {
 } from './db/messages-in.js';
 import { hasTaskSend, writeMessageOut } from './db/messages-out.js';
 import { touchHeartbeat, clearStaleProcessingAcks } from './db/connection.js';
+import { isCorruptionError } from './db/sqlite-errors.js';
 import {
   clearTurnSends,
   clearContinuation,
@@ -50,21 +51,6 @@ const ACTIVE_POLL_INTERVAL_MS = 500;
  */
 const CORRUPTION_STREAK_EXIT = 10;
 
-/**
- * True for SQLite errors that indicate a corrupt READ view — almost always a
- * cross-mount page-cache coherency issue on Docker Desktop macOS rather than
- * actual file damage (host-side integrity_check passes). Reopening the DB
- * handle inside this process does NOT recover; only a fresh container mount
- * does. Caller's job is to exit so host-sweep respawns the container.
- */
-export function isCorruptionError(msg: string): boolean {
-  return (
-    msg.includes('database disk image is malformed') ||
-    msg.includes('SQLITE_CORRUPT') ||
-    msg.includes('file is not a database')
-  );
-}
-
 function log(msg: string): void {
   console.error(`[poll-loop] ${msg}`);
 }