fix claude turn teardown and start CI watchers immediately

runners/agent-runner/src/index.ts

@@ -394,7 +394,11 @@ async function runQuery(
   mcpServerPath: string,
   containerInput: ContainerInput,
   sdkEnv: Record<string, string | undefined>,
-): Promise<{ newSessionId?: string; closedDuringQuery: boolean }> {
+): Promise<{
+  newSessionId?: string;
+  closedDuringQuery: boolean;
+  terminalResultObserved: boolean;
+}> {
   const stream = new MessageStream();
   stream.push(prompt);
 
@@ -422,6 +426,7 @@ async function runQuery(
   let newSessionId: string | undefined;
   let messageCount = 0;
   let resultCount = 0;
+  let terminalResultObserved = false;
 
   // Discover additional directories
   const extraDirs: string[] = [];
@@ -554,6 +559,15 @@ async function runQuery(
           newSessionId
         });
       }
+
+      // Single-turn runtimes must terminate the query after the first
+      // terminal result. Leaving the message stream open can keep the SDK
+      // query alive indefinitely, which pins the host queue after a reply.
+      terminalResultObserved = true;
+      ipcPolling = false;
+      stream.end();
+      log('Terminal result observed, ending query stream');
+      break;
     }
   }
 
@@ -561,7 +575,7 @@ async function runQuery(
   log(
     `Query done. Messages: ${messageCount}, results: ${resultCount}, closedDuringQuery: ${closedDuringQuery}`,
   );
-  return { newSessionId, closedDuringQuery };
+  return { newSessionId, closedDuringQuery, terminalResultObserved };
 }
 
 async function main(): Promise<void> {
@@ -726,8 +740,10 @@ async function main(): Promise<void> {
       sessionId = queryResult.newSessionId;
     }
 
-    if (!queryResult.closedDuringQuery) {
+    if (!queryResult.closedDuringQuery && !queryResult.terminalResultObserved) {
       writeOutput({ status: 'success', result: null, newSessionId: sessionId });
+    } else if (queryResult.terminalResultObserved) {
+      log('Terminal result already emitted, exiting single-turn runtime');
     } else {
       log('Close sentinel consumed during query, exiting');
     }

runners/agent-runner/src/ipc-mcp-stdio.ts

@@ -17,8 +17,9 @@ import {
 } from './watch-ci.js';
 
 const IPC_DIR = process.env.EJCLAW_IPC_DIR || '/workspace/ipc';
-const MESSAGES_DIR = path.join(IPC_DIR, 'messages');
+const HOST_IPC_DIR = process.env.EJCLAW_HOST_IPC_DIR || IPC_DIR;
-const TASKS_DIR = path.join(IPC_DIR, 'tasks');
+const MESSAGES_DIR = path.join(HOST_IPC_DIR, 'messages');
+const TASKS_DIR = path.join(HOST_IPC_DIR, 'tasks');
 
 // Context from environment variables (set by the agent runner)
 const chatJid = process.env.EJCLAW_CHAT_JID!;

src/agent-runner-environment.ts

@@ -60,6 +60,7 @@ function buildBaseRunnerEnv(args: {
   isMain: boolean;
   groupDir: string;
   groupIpcDir: string;
+  hostIpcDir: string;
   globalDir: string;
   groupSessionsDir: string;
   agentType: AgentType;
@@ -89,6 +90,7 @@ function buildBaseRunnerEnv(args: {
     HOME: os.homedir(),
     EJCLAW_GROUP_DIR: args.groupDir,
     EJCLAW_IPC_DIR: args.groupIpcDir,
+    EJCLAW_HOST_IPC_DIR: args.hostIpcDir,
     EJCLAW_GLOBAL_DIR: args.globalDir,
     ...(args.group.workDir ? { EJCLAW_WORK_DIR: args.group.workDir } : {}),
     EJCLAW_CHAT_JID: args.chatJid,
@@ -329,6 +331,7 @@ export function prepareGroupEnvironment(
   const groupIpcDir = runtimeTaskId
     ? resolveTaskRuntimeIpcPath(group.folder, runtimeTaskId)
     : resolveGroupIpcPath(group.folder);
+  const hostIpcDir = resolveGroupIpcPath(group.folder);
   fs.mkdirSync(path.join(groupIpcDir, 'messages'), { recursive: true });
   fs.mkdirSync(path.join(groupIpcDir, 'tasks'), { recursive: true });
   fs.mkdirSync(path.join(groupIpcDir, 'input'), { recursive: true });
@@ -388,6 +391,7 @@ export function prepareGroupEnvironment(
     isMain,
     groupDir,
     groupIpcDir,
+    hostIpcDir,
     globalDir,
     groupSessionsDir,
     agentType,

src/agent-runner.test.ts

@@ -325,6 +325,9 @@ describe('agent-runner timeout behavior', () => {
     expect(spawnEnv?.EJCLAW_IPC_DIR).toBe(
       '/tmp/ejclaw-test-data/ipc/test-group/tasks/task-123',
     );
+    expect(spawnEnv?.EJCLAW_HOST_IPC_DIR).toBe(
+      '/tmp/ejclaw-test-data/ipc/test-group',
+    );
     expect(spawnEnv?.CLAUDE_CONFIG_DIR).toBe(
       '/tmp/ejclaw-test-data/sessions/test-group/tasks/task-123/.claude',
     );
@@ -356,6 +359,9 @@ describe('agent-runner timeout behavior', () => {
     expect(spawnEnv?.EJCLAW_IPC_DIR).toBe(
       '/tmp/ejclaw-test-data/ipc/test-group/tasks/task-watch-group',
     );
+    expect(spawnEnv?.EJCLAW_HOST_IPC_DIR).toBe(
+      '/tmp/ejclaw-test-data/ipc/test-group',
+    );
     expect(spawnEnv?.CLAUDE_CONFIG_DIR).toBe(
       '/tmp/ejclaw-test-data/sessions/test-group/.claude',
     );

src/group-queue.test.ts

@@ -28,6 +28,7 @@ describe('GroupQueue', () => {
 
   beforeEach(() => {
     vi.useFakeTimers();
+    vi.clearAllMocks();
     queue = new GroupQueue();
   });
 
@@ -87,6 +88,38 @@ describe('GroupQueue', () => {
     await vi.advanceTimersByTimeAsync(10);
   });
 
+  it('does not pipe follow-up messages after stdin close was requested', async () => {
+    const ipcDir = '/tmp/ejclaw-test-data/ipc/group-folder';
+    let releaseRun!: (value: boolean) => void;
+    const blocker = new Promise<boolean>((resolve) => {
+      releaseRun = resolve;
+    });
+
+    const processMessages = vi.fn(async () => await blocker);
+
+    queue.setProcessMessagesFn(processMessages);
+    queue.enqueueMessageCheck('group1@g.us', ipcDir);
+    await vi.advanceTimersByTimeAsync(10);
+
+    vi.mocked(fs.mkdirSync).mockClear();
+    vi.mocked(fs.writeFileSync).mockClear();
+    vi.mocked(fs.renameSync).mockClear();
+
+    queue.closeStdin('group1@g.us');
+
+    vi.mocked(fs.mkdirSync).mockClear();
+    vi.mocked(fs.writeFileSync).mockClear();
+    vi.mocked(fs.renameSync).mockClear();
+
+    expect(queue.sendMessage('group1@g.us', '후속 메시지')).toBe(false);
+    expect(fs.mkdirSync).not.toHaveBeenCalled();
+    expect(fs.writeFileSync).not.toHaveBeenCalled();
+    expect(fs.renameSync).not.toHaveBeenCalled();
+
+    releaseRun(true);
+    await vi.advanceTimersByTimeAsync(10);
+  });
+
   // --- Global concurrency limit ---
 
   it('respects global concurrency limit', async () => {

src/group-queue.ts

@@ -209,7 +209,12 @@ export class GroupQueue {
 
   sendMessage(groupJid: string, text: string): boolean {
     const state = this.getGroup(groupJid);
-    if (!state.active || !state.ipcDir || state.isTaskProcess) {
+    if (
+      !state.active ||
+      !state.ipcDir ||
+      state.isTaskProcess ||
+      state.closingStdin
+    ) {
       logger.debug(
         {
           groupJid,
@@ -217,6 +222,7 @@ export class GroupQueue {
           active: state.active,
           ipcDir: state.ipcDir,
           isTaskProcess: state.isTaskProcess,
+          closingStdin: state.closingStdin,
         },
         'Cannot pipe follow-up message to active agent',
       );

src/index.ts

@@ -58,7 +58,7 @@ import {
   shouldDropMessage,
 } from './sender-allowlist.js';
 import { createMessageRuntime } from './message-runtime.js';
-import { startSchedulerLoop } from './task-scheduler.js';
+import { nudgeSchedulerLoop, startSchedulerLoop } from './task-scheduler.js';
 import { startUnifiedDashboard } from './unified-dashboard.js';
 import { Channel, NewMessage, RegisteredGroup } from './types.js';
 import { logger } from './logger.js';
@@ -409,6 +409,7 @@ async function main(): Promise<void> {
       if (!channel) throw new Error(`No channel for JID: ${jid}`);
       return channel.sendMessage(jid, text);
     },
+    nudgeScheduler: nudgeSchedulerLoop,
     registeredGroups: () => registeredGroups,
     registerGroup,
     syncGroups: async (force: boolean) => {

src/ipc-auth.test.ts

@@ -1,4 +1,4 @@
-import { describe, it, expect, beforeEach } from 'vitest';
+import { describe, it, expect, beforeEach, vi } from 'vitest';
 
 import {
   _initTestDatabase,
@@ -53,6 +53,7 @@ beforeEach(() => {
 
   deps = {
     sendMessage: async () => {},
+    nudgeScheduler: vi.fn(),
     registeredGroups: () => groups,
     registerGroup: (jid, group) => {
       groups[jid] = group;
@@ -549,6 +550,42 @@ describe('schedule_task schedule types', () => {
     expect(nextRun).toBeLessThanOrEqual(Date.now() + 3600000 + 1000);
   });
 
+  it('starts watch_ci interval tasks immediately and nudges the scheduler', async () => {
+    const before = Date.now();
+
+    await processTaskIpc(
+      {
+        type: 'schedule_task',
+        prompt: `
+[BACKGROUND CI WATCH]
+
+Watch target:
+GitHub Actions run 123456
+
+Task ID:
+task-watch-ci
+
+Check instructions:
+Check the run.
+        `.trim(),
+        schedule_type: 'interval',
+        schedule_value: '60000',
+        targetJid: 'other@g.us',
+      },
+      'whatsapp_main',
+      true,
+      deps,
+    );
+
+    const tasks = getAllTasks();
+    expect(tasks).toHaveLength(1);
+    expect(tasks[0].schedule_type).toBe('interval');
+    const nextRun = new Date(tasks[0].next_run!).getTime();
+    expect(nextRun).toBeGreaterThanOrEqual(before - 1000);
+    expect(nextRun).toBeLessThanOrEqual(Date.now() + 1000);
+    expect(deps.nudgeScheduler).toHaveBeenCalledTimes(1);
+  });
+
   it('rejects invalid interval (non-numeric)', async () => {
     await processTaskIpc(
       {

src/ipc.ts

@@ -13,10 +13,12 @@ import { AvailableGroup } from './agent-runner.js';
 import { createTask, deleteTask, getTaskById, updateTask } from './db.js';
 import { isValidGroupFolder } from './group-folder.js';
 import { logger } from './logger.js';
+import { isWatchCiTask } from './task-watch-status.js';
 import { RegisteredGroup } from './types.js';
 
 export interface IpcDeps {
   sendMessage: (jid: string, text: string) => Promise<void>;
+  nudgeScheduler?: () => void;
   registeredGroups: () => Record<string, RegisteredGroup>;
   registerGroup: (jid: string, group: RegisteredGroup) => void;
   syncGroups: (force: boolean) => Promise<void>;
@@ -256,7 +258,9 @@ export async function processTaskIpc(
             );
             break;
           }
-          nextRun = new Date(Date.now() + ms).toISOString();
+          nextRun = isWatchCiTask({ prompt: data.prompt })
+            ? new Date().toISOString()
+            : new Date(Date.now() + ms).toISOString();
         } else if (scheduleType === 'once') {
           const date = new Date(data.schedule_value);
           if (isNaN(date.getTime())) {
@@ -299,6 +303,9 @@ export async function processTaskIpc(
           },
           'Task created via IPC',
         );
+        if (nextRun && new Date(nextRun).getTime() <= Date.now()) {
+          deps.nudgeScheduler?.();
+        }
       }
       break;
 

src/task-scheduler.test.ts

@@ -28,6 +28,7 @@ import {
   computeNextRun,
   extractWatchCiTarget,
   isWatchCiTask,
+  nudgeSchedulerLoop,
   renderWatchCiStatusMessage,
   startSchedulerLoop,
 } from './task-scheduler.js';
@@ -231,6 +232,65 @@ Check the run.
     expect(enqueueTask.mock.calls[0][1]).toBe('task-watch-group');
   });
 
+  it('picks up newly due tasks immediately when nudged', async () => {
+    const enqueueTask = vi.fn();
+
+    startSchedulerLoop({
+      serviceAgentType: 'codex',
+      registeredGroups: () => ({
+        'shared@g.us': {
+          name: 'Shared',
+          folder: 'shared-group',
+          trigger: '@Codex',
+          added_at: '2026-02-22T00:00:00.000Z',
+          agentType: 'codex',
+        },
+      }),
+      getSessions: () => ({}),
+      queue: { enqueueTask } as any,
+      onProcess: () => {},
+      sendMessage: async () => {},
+    });
+
+    await vi.advanceTimersByTimeAsync(10);
+    expect(enqueueTask).not.toHaveBeenCalled();
+
+    createTask({
+      id: 'task-watch-immediate',
+      group_folder: 'shared-group',
+      chat_jid: 'shared@g.us',
+      agent_type: 'codex',
+      prompt: `
+[BACKGROUND CI WATCH]
+
+Watch target:
+GitHub Actions run 654321
+
+Task ID:
+task-watch-immediate
+
+Check instructions:
+Check the run.
+      `.trim(),
+      schedule_type: 'interval',
+      schedule_value: '60000',
+      context_mode: 'isolated',
+      next_run: new Date(Date.now() - 1000).toISOString(),
+      status: 'active',
+      created_at: '2026-02-22T00:00:00.000Z',
+    });
+
+    nudgeSchedulerLoop();
+    await vi.advanceTimersByTimeAsync(10);
+
+    expect(enqueueTask).toHaveBeenCalledTimes(1);
+    expect(enqueueTask).toHaveBeenCalledWith(
+      'shared@g.us::task:task-watch-immediate',
+      'task-watch-immediate',
+      expect.any(Function),
+    );
+  });
+
   it('uses dedicated IPC but shared session state for group-context CI watchers', async () => {
     const dueAt = new Date(Date.now() - 60_000).toISOString();
     createTask({

src/task-scheduler.ts

@@ -331,6 +331,30 @@ async function runTask(
 }
 
 let schedulerRunning = false;
+let schedulerTimer: ReturnType<typeof setTimeout> | null = null;
+let schedulerLoopFn: (() => Promise<void>) | null = null;
+let schedulerTickInFlight = false;
+let schedulerTickPending = false;
+
+function scheduleSchedulerTick(delayMs: number): void {
+  if (!schedulerRunning || !schedulerLoopFn) return;
+  if (schedulerTimer) {
+    clearTimeout(schedulerTimer);
+  }
+  schedulerTimer = setTimeout(() => {
+    schedulerTimer = null;
+    void schedulerLoopFn?.();
+  }, delayMs);
+}
+
+export function nudgeSchedulerLoop(): void {
+  if (!schedulerRunning || !schedulerLoopFn) return;
+  if (schedulerTickInFlight) {
+    schedulerTickPending = true;
+    return;
+  }
+  scheduleSchedulerTick(0);
+}
 
 export function startSchedulerLoop(deps: SchedulerDependencies): void {
   if (schedulerRunning) {
@@ -341,6 +365,12 @@ export function startSchedulerLoop(deps: SchedulerDependencies): void {
   logger.info('Scheduler loop started');
 
   const loop = async () => {
+    if (schedulerTickInFlight) {
+      schedulerTickPending = true;
+      return;
+    }
+    schedulerTickInFlight = true;
+
     try {
       const dueTasks = getDueTasks(deps.serviceAgentType || SERVICE_AGENT_TYPE);
       if (dueTasks.length > 0) {
@@ -362,15 +392,35 @@ export function startSchedulerLoop(deps: SchedulerDependencies): void {
       }
     } catch (err) {
       logger.error({ err }, 'Error in scheduler loop');
+    } finally {
+      schedulerTickInFlight = false;
     }
 
-    setTimeout(loop, SCHEDULER_POLL_INTERVAL);
+    if (!schedulerRunning) {
+      return;
+    }
+
+    if (schedulerTickPending) {
+      schedulerTickPending = false;
+      scheduleSchedulerTick(0);
+      return;
+    }
+
+    scheduleSchedulerTick(SCHEDULER_POLL_INTERVAL);
   };
 
-  loop();
+  schedulerLoopFn = loop;
+  void loop();
 }
 
 /** @internal - for tests only. */
 export function _resetSchedulerLoopForTests(): void {
   schedulerRunning = false;
+  schedulerLoopFn = null;
+  schedulerTickInFlight = false;
+  schedulerTickPending = false;
+  if (schedulerTimer) {
+    clearTimeout(schedulerTimer);
+    schedulerTimer = null;
+  }
 }