From 807828e0ce8d463ee23afe258a97eace919b35f0 Mon Sep 17 00:00:00 2001 From: Eyejoker Date: Wed, 3 Jun 2026 17:58:24 +0800 Subject: [PATCH] fix: drain sdk follow-up turns (#219) --- src/agent-runner.test.ts | 49 ++++++++++++++++++++ src/agent-runner.ts | 32 ++++++++++++- src/group-queue-state.ts | 7 ++- src/group-queue.test.ts | 46 +++++++++++++++++- src/group-queue.ts | 8 ++-- src/message-agent-executor-attempt-runner.ts | 1 + src/message-agent-executor.ts | 4 +- 7 files changed, 137 insertions(+), 10 deletions(-) diff --git a/src/agent-runner.test.ts b/src/agent-runner.test.ts index 9ab63c3..b699d0f 100644 --- a/src/agent-runner.test.ts +++ b/src/agent-runner.test.ts @@ -852,4 +852,53 @@ describe('agent-runner Codex goals', () => { | undefined; expect(spawnEnv?.CODEX_GOALS).toBe('true'); }); + + it('marks SDK codex processes so queued IPC follow-ups drain after the active run', async () => { + const previousCodexGoals = process.env.CODEX_GOALS; + delete process.env.CODEX_GOALS; + const onProcess = vi.fn(); + const codexGroup: RegisteredGroup = { + ...testGroup, + agentType: 'codex', + }; + + try { + const resultPromise = runAgentProcess( + codexGroup, + { + ...testInput, + roomRoleContext: { + serviceId: 'codex-main', + role: 'owner', + ownerServiceId: 'codex-main', + reviewerServiceId: 'claude', + failoverOwner: false, + }, + }, + onProcess, + async () => {}, + { + CODEX_RUNTIME: 'sdk', + CODEX_RUNTIME_SDK_ROLES: 'owner', + }, + ); + + fakeProc.emit('close', 0); + const result = await resultPromise; + expect(result.status).toBe('success'); + + expect(onProcess).toHaveBeenCalledWith( + fakeProc, + expect.any(String), + expect.any(String), + { drainFollowUpsAfterRun: true }, + ); + } finally { + if (previousCodexGoals === undefined) { + delete process.env.CODEX_GOALS; + } else { + process.env.CODEX_GOALS = previousCodexGoals; + } + } + }); }); diff --git a/src/agent-runner.ts b/src/agent-runner.ts index 3b7a7dd..1ecd1c5 100644 --- a/src/agent-runner.ts +++ b/src/agent-runner.ts @@ -61,6 +61,30 @@ export interface AgentOutput { }; } +interface ActiveProcessOptions { + drainFollowUpsAfterRun?: boolean; +} + +function isCodexSdkRuntimeMode( + env: Record, + input: AgentInput, +): boolean { + const runtime = env.CODEX_RUNTIME?.trim().toLowerCase(); + if (runtime !== 'sdk') return false; + if (input.prompt.trim() === '/compact') return false; + if (input.codexGoals === true || env.CODEX_GOALS === 'true') return false; + + const rawRoles = env.CODEX_RUNTIME_SDK_ROLES?.trim(); + if (!rawRoles) return true; + const role = input.roomRoleContext?.role?.trim().toLowerCase(); + if (!role) return false; + return rawRoles + .split(',') + .map((item) => item.trim().toLowerCase()) + .filter(Boolean) + .includes(role); +} + function readRoomSkillOverridesForRunner( chatJid: string, ): StoredRoomSkillOverride[] { @@ -126,6 +150,7 @@ export async function runAgentProcess( proc: ChildProcess, processName: string, runtimeIpcDir: string, + options?: ActiveProcessOptions, ) => void, onOutput?: (output: AgentOutput) => Promise, envOverrides?: Record, @@ -186,6 +211,11 @@ export async function runAgentProcess( const safeName = group.folder.replace(/[^a-zA-Z0-9-]/g, '-'); const processSuffix = input.runId || `${Date.now()}`; const processName = `ejclaw-${safeName}-${processSuffix}`; + const activeProcessOptions: ActiveProcessOptions = { + drainFollowUpsAfterRun: + (group.agentType || 'claude-code') === 'codex' && + isCodexSdkRuntimeMode(env, input), + }; const finalizeCodexAuthSessionOnce = createCodexAuthSessionFinalizer( () => codexSessionAuth, ); @@ -228,7 +258,7 @@ export async function runAgentProcess( env, }); - onProcess(proc, processName, env[EJCLAW_ENV.ipcDir]); + onProcess(proc, processName, env[EJCLAW_ENV.ipcDir], activeProcessOptions); const runnerInput: AgentInput = { ...input, diff --git a/src/group-queue-state.ts b/src/group-queue-state.ts index 8719b6f..12e6f03 100644 --- a/src/group-queue-state.ts +++ b/src/group-queue-state.ts @@ -37,6 +37,7 @@ export interface GroupState { pendingTasks: QueuedTask[]; process: ChildProcess | null; processName: string | null; + drainFollowUpsAfterRun: boolean; ipcDir: string | null; retryCount: number; retryTimer: ReturnType | null; @@ -60,6 +61,7 @@ export function createGroupState(): GroupState { pendingTasks: [], process: null, processName: null, + drainFollowUpsAfterRun: false, ipcDir: null, retryCount: 0, retryTimer: null, @@ -145,6 +147,7 @@ export function resetRunState(state: GroupState, groupJid: string): void { state.startedAt = null; state.process = null; state.processName = null; + state.drainFollowUpsAfterRun = false; state.ipcDir = null; state.directTerminalDeliveries.clear(); transitionRunPhase(state, groupJid, 'idle'); @@ -160,7 +163,8 @@ export function assertRunPhaseInvariants( state.currentRunId != null || state.runningTaskId != null || state.process != null || - state.processName != null + state.processName != null || + state.drainFollowUpsAfterRun ) { logger.error( { @@ -170,6 +174,7 @@ export function assertRunPhaseInvariants( runningTaskId: state.runningTaskId, hasProcess: state.process != null, processName: state.processName, + drainFollowUpsAfterRun: state.drainFollowUpsAfterRun, }, 'Invariant violation: idle phase has stale run/task ID or process', ); diff --git a/src/group-queue.test.ts b/src/group-queue.test.ts index d2991d1..76b2a60 100644 --- a/src/group-queue.test.ts +++ b/src/group-queue.test.ts @@ -40,8 +40,6 @@ describe('GroupQueue', () => { vi.useRealTimers(); }); - // --- Single group at a time --- - it('only runs one agent per group at a time', async () => { let concurrentCount = 0; let maxConcurrent = 0; @@ -90,6 +88,7 @@ describe('GroupQueue', () => { releaseRun(true); await vi.advanceTimersByTimeAsync(10); + expect(processMessages).toHaveBeenCalledTimes(1); }); it('does not pipe follow-up messages after stdin close was requested', async () => { @@ -791,3 +790,46 @@ describe('GroupQueue close reason tracking', () => { expect(queue.getCloseReasonForRun('group1@g.us', activeRunId)).toBeNull(); }); }); + +describe('GroupQueue SDK follow-up draining', () => { + let queue: GroupQueue; + + beforeEach(() => { + vi.useFakeTimers(); + vi.clearAllMocks(); + queue = new GroupQueue(); + }); + + afterEach(() => { + vi.useRealTimers(); + }); + + it('drains follow-up messages after active processes that cannot consume queued IPC', async () => { + const ipcDir = '/tmp/ejclaw-test-data/ipc/group-folder'; + let releaseRun!: (value: boolean) => void; + let activeRun = true; + const blocker = new Promise((resolve) => { + releaseRun = resolve; + }); + + const processMessages = vi.fn(async () => { + if (!activeRun) return true; + activeRun = false; + return await blocker; + }); + + queue.setProcessMessagesFn(processMessages); + queue.enqueueMessageCheck('group1@g.us', ipcDir); + await vi.advanceTimersByTimeAsync(10); + + queue.registerProcess('group1@g.us', {} as any, 'sdk-agent', ipcDir, { + drainFollowUpsAfterRun: true, + }); + expect(queue.sendMessage('group1@g.us', '후속 메시지')).toBe(true); + + releaseRun(true); + await vi.advanceTimersByTimeAsync(10); + + expect(processMessages).toHaveBeenCalledTimes(2); + }); +}); diff --git a/src/group-queue.ts b/src/group-queue.ts index c92d576..3d141f0 100644 --- a/src/group-queue.ts +++ b/src/group-queue.ts @@ -65,7 +65,6 @@ export class GroupQueue { this.processMessagesFn = fn; } - /** Limit concurrency after restart to avoid API rate-limit storms. */ enterRecoveryMode(): void { this.recoveryMode = true; logger.info( @@ -101,7 +100,6 @@ export class GroupQueue { const state = this.getGroup(groupJid); - // Pre-set IPC dir so sendMessage can pipe follow-ups while agent process starts if (ipcDir && !state.ipcDir) { state.ipcDir = ipcDir; } @@ -157,7 +155,6 @@ export class GroupQueue { const state = this.getGroup(groupJid); - // Prevent double-queuing: check both pending and currently-running task if (state.runningTaskId === taskId) { logger.debug({ groupJid, taskId }, 'Task already running, skipping'); return; @@ -196,7 +193,6 @@ export class GroupQueue { return; } - // Run immediately this.runTask(groupJid, { id: taskId, groupJid, fn }).catch((err) => logger.error({ groupJid, taskId, err }, 'Unhandled error in runTask'), ); @@ -207,10 +203,12 @@ export class GroupQueue { proc: ChildProcess, processName: string, ipcDir?: string, + options?: { drainFollowUpsAfterRun?: boolean }, ): void { const state = this.getGroup(groupJid); state.process = proc; state.processName = processName; + state.drainFollowUpsAfterRun = options?.drainFollowUpsAfterRun === true; if (ipcDir) state.ipcDir = ipcDir; logger.info( { @@ -241,6 +239,7 @@ export class GroupQueue { try { const filename = queueFollowUpMessage(state.ipcDir, text); + state.pendingMessages ||= state.drainFollowUpsAfterRun; logger.info( { groupJid, @@ -248,6 +247,7 @@ export class GroupQueue { ipcDir: state.ipcDir, textLength: text.length, filename, + pendingMessages: state.pendingMessages, }, 'Queued follow-up message for active agent', ); diff --git a/src/message-agent-executor-attempt-runner.ts b/src/message-agent-executor-attempt-runner.ts index 64a9126..2b07de4 100644 --- a/src/message-agent-executor-attempt-runner.ts +++ b/src/message-agent-executor-attempt-runner.ts @@ -100,6 +100,7 @@ interface RunMessageAgentAttemptArgs { : never, processName: string, ipcDir?: string, + options?: { drainFollowUpsAfterRun?: boolean }, ) => void; onOutput?: (output: AgentOutput) => Promise; pairedExecutionLifecycle: { diff --git a/src/message-agent-executor.ts b/src/message-agent-executor.ts index 6b85cdb..846e1ce 100644 --- a/src/message-agent-executor.ts +++ b/src/message-agent-executor.ts @@ -274,8 +274,8 @@ export async function runAgentForGroup( deps.persistSession(sessionFolder, sessionId); currentSessionId = sessionId; }, - registerProcess: (proc, processName, ipcDir) => - deps.queue.registerProcess(chatJid, proc, processName, ipcDir), + registerProcess: (proc, processName, ipcDir, options) => + deps.queue.registerProcess(chatJid, proc, processName, ipcDir, options), onOutput, pairedExecutionLifecycle, log,