feat: block follow-up IPC after closeStdin with closingStdin flag

Prevent piping new messages to an agent that is shutting down.
Reset flag on new process start and cleanup.

src/group-queue.test.ts

@@ -413,6 +413,34 @@ describe('GroupQueue', () => {
     await vi.advanceTimersByTimeAsync(10);
   });
 
+  it('does not pipe follow-up messages to an agent after closeStdin', async () => {
+    let resolveProcess: () => void;
+
+    const processMessages = vi.fn(async () => {
+      await new Promise<void>((resolve) => {
+        resolveProcess = resolve;
+      });
+      return true;
+    });
+
+    queue.setProcessMessagesFn(processMessages);
+    queue.enqueueMessageCheck('group1@g.us', 'test-group');
+    await vi.advanceTimersByTimeAsync(10);
+
+    queue.registerProcess('group1@g.us', {} as any, 'agent-1', 'test-group');
+    queue.notifyIdle('group1@g.us');
+    queue.closeStdin('group1@g.us');
+
+    expect(queue.sendMessage('group1@g.us', 'hello after clear')).toBe(false);
+
+    queue.enqueueMessageCheck('group1@g.us', 'test-group');
+
+    resolveProcess!();
+    await vi.advanceTimersByTimeAsync(10);
+
+    expect(processMessages).toHaveBeenCalledTimes(2);
+  });
+
   it('preempts when idle arrives with pending tasks', async () => {
     const fs = await import('fs');
     let resolveProcess: () => void;

src/group-queue.ts

@@ -22,6 +22,7 @@ const BASE_RETRY_MS = 5000;
 interface GroupState {
   active: boolean;
   idleWaiting: boolean;
+  closingStdin: boolean;
   isTaskProcess: boolean;
   runningTaskId: string | null;
   currentRunId: string | null;
@@ -57,6 +58,7 @@ export class GroupQueue {
       state = {
         active: false,
         idleWaiting: false,
+        closingStdin: false,
         isTaskProcess: false,
         runningTaskId: null,
         currentRunId: null,
@@ -215,6 +217,7 @@ export class GroupQueue {
           groupJid,
           runId: state.currentRunId,
           active: state.active,
+          closingStdin: state.closingStdin,
           groupFolder: state.groupFolder,
           isTaskProcess: state.isTaskProcess,
         },
@@ -222,6 +225,13 @@ export class GroupQueue {
       );
       return false;
     }
+    if (state.closingStdin) {
+      logger.info(
+        { groupJid, runId: state.currentRunId, groupFolder: state.groupFolder },
+        'Skipping follow-up IPC because active agent is closing',
+      );
+      return false;
+    }
     state.idleWaiting = false; // Agent is about to receive work, no longer idle
 
     const inputDir = path.join(DATA_DIR, 'ipc', state.groupFolder, 'input');
@@ -266,6 +276,8 @@ export class GroupQueue {
   ): void {
     const state = this.getGroup(groupJid);
     if (!state.active || !state.groupFolder) return;
+    state.closingStdin = true;
+    state.idleWaiting = false;
 
     const inputDir = path.join(DATA_DIR, 'ipc', state.groupFolder, 'input');
     try {
@@ -302,6 +314,7 @@ export class GroupQueue {
     const runId = this.createRunId();
     state.active = true;
     state.idleWaiting = false;
+    state.closingStdin = false;
     state.isTaskProcess = false;
     state.currentRunId = runId;
     state.pendingMessages = false;
@@ -350,6 +363,8 @@ export class GroupQueue {
       );
       state.active = false;
       state.startedAt = null;
+      state.idleWaiting = false;
+      state.closingStdin = false;
       state.process = null;
       state.processName = null;
       state.groupFolder = null;
@@ -363,6 +378,7 @@ export class GroupQueue {
     const state = this.getGroup(groupJid);
     state.active = true;
     state.idleWaiting = false;
+    state.closingStdin = false;
     state.isTaskProcess = true;
     state.runningTaskId = task.id;
     state.startedAt = Date.now();
@@ -382,6 +398,8 @@ export class GroupQueue {
       state.isTaskProcess = false;
       state.runningTaskId = null;
       state.startedAt = null;
+      state.idleWaiting = false;
+      state.closingStdin = false;
       state.process = null;
       state.processName = null;
       state.groupFolder = null;