claude-bot/EJClaw
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: make runners single-turn

Browse Source
This commit is contained in:
Eyejoker
2026-03-23 04:51:54 +09:00
parent 83d7bf3d51
commit 02408aae8e
3 changed files with 39 additions and 166 deletions
Download Patch File Download Diff File Expand all files Collapse all files

85
runners/agent-runner/src/index.ts
View File

@@ -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<string | null> {
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<string, string | undefined>,
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<void> {
@@ -740,49 +712,24 @@ async function main(): Promise<void> {
}
// --- 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);
const queryResult = await runQuery(
prompt,
sessionId,
mcpServerPath,
containerInput,
sdkEnv,
);
if (queryResult.newSessionId) {
sessionId = queryResult.newSessionId;
}
if (queryResult.lastAssistantUuid) {
resumeAt = queryResult.lastAssistantUuid;
}
// 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);

58
runners/codex-runner/src/index.ts
View File

@@ -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<string | null> {
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,34 +345,9 @@ 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 ${turnCount}/${MAX_TURNS}...`);
log('Starting app-server turn...');
const { result, error } = await executeAppServerTurn(client, threadId, prompt);
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({
@@ -409,16 +365,8 @@ async function runAppServerSession(
});
}
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();

22
src/task-scheduler.ts
View File

@@ -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<typeof setTimeout> | 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');
}