fix(slack): preserve active prefetch claims under cap

src/slack/bot.ts

@@ -22,7 +22,8 @@ import { SlackSocketClient, type SlackEnvelope } from './socket.js';
 import { resolveEventText, shouldAttachSlack, shouldProcessSlackEvent, type SlackMessageEvent } from './events.js';
 import {
     isThreadParticipated, markThreadParticipated,
-    claimThreadPrefetch, releaseThreadPrefetch, resetThreadPrefetchClaims,
+    claimThreadPrefetch, commitThreadPrefetch,
+    releaseThreadPrefetch, resetThreadPrefetchClaims,
 } from './thread-tracker.js';
 import { sendSlackText, getSlackSendClient } from './send-only-client.js';
 import { startSlackProgress, statusFromToolEvent } from './progress.js';
@@ -237,7 +238,11 @@ export async function processSlackMessageEvent(
     // rather than a return-by-return audit.
     let prefetchCommitted = false;
     try {
-        await runSlackMessageEvent(event, target, text, signal, opts, () => { prefetchCommitted = true; });
+        await runSlackMessageEvent(event, target, text, signal, opts, () => {
+            prefetchCommitted = Boolean(opts.prefetchToken) && commitThreadPrefetch(
+                event.channel || '', event.thread_ts || '', opts.prefetchToken || 0,
+            );
+        });
     } finally {
         if (opts.prefetchToken && !prefetchCommitted) {
             releaseThreadPrefetch(event.channel || '', event.thread_ts || '', opts.prefetchToken);
@@ -454,7 +459,8 @@ export async function handleSlackEnvelope(envelope: SlackEnvelope): Promise<void
             // A thread WE start needs no history: the parent mention and our
             // reply are already the session's own context. Spend the claim now
             // so the first follow-up does not re-inject what the agent said.
-            claimThreadPrefetch(event.channel, event.ts);
+            const token = claimThreadPrefetch(event.channel, event.ts);
+            commitThreadPrefetch(event.channel, event.ts, token);
         }
     }
     // Claim the one-time thread prefetch HERE, synchronously, before the ingress

src/slack/thread-tracker.ts

@@ -92,8 +92,9 @@ export function isThreadParticipated(channel: string, threadTs: string): boolean
 // too, so re-injecting the thread's history is the RIGHT behavior. Persisting
 // the claim would leave a context-less session permanently without context.
 
-/** key -> the token of the claim currently holding it. */
-const prefetchClaimed = new Map<string, number>();
+type PrefetchClaim = { token: number; committed: boolean };
+/** key -> current owner and whether history was actually injected. */
+const prefetchClaimed = new Map<string, PrefetchClaim>();
 const PREFETCH_CLAIM_CAP = 500;
 let prefetchToken = 0;
 
@@ -115,17 +116,36 @@ export function claimThreadPrefetch(channel: string, threadTs: string): number {
     const key = threadKey(channel, threadTs);
     if (prefetchClaimed.has(key)) return 0;
     if (prefetchClaimed.size >= PREFETCH_CLAIM_CAP) {
-        // Oldest half by insertion order — a claim is never refreshed, so
-        // insertion order IS recency here.
-        for (const [stale] of [...prefetchClaimed].slice(0, Math.floor(PREFETCH_CLAIM_CAP / 2))) {
+        // Active owners are singleflight locks, not cache entries. Evicting one
+        // lets another envelope claim the same live thread and inject history
+        // twice. Only completed claims may give ground under pressure.
+        let removed = 0;
+        const target = Math.floor(PREFETCH_CLAIM_CAP / 2);
+        for (const [stale, claim] of prefetchClaimed) {
+            if (!claim.committed) continue;
             prefetchClaimed.delete(stale);
+            removed += 1;
+            if (removed >= target) break;
         }
+        // All bounded slots can legitimately be in flight. Decline rather than
+        // queue or violate singleflight; a later message can retry after one
+        // owner commits or releases.
+        if (prefetchClaimed.size >= PREFETCH_CLAIM_CAP) return 0;
     }
     const token = ++prefetchToken;
-    prefetchClaimed.set(key, token);
+    prefetchClaimed.set(key, { token, committed: false });
     return token;
 }
 
+/** Mark that this owner actually injected history; completed claims are evictable. */
+export function commitThreadPrefetch(channel: string, threadTs: string, token: number): boolean {
+    if (!channel || !threadTs || !token) return false;
+    const claim = prefetchClaimed.get(threadKey(channel, threadTs));
+    if (!claim || claim.token !== token) return false;
+    claim.committed = true;
+    return true;
+}
+
 /**
  * Give a claim back when no history was actually injected.
  *
@@ -138,7 +158,7 @@ export function claimThreadPrefetch(channel: string, threadTs: string): number {
 export function releaseThreadPrefetch(channel: string, threadTs: string, token: number): void {
     if (!channel || !threadTs || !token) return;
     const key = threadKey(channel, threadTs);
-    if (prefetchClaimed.get(key) !== token) return;
+    if (prefetchClaimed.get(key)?.token !== token) return;
     prefetchClaimed.delete(key);
 }
 

structure/str_func.md

@@ -239,11 +239,11 @@ cli-jaw/
 │   │   └── discord-file.ts   ← Discord 파일 전송 (67L)
 │   ├── slack/                ← Slack 인터페이스 (20 files, Socket Mode + Web API, SDK 없음)
 │   │   ├── socket.ts         ← Socket Mode client (apps.connections.open → wss, ack-before-work, envelope dedupe TTL, hello deadline, backoff 재연결) (372L)
-│   │   ├── bot.ts            ← Slack 봇 lifecycle + envelope routing + orchestrate 경로 + queued-result waiter (627L)
+│   │   ├── bot.ts            ← Slack 봇 lifecycle + envelope routing + orchestrate 경로 + queued-result waiter (633L)
 │   │   ├── api.ts            ← Slack Web API fetch wrapper (HTTP 200 + ok:false를 실패로 처리, credential/URL redaction) (168L)
 │   │   ├── format.ts         ← CommonMark → mrkdwn 변환 + code-fence 보존 chunking (62L)
 │   │   ├── events.ts         ← inbound gating (self-echo/bot/subtype/allowlist/mention) + Block Kit 텍스트 추출 (216L)
-│   │   ├── thread-tracker.ts ← 참여 스레드 영속 추적 (mention/봇응답 마킹, 캡드 셋, 무멘션 스레드 연속 대화 게이트 지원) (154L)
+│   │   ├── thread-tracker.ts ← 참여 스레드 영속 추적 (mention/봇응답 마킹, 캡드 셋, 무멘션 스레드 연속 대화 게이트 지원) (174L)
 │   │   ├── enrichment-cache.ts ← 공용 동시성 프리미티브 (TTL/cap 캐시, 원인별 억제, 능력 잠금 단일 재탐침, in-flight 합류, 집계 취소, 세대 무효화) (425L)
 │   │   ├── conversation.ts   ← 대화/스레드 컨텍스트 (conversations.info + replies, 참여자는 author 유도, method별 억제·시작률) (347L)
 │   │   ├── context.ts        ← 프롬프트 컨텍스트 블록 조립 (채널 id·thread_ts 무절단, 섹션별 코드포인트 예산, 신뢰 경계 문구 보존) (239L)

tests/unit/slack-thread-prefetch.test.ts

@@ -14,6 +14,7 @@ import assert from 'node:assert/strict';
 
 import {
     claimThreadPrefetch,
+    commitThreadPrefetch,
     releaseThreadPrefetch,
     resetThreadPrefetchClaims,
 } from '../../src/slack/thread-tracker.ts';
@@ -74,6 +75,34 @@ test('reset clears every claim', () => {
     assert.ok(claimThreadPrefetch('C1', '100.1') > 0, 'a new runtime re-injects history');
 });
 
+test('capacity pressure never evicts an active claim', () => {
+    const tokens: number[] = [];
+    for (let i = 0; i < 500; i += 1) {
+        tokens.push(claimThreadPrefetch('C1', `${i}.1`));
+    }
+    assert.ok(tokens.every(token => token > 0));
+    assert.equal(
+        claimThreadPrefetch('C1', 'overflow.1'), 0,
+        'a new prefetch must degrade while every bounded slot is active',
+    );
+    assert.equal(
+        claimThreadPrefetch('C1', '0.1'), 0,
+        'the oldest live owner must remain claimed under pressure',
+    );
+});
+
+test('capacity pressure may evict completed claims but preserves active ones', () => {
+    const active = claimThreadPrefetch('C1', 'active.1');
+    for (let i = 0; i < 499; i += 1) {
+        const ts = `done-${i}.1`;
+        const token = claimThreadPrefetch('C1', ts);
+        assert.ok(commitThreadPrefetch('C1', ts, token));
+    }
+    assert.ok(claimThreadPrefetch('C1', 'new.1') > 0, 'completed entries make bounded room');
+    assert.equal(claimThreadPrefetch('C1', 'active.1'), 0, 'the live owner is never evicted');
+    releaseThreadPrefetch('C1', 'active.1', active);
+});
+
 // ─── preamble rendering ─────────────────────────────
 
 test('the preamble is delimited and labelled with the reply count', () => {