diff --git a/runners/agent-runner/src/index.ts b/runners/agent-runner/src/index.ts index 149977b..69702e2 100644 --- a/runners/agent-runner/src/index.ts +++ b/runners/agent-runner/src/index.ts @@ -394,7 +394,11 @@ async function runQuery( mcpServerPath: string, containerInput: ContainerInput, sdkEnv: Record, -): 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 { @@ -726,8 +740,10 @@ async function main(): Promise { 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'); } diff --git a/runners/agent-runner/src/ipc-mcp-stdio.ts b/runners/agent-runner/src/ipc-mcp-stdio.ts index 2c6322e..e7b581e 100644 --- a/runners/agent-runner/src/ipc-mcp-stdio.ts +++ b/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 TASKS_DIR = path.join(IPC_DIR, 'tasks'); +const HOST_IPC_DIR = process.env.EJCLAW_HOST_IPC_DIR || IPC_DIR; +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!; diff --git a/src/agent-runner-environment.ts b/src/agent-runner-environment.ts index 34be1e7..1f3f204 100644 --- a/src/agent-runner-environment.ts +++ b/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, diff --git a/src/agent-runner.test.ts b/src/agent-runner.test.ts index 18204dc..60238c2 100644 --- a/src/agent-runner.test.ts +++ b/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', ); diff --git a/src/group-queue.test.ts b/src/group-queue.test.ts index 0e90223..07e1674 100644 --- a/src/group-queue.test.ts +++ b/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((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 () => { diff --git a/src/group-queue.ts b/src/group-queue.ts index 18994ef..fcd2d28 100644 --- a/src/group-queue.ts +++ b/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', ); diff --git a/src/index.ts b/src/index.ts index 863443d..8a33e6e 100644 --- a/src/index.ts +++ b/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 { if (!channel) throw new Error(`No channel for JID: ${jid}`); return channel.sendMessage(jid, text); }, + nudgeScheduler: nudgeSchedulerLoop, registeredGroups: () => registeredGroups, registerGroup, syncGroups: async (force: boolean) => { diff --git a/src/ipc-auth.test.ts b/src/ipc-auth.test.ts index 231bdb2..059c2fc 100644 --- a/src/ipc-auth.test.ts +++ b/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( { diff --git a/src/ipc.ts b/src/ipc.ts index 4b4d3f7..c17f3c1 100644 --- a/src/ipc.ts +++ b/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; + nudgeScheduler?: () => void; registeredGroups: () => Record; registerGroup: (jid: string, group: RegisteredGroup) => void; syncGroups: (force: boolean) => Promise; @@ -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; diff --git a/src/task-scheduler.test.ts b/src/task-scheduler.test.ts index 19993a3..97fa40d 100644 --- a/src/task-scheduler.test.ts +++ b/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({ diff --git a/src/task-scheduler.ts b/src/task-scheduler.ts index ff18053..1167e1b 100644 --- a/src/task-scheduler.ts +++ b/src/task-scheduler.ts @@ -331,6 +331,30 @@ async function runTask( } let schedulerRunning = false; +let schedulerTimer: ReturnType | null = null; +let schedulerLoopFn: (() => Promise) | 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; + } }