From 9def49f7fb5703fa26fdcb2bca79ee6f5b2e44eb Mon Sep 17 00:00:00 2001 From: Eyejoker Date: Mon, 23 Mar 2026 07:16:07 +0900 Subject: [PATCH] fix: isolate watcher task runtimes --- src/agent-runner.test.ts | 62 ++++++++++++++ src/agent-runner.ts | 62 +++++++++----- src/group-folder.test.ts | 19 +++++ src/group-folder.ts | 42 ++++++++++ src/group-queue-ipc.ts | 15 ++-- src/group-queue.test.ts | 5 +- src/group-queue.ts | 40 ++++----- src/index.ts | 4 +- src/message-agent-executor.ts | 4 +- src/message-runtime.test.ts | 6 +- src/message-runtime.ts | 7 +- src/task-scheduler.test.ts | 149 ++++++++++++++++++++++++++++++++++ src/task-scheduler.ts | 39 ++++++++- src/task-watch-status.ts | 14 ++++ 14 files changed, 407 insertions(+), 61 deletions(-) diff --git a/src/agent-runner.test.ts b/src/agent-runner.test.ts index 715f432..95e189e 100644 --- a/src/agent-runner.test.ts +++ b/src/agent-runner.test.ts @@ -299,6 +299,68 @@ describe('agent-runner timeout behavior', () => { ); }); + it('isolates IPC and session directories for isolated scheduled tasks', async () => { + vi.useRealTimers(); + fakeProc = createFakeProcess(); + + const resultPromise = runAgentProcess( + testGroup, + { + ...testInput, + isScheduledTask: true, + runtimeTaskId: 'task-123', + useTaskScopedSession: true, + }, + () => {}, + async () => {}, + ); + + fakeProc.emit('close', 0); + const result = await resultPromise; + expect(result.status).toBe('success'); + + const spawnEnv = vi.mocked(spawn).mock.calls.at(-1)?.[2]?.env as + | Record + | undefined; + expect(spawnEnv?.NANOCLAW_IPC_DIR).toBe( + '/tmp/nanoclaw-test-data/ipc/test-group/tasks/task-123', + ); + expect(spawnEnv?.CLAUDE_CONFIG_DIR).toBe( + '/tmp/nanoclaw-test-data/sessions/test-group/tasks/task-123/.claude', + ); + }); + + it('keeps shared session history for group-context task runtimes while isolating IPC', async () => { + vi.useRealTimers(); + fakeProc = createFakeProcess(); + + const resultPromise = runAgentProcess( + testGroup, + { + ...testInput, + isScheduledTask: true, + runtimeTaskId: 'task-watch-group', + useTaskScopedSession: false, + }, + () => {}, + async () => {}, + ); + + fakeProc.emit('close', 0); + const result = await resultPromise; + expect(result.status).toBe('success'); + + const spawnEnv = vi.mocked(spawn).mock.calls.at(-1)?.[2]?.env as + | Record + | undefined; + expect(spawnEnv?.NANOCLAW_IPC_DIR).toBe( + '/tmp/nanoclaw-test-data/ipc/test-group/tasks/task-watch-group', + ); + expect(spawnEnv?.CLAUDE_CONFIG_DIR).toBe( + '/tmp/nanoclaw-test-data/sessions/test-group/.claude', + ); + }); + it('merges a per-group codex config overlay before injecting managed MCP servers', async () => { vi.useRealTimers(); fakeProc = createFakeProcess(); diff --git a/src/agent-runner.ts b/src/agent-runner.ts index ce04cd1..27b4ffe 100644 --- a/src/agent-runner.ts +++ b/src/agent-runner.ts @@ -10,12 +10,17 @@ import path from 'path'; import { AGENT_MAX_OUTPUT_SIZE, AGENT_TIMEOUT, - DATA_DIR, GROUPS_DIR, IDLE_TIMEOUT, TIMEZONE, } from './config.js'; -import { resolveGroupFolderPath, resolveGroupIpcPath } from './group-folder.js'; +import { + resolveGroupFolderPath, + resolveGroupIpcPath, + resolveGroupSessionsPath, + resolveTaskRuntimeIpcPath, + resolveTaskSessionsPath, +} from './group-folder.js'; import { logger } from './logger.js'; import { readEnvFile } from './env.js'; import { isPairedRoomJid } from './db.js'; @@ -37,6 +42,8 @@ export interface AgentInput { runId?: string; isMain: boolean; isScheduledTask?: boolean; + runtimeTaskId?: string; + useTaskScopedSession?: boolean; assistantName?: string; agentType?: 'claude-code' | 'codex'; } @@ -182,6 +189,7 @@ function prepareCodexSessionEnvironment(args: { projectRoot: string; group: RegisteredGroup; groupDir: string; + sessionRootDir: string; chatJid: string; isMain: boolean; isPairedRoom: boolean; @@ -206,12 +214,7 @@ function prepareCodexSessionEnvironment(args: { if (codexEffort) args.env.CODEX_EFFORT = codexEffort; const hostCodexDir = path.join(os.homedir(), '.codex'); - const sessionCodexDir = path.join( - DATA_DIR, - 'sessions', - args.group.folder, - '.codex', - ); + const sessionCodexDir = path.join(args.sessionRootDir, '.codex'); fs.mkdirSync(sessionCodexDir, { recursive: true }); const authSrc = path.join(hostCodexDir, 'auth.json'); @@ -328,18 +331,25 @@ function prepareGroupEnvironment( group: RegisteredGroup, isMain: boolean, chatJid: string, + options?: { + runtimeTaskId?: string; + useTaskScopedSession?: boolean; + }, ): { env: Record; groupDir: string; runnerDir: string } { const projectRoot = process.cwd(); const groupDir = resolveGroupFolderPath(group.folder); fs.mkdirSync(groupDir, { recursive: true }); - // Per-group Claude sessions directory - const groupSessionsDir = path.join( - DATA_DIR, - 'sessions', - group.folder, - '.claude', - ); + const runtimeTaskId = options?.runtimeTaskId; + const useTaskScopedSession = + options?.useTaskScopedSession === true && Boolean(runtimeTaskId); + const sessionRootDir = + runtimeTaskId && useTaskScopedSession + ? resolveTaskSessionsPath(group.folder, runtimeTaskId) + : resolveGroupSessionsPath(group.folder); + + // Per-runtime Claude sessions directory + const groupSessionsDir = path.join(sessionRootDir, '.claude'); fs.mkdirSync(groupSessionsDir, { recursive: true }); ensureClaudeSessionSettings(groupSessionsDir); @@ -356,7 +366,9 @@ function prepareGroupEnvironment( syncDirectoryEntries(skillSources, path.join(groupSessionsDir, 'skills')); // Per-group IPC namespace - const groupIpcDir = resolveGroupIpcPath(group.folder); + const groupIpcDir = runtimeTaskId + ? resolveTaskRuntimeIpcPath(group.folder, runtimeTaskId) + : 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 }); @@ -433,6 +445,7 @@ function prepareGroupEnvironment( projectRoot, group, groupDir, + sessionRootDir, chatJid, isMain, isPairedRoom, @@ -447,7 +460,11 @@ function prepareGroupEnvironment( export async function runAgentProcess( group: RegisteredGroup, input: AgentInput, - onProcess: (proc: ChildProcess, processName: string) => void, + onProcess: ( + proc: ChildProcess, + processName: string, + runtimeIpcDir: string, + ) => void, onOutput?: (output: AgentOutput) => Promise, envOverrides?: Record, ): Promise { @@ -456,6 +473,10 @@ export async function runAgentProcess( group, input.isMain, input.chatJid, + { + runtimeTaskId: input.runtimeTaskId, + useTaskScopedSession: input.useTaskScopedSession, + }, ); // Apply provider fallback overrides (e.g. Kimi env vars when Claude is in cooldown) @@ -509,7 +530,7 @@ export async function runAgentProcess( env, }); - onProcess(proc, processName); + onProcess(proc, processName, env.NANOCLAW_IPC_DIR); let stdout = ''; let stderr = ''; @@ -860,8 +881,11 @@ export function writeTasksSnapshot( status: string; next_run: string | null; }>, + runtimeTaskId?: string, ): void { - const groupIpcDir = resolveGroupIpcPath(groupFolder); + const groupIpcDir = runtimeTaskId + ? resolveTaskRuntimeIpcPath(groupFolder, runtimeTaskId) + : resolveGroupIpcPath(groupFolder); fs.mkdirSync(groupIpcDir, { recursive: true }); const filteredTasks = isMain ? tasks diff --git a/src/group-folder.test.ts b/src/group-folder.test.ts index 99c4b0c..9e4bcd5 100644 --- a/src/group-folder.test.ts +++ b/src/group-folder.test.ts @@ -7,6 +7,9 @@ import { isValidGroupFolder, resolveGroupFolderPath, resolveGroupIpcPath, + resolveGroupSessionsPath, + resolveTaskRuntimeIpcPath, + resolveTaskSessionsPath, } from './group-folder.js'; describe('group folder validation', () => { @@ -33,8 +36,24 @@ describe('group folder validation', () => { expect(resolved).toBe(path.resolve(DATA_DIR, 'ipc', 'family-chat')); }); + it('resolves safe paths under data sessions directory', () => { + const resolved = resolveGroupSessionsPath('family-chat'); + expect(resolved).toBe(path.resolve(DATA_DIR, 'sessions', 'family-chat')); + }); + + it('resolves task-scoped IPC and session paths under the group namespace', () => { + expect(resolveTaskRuntimeIpcPath('family-chat', 'task-123')).toBe( + path.resolve(DATA_DIR, 'ipc', 'family-chat', 'tasks', 'task-123'), + ); + expect(resolveTaskSessionsPath('family-chat', 'task-123')).toBe( + path.resolve(DATA_DIR, 'sessions', 'family-chat', 'tasks', 'task-123'), + ); + }); + it('throws for unsafe folder names', () => { expect(() => resolveGroupFolderPath('../../etc')).toThrow(); expect(() => resolveGroupIpcPath('/tmp')).toThrow(); + expect(() => resolveTaskRuntimeIpcPath('family-chat', '../../etc')).toThrow(); + expect(() => resolveTaskSessionsPath('family-chat', '/tmp')).toThrow(); }); }); diff --git a/src/group-folder.ts b/src/group-folder.ts index 5745954..1788c98 100644 --- a/src/group-folder.ts +++ b/src/group-folder.ts @@ -3,6 +3,7 @@ import path from 'path'; import { DATA_DIR, GROUPS_DIR } from './config.js'; const GROUP_FOLDER_PATTERN = /^[A-Za-z0-9][A-Za-z0-9_-]{0,63}$/; +const RUNTIME_SEGMENT_PATTERN = /^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$/; const RESERVED_FOLDERS = new Set(['global']); export function isValidGroupFolder(folder: string): boolean { @@ -21,6 +22,18 @@ export function assertValidGroupFolder(folder: string): void { } } +function assertValidRuntimeSegment(segment: string, label: string): void { + if (!segment || segment !== segment.trim()) { + throw new Error(`Invalid ${label} "${segment}"`); + } + if (!RUNTIME_SEGMENT_PATTERN.test(segment)) { + throw new Error(`Invalid ${label} "${segment}"`); + } + if (segment.includes('/') || segment.includes('\\') || segment.includes('..')) { + throw new Error(`Invalid ${label} "${segment}"`); + } +} + function ensureWithinBase(baseDir: string, resolvedPath: string): void { const rel = path.relative(baseDir, resolvedPath); if (rel.startsWith('..') || path.isAbsolute(rel)) { @@ -42,3 +55,32 @@ export function resolveGroupIpcPath(folder: string): string { ensureWithinBase(ipcBaseDir, ipcPath); return ipcPath; } + +export function resolveGroupSessionsPath(folder: string): string { + assertValidGroupFolder(folder); + const sessionsBaseDir = path.resolve(DATA_DIR, 'sessions'); + const sessionsPath = path.resolve(sessionsBaseDir, folder); + ensureWithinBase(sessionsBaseDir, sessionsPath); + return sessionsPath; +} + +export function resolveTaskRuntimeIpcPath( + folder: string, + taskId: string, +): string { + assertValidGroupFolder(folder); + assertValidRuntimeSegment(taskId, 'task ID'); + const ipcBaseDir = path.resolve(DATA_DIR, 'ipc'); + const ipcPath = path.resolve(ipcBaseDir, folder, 'tasks', taskId); + ensureWithinBase(ipcBaseDir, ipcPath); + return ipcPath; +} + +export function resolveTaskSessionsPath(folder: string, taskId: string): string { + assertValidGroupFolder(folder); + assertValidRuntimeSegment(taskId, 'task ID'); + const sessionsBaseDir = path.resolve(DATA_DIR, 'sessions'); + const sessionsPath = path.resolve(sessionsBaseDir, folder, 'tasks', taskId); + ensureWithinBase(sessionsBaseDir, sessionsPath); + return sessionsPath; +} diff --git a/src/group-queue-ipc.ts b/src/group-queue-ipc.ts index ba521ec..11a69d8 100644 --- a/src/group-queue-ipc.ts +++ b/src/group-queue-ipc.ts @@ -1,13 +1,12 @@ import fs from 'fs'; import path from 'path'; -import { DATA_DIR } from './config.js'; +function resolveInputDir(ipcDir: string): string { + return path.join(ipcDir, 'input'); +} -export function queueFollowUpMessage( - groupFolder: string, - text: string, -): string { - const inputDir = path.join(DATA_DIR, 'ipc', groupFolder, 'input'); +export function queueFollowUpMessage(ipcDir: string, text: string): string { + const inputDir = resolveInputDir(ipcDir); fs.mkdirSync(inputDir, { recursive: true }); const filename = `${Date.now()}-${Math.random().toString(36).slice(2, 6)}.json`; const filepath = path.join(inputDir, filename); @@ -17,8 +16,8 @@ export function queueFollowUpMessage( return filename; } -export function writeCloseSentinel(groupFolder: string): void { - const inputDir = path.join(DATA_DIR, 'ipc', groupFolder, 'input'); +export function writeCloseSentinel(ipcDir: string): void { + const inputDir = resolveInputDir(ipcDir); fs.mkdirSync(inputDir, { recursive: true }); fs.writeFileSync(path.join(inputDir, '_close'), ''); } diff --git a/src/group-queue.test.ts b/src/group-queue.test.ts index a027cf2..3cc7e3a 100644 --- a/src/group-queue.test.ts +++ b/src/group-queue.test.ts @@ -64,6 +64,7 @@ describe('GroupQueue', () => { }); it('pipes follow-up messages to an active non-task process', async () => { + const ipcDir = '/tmp/ejclaw-test-data/ipc/group-folder'; let releaseRun!: (value: boolean) => void; const blocker = new Promise((resolve) => { releaseRun = resolve; @@ -72,12 +73,12 @@ describe('GroupQueue', () => { const processMessages = vi.fn(async () => await blocker); queue.setProcessMessagesFn(processMessages); - queue.enqueueMessageCheck('group1@g.us', 'group-folder'); + queue.enqueueMessageCheck('group1@g.us', ipcDir); await vi.advanceTimersByTimeAsync(10); expect(queue.sendMessage('group1@g.us', '후속 메시지')).toBe(true); expect(fs.mkdirSync).toHaveBeenCalledWith( - '/tmp/ejclaw-test-data/ipc/group-folder/input', + `${ipcDir}/input`, { recursive: true }, ); expect(fs.writeFileSync).toHaveBeenCalled(); diff --git a/src/group-queue.ts b/src/group-queue.ts index 7902d55..cbf6f96 100644 --- a/src/group-queue.ts +++ b/src/group-queue.ts @@ -28,7 +28,7 @@ interface GroupState { pendingTasks: QueuedTask[]; process: ChildProcess | null; processName: string | null; - groupFolder: string | null; + ipcDir: string | null; retryCount: number; retryTimer: ReturnType | null; retryScheduledAt: number | null; @@ -65,7 +65,7 @@ export class GroupQueue { pendingTasks: [], process: null, processName: null, - groupFolder: null, + ipcDir: null, retryCount: 0, retryTimer: null, retryScheduledAt: null, @@ -86,14 +86,14 @@ export class GroupQueue { return `run-${Date.now().toString(36)}-${Math.random().toString(36).slice(2, 8)}`; } - enqueueMessageCheck(groupJid: string, groupFolder?: string): void { + enqueueMessageCheck(groupJid: string, ipcDir?: string): void { if (this.shuttingDown) return; const state = this.getGroup(groupJid); - // Pre-set groupFolder so sendMessage can pipe IPC while agent process starts - if (groupFolder && !state.groupFolder) { - state.groupFolder = groupFolder; + // Pre-set IPC dir so sendMessage can pipe follow-ups while agent process starts + if (ipcDir && !state.ipcDir) { + state.ipcDir = ipcDir; } if (state.active) { @@ -178,18 +178,18 @@ export class GroupQueue { groupJid: string, proc: ChildProcess, processName: string, - groupFolder?: string, + ipcDir?: string, ): void { const state = this.getGroup(groupJid); state.process = proc; state.processName = processName; - if (groupFolder) state.groupFolder = groupFolder; + if (ipcDir) state.ipcDir = ipcDir; logger.info( { groupJid, runId: state.currentRunId, processName, - groupFolder: state.groupFolder, + ipcDir: state.ipcDir, isTaskProcess: state.isTaskProcess, }, 'Registered active process for group', @@ -198,13 +198,13 @@ export class GroupQueue { sendMessage(groupJid: string, text: string): boolean { const state = this.getGroup(groupJid); - if (!state.active || !state.groupFolder || state.isTaskProcess) { + if (!state.active || !state.ipcDir || state.isTaskProcess) { logger.debug( { groupJid, runId: state.currentRunId, active: state.active, - groupFolder: state.groupFolder, + ipcDir: state.ipcDir, isTaskProcess: state.isTaskProcess, }, 'Cannot pipe follow-up message to active agent', @@ -213,12 +213,12 @@ export class GroupQueue { } try { - const filename = queueFollowUpMessage(state.groupFolder, text); + const filename = queueFollowUpMessage(state.ipcDir, text); logger.info( { groupJid, runId: state.currentRunId, - groupFolder: state.groupFolder, + ipcDir: state.ipcDir, textLength: text.length, filename, }, @@ -230,7 +230,7 @@ export class GroupQueue { { groupJid, runId: state.currentRunId, - groupFolder: state.groupFolder, + ipcDir: state.ipcDir, err, }, 'Failed to queue follow-up message for active agent', @@ -247,16 +247,16 @@ export class GroupQueue { metadata?: { runId?: string; reason?: string }, ): void { const state = this.getGroup(groupJid); - if (!state.active || !state.groupFolder) return; + if (!state.active || !state.ipcDir) return; state.closingStdin = true; try { - writeCloseSentinel(state.groupFolder); + writeCloseSentinel(state.ipcDir); logger.info( { groupJid, runId: metadata?.runId ?? state.currentRunId, - groupFolder: state.groupFolder, + ipcDir: state.ipcDir, reason: metadata?.reason ?? 'unspecified', }, 'Signaled active agent to close stdin', @@ -266,7 +266,7 @@ export class GroupQueue { { groupJid, runId: metadata?.runId ?? state.currentRunId, - groupFolder: state.groupFolder, + ipcDir: state.ipcDir, reason: metadata?.reason ?? 'unspecified', err, }, @@ -335,7 +335,7 @@ export class GroupQueue { state.closingStdin = false; state.process = null; state.processName = null; - state.groupFolder = null; + state.ipcDir = null; state.currentRunId = null; this.activeCount--; this.drainGroup(groupJid); @@ -368,7 +368,7 @@ export class GroupQueue { state.closingStdin = false; state.process = null; state.processName = null; - state.groupFolder = null; + state.ipcDir = null; this.activeCount--; this.drainGroup(groupJid); } diff --git a/src/index.ts b/src/index.ts index 826b18a..588eabb 100644 --- a/src/index.ts +++ b/src/index.ts @@ -394,8 +394,8 @@ async function main(): Promise { registeredGroups: () => registeredGroups, getSessions: () => sessions, queue, - onProcess: (groupJid, proc, processName, groupFolder) => - queue.registerProcess(groupJid, proc, processName, groupFolder), + onProcess: (groupJid, proc, processName, ipcDir) => + queue.registerProcess(groupJid, proc, processName, ipcDir), sendMessage: (jid, rawText) => sendFormattedChannelMessage(channels, jid, rawText), sendTrackedMessage: (jid, rawText) => diff --git a/src/message-agent-executor.ts b/src/message-agent-executor.ts index 3debef4..3dda543 100644 --- a/src/message-agent-executor.ts +++ b/src/message-agent-executor.ts @@ -187,8 +187,8 @@ export async function runAgentForGroup( ...agentInput, sessionId: persistSessionIds ? sessionId : undefined, }, - (proc, processName) => - deps.queue.registerProcess(chatJid, proc, processName, group.folder), + (proc, processName, ipcDir) => + deps.queue.registerProcess(chatJid, proc, processName, ipcDir), wrappedOnOutput, provider === 'claude' ? undefined : getFallbackEnvOverrides(), ); diff --git a/src/message-runtime.test.ts b/src/message-runtime.test.ts index ad306f1..a60d808 100644 --- a/src/message-runtime.test.ts +++ b/src/message-runtime.test.ts @@ -143,6 +143,7 @@ vi.mock('./session-commands.js', () => ({ import * as agentRunner from './agent-runner.js'; import * as db from './db.js'; +import { resolveGroupIpcPath } from './group-folder.js'; import { createMessageRuntime } from './message-runtime.js'; import * as providerFallback from './provider-fallback.js'; import type { Channel, RegisteredGroup } from './types.js'; @@ -1989,7 +1990,10 @@ describe('createMessageRuntime', () => { runtime.recoverPendingMessages(); - expect(enqueueMessageCheck).toHaveBeenCalledWith(chatJid, group.folder); + expect(enqueueMessageCheck).toHaveBeenCalledWith( + chatJid, + resolveGroupIpcPath(group.folder), + ); expect(db.getMessagesSinceSeq).not.toHaveBeenCalled(); }); }); diff --git a/src/message-runtime.ts b/src/message-runtime.ts index 8c00d50..2969e28 100644 --- a/src/message-runtime.ts +++ b/src/message-runtime.ts @@ -30,6 +30,7 @@ import { } from './session-commands.js'; import { Channel, NewMessage, RegisteredGroup } from './types.js'; import { logger } from './logger.js'; +import { resolveGroupIpcPath } from './group-folder.js'; export interface MessageRuntimeDeps { assistantName: string; @@ -532,7 +533,7 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { continue; } - deps.queue.enqueueMessageCheck(chatJid, group.folder); + deps.queue.enqueueMessageCheck(chatJid, resolveGroupIpcPath(group.folder)); } } } catch (err) { @@ -554,7 +555,7 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { { chatJid, group: group.name, workItemId: openWorkItem.id }, 'Recovery: found open work item awaiting delivery', ); - deps.queue.enqueueMessageCheck(chatJid, group.folder); + deps.queue.enqueueMessageCheck(chatJid, resolveGroupIpcPath(group.folder)); continue; } @@ -575,7 +576,7 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { { group: group.name, pendingCount: pending.length }, 'Recovery: found unprocessed messages', ); - deps.queue.enqueueMessageCheck(chatJid, group.folder); + deps.queue.enqueueMessageCheck(chatJid, resolveGroupIpcPath(group.folder)); } else if (rawPending.length > 0) { const endSeq = rawPending[rawPending.length - 1].seq; if (endSeq != null) { diff --git a/src/task-scheduler.test.ts b/src/task-scheduler.test.ts index 3a52a58..511d85a 100644 --- a/src/task-scheduler.test.ts +++ b/src/task-scheduler.test.ts @@ -1,5 +1,23 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +const { runAgentProcessMock, writeTasksSnapshotMock } = vi.hoisted(() => ({ + runAgentProcessMock: vi.fn(async () => ({ + status: 'success' as const, + result: 'done', + })), + writeTasksSnapshotMock: vi.fn(), +})); + +vi.mock('./agent-runner.js', async () => { + const actual = + await vi.importActual('./agent-runner.js'); + return { + ...actual, + runAgentProcess: runAgentProcessMock, + writeTasksSnapshot: writeTasksSnapshotMock, + }; +}); + import { _initTestDatabase, createTask, getTaskById } from './db.js'; import { createTaskStatusTracker } from './task-status-tracker.js'; import { TASK_STATUS_MESSAGE_PREFIX } from './task-watch-status.js'; @@ -16,6 +34,8 @@ describe('task scheduler', () => { beforeEach(() => { _initTestDatabase(); _resetSchedulerLoopForTests(); + runAgentProcessMock.mockClear(); + writeTasksSnapshotMock.mockClear(); vi.useFakeTimers(); }); @@ -209,6 +229,135 @@ Check the run. expect(enqueueTask.mock.calls[0][1]).toBe('task-watch-group'); }); + it('uses dedicated IPC but shared session state for group-context CI watchers', async () => { + const dueAt = new Date(Date.now() - 60_000).toISOString(); + createTask({ + id: 'task-watch-runtime', + group_folder: 'shared-group', + chat_jid: 'shared@g.us', + agent_type: 'codex', + prompt: ` +[BACKGROUND CI WATCH] + +Watch target: +GitHub Actions run 123456 + +Task ID: +task-watch-runtime + +Check instructions: +Check the run. + `.trim(), + schedule_type: 'interval', + schedule_value: '60000', + context_mode: 'group', + next_run: dueAt, + status: 'active', + created_at: '2026-02-22T00:00:00.000Z', + }); + + const enqueueTask = vi.fn( + async (_groupJid: string, _taskId: string, fn: () => Promise) => { + await 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: () => ({ 'shared-group': 'session-123' }), + queue: { enqueueTask } as any, + onProcess: () => {}, + sendMessage: async () => {}, + }); + + await vi.advanceTimersByTimeAsync(10); + + expect(runAgentProcessMock).toHaveBeenCalledWith( + expect.anything(), + expect.objectContaining({ + runtimeTaskId: 'task-watch-runtime', + useTaskScopedSession: false, + sessionId: 'session-123', + }), + expect.any(Function), + expect.any(Function), + ); + expect(writeTasksSnapshotMock).toHaveBeenCalledWith( + 'shared-group', + false, + expect.any(Array), + 'task-watch-runtime', + ); + }); + + it('isolates both IPC and session state for isolated tasks', async () => { + const dueAt = new Date(Date.now() - 60_000).toISOString(); + createTask({ + id: 'task-isolated-runtime', + group_folder: 'shared-group', + chat_jid: 'shared@g.us', + agent_type: 'claude-code', + prompt: 'run isolated task', + schedule_type: 'once', + schedule_value: dueAt, + context_mode: 'isolated', + next_run: dueAt, + status: 'active', + created_at: '2026-02-22T00:00:00.000Z', + }); + + const enqueueTask = vi.fn( + async (_groupJid: string, _taskId: string, fn: () => Promise) => { + await fn(); + }, + ); + + startSchedulerLoop({ + serviceAgentType: 'claude-code', + registeredGroups: () => ({ + 'shared@g.us': { + name: 'Shared', + folder: 'shared-group', + trigger: '@Claude', + added_at: '2026-02-22T00:00:00.000Z', + agentType: 'claude-code', + }, + }), + getSessions: () => ({ 'shared-group': 'session-123' }), + queue: { enqueueTask } as any, + onProcess: () => {}, + sendMessage: async () => {}, + }); + + await vi.advanceTimersByTimeAsync(10); + + expect(runAgentProcessMock).toHaveBeenCalledWith( + expect.anything(), + expect.objectContaining({ + runtimeTaskId: 'task-isolated-runtime', + useTaskScopedSession: true, + sessionId: undefined, + }), + expect.any(Function), + expect.any(Function), + ); + expect(writeTasksSnapshotMock).toHaveBeenCalledWith( + 'shared-group', + false, + expect.any(Array), + 'task-isolated-runtime', + ); + }); + it('renders watcher heartbeat messages with target and timing', () => { const prompt = ` [BACKGROUND CI WATCH] diff --git a/src/task-scheduler.ts b/src/task-scheduler.ts index 2ab8f00..ff18053 100644 --- a/src/task-scheduler.ts +++ b/src/task-scheduler.ts @@ -22,17 +22,27 @@ import { updateTaskAfterRun, } from './db.js'; import { GroupQueue } from './group-queue.js'; -import { resolveGroupFolderPath } from './group-folder.js'; +import { + resolveGroupFolderPath, + resolveGroupIpcPath, + resolveTaskRuntimeIpcPath, +} from './group-folder.js'; import { logger } from './logger.js'; import { createTaskStatusTracker } from './task-status-tracker.js'; -import { getTaskQueueJid } from './task-watch-status.js'; +import { + getTaskQueueJid, + getTaskRuntimeTaskId, + shouldUseTaskScopedSession, +} from './task-watch-status.js'; import { AgentType, RegisteredGroup, ScheduledTask } from './types.js'; export { extractWatchCiTarget, getTaskQueueJid, + getTaskRuntimeTaskId, isTaskStatusControlMessage, isWatchCiTask, renderWatchCiStatusMessage, + shouldUseTaskScopedSession, } from './task-watch-status.js'; /** @@ -85,7 +95,7 @@ export interface SchedulerDependencies { groupJid: string, proc: ChildProcess, processName: string, - groupFolder: string, + ipcDir: string, ) => void; sendMessage: (jid: string, text: string) => Promise; sendTrackedMessage?: (jid: string, text: string) => Promise; @@ -101,7 +111,10 @@ interface TaskExecutionContext { groupDir: string; isMain: boolean; queueJid: string; + runtimeIpcDir: string; + runtimeTaskId?: string; sessionId?: string; + useTaskScopedSession: boolean; taskAgentType: AgentType; } @@ -124,14 +137,22 @@ function resolveTaskExecutionContext( const taskAgentType = task.agent_type || deps.serviceAgentType || SERVICE_AGENT_TYPE; const sessions = deps.getSessions(); + const runtimeTaskId = getTaskRuntimeTaskId(task); + const useTaskScopedSession = shouldUseTaskScopedSession(task); + const runtimeIpcDir = runtimeTaskId + ? resolveTaskRuntimeIpcPath(task.group_folder, runtimeTaskId) + : resolveGroupIpcPath(task.group_folder); return { group, groupDir, isMain, queueJid: getTaskQueueJid(task), + runtimeIpcDir, + runtimeTaskId, sessionId: task.context_mode === 'group' ? sessions[task.group_folder] : undefined, + useTaskScopedSession, taskAgentType, }; } @@ -140,6 +161,7 @@ function writeTaskSnapshotForGroup( taskAgentType: AgentType, groupFolder: string, isMain: boolean, + runtimeTaskId?: string, ): void { const tasks = getAllTasks(taskAgentType); writeTasksSnapshot( @@ -154,6 +176,7 @@ function writeTaskSnapshotForGroup( status: task.status, next_run: task.next_run, })), + runtimeTaskId, ); } @@ -201,6 +224,7 @@ async function runTask( context.taskAgentType, task.group_folder, context.isMain, + context.runtimeTaskId, ); let result: string | null = null; @@ -222,10 +246,17 @@ async function runTask( chatJid: task.chat_jid, isMain: context.isMain, isScheduledTask: true, + runtimeTaskId: context.runtimeTaskId, + useTaskScopedSession: context.useTaskScopedSession, assistantName: ASSISTANT_NAME, }, (proc, processName) => - deps.onProcess(context.queueJid, proc, processName, task.group_folder), + deps.onProcess( + context.queueJid, + proc, + processName, + context.runtimeIpcDir, + ), async (streamedOutput: AgentOutput) => { if (streamedOutput.phase === 'progress') { return; diff --git a/src/task-watch-status.ts b/src/task-watch-status.ts index 9abab5d..ea198db 100644 --- a/src/task-watch-status.ts +++ b/src/task-watch-status.ts @@ -126,3 +126,17 @@ export function getTaskQueueJid( ? `${task.chat_jid}::task:${task.id}` : task.chat_jid; } + +export function getTaskRuntimeTaskId( + task: Pick, +): string | undefined { + return task.context_mode === 'isolated' || isWatchCiTask(task) + ? task.id + : undefined; +} + +export function shouldUseTaskScopedSession( + task: Pick, +): boolean { + return task.context_mode === 'isolated'; +}