fix: remove idle-exit loop and add graceful worker swap (#6)

* fix: replace idle-exit with socket health watchdog and graceful worker swap

* fix: address PR review for reconnect watchdog and worker swap

---------

Co-authored-by: leecoder <leecoder@users.noreply.github.com>

Author: leecoder

dist/plugin.js

@@ -12431,6 +12431,7 @@ import { spawn } from "child_process";
 import { appendFileSync, readFileSync, writeFileSync, existsSync } from "fs";
 import { dirname, join } from "path";
 import { fileURLToPath } from "url";
+import { homedir } from "os";
 var LOG_FILE = "/tmp/slack-agent-plugin.log";
 var SLACK_MSG_LIMIT = 3900;
 var RESPONSE_STALL_MS = 45e3;
@@ -12441,6 +12442,7 @@ var BG_OUTPUT_CANDIDATE_PATHS = ["/background/output", "/background_output", "/a
 var BG_TASK_ID_PATTERN = /^bg_[A-Za-z0-9][A-Za-z0-9_-]*$/;
 var AUTO_ATTACH_MAX_TASKS_PER_MESSAGE = 3;
 var AUTO_ATTACH_MAX_MONITOR_BUDGET_MS = 12e4;
+var INBOUND_EVENT_DEDUPE_TTL_MS = 10 * 60 * 1e3;
 var initialized = false;
 var worker = null;
 var pluginClient = null;
@@ -12453,8 +12455,14 @@ var allowedUsers = null;
 var allowlistReady = true;
 var attachBgTimeoutMs = DEFAULT_ATTACH_TIMEOUT_SEC * 1e3;
 var cachedBackgroundOutputPath = null;
+var seenInboundEventKeys = /* @__PURE__ */ new Map();
 var pendingPermissions = /* @__PURE__ */ new Map();
 var pendingQuestions = /* @__PURE__ */ new Map();
+function expandTilde(p) {
+  if (p === "~") return homedir();
+  if (p.startsWith("~/")) return join(homedir(), p.slice(2));
+  return p;
+}
 function log(m) {
   try {
     appendFileSync(LOG_FILE, `[${(/* @__PURE__ */ new Date()).toISOString()}] plugin: ${m}
@@ -12519,6 +12527,42 @@ function markdownToSlackMrkdwn(text) {
 function sendIPC(msg) {
   if (worker && worker.connected) worker.send(msg);
 }
+function buildInboundEventKey(msg) {
+  const eventId = typeof msg?.eventId === "string" ? msg.eventId : "";
+  if (eventId) return `event:${eventId}`;
+  const channel = typeof msg?.channel === "string" ? msg.channel : "";
+  const messageTs = typeof msg?.messageTs === "string" ? msg.messageTs : "";
+  if (channel && messageTs) return `msg:${channel}:${messageTs}`;
+  return "";
+}
+function inboundMeta(msg) {
+  const eventId = typeof msg?.eventId === "string" ? msg.eventId : "-";
+  const channel = typeof msg?.channel === "string" ? msg.channel : "-";
+  const threadTs = typeof msg?.threadTs === "string" ? msg.threadTs : "-";
+  const messageTs = typeof msg?.messageTs === "string" ? msg.messageTs : "-";
+  const user = typeof msg?.user === "string" ? msg.user : "-";
+  return `event_id=${eventId} channel=${channel} thread_ts=${threadTs} message_ts=${messageTs} user=${user}`;
+}
+function shouldProcessInboundEvent(msg) {
+  const key = buildInboundEventKey(msg);
+  if (!key) {
+    log(`inbound key_missing ${inboundMeta(msg)}`);
+    return true;
+  }
+  if (seenInboundEventKeys.has(key)) {
+    log(`inbound duplicate skipped key=${key} ${inboundMeta(msg)}`);
+    return false;
+  }
+  seenInboundEventKeys.set(key, Date.now() + INBOUND_EVENT_DEDUPE_TTL_MS);
+  log(`inbound accepted key=${key} ${inboundMeta(msg)}`);
+  return true;
+}
+setInterval(() => {
+  const now = Date.now();
+  for (const [key, expiresAt] of seenInboundEventKeys) {
+    if (expiresAt <= now) seenInboundEventKeys.delete(key);
+  }
+}, 6e4);
 function slackSend(channel, text, threadTs) {
   sendIPC({ type: "slack_send", channel, text: markdownToSlackMrkdwn(text), threadTs });
 }
@@ -13287,13 +13331,14 @@ async function handleCommand(channel, text, ts) {
       slackSend(channel, `*\uD604\uC7AC \uC6CC\uD06C\uC2A4\uD398\uC774\uC2A4:* \`${currentDir}\``, ts);
       return true;
     }
+    const resolvedDir = expandTilde(arg);
     if (!sessions[ts]) {
-      sessions[ts] = { sessionId: "", channel, lastUsed: Date.now(), directory: arg };
+      sessions[ts] = { sessionId: "", channel, lastUsed: Date.now(), directory: resolvedDir };
     } else {
-      sessions[ts].directory = arg;
+      sessions[ts].directory = resolvedDir;
     }
     saveSessions();
-    slackSend(channel, `\u2705 \uC6CC\uD06C\uC2A4\uD398\uC774\uC2A4 \uBCC0\uACBD: \`${arg}\`
+    slackSend(channel, `\u2705 \uC6CC\uD06C\uC2A4\uD398\uC774\uC2A4 \uBCC0\uACBD: \`${resolvedDir}\`
 \uB2E4\uC74C \uBA54\uC2DC\uC9C0\uBD80\uD130 \uC774 \uB514\uB809\uD1A0\uB9AC\uC5D0\uC11C \uC138\uC158 \uC0DD\uC131.`, ts);
     return true;
   }
@@ -13404,22 +13449,20 @@ ${textParts}`, ts);
   }
   return false;
 }
-function startWorker(env) {
-  const workerPath = join(dirname(fileURLToPath(import.meta.url)), "socket-worker.js");
-  log(`starting worker: ${workerPath}`);
-  worker = spawn("node", [workerPath], {
-    env: { ...process.env, ...env },
-    stdio: ["ignore", "ignore", "ignore", "ipc"],
-    detached: false
-  });
-  worker.on("message", (msg) => {
+function attachWorkerHandlers(w) {
+  w.on("message", (msg) => {
     if (msg?.type === "slack_event") {
+      log(`inbound received ${inboundMeta(msg)}`);
+      if (!shouldProcessInboundEvent(msg)) {
+        return;
+      }
       const isAllowed = !allowedUsers || !allowlistReady || allowedUsers.has(msg.user);
       const isThreadReply = msg.threadTs !== msg.messageTs;
       if (!isAllowed && !isThreadReply) {
-        log(`blocked user: ${msg.user} (new thread)`);
+        log(`inbound blocked_user ${inboundMeta(msg)} reason=new_thread_not_allowlisted`);
         return;
       }
+      log(`inbound dispatch_handleMessage ${inboundMeta(msg)}`);
       handleMessage(msg.channel, msg.text, msg.threadTs, msg.messageTs, isAllowed);
     } else if (msg?.type === "resolved_emails") {
       if (allowedUsers && msg.users) {
@@ -13429,18 +13472,61 @@ function startWorker(env) {
         log(`resolved email users: ${msg.users.join(", ")}`);
       }
       allowlistReady = true;
+    } else if (msg?.type === "worker_connected") {
     }
   });
-  worker.on("exit", (code) => {
+}
+var GRACEFUL_RESTART_TIMEOUT_MS = 3e4;
+var workerEnvCache = null;
+function startWorker(env, dyingWorker) {
+  workerEnvCache = env;
+  const workerPath = join(dirname(fileURLToPath(import.meta.url)), "socket-worker.js");
+  log(`starting worker: ${workerPath}`);
+  const newWorker = spawn("node", [workerPath], {
+    env: { ...process.env, ...env },
+    stdio: ["ignore", "ignore", "ignore", "ipc"],
+    detached: false
+  });
+  attachWorkerHandlers(newWorker);
+  const oldWorker = dyingWorker ?? null;
+  let swapped = false;
+  const doSwap = (reason) => {
+    if (swapped) return;
+    swapped = true;
+    worker = newWorker;
+    if (oldWorker && !oldWorker.killed) {
+      log(`graceful swap: ${reason} \u2014 terminating old worker`);
+      oldWorker.removeAllListeners();
+      oldWorker.kill("SIGTERM");
+    } else {
+      log(`worker ready: ${reason} (no old worker to terminate)`);
+    }
+  };
+  const onNewWorkerMessage = (msg) => {
+    if (msg?.type === "worker_connected") {
+      doSwap("new worker connected");
+    }
+  };
+  newWorker.on("message", onNewWorkerMessage);
+  const swapTimer = setTimeout(() => {
+    doSwap("timeout \u2014 forcing swap");
+  }, GRACEFUL_RESTART_TIMEOUT_MS);
+  newWorker.on("exit", (code) => {
+    clearTimeout(swapTimer);
     log(`worker exited: code=${code}`);
-    worker = null;
-    if (initialized) {
+    if (worker === newWorker) {
+      worker = null;
+    }
+    if (initialized && workerEnvCache) {
       setTimeout(() => {
         log("restarting worker...");
-        startWorker(env);
+        startWorker(workerEnvCache, newWorker);
       }, 5e3);
     }
   });
+  if (!oldWorker) {
+    doSwap("initial start");
+  }
   log("worker spawned");
 }
 function stopWorker() {
@@ -13479,7 +13565,7 @@ var pluginModule = {
       return { tool: { slack_status: slackStatusTool } };
     }
     pluginClient = input.client;
-    defaultDirectory = options?.DEFAULT_DIRECTORY || process.env.SLACK_DEFAULT_DIRECTORY || input.directory;
+    defaultDirectory = expandTilde(options?.DEFAULT_DIRECTORY || process.env.SLACK_DEFAULT_DIRECTORY || input.directory);
     attachBgTimeoutMs = resolveAttachTimeoutMs(options?.ATTACH_TIMEOUT_SEC);
     log(`attach timeout set to ${attachBgTimeoutMs}ms`);
     sessionsPath = join(input.directory, "slack-sessions.json");

dist/socket-worker.js

@@ -28,6 +28,38 @@ const slack = new WebClient(botToken, agent ? { agent, tls: { ca: agent.options.
 const socketClient = new SocketModeClient({ appToken });
 
 let botUserId;
+const EVENT_DEDUPE_TTL_MS = 10 * 60 * 1000;
+const seenEventKeys = new Map();
+
+function formatInboundMeta(body, ev) {
+  const eventId = typeof body?.event_id === "string" ? body.event_id : "-";
+  const eventType = typeof ev?.type === "string" ? ev.type : "-";
+  const channel = typeof ev?.channel === "string" ? ev.channel : "-";
+  const ts = typeof ev?.ts === "string" ? ev.ts : "-";
+  const threadTs = typeof ev?.thread_ts === "string" ? ev.thread_ts : "-";
+  const user = typeof ev?.user === "string" ? ev.user : "-";
+  return `event_id=${eventId} type=${eventType} channel=${channel} ts=${ts} thread_ts=${threadTs} user=${user}`;
+}
+
+function buildEventKey(body, ev) {
+  const channel = typeof ev?.channel === "string" ? ev.channel : "";
+  const ts = typeof ev?.ts === "string" ? ev.ts : "";
+  if (channel && ts) return `msg:${channel}:${ts}`;
+  const eventId = typeof body?.event_id === "string" ? body.event_id : "";
+  if (eventId) return `event:${eventId}`;
+  return "";
+}
+
+function shouldProcessEvent(eventKey) {
+  if (!eventKey) return true;
+  const now = Date.now();
+  for (const [key, expiresAt] of seenEventKeys) {
+    if (expiresAt <= now) seenEventKeys.delete(key);
+  }
+  if (seenEventKeys.has(eventKey)) return false;
+  seenEventKeys.set(eventKey, now + EVENT_DEDUPE_TTL_MS);
+  return true;
+}
 
 async function init() {
   const auth = await slack.auth.test();
@@ -106,42 +138,92 @@ process.on("message", async (msg) => {
 });
 
 socketClient.on("slack_event", async ({ body, ack }) => {
-  if (ack) await ack();
   const ev = body?.event;
+  log(`ingress received ${formatInboundMeta(body, ev)}`);
+
+  if (ack) {
+    try {
+      await ack();
+      log(`ingress ack sent ${formatInboundMeta(body, ev)}`);
+    } catch (e) {
+      log(`ingress ack failed ${formatInboundMeta(body, ev)} err=${e?.message || String(e)}`);
+    }
+  }
+
   if (!ev) return;
 
+  const eventKey = buildEventKey(body, ev);
+  if (!shouldProcessEvent(eventKey)) {
+    log(`ingress duplicate skipped key=${eventKey} ${formatInboundMeta(body, ev)}`);
+    return;
+  }
+
   if (ev.type === "message" || ev.type === "app_mention") {
-    if (ev.bot_id || ev.user === botUserId) return;
-    if (ev.subtype && ev.subtype !== "file_share") return;
-    if (!ev.channel || !ev.text || !ev.ts) return;
+    if (ev.bot_id || ev.user === botUserId) {
+      log(`ingress dropped reason=self_or_bot ${formatInboundMeta(body, ev)}`);
+      return;
+    }
+    if (ev.subtype && ev.subtype !== "file_share") {
+      log(`ingress dropped reason=subtype_${ev.subtype} ${formatInboundMeta(body, ev)}`);
+      return;
+    }
+    if (!ev.channel || !ev.text || !ev.ts) {
+      log(`ingress dropped reason=missing_required_fields ${formatInboundMeta(body, ev)}`);
+      return;
+    }
 
     const text = ev.type === "app_mention"
       ? ev.text.replace(/<@[A-Z0-9]+>/g, "").trim()
       : ev.text;
 
     if (text && process.send) {
-      process.send({ type: "slack_event", channel: ev.channel, text, ts: ev.ts, threadTs: ev.thread_ts || ev.ts, messageTs: ev.ts, user: ev.user });
-      log(`event sent via IPC: ${text.slice(0, 50)}`);
+      process.send({
+        type: "slack_event",
+        channel: ev.channel,
+        text,
+        ts: ev.ts,
+        threadTs: ev.thread_ts || ev.ts,
+        messageTs: ev.ts,
+        eventId: typeof body?.event_id === "string" ? body.event_id : undefined,
+        user: ev.user,
+      });
+      log(`ingress forwarded_to_plugin key=${eventKey || "-"} ${formatInboundMeta(body, ev)} text_preview=${text.slice(0, 50)}`);
     }
   }
 });
 
-socketClient.on("connected", () => log("Socket Mode connected"));
-socketClient.on("disconnected", () => log("Socket Mode disconnected"));
+// --- Connection health monitoring ---
+// SDK handles ping/pong internally (serverPingTimeout=30s, clientPingTimeout=5s).
+// We only exit if SDK fires 'disconnected' AND fails to reconnect within RECONNECT_GRACE_MS.
+const RECONNECT_GRACE_MS = 60000;
+let outageStartedAt = null;
 
-await init();
-await socketClient.start();
-log("worker running");
+socketClient.on("connected", () => {
+  outageStartedAt = null;
+  log("Socket Mode connected");
+  if (process.send) process.send({ type: "worker_connected" });
+});
 
-const STALE_TIMEOUT = 120000;
-let lastEventTime = Date.now();
+socketClient.on("disconnected", () => {
+  if (!outageStartedAt) outageStartedAt = Date.now();
+  log("Socket Mode disconnected — waiting for SDK auto-reconnect");
+});
 
-socketClient.on("slack_event", () => { lastEventTime = Date.now(); });
+socketClient.on("reconnecting", () => {
+  if (!outageStartedAt) outageStartedAt = Date.now();
+  log("Socket Mode reconnecting...");
+});
+
+await init();
+await socketClient.start();
+log("worker running (idle-exit disabled, relying on SDK ping/pong health check)");
 
 setInterval(() => {
-  const elapsed = Date.now() - lastEventTime;
-  if (elapsed > STALE_TIMEOUT) {
-    log(`no events for ${Math.round(elapsed / 1000)}s, exiting for restart`);
-    process.exit(1);
+  if (outageStartedAt) {
+    const elapsed = Date.now() - outageStartedAt;
+    if (elapsed > RECONNECT_GRACE_MS) {
+      log(`connection unhealthy for ${Math.round(elapsed / 1000)}s with no reconnect — exiting for restart`);
+      process.exit(1);
+    }
   }
-}, 30000);
+}, 15000);