diff --git a/runners/agent-runner/src/index.ts b/runners/agent-runner/src/index.ts index dc54b84..71482e0 100644 --- a/runners/agent-runner/src/index.ts +++ b/runners/agent-runner/src/index.ts @@ -66,7 +66,6 @@ const IPC_DIR = process.env.NANOCLAW_IPC_DIR || '/workspace/ipc'; const WORK_DIR = process.env.NANOCLAW_WORK_DIR || ''; const IPC_INPUT_DIR = path.join(IPC_DIR, 'input'); -const MAX_TURNS = 100; const IPC_INPUT_CLOSE_SENTINEL = path.join(IPC_INPUT_DIR, '_close'); const IPC_POLL_MS = 500; @@ -383,28 +382,6 @@ function drainIpcInput(): string[] { } } -/** - * Wait for a new IPC message or _close sentinel. - * Returns the messages as a single string, or null if _close. - */ -function waitForIpcMessage(): Promise { - return new Promise((resolve) => { - const poll = () => { - if (shouldClose()) { - resolve(null); - return; - } - const messages = drainIpcInput(); - if (messages.length > 0) { - resolve(messages.join('\n')); - return; - } - setTimeout(poll, IPC_POLL_MS); - }; - poll(); - }); -} - /** * Run a single query and stream results via writeOutput. * Uses MessageStream (AsyncIterable) to keep isSingleUserTurn=false, @@ -417,8 +394,7 @@ async function runQuery( mcpServerPath: string, containerInput: ContainerInput, sdkEnv: Record, - resumeAt?: string, -): Promise<{ newSessionId?: string; lastAssistantUuid?: string; closedDuringQuery: boolean }> { +): Promise<{ newSessionId?: string; closedDuringQuery: boolean }> { const stream = new MessageStream(); stream.push(prompt); @@ -444,7 +420,6 @@ async function runQuery( setTimeout(pollIpcDuringQuery, IPC_POLL_MS); let newSessionId: string | undefined; - let lastAssistantUuid: string | undefined; let messageCount = 0; let resultCount = 0; @@ -486,7 +461,6 @@ async function runQuery( effort, additionalDirectories: extraDirs.length > 0 ? extraDirs : undefined, resume: sessionId, - resumeSessionAt: resumeAt, allowedTools: [ 'Bash', 'Read', 'Write', 'Edit', 'Glob', 'Grep', @@ -536,10 +510,6 @@ async function runQuery( const msgType = message.type === 'system' ? `system/${(message as { subtype?: string }).subtype}` : message.type; log(`[msg #${messageCount}] type=${msgType}`); - if (message.type === 'assistant' && 'uuid' in message) { - lastAssistantUuid = (message as { uuid: string }).uuid; - } - if (message.type === 'system' && message.subtype === 'init') { newSessionId = message.session_id; log(`Session initialized: ${newSessionId}`); @@ -588,8 +558,10 @@ async function runQuery( } ipcPolling = false; - log(`Query done. Messages: ${messageCount}, results: ${resultCount}, lastAssistantUuid: ${lastAssistantUuid || 'none'}, closedDuringQuery: ${closedDuringQuery}`); - return { newSessionId, lastAssistantUuid, closedDuringQuery }; + log( + `Query done. Messages: ${messageCount}, results: ${resultCount}, closedDuringQuery: ${closedDuringQuery}`, + ); + return { newSessionId, closedDuringQuery }; } async function main(): Promise { @@ -740,49 +712,24 @@ async function main(): Promise { } // --- End slash command handling --- - // Query loop: run query → wait for IPC message → run new query → repeat - let resumeAt: string | undefined; - let turnCount = 0; try { - while (true) { - turnCount++; - if (turnCount > MAX_TURNS) { - log(`Turn limit reached (${MAX_TURNS}), exiting`); - writeOutput({ status: 'success', result: '[세션 턴 제한 도달. 새 메시지로 다시 시작됩니다.]', newSessionId: sessionId }); - break; - } - log(`Starting query (session: ${sessionId || 'new'}, turn: ${turnCount}/${MAX_TURNS}, resumeAt: ${resumeAt || 'latest'})...`); + log(`Starting query (session: ${sessionId || 'new'})...`); - const queryResult = await runQuery(prompt, sessionId, mcpServerPath, containerInput, sdkEnv, resumeAt); - if (queryResult.newSessionId) { - sessionId = queryResult.newSessionId; - } - if (queryResult.lastAssistantUuid) { - resumeAt = queryResult.lastAssistantUuid; - } + const queryResult = await runQuery( + prompt, + sessionId, + mcpServerPath, + containerInput, + sdkEnv, + ); + if (queryResult.newSessionId) { + sessionId = queryResult.newSessionId; + } - // If _close was consumed during the query, exit immediately. - // Don't emit a session-update marker (it would reset the host's - // idle timer and cause a 30-min delay before the next _close). - if (queryResult.closedDuringQuery) { - log('Close sentinel consumed during query, exiting'); - break; - } - - // Emit session update so host can track it + if (!queryResult.closedDuringQuery) { writeOutput({ status: 'success', result: null, newSessionId: sessionId }); - - log('Query ended, waiting for next IPC message...'); - - // Wait for the next message or _close sentinel - const nextMessage = await waitForIpcMessage(); - if (nextMessage === null) { - log('Close sentinel received, exiting'); - break; - } - - log(`Got new message (${nextMessage.length} chars), starting new query`); - prompt = nextMessage; + } else { + log('Close sentinel consumed during query, exiting'); } } catch (err) { const errorMessage = err instanceof Error ? err.message : String(err); diff --git a/runners/codex-runner/src/index.ts b/runners/codex-runner/src/index.ts index 4395f1c..27da4cd 100644 --- a/runners/codex-runner/src/index.ts +++ b/runners/codex-runner/src/index.ts @@ -49,7 +49,6 @@ const WORK_DIR = process.env.NANOCLAW_WORK_DIR || ''; const IPC_INPUT_DIR = path.join(IPC_DIR, 'input'); const IPC_INPUT_CLOSE_SENTINEL = path.join(IPC_INPUT_DIR, '_close'); const IPC_POLL_MS = 500; -const MAX_TURNS = 100; const OUTPUT_START_MARKER = '---NANOCLAW_OUTPUT_START---'; const OUTPUT_END_MARKER = '---NANOCLAW_OUTPUT_END---'; @@ -134,24 +133,6 @@ function drainIpcInput(): string[] { } } -function waitForIpcMessage(): Promise { - return new Promise((resolve) => { - const poll = () => { - if (consumeCloseSentinel()) { - resolve(null); - return; - } - const messages = drainIpcInput(); - if (messages.length > 0) { - resolve(messages.join('\n')); - return; - } - setTimeout(poll, IPC_POLL_MS); - }; - poll(); - }); -} - function extractImagePaths(text: string): { cleanText: string; imagePaths: string[] } { const imagePattern = /\[Image:\s*(\/[^\]]+)\]/g; const imagePaths: string[] = []; @@ -364,61 +345,28 @@ async function runAppServerSession( return; } - let turnCount = 0; - while (true) { - turnCount++; - if (turnCount > MAX_TURNS) { - log(`Turn limit reached (${MAX_TURNS}), exiting`); - writeOutput({ - status: 'success', - result: '[세션 턴 제한 도달. 새 메시지로 다시 시작됩니다.]', - newSessionId: threadId, - }); - break; - } + log('Starting app-server turn...'); + const { result, error } = await executeAppServerTurn(client, threadId, prompt); - log(`Starting app-server turn ${turnCount}/${MAX_TURNS}...`); - const { result, error } = await executeAppServerTurn(client, threadId, prompt); + if (error) { + log(`App-server turn error: ${error}`); + writeOutput({ + status: 'error', + result: result || null, + newSessionId: threadId, + error, + }); + } else { + writeOutput({ + status: 'success', + result: result || null, + ...(result ? { phase: 'final' as const } : {}), + newSessionId: threadId, + }); + } - if (consumeCloseSentinel()) { - if (result) { - writeOutput({ - status: 'success', - result, - newSessionId: threadId, - }); - } - log('Close sentinel detected, exiting app-server runtime'); - break; - } - - if (error) { - log(`App-server turn error: ${error}`); - writeOutput({ - status: 'error', - result: result || null, - newSessionId: threadId, - error, - }); - } else { - writeOutput({ - status: 'success', - result: result || null, - ...(result ? { phase: 'final' as const } : {}), - newSessionId: threadId, - }); - } - - log('App-server turn done, waiting for next IPC message...'); - - const nextMessage = await waitForIpcMessage(); - if (nextMessage === null) { - log('Close sentinel received, exiting app-server runtime'); - break; - } - - log(`Got new app-server message (${nextMessage.length} chars)`); - prompt = nextMessage; + if (consumeCloseSentinel()) { + log('Close sentinel detected, exiting app-server runtime'); } } finally { await client.close(); diff --git a/src/task-scheduler.ts b/src/task-scheduler.ts index db96398..c98372b 100644 --- a/src/task-scheduler.ts +++ b/src/task-scheduler.ts @@ -259,20 +259,6 @@ async function runTask( const sessionId = task.context_mode === 'group' ? sessions[task.group_folder] : undefined; - // After the task produces a result, close the agent promptly. - // Tasks are single-turn — no need to wait IDLE_TIMEOUT (30 min) for the - // query loop to time out. A short delay handles any final MCP calls. - const TASK_CLOSE_DELAY_MS = 10000; - let closeTimer: ReturnType | null = null; - - const scheduleClose = () => { - if (closeTimer) return; // already scheduled - closeTimer = setTimeout(() => { - logger.debug({ taskId: task.id }, 'Closing task agent after result'); - deps.queue.closeStdin(task.chat_jid); - }, TASK_CLOSE_DELAY_MS); - }; - const shouldTrackStatus = isWatchCiTask(task) && typeof deps.sendTrackedMessage === 'function' && @@ -350,11 +336,6 @@ async function runTask( result = streamedOutput.result; // Forward result to user (sendMessage handles formatting) await deps.sendMessage(task.chat_jid, streamedOutput.result); - scheduleClose(); - } - if (streamedOutput.status === 'success') { - deps.queue.notifyIdle(task.chat_jid); - scheduleClose(); // Close promptly even when result is null (e.g. IPC-only tasks) } if (streamedOutput.status === 'error') { error = streamedOutput.error || 'Unknown error'; @@ -362,8 +343,6 @@ async function runTask( }, ); - if (closeTimer) clearTimeout(closeTimer); - if (output.status === 'error') { error = output.error || 'Unknown error'; } else if (output.result) { @@ -380,7 +359,6 @@ async function runTask( 'Task completed', ); } catch (err) { - if (closeTimer) clearTimeout(closeTimer); error = err instanceof Error ? err.message : String(err); logger.error({ taskId: task.id, error }, 'Task failed'); }