Skip to content

fix(slack): preserve active prefetch claims under cap #325

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
parkjs101 merged 6 commits into from
Aug 12, 2026
Merged

fix(slack): preserve active prefetch claims under cap #325

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
b5b34bf
fix(slack): preserve active prefetch claims under cap
parkjs101 Aug 12, 2026
7400e95
Merge remote-tracking branch 'origin/dev' into codex/issue-317-slack-…
parkjs101 Aug 12, 2026
4afc972
Merge remote-tracking branch 'origin/dev' into codex/issue-317-slack-…
parkjs101 Aug 12, 2026
ae7f055
fix(slack): release prefetch claims before handoff
parkjs101 Aug 12, 2026
9f5c346
Merge remote-tracking branch 'origin/dev' into codex/issue-317-slack-…
parkjs101 Aug 12, 2026
4cf9868
test(install): scope native Windows README guard
parkjs101 Aug 12, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
98 changes: 55 additions & 43 deletions src/slack/bot.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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';
Expand Down Expand Up @@ -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,
);
});
Comment on lines +241 to +245

@coderabbitai coderabbitai Bot Aug 12, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🎯 Functional Correctness | 🟠 Major | ⚡ Quick win

Release claims on every pre-enqueue exit.

processSlackMessageEvent releases only claims that reach its finally. handleSlackEnvelope acquires prefetchToken before attachment recovery, then can return at Line 500 or Lines 503-511 without passing the claim to this function. An attachment-recovery exception has the same result.

These paths leave an uncommitted claim active. The new capacity logic cannot evict active claims. After 500 such threaded events, all later prefetch claims are declined until reset.

Wrap post-claim work in a try/finally. Release the token unless enqueueSlackIngress accepts ownership.

Proposed lifecycle guard 
 const prefetchToken = event.thread_ts
     ? claimThreadPrefetch(event.channel || '', event.thread_ts)
     : 0;
+let prefetchHandedOff = false;
 
-const target = buildSlackTarget(event);
+try {
+const target = buildSlackTarget(event);
 // ... attachment recovery and early-return paths ...
 
 enqueueSlackIngress(slackIngressLaneKey(target), signal =>
     processSlackMessageEvent(event, target, text, signal, {
         prefetchToken,
         // ...
     }));
+prefetchHandedOff = true;
+} finally {
+    if (prefetchToken && !prefetchHandedOff) {
+        releaseThreadPrefetch(event.channel || '', event.thread_ts || '', prefetchToken);
+    }
+}
🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@src/slack/bot.ts` around lines 241 - 245, Update handleSlackEnvelope’s
post-prefetch-token flow to use a try/finally that releases the acquired claim
on every early return and attachment-recovery exception. Track whether
enqueueSlackIngress (or the existing ownership-transfer path) accepted the
token, and only skip release when ownership was accepted; preserve normal
processing and commit behavior otherwise.

} finally {
if (opts.prefetchToken && !prefetchCommitted) {
releaseThreadPrefetch(event.channel || '', event.thread_ts || '', opts.prefetchToken);
Expand Down Expand Up @@ -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
Expand All @@ -467,50 +473,56 @@ export async function handleSlackEnvelope(envelope: SlackEnvelope): Promise<void
const prefetchToken = event.thread_ts
? claimThreadPrefetch(event.channel || '', event.thread_ts)
: 0;

const target = buildSlackTarget(event);
setLastActiveTarget('slack', target);
setLatestSeenTarget('slack', target);

const text = resolveEventText(event, selfUserId);
let hasFiles = Boolean(event.files?.length);
// app_mention 봉투에는 files 가 없고, 첨부를 가진 message 사본은 위
// shouldProcessSlackEvent 에서 mention_via_app_mention 으로 드롭된다.
// 그래서 멘션과 함께 올린 파일은 여기서 되찾지 않으면 영영 사라진다.
if (!hasFiles && event.type === 'app_mention' && event.channel && event.ts) {
const recoverToken = getSlackSendClient().token;
if (recoverToken) {
const recovered = await recoverSlackAttachments(
recoverToken, event.channel, event.ts,
event.thread_ts ? { threadTs: event.thread_ts } : {},
);
if (recovered.length) {
event.files = recovered;
hasFiles = true;
log.info(`[slack:recover] ${event.channel} ts=${event.ts}: ${recovered.length} attachment(s)`);
let prefetchHandedOff = false;
try {
const target = buildSlackTarget(event);
setLastActiveTarget('slack', target);
setLatestSeenTarget('slack', target);

const text = resolveEventText(event, selfUserId);
let hasFiles = Boolean(event.files?.length);
// app_mention 봉투에는 files 가 없고, 첨부를 가진 message 사본은 위
// shouldProcessSlackEvent 에서 mention_via_app_mention 으로 드롭된다.
// 그래서 멘션과 함께 올린 파일은 여기서 되찾지 않으면 영영 사라진다.
if (!hasFiles && event.type === 'app_mention' && event.channel && event.ts) {
const recoverToken = getSlackSendClient().token;
if (recoverToken) {
const recovered = await recoverSlackAttachments(
recoverToken, event.channel, event.ts,
event.thread_ts ? { threadTs: event.thread_ts } : {},
);
if (recovered.length) {
event.files = recovered;
hasFiles = true;
log.info(`[slack:recover] ${event.channel} ts=${event.ts}: ${recovered.length} attachment(s)`);
}
}
}
}
if (!text && !hasFiles) return;
if (text) log.info(`[slack:in] ${event.channel}: ${redactOutboundText(text).slice(0, 80)}`);

if (!hasFiles && isResetIntent(text)) {
const client = getSlackSendClient();
const result = submitMessage(text, { origin: 'slack', target });
if (client.token) {
await sendSlackText(client.token, target, result.action === 'rejected'
? t('ws.agentBusy', {}, currentLocale())
: t('tg.resetDone', {}, currentLocale()));
if (!text && !hasFiles) return;
if (text) log.info(`[slack:in] ${event.channel}: ${redactOutboundText(text).slice(0, 80)}`);

if (!hasFiles && isResetIntent(text)) {
const client = getSlackSendClient();
const result = submitMessage(text, { origin: 'slack', target });
if (client.token) {
await sendSlackText(client.token, target, result.action === 'rejected'
? t('ws.agentBusy', {}, currentLocale())
: t('tg.resetDone', {}, currentLocale()));
}
return;
}
return;
}

enqueueSlackIngress(slackIngressLaneKey(target), signal =>
processSlackMessageEvent(event, target, text, signal, {
prefetchToken,
...(reservedEventKey ? { eventKey: reservedEventKey } : {}),
...(reservationGeneration !== undefined ? { reservationGeneration } : {}),
}));
prefetchHandedOff = enqueueSlackIngress(slackIngressLaneKey(target), signal =>
processSlackMessageEvent(event, target, text, signal, {
prefetchToken,
...(reservedEventKey ? { eventKey: reservedEventKey } : {}),
...(reservationGeneration !== undefined ? { reservationGeneration } : {}),
}));
} finally {
if (prefetchToken && !prefetchHandedOff) {
releaseThreadPrefetch(event.channel || '', event.thread_ts || '', prefetchToken);
}
}
}

// ─── Init / Shutdown ────────────────────────────────
Expand Down
5 changes: 3 additions & 2 deletions src/slack/ingress.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -180,8 +180,8 @@ export function slackIngressLaneKey(target: RemoteTarget): string {
export function enqueueSlackIngress(
laneKey: string,
task: (signal: AbortSignal) => Promise<void>,
): void {
if (resetting) return;
): boolean {
if (resetting) return false;
const taskGeneration = generation;
const controller = new AbortController();
controllers.add(controller);
Expand All @@ -201,6 +201,7 @@ export function enqueueSlackIngress(
void tail.then(() => {
if (ingressTails.get(laneKey) === tail) ingressTails.delete(laneKey);
});
return true;
}

export type SlackRunContext = {
Expand Down
34 changes: 27 additions & 7 deletions src/slack/thread-tracker.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;

Expand All @@ -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.
*
Expand All @@ -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);
}

Expand Down
6 changes: 3 additions & 3 deletions structure/str_func.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -239,19 +239,19 @@ 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 (639L)
│ │ ├── 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)
│ │ ├── history.ts ← 동적 조회 (conversations.history/replies form-encoded 래퍼 + 재시도 + 에이전트용 포맷/redact) (218L)
│ │ ├── attachment-recovery.ts ← app_mention 봉투에 없는 첨부를 channel+ts 재조회로 복구 (oldest+inclusive+limit=1) (53L)
│ │ ├── commands.ts ← slash command → 공유 parseCommand/executeCommand 파이프라인 (148L)
│ │ ├── slack-file.ts ← files.getUploadURLExternal → upload → completeUploadExternal 3단계 업로드 (97L)
│ │ ├── ingress.ts ← 세션별 ingress lane + admitSlackRun 동기 실행 예약(sessionLanes) + 전역 다운로드 세마포어 + shutdown abort/drain (275L) ✨
│ │ ├── ingress.ts ← 세션별 ingress lane + admitSlackRun 동기 실행 예약(sessionLanes) + 전역 다운로드 세마포어 + shutdown abort/drain (276L) ✨
│ │ ├── inbound-file.ts ← 인바운드 첨부 단일 IO owner (files.info → 인증 스트리밍 다운로드 → saveUpload, 파일/메시지 바이트 예산, 고정 error code) (280L) ✨
│ │ ├── inbound-url.ts ← 인바운드 다운로드 URL 검증 (Slack host allowlist + https-only hop + 사설망 거부) (44L) ✨
│ │ ├── send-only-client.ts ← bot-token 전용 outbound + conversations.open DM 해석 (69L)
Expand Down
7 changes: 6 additions & 1 deletion tests/unit/safe-install.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -459,10 +459,15 @@ test('SAF-004g: postinstall child processes use service-safe PATH consistently',
});

test('SAF-004h: README scopes Windows installation support to WSL', () => {
const nativeWindowsBetaStart = readmeSrc.indexOf('<summary><b>Native Windows (PowerShell beta)</b>');
const nativeWindowsBetaEnd = readmeSrc.indexOf('</details>', nativeWindowsBetaStart);
const readmeOutsideNativeWindowsBeta = nativeWindowsBetaStart >= 0 && nativeWindowsBetaEnd >= 0
? readmeSrc.slice(0, nativeWindowsBetaStart) + readmeSrc.slice(nativeWindowsBetaEnd + '</details>'.length)
: readmeSrc;
assert.ok(readmeSrc.includes('wsl --install'), 'README should document Windows setup through WSL');
assert.ok(readmeSrc.includes('wsl.exe -d Ubuntu -- bash -lc "jaw dashboard"'), 'README should document PowerShell-to-WSL login-shell invocation');
assert.ok(readmeSrc.includes('macOS / Linux / WSL with Node.js 22+ already installed'), 'README default npm install block should be OS-scoped');
assert.equal(readmeSrc.includes('Get-Command jaw'), false, 'README must not troubleshoot native PowerShell jaw resolution as a supported path');
assert.equal(readmeOutsideNativeWindowsBeta.includes('Get-Command jaw'), false, 'README must keep native PowerShell jaw resolution inside the explicitly scoped beta section');
assert.equal(localizedReadmeSrc.includes('$env:JAW_SAFE="1"; npm install -g cli-jaw'), false, 'localized READMEs must not advertise native PowerShell safe install');
assert.equal(localizedReadmeSrc.includes('# Windows PowerShell'), false, 'localized READMEs must not present native PowerShell install snippets');
assert.equal(localizedReadmeSrc.includes('npm bin -g'), false, 'localized README troubleshooting should use npm prefix -g, not removed npm bin -g');
Expand Down
Loading
Loading