refactor: split Claude agent runner entrypoint (#200)
This commit is contained in:
Eyejoker
10
runners/agent-runner/src/claude-cli.ts
Normal file
10
runners/agent-runner/src/claude-cli.ts
Normal file
@@ -0,0 +1,10 @@
|
||||
import { resolveBundledClaudeCodeExecutable } from './bundled-cli-path.js';
|
||||
|
||||
let cachedClaudeCliPath: string | null = null;
|
||||
|
||||
export function getClaudeCliPath(log: (message: string) => void): string {
|
||||
if (cachedClaudeCliPath) return cachedClaudeCliPath;
|
||||
cachedClaudeCliPath = resolveBundledClaudeCodeExecutable();
|
||||
log(`Resolved bundled Claude Code CLI: ${cachedClaudeCliPath}`);
|
||||
return cachedClaudeCliPath;
|
||||
}
|
||||
596
runners/agent-runner/src/claude-query-runner.ts
Normal file
596
runners/agent-runner/src/claude-query-runner.ts
Normal file
@@ -0,0 +1,596 @@
|
||||
import { query } from '@anthropic-ai/claude-agent-sdk';
|
||||
import { EJCLAW_ENV, IPC_POLL_MS } from 'ejclaw-runners-shared';
|
||||
|
||||
import { compactBoundaryFromMessage } from './compaction-boundary.js';
|
||||
import { getClaudeCliPath } from './claude-cli.js';
|
||||
import { drainIpcInput, shouldClose } from './ipc-input.js';
|
||||
import { buildEjclawMcpServerConfig } from './mcp-config.js';
|
||||
import {
|
||||
MessageStream,
|
||||
buildCompactionOutput,
|
||||
buildMultimodalContent,
|
||||
extractAssistantText,
|
||||
normalizeStructuredOutput,
|
||||
type RunnerCompaction,
|
||||
writeOutput,
|
||||
} from './output-protocol.js';
|
||||
import { buildClaudeReadonlySandboxSettings } from './reviewer-runtime.js';
|
||||
import {
|
||||
createPreCompactHook,
|
||||
createReviewerBashGuardHook,
|
||||
createSanitizeBashHook,
|
||||
} from './runner-hooks.js';
|
||||
import type { RunnerInput } from './runner-input.js';
|
||||
import {
|
||||
TopLevelAgentTaskTracker,
|
||||
buildTaskNotificationOutput,
|
||||
buildTaskProgressOutput,
|
||||
buildTaskStartedOutput,
|
||||
} from './task-progress-mapping.js';
|
||||
|
||||
export interface ClaudeQueryPaths {
|
||||
groupDir: string;
|
||||
groupFolder: string;
|
||||
hostTasksDir: string;
|
||||
ipcInputCloseSentinel: string;
|
||||
ipcInputDir: string;
|
||||
workDir: string;
|
||||
}
|
||||
|
||||
export interface RunClaudeQueryArgs {
|
||||
prompt: string;
|
||||
sessionId: string | undefined;
|
||||
mcpServerPath: string;
|
||||
runnerInput: RunnerInput;
|
||||
sdkEnv: Record<string, string | undefined>;
|
||||
reviewerRuntime: boolean;
|
||||
claudeReadonlyReviewerRuntime: boolean;
|
||||
claudeReadonlySandboxMode: 'strict' | 'best-effort' | null;
|
||||
abortController: AbortController;
|
||||
paths: ClaudeQueryPaths;
|
||||
log: (message: string) => void;
|
||||
}
|
||||
|
||||
export interface ClaudeQueryResult {
|
||||
newSessionId?: string;
|
||||
closedDuringQuery: boolean;
|
||||
terminalResultObserved: boolean;
|
||||
compaction?: RunnerCompaction;
|
||||
}
|
||||
|
||||
class ClaudeQueryRunner {
|
||||
private readonly stream: MessageStream;
|
||||
private readonly trackedAgentTasks = new TopLevelAgentTaskTracker();
|
||||
private readonly extraDirs: string[] = [];
|
||||
private readonly effectiveCwd: string;
|
||||
private ipcPolling = true;
|
||||
private closedDuringQuery = false;
|
||||
private newSessionId: string | undefined;
|
||||
private messageCount = 0;
|
||||
private resultCount = 0;
|
||||
private terminalResultObserved = false;
|
||||
private pendingProgressText: string | null = null;
|
||||
private compaction: RunnerCompaction | undefined;
|
||||
|
||||
constructor(private readonly args: RunClaudeQueryArgs) {
|
||||
this.stream = new MessageStream((text) =>
|
||||
buildMultimodalContent(text, args.log),
|
||||
);
|
||||
this.effectiveCwd = args.paths.workDir || args.paths.groupDir;
|
||||
if (args.paths.workDir && args.paths.workDir !== args.paths.groupDir) {
|
||||
this.extraDirs.push(args.paths.groupDir);
|
||||
args.log(
|
||||
`Work directory override: ${args.paths.workDir} (group dir added to additionalDirectories)`,
|
||||
);
|
||||
}
|
||||
if (this.extraDirs.length > 0) {
|
||||
args.log(`Additional directories: ${this.extraDirs.join(', ')}`);
|
||||
}
|
||||
}
|
||||
|
||||
async run(): Promise<ClaudeQueryResult> {
|
||||
this.stream.push(this.args.prompt);
|
||||
this.startIpcPolling();
|
||||
this.logRuntimeConfig();
|
||||
|
||||
for await (const message of query({
|
||||
prompt: this.stream,
|
||||
options: this.buildQueryOptions(),
|
||||
})) {
|
||||
this.messageCount++;
|
||||
const msgType =
|
||||
typeof message === 'object' &&
|
||||
message !== null &&
|
||||
(message as { type?: string }).type === 'system'
|
||||
? `system/${(message as { subtype?: string }).subtype}`
|
||||
: (message as { type?: string }).type;
|
||||
this.args.log(`[msg #${this.messageCount}] type=${msgType}`);
|
||||
|
||||
if (this.handleMessage(message)) break;
|
||||
}
|
||||
|
||||
this.flushRemainingPendingText();
|
||||
this.ipcPolling = false;
|
||||
this.args.log(
|
||||
`Query done. Messages: ${this.messageCount}, results: ${this.resultCount}, closedDuringQuery: ${this.closedDuringQuery}`,
|
||||
);
|
||||
return {
|
||||
newSessionId: this.newSessionId,
|
||||
closedDuringQuery: this.closedDuringQuery,
|
||||
terminalResultObserved: this.terminalResultObserved,
|
||||
compaction: this.compaction,
|
||||
};
|
||||
}
|
||||
|
||||
private startIpcPolling(): void {
|
||||
const pollIpcDuringQuery = () => {
|
||||
if (!this.ipcPolling) return;
|
||||
if (shouldClose(this.args.paths.ipcInputCloseSentinel)) {
|
||||
this.handleCloseSentinel();
|
||||
return;
|
||||
}
|
||||
const messages = drainIpcInput(
|
||||
this.args.paths.ipcInputDir,
|
||||
this.args.log,
|
||||
);
|
||||
for (const text of messages) {
|
||||
this.args.log(
|
||||
`Piping IPC message into active query (${text.length} chars)`,
|
||||
);
|
||||
this.stream.push(text);
|
||||
}
|
||||
setTimeout(pollIpcDuringQuery, IPC_POLL_MS);
|
||||
};
|
||||
setTimeout(pollIpcDuringQuery, IPC_POLL_MS);
|
||||
}
|
||||
|
||||
private handleCloseSentinel(): void {
|
||||
this.args.log('Close sentinel detected during query, ending stream');
|
||||
if (this.pendingProgressText && !this.terminalResultObserved) {
|
||||
this.args.log(
|
||||
`Flushing pending text before close (${this.pendingProgressText.length} chars)`,
|
||||
);
|
||||
writeOutput({
|
||||
status: 'success',
|
||||
...normalizeStructuredOutput(this.pendingProgressText),
|
||||
newSessionId: this.newSessionId,
|
||||
});
|
||||
this.pendingProgressText = null;
|
||||
this.terminalResultObserved = true;
|
||||
this.resultCount++;
|
||||
}
|
||||
this.closedDuringQuery = true;
|
||||
this.stream.end();
|
||||
this.ipcPolling = false;
|
||||
}
|
||||
|
||||
private buildQueryOptions() {
|
||||
const readonlyReviewerRuntime =
|
||||
this.args.reviewerRuntime || this.args.claudeReadonlyReviewerRuntime;
|
||||
const readonlyProtectedPaths = [
|
||||
this.effectiveCwd,
|
||||
...this.extraDirs,
|
||||
].filter((value): value is string => Boolean(value));
|
||||
const claudeReadonlySandboxSettings =
|
||||
this.args.claudeReadonlyReviewerRuntime &&
|
||||
this.args.claudeReadonlySandboxMode
|
||||
? buildClaudeReadonlySandboxSettings(
|
||||
readonlyProtectedPaths,
|
||||
undefined,
|
||||
this.args.claudeReadonlySandboxMode,
|
||||
)
|
||||
: undefined;
|
||||
|
||||
return {
|
||||
pathToClaudeCodeExecutable: getClaudeCliPath(this.args.log),
|
||||
cwd: this.effectiveCwd,
|
||||
model: this.resolveModel(),
|
||||
thinking: this.resolveThinking(),
|
||||
effort: this.resolveEffort(),
|
||||
additionalDirectories:
|
||||
this.extraDirs.length > 0 ? this.extraDirs : undefined,
|
||||
resume: this.args.sessionId,
|
||||
allowedTools: this.buildAllowedTools(readonlyReviewerRuntime),
|
||||
env: this.args.sdkEnv,
|
||||
permissionMode: 'bypassPermissions' as const,
|
||||
allowDangerouslySkipPermissions: true,
|
||||
...(claudeReadonlySandboxSettings
|
||||
? { sandbox: claudeReadonlySandboxSettings }
|
||||
: {}),
|
||||
settingSources: ['project', 'user'] as ('project' | 'user')[],
|
||||
abortController: this.args.abortController,
|
||||
mcpServers: {
|
||||
ejclaw: buildEjclawMcpServerConfig(this.args.mcpServerPath, {
|
||||
chatJid: this.args.runnerInput.chatJid,
|
||||
groupFolder: this.args.runnerInput.groupFolder,
|
||||
isMain: this.args.runnerInput.isMain,
|
||||
agentType: process.env[EJCLAW_ENV.agentType] || 'claude-code',
|
||||
roomRole: this.args.runnerInput.roomRoleContext?.role || '',
|
||||
ipcDir: process.env[EJCLAW_ENV.ipcDir],
|
||||
hostIpcDir: process.env[EJCLAW_ENV.hostIpcDir],
|
||||
}),
|
||||
},
|
||||
hooks: {
|
||||
PreCompact: [
|
||||
{
|
||||
hooks: [
|
||||
createPreCompactHook({
|
||||
assistantName: this.args.runnerInput.assistantName,
|
||||
groupDir: this.args.paths.groupDir,
|
||||
groupFolder: this.args.paths.groupFolder,
|
||||
hostTasksDir: this.args.paths.hostTasksDir,
|
||||
log: this.args.log,
|
||||
writeOutput,
|
||||
}),
|
||||
],
|
||||
},
|
||||
],
|
||||
PreToolUse: [
|
||||
{
|
||||
matcher: 'Bash',
|
||||
hooks: readonlyReviewerRuntime
|
||||
? [createReviewerBashGuardHook(), createSanitizeBashHook()]
|
||||
: [createSanitizeBashHook()],
|
||||
},
|
||||
],
|
||||
},
|
||||
agentProgressSummaries: true,
|
||||
};
|
||||
}
|
||||
|
||||
private logRuntimeConfig(): void {
|
||||
const model = this.resolveModel();
|
||||
const thinking = this.resolveThinking();
|
||||
const effort = this.resolveEffort();
|
||||
if (model) this.args.log(`Using model: ${model}`);
|
||||
if (thinking) this.args.log(`Thinking config: ${JSON.stringify(thinking)}`);
|
||||
if (effort) this.args.log(`Effort: ${effort}`);
|
||||
if (this.args.reviewerRuntime) {
|
||||
this.args.log('Reviewer runtime restrictions enabled');
|
||||
}
|
||||
if (this.args.claudeReadonlyReviewerRuntime) {
|
||||
this.args.log(
|
||||
`Claude host reviewer read-only sandbox enabled (${this.args.claudeReadonlySandboxMode || 'unknown'})`,
|
||||
);
|
||||
}
|
||||
if (this.args.claudeReadonlySandboxMode === 'best-effort') {
|
||||
this.args.log(
|
||||
'Claude host reviewer sandbox capability unavailable on this host, using best-effort read-only mode',
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
private resolveModel(): string | undefined {
|
||||
return process.env.CLAUDE_MODEL || undefined;
|
||||
}
|
||||
|
||||
private resolveThinking():
|
||||
| { type: 'adaptive' }
|
||||
| { type: 'enabled'; budgetTokens?: number }
|
||||
| { type: 'disabled' }
|
||||
| undefined {
|
||||
const thinkingType = process.env.CLAUDE_THINKING || undefined;
|
||||
const thinkingBudget = process.env.CLAUDE_THINKING_BUDGET
|
||||
? parseInt(process.env.CLAUDE_THINKING_BUDGET, 10)
|
||||
: undefined;
|
||||
return thinkingType === 'adaptive'
|
||||
? { type: 'adaptive' }
|
||||
: thinkingType === 'enabled'
|
||||
? { type: 'enabled', budgetTokens: thinkingBudget }
|
||||
: thinkingType === 'disabled'
|
||||
? { type: 'disabled' }
|
||||
: undefined;
|
||||
}
|
||||
|
||||
private resolveEffort(): 'low' | 'medium' | 'high' | 'max' | undefined {
|
||||
return (
|
||||
(process.env.CLAUDE_EFFORT as 'low' | 'medium' | 'high' | 'max') ||
|
||||
undefined
|
||||
);
|
||||
}
|
||||
|
||||
private buildAllowedTools(readonlyRuntime: boolean): string[] {
|
||||
const readonlyTools = [
|
||||
'Bash',
|
||||
'Read',
|
||||
'Glob',
|
||||
'Grep',
|
||||
'WebSearch',
|
||||
'WebFetch',
|
||||
'Task',
|
||||
'TaskOutput',
|
||||
'TaskStop',
|
||||
'TeamCreate',
|
||||
'TeamDelete',
|
||||
'SendMessage',
|
||||
'TodoWrite',
|
||||
'ToolSearch',
|
||||
'Skill',
|
||||
'mcp__ejclaw__*',
|
||||
];
|
||||
const writeTools = [
|
||||
'Bash',
|
||||
'Read',
|
||||
'Write',
|
||||
'Edit',
|
||||
'Glob',
|
||||
'Grep',
|
||||
'WebSearch',
|
||||
'WebFetch',
|
||||
'Task',
|
||||
'TaskOutput',
|
||||
'TaskStop',
|
||||
'TeamCreate',
|
||||
'TeamDelete',
|
||||
'SendMessage',
|
||||
'TodoWrite',
|
||||
'ToolSearch',
|
||||
'Skill',
|
||||
'NotebookEdit',
|
||||
'mcp__ejclaw__*',
|
||||
];
|
||||
return readonlyRuntime ? readonlyTools : writeTools;
|
||||
}
|
||||
|
||||
private handleMessage(message: unknown): boolean {
|
||||
const type = (message as { type?: string }).type;
|
||||
if (type !== 'assistant' && this.pendingProgressText) {
|
||||
this.flushPendingProgressAsIntermediate();
|
||||
}
|
||||
if (type === 'system') this.handleSystemMessage(message);
|
||||
if (type === 'tool_progress') this.handleToolProgress(message);
|
||||
if (type === 'tool_use_summary') this.handleToolUseSummary(message);
|
||||
if (type === 'result') return this.handleResultMessage(message);
|
||||
if (type === 'assistant') return this.handleAssistantMessage(message);
|
||||
return false;
|
||||
}
|
||||
|
||||
private handleSystemMessage(message: unknown): void {
|
||||
const subtype = (message as { subtype?: string }).subtype;
|
||||
if (subtype === 'init') {
|
||||
this.newSessionId = (message as { session_id?: string }).session_id;
|
||||
this.args.log(`Session initialized: ${this.newSessionId}`);
|
||||
}
|
||||
|
||||
this.compaction =
|
||||
compactBoundaryFromMessage(message, this.args.log) ?? this.compaction;
|
||||
|
||||
if (subtype === 'task_notification') {
|
||||
const tn = message as {
|
||||
task_id: string;
|
||||
tool_use_id?: string;
|
||||
status: string;
|
||||
summary: string;
|
||||
};
|
||||
this.args.log(
|
||||
`Task notification: task=${tn.task_id} status=${tn.status} summary=${tn.summary}`,
|
||||
);
|
||||
const mapped = buildTaskNotificationOutput(
|
||||
this.trackedAgentTasks,
|
||||
tn,
|
||||
this.newSessionId,
|
||||
);
|
||||
if (mapped) writeOutput(mapped);
|
||||
}
|
||||
|
||||
if (subtype === 'task_progress') {
|
||||
const tp = message as Record<string, unknown>;
|
||||
const description =
|
||||
typeof tp.description === 'string' ? tp.description : '';
|
||||
const mapped = buildTaskProgressOutput(
|
||||
this.trackedAgentTasks,
|
||||
tp,
|
||||
this.newSessionId,
|
||||
);
|
||||
if (mapped) {
|
||||
writeOutput(mapped);
|
||||
} else if (description) {
|
||||
this.args.log(
|
||||
`Skipping long task_progress description (${description.length} chars)`,
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
if (subtype === 'task_started') {
|
||||
const ts = message as { task_id: string; description?: string };
|
||||
const desc = ts.description || '';
|
||||
this.args.log(
|
||||
`Subagent started: task=${ts.task_id} desc=${desc.slice(0, 200)}`,
|
||||
);
|
||||
const mapped = buildTaskStartedOutput(
|
||||
this.trackedAgentTasks,
|
||||
ts,
|
||||
this.newSessionId,
|
||||
);
|
||||
if (mapped) writeOutput(mapped);
|
||||
}
|
||||
}
|
||||
|
||||
private handleToolProgress(message: unknown): void {
|
||||
const tp = message as {
|
||||
tool_name: string;
|
||||
elapsed_time_seconds: number;
|
||||
};
|
||||
const label = `${tp.tool_name} (${Math.round(tp.elapsed_time_seconds)}s)`;
|
||||
this.args.log(`Tool progress: ${label}`);
|
||||
writeOutput({
|
||||
status: 'success',
|
||||
phase: 'progress',
|
||||
...normalizeStructuredOutput(label),
|
||||
newSessionId: this.newSessionId,
|
||||
});
|
||||
}
|
||||
|
||||
private handleToolUseSummary(message: unknown): void {
|
||||
const ts = message as { summary: string };
|
||||
this.args.log(`Tool use summary: ${ts.summary.slice(0, 200)}`);
|
||||
writeOutput({
|
||||
status: 'success',
|
||||
phase: 'progress',
|
||||
...normalizeStructuredOutput(ts.summary),
|
||||
newSessionId: this.newSessionId,
|
||||
});
|
||||
}
|
||||
|
||||
private handleResultMessage(message: unknown): boolean {
|
||||
const resultMessage = message as {
|
||||
subtype?: string;
|
||||
result?: string;
|
||||
errors?: unknown;
|
||||
stop_reason?: unknown;
|
||||
duration_ms?: unknown;
|
||||
duration_api_ms?: unknown;
|
||||
session_id?: unknown;
|
||||
};
|
||||
this.resultCount++;
|
||||
let textResult = resultMessage.result || null;
|
||||
const isError = resultMessage.subtype?.startsWith('error');
|
||||
|
||||
if (this.pendingProgressText && textResult === this.pendingProgressText) {
|
||||
this.args.log('Discarding pending progress (matches result)');
|
||||
this.pendingProgressText = null;
|
||||
} else if (this.pendingProgressText) {
|
||||
if (!textResult) {
|
||||
this.args.log(
|
||||
`Promoting pending progress text to result (${this.pendingProgressText.length} chars)`,
|
||||
);
|
||||
textResult = this.pendingProgressText;
|
||||
} else {
|
||||
writeOutput({
|
||||
status: 'success',
|
||||
phase: 'intermediate',
|
||||
result: this.pendingProgressText,
|
||||
newSessionId: this.newSessionId,
|
||||
});
|
||||
}
|
||||
this.pendingProgressText = null;
|
||||
}
|
||||
|
||||
this.args.log(
|
||||
`Result #${this.resultCount}: subtype=${resultMessage.subtype}${textResult ? ` text=${textResult.slice(0, 200)}` : ''}`,
|
||||
);
|
||||
if (isError) {
|
||||
this.writeErrorResult(resultMessage, textResult);
|
||||
} else {
|
||||
this.writeSuccessResult(textResult);
|
||||
}
|
||||
this.terminalResultObserved = true;
|
||||
this.ipcPolling = false;
|
||||
this.stream.end();
|
||||
this.args.log('Terminal result observed, ending query stream');
|
||||
return true;
|
||||
}
|
||||
|
||||
private writeErrorResult(
|
||||
message: {
|
||||
subtype?: string;
|
||||
errors?: unknown;
|
||||
stop_reason?: unknown;
|
||||
duration_ms?: unknown;
|
||||
duration_api_ms?: unknown;
|
||||
session_id?: unknown;
|
||||
},
|
||||
textResult: string | null,
|
||||
): void {
|
||||
const sdkErrors = Array.isArray(message.errors)
|
||||
? (message.errors as string[])
|
||||
: [];
|
||||
this.args.log(
|
||||
`Error result detail: ${JSON.stringify({
|
||||
subtype: message.subtype,
|
||||
result: textResult?.slice(0, 500),
|
||||
errors: sdkErrors,
|
||||
stop_reason: message.stop_reason,
|
||||
duration_ms: message.duration_ms,
|
||||
duration_api_ms: message.duration_api_ms,
|
||||
session_id: message.session_id,
|
||||
})}`,
|
||||
);
|
||||
writeOutput({
|
||||
status: 'error',
|
||||
result: textResult || null,
|
||||
newSessionId: this.newSessionId,
|
||||
error:
|
||||
sdkErrors.length > 0
|
||||
? sdkErrors.join('; ')
|
||||
: `Agent error: ${message.subtype}`,
|
||||
...buildCompactionOutput(this.compaction),
|
||||
});
|
||||
}
|
||||
|
||||
private writeSuccessResult(textResult: string | null): void {
|
||||
writeOutput({
|
||||
status: 'success',
|
||||
...normalizeStructuredOutput(textResult || null),
|
||||
newSessionId: this.newSessionId,
|
||||
...buildCompactionOutput(this.compaction),
|
||||
});
|
||||
}
|
||||
|
||||
private handleAssistantMessage(message: unknown): boolean {
|
||||
this.trackedAgentTasks.rememberAssistantMessage(message);
|
||||
const stopReason = (message as { stop_reason?: string }).stop_reason;
|
||||
const textResult = extractAssistantText(message);
|
||||
if (textResult || stopReason === 'end_turn') {
|
||||
this.args.log(
|
||||
`Assistant: stop=${stopReason} text=${textResult ? textResult.length + ' chars' : 'null'}`,
|
||||
);
|
||||
}
|
||||
if (stopReason === 'end_turn' && textResult) {
|
||||
this.resultCount++;
|
||||
this.args.log(
|
||||
`Terminal assistant turn observed without result event (${textResult.length} chars), ending query stream`,
|
||||
);
|
||||
writeOutput({
|
||||
status: 'success',
|
||||
...normalizeStructuredOutput(textResult),
|
||||
newSessionId: this.newSessionId,
|
||||
...buildCompactionOutput(this.compaction),
|
||||
});
|
||||
this.terminalResultObserved = true;
|
||||
this.ipcPolling = false;
|
||||
this.stream.end();
|
||||
return true;
|
||||
}
|
||||
if (stopReason !== 'end_turn' && textResult) {
|
||||
if (this.pendingProgressText) this.flushPendingProgressAsIntermediate();
|
||||
this.pendingProgressText = textResult;
|
||||
this.args.log(
|
||||
`Intermediate assistant text buffered (${textResult.length} chars, stop=${stopReason})`,
|
||||
);
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
private flushPendingProgressAsIntermediate(): void {
|
||||
if (!this.pendingProgressText) return;
|
||||
writeOutput({
|
||||
status: 'success',
|
||||
phase: 'intermediate',
|
||||
...normalizeStructuredOutput(this.pendingProgressText),
|
||||
newSessionId: this.newSessionId,
|
||||
});
|
||||
this.pendingProgressText = null;
|
||||
}
|
||||
|
||||
private flushRemainingPendingText(): void {
|
||||
if (!this.pendingProgressText || this.terminalResultObserved) return;
|
||||
this.args.log(
|
||||
`Flushing remaining pending progress text as final output (${this.pendingProgressText.length} chars)`,
|
||||
);
|
||||
writeOutput({
|
||||
status: 'success',
|
||||
...normalizeStructuredOutput(this.pendingProgressText),
|
||||
newSessionId: this.newSessionId,
|
||||
...buildCompactionOutput(this.compaction),
|
||||
});
|
||||
this.terminalResultObserved = true;
|
||||
this.resultCount++;
|
||||
}
|
||||
}
|
||||
|
||||
export function runClaudeQuery(
|
||||
args: RunClaudeQueryArgs,
|
||||
): Promise<ClaudeQueryResult> {
|
||||
return new ClaudeQueryRunner(args).run();
|
||||
}
|
||||
29
runners/agent-runner/src/compaction-boundary.ts
Normal file
29
runners/agent-runner/src/compaction-boundary.ts
Normal file
@@ -0,0 +1,29 @@
|
||||
import {
|
||||
compactBoundaryOutput,
|
||||
type RunnerCompaction,
|
||||
} from './output-protocol.js';
|
||||
|
||||
export function compactBoundaryFromMessage(
|
||||
message: unknown,
|
||||
log: (message: string) => void,
|
||||
): RunnerCompaction | undefined {
|
||||
if (
|
||||
!(
|
||||
typeof message === 'object' &&
|
||||
message !== null &&
|
||||
(message as { type?: string }).type === 'system' &&
|
||||
(message as { subtype?: string }).subtype === 'compact_boundary'
|
||||
)
|
||||
) {
|
||||
return undefined;
|
||||
}
|
||||
const meta = (
|
||||
message as {
|
||||
compact_metadata?: { trigger?: string; pre_tokens?: number };
|
||||
}
|
||||
).compact_metadata;
|
||||
log(
|
||||
`Compact boundary — trigger=${meta?.trigger || '?'} pre_tokens=${meta?.pre_tokens ?? '?'}`,
|
||||
);
|
||||
return compactBoundaryOutput(meta?.trigger);
|
||||
}
|
||||
@@ -16,23 +16,16 @@
|
||||
|
||||
import fs from 'fs';
|
||||
import path from 'path';
|
||||
import { query } from '@anthropic-ai/claude-agent-sdk';
|
||||
import {
|
||||
EJCLAW_ENV,
|
||||
IPC_CLOSE_SENTINEL,
|
||||
IPC_INPUT_SUBDIR,
|
||||
IPC_POLL_MS,
|
||||
} from 'ejclaw-runners-shared';
|
||||
import { fileURLToPath } from 'url';
|
||||
|
||||
import { resolveBundledClaudeCodeExecutable } from './bundled-cli-path.js';
|
||||
import {
|
||||
prependRoomRoleHeader,
|
||||
type RoomRoleContext,
|
||||
} from './room-role-context.js';
|
||||
import { prependRoomRoleHeader } from './room-role-context.js';
|
||||
import {
|
||||
assertReadonlyWorkspaceRepoConnectivity,
|
||||
buildClaudeReadonlySandboxSettings,
|
||||
buildReviewerGitGuardEnv,
|
||||
getClaudeReadonlySandboxMode,
|
||||
isArbiterRuntimeEnvEnabled,
|
||||
@@ -40,43 +33,16 @@ import {
|
||||
isReviewerRuntime,
|
||||
isReviewerRuntimeEnvEnabled,
|
||||
} from './reviewer-runtime.js';
|
||||
import { drainIpcInput, shouldClose } from './ipc-input.js';
|
||||
import { drainIpcInput } from './ipc-input.js';
|
||||
import {
|
||||
MessageStream,
|
||||
buildCompactionOutput,
|
||||
buildMultimodalContent,
|
||||
compactBoundaryOutput,
|
||||
extractAssistantText,
|
||||
normalizeStructuredOutput,
|
||||
readStdin,
|
||||
type RunnerCompaction,
|
||||
writeOutput,
|
||||
} from './output-protocol.js';
|
||||
import {
|
||||
TopLevelAgentTaskTracker,
|
||||
buildTaskNotificationOutput,
|
||||
buildTaskProgressOutput,
|
||||
buildTaskStartedOutput,
|
||||
} from './task-progress-mapping.js';
|
||||
import {
|
||||
createPreCompactHook,
|
||||
createReviewerBashGuardHook,
|
||||
createSanitizeBashHook,
|
||||
} from './runner-hooks.js';
|
||||
import { buildClaudeSdkEnv } from './sdk-env.js';
|
||||
import { buildEjclawMcpServerConfig } from './mcp-config.js';
|
||||
|
||||
interface RunnerInput {
|
||||
prompt: string;
|
||||
sessionId?: string;
|
||||
groupFolder: string;
|
||||
chatJid: string;
|
||||
isMain: boolean;
|
||||
isScheduledTask?: boolean;
|
||||
assistantName?: string;
|
||||
secrets?: Record<string, string>;
|
||||
roomRoleContext?: RoomRoleContext;
|
||||
}
|
||||
import { runSessionCommand } from './session-command.js';
|
||||
import { runClaudeQuery } from './claude-query-runner.js';
|
||||
import type { RunnerInput } from './runner-input.js';
|
||||
|
||||
// Paths configurable via env vars.
|
||||
const GROUP_DIR = process.env[EJCLAW_ENV.groupDir] || '/workspace/group';
|
||||
@@ -94,41 +60,6 @@ function log(message: string): void {
|
||||
console.error(`[agent-runner] ${message}`);
|
||||
}
|
||||
|
||||
function compactBoundaryFromMessage(
|
||||
message: unknown,
|
||||
): RunnerCompaction | undefined {
|
||||
if (
|
||||
!(
|
||||
typeof message === 'object' &&
|
||||
message !== null &&
|
||||
(message as { type?: string }).type === 'system' &&
|
||||
(message as { subtype?: string }).subtype === 'compact_boundary'
|
||||
)
|
||||
) {
|
||||
return undefined;
|
||||
}
|
||||
const meta = (
|
||||
message as {
|
||||
compact_metadata?: { trigger?: string; pre_tokens?: number };
|
||||
}
|
||||
).compact_metadata;
|
||||
log(
|
||||
`Compact boundary — trigger=${meta?.trigger || '?'} pre_tokens=${meta?.pre_tokens ?? '?'}`,
|
||||
);
|
||||
return compactBoundaryOutput(meta?.trigger);
|
||||
}
|
||||
|
||||
// 번들 CLI binary를 명시해야 SDK가 musl/glibc 잘못 탐색하는 걸 우회함.
|
||||
// (SDK 0.2.114의 W7() 헬퍼는 linux-x64-musl를 linux-x64보다 먼저 시도하므로
|
||||
// glibc 호스트에서 musl 패키지가 빈 껍데기로 설치돼 있으면 실패한다.)
|
||||
let cachedClaudeCliPath: string | null = null;
|
||||
function getClaudeCliPath(): string {
|
||||
if (cachedClaudeCliPath) return cachedClaudeCliPath;
|
||||
cachedClaudeCliPath = resolveBundledClaudeCodeExecutable();
|
||||
log(`Resolved bundled Claude Code CLI: ${cachedClaudeCliPath}`);
|
||||
return cachedClaudeCliPath;
|
||||
}
|
||||
|
||||
// Graceful shutdown: SIGTERM → abort SDK query, allowing cleanup
|
||||
const agentAbortController = new AbortController();
|
||||
process.on('SIGTERM', () => {
|
||||
@@ -136,509 +67,6 @@ process.on('SIGTERM', () => {
|
||||
agentAbortController.abort();
|
||||
});
|
||||
|
||||
/**
|
||||
* Run a single query and stream results via writeOutput.
|
||||
* Uses MessageStream (AsyncIterable) to keep isSingleUserTurn=false,
|
||||
* allowing agent teams subagents to run to completion.
|
||||
* Also pipes IPC messages into the stream during the query.
|
||||
*/
|
||||
async function runQuery(
|
||||
prompt: string,
|
||||
sessionId: string | undefined,
|
||||
mcpServerPath: string,
|
||||
runnerInput: RunnerInput,
|
||||
sdkEnv: Record<string, string | undefined>,
|
||||
reviewerRuntime: boolean,
|
||||
claudeReadonlyReviewerRuntime: boolean,
|
||||
claudeReadonlySandboxMode: 'strict' | 'best-effort' | null,
|
||||
): Promise<{
|
||||
newSessionId?: string;
|
||||
closedDuringQuery: boolean;
|
||||
terminalResultObserved: boolean;
|
||||
compaction?: RunnerCompaction;
|
||||
}> {
|
||||
const stream = new MessageStream((text) => buildMultimodalContent(text, log));
|
||||
stream.push(prompt);
|
||||
|
||||
// Poll IPC for follow-up messages and _close sentinel during the query
|
||||
let ipcPolling = true;
|
||||
let closedDuringQuery = false;
|
||||
const pollIpcDuringQuery = () => {
|
||||
if (!ipcPolling) return;
|
||||
if (shouldClose(IPC_INPUT_CLOSE_SENTINEL)) {
|
||||
log('Close sentinel detected during query, ending stream');
|
||||
// Flush any buffered text before closing — the for-await loop may not
|
||||
// reach the post-loop flush code after stream.end().
|
||||
if (pendingProgressText && !terminalResultObserved) {
|
||||
log(
|
||||
`Flushing pending text before close (${pendingProgressText.length} chars)`,
|
||||
);
|
||||
writeOutput({
|
||||
status: 'success',
|
||||
...normalizeStructuredOutput(pendingProgressText),
|
||||
newSessionId,
|
||||
});
|
||||
pendingProgressText = null;
|
||||
terminalResultObserved = true;
|
||||
resultCount++;
|
||||
}
|
||||
closedDuringQuery = true;
|
||||
stream.end();
|
||||
ipcPolling = false;
|
||||
return;
|
||||
}
|
||||
const messages = drainIpcInput(IPC_INPUT_DIR, log);
|
||||
for (const text of messages) {
|
||||
log(`Piping IPC message into active query (${text.length} chars)`);
|
||||
stream.push(text);
|
||||
}
|
||||
setTimeout(pollIpcDuringQuery, IPC_POLL_MS);
|
||||
};
|
||||
setTimeout(pollIpcDuringQuery, IPC_POLL_MS);
|
||||
|
||||
let newSessionId: string | undefined;
|
||||
let messageCount = 0;
|
||||
let resultCount = 0;
|
||||
let terminalResultObserved = false;
|
||||
let pendingProgressText: string | null = null;
|
||||
let compaction: RunnerCompaction | undefined;
|
||||
const trackedAgentTasks = new TopLevelAgentTaskTracker();
|
||||
|
||||
// Discover additional directories
|
||||
const extraDirs: string[] = [];
|
||||
|
||||
// When WORK_DIR is set, use it as cwd and include GROUP_DIR as additional directory
|
||||
const effectiveCwd = WORK_DIR || GROUP_DIR;
|
||||
if (WORK_DIR && WORK_DIR !== GROUP_DIR) {
|
||||
extraDirs.push(GROUP_DIR);
|
||||
log(
|
||||
`Work directory override: ${WORK_DIR} (group dir added to additionalDirectories)`,
|
||||
);
|
||||
}
|
||||
if (extraDirs.length > 0) {
|
||||
log(`Additional directories: ${extraDirs.join(', ')}`);
|
||||
}
|
||||
|
||||
// Model and thinking configuration from environment
|
||||
const model = process.env.CLAUDE_MODEL || undefined;
|
||||
const thinkingType = process.env.CLAUDE_THINKING || undefined; // 'adaptive' | 'enabled' | 'disabled'
|
||||
const thinkingBudget = process.env.CLAUDE_THINKING_BUDGET
|
||||
? parseInt(process.env.CLAUDE_THINKING_BUDGET, 10)
|
||||
: undefined;
|
||||
const effort =
|
||||
(process.env.CLAUDE_EFFORT as 'low' | 'medium' | 'high' | 'max') ||
|
||||
undefined;
|
||||
const thinking =
|
||||
thinkingType === 'adaptive'
|
||||
? { type: 'adaptive' as const }
|
||||
: thinkingType === 'enabled'
|
||||
? { type: 'enabled' as const, budgetTokens: thinkingBudget }
|
||||
: thinkingType === 'disabled'
|
||||
? { type: 'disabled' as const }
|
||||
: undefined;
|
||||
|
||||
if (model) log(`Using model: ${model}`);
|
||||
if (thinking) log(`Thinking config: ${JSON.stringify(thinking)}`);
|
||||
if (effort) log(`Effort: ${effort}`);
|
||||
if (reviewerRuntime) log('Reviewer runtime restrictions enabled');
|
||||
if (claudeReadonlyReviewerRuntime) {
|
||||
log(
|
||||
`Claude host reviewer read-only sandbox enabled (${claudeReadonlySandboxMode || 'unknown'})`,
|
||||
);
|
||||
}
|
||||
if (claudeReadonlySandboxMode === 'best-effort') {
|
||||
log(
|
||||
'Claude host reviewer sandbox capability unavailable on this host, using best-effort read-only mode',
|
||||
);
|
||||
}
|
||||
|
||||
const readonlyReviewerRuntime =
|
||||
reviewerRuntime || claudeReadonlyReviewerRuntime;
|
||||
const allowedTools = readonlyReviewerRuntime
|
||||
? [
|
||||
'Bash',
|
||||
'Read',
|
||||
'Glob',
|
||||
'Grep',
|
||||
'WebSearch',
|
||||
'WebFetch',
|
||||
'Task',
|
||||
'TaskOutput',
|
||||
'TaskStop',
|
||||
'TeamCreate',
|
||||
'TeamDelete',
|
||||
'SendMessage',
|
||||
'TodoWrite',
|
||||
'ToolSearch',
|
||||
'Skill',
|
||||
'mcp__ejclaw__*',
|
||||
]
|
||||
: [
|
||||
'Bash',
|
||||
'Read',
|
||||
'Write',
|
||||
'Edit',
|
||||
'Glob',
|
||||
'Grep',
|
||||
'WebSearch',
|
||||
'WebFetch',
|
||||
'Task',
|
||||
'TaskOutput',
|
||||
'TaskStop',
|
||||
'TeamCreate',
|
||||
'TeamDelete',
|
||||
'SendMessage',
|
||||
'TodoWrite',
|
||||
'ToolSearch',
|
||||
'Skill',
|
||||
'NotebookEdit',
|
||||
'mcp__ejclaw__*',
|
||||
];
|
||||
|
||||
const readonlyProtectedPaths = [effectiveCwd, ...extraDirs].filter(
|
||||
(value): value is string => Boolean(value),
|
||||
);
|
||||
const claudeReadonlySandboxSettings =
|
||||
claudeReadonlyReviewerRuntime && claudeReadonlySandboxMode
|
||||
? buildClaudeReadonlySandboxSettings(
|
||||
readonlyProtectedPaths,
|
||||
undefined,
|
||||
claudeReadonlySandboxMode,
|
||||
)
|
||||
: undefined;
|
||||
|
||||
for await (const message of query({
|
||||
prompt: stream,
|
||||
options: {
|
||||
// 번들 CLI binary를 명시해야 SDK가 musl/glibc 잘못 탐색하는 걸 우회함.
|
||||
pathToClaudeCodeExecutable: getClaudeCliPath(),
|
||||
cwd: effectiveCwd,
|
||||
model,
|
||||
thinking,
|
||||
effort,
|
||||
additionalDirectories: extraDirs.length > 0 ? extraDirs : undefined,
|
||||
resume: sessionId,
|
||||
allowedTools,
|
||||
env: sdkEnv,
|
||||
permissionMode: 'bypassPermissions',
|
||||
allowDangerouslySkipPermissions: true,
|
||||
...(claudeReadonlySandboxSettings
|
||||
? {
|
||||
sandbox: claudeReadonlySandboxSettings,
|
||||
}
|
||||
: {}),
|
||||
settingSources: ['project', 'user'],
|
||||
abortController: agentAbortController,
|
||||
mcpServers: {
|
||||
ejclaw: buildEjclawMcpServerConfig(mcpServerPath, {
|
||||
chatJid: runnerInput.chatJid,
|
||||
groupFolder: runnerInput.groupFolder,
|
||||
isMain: runnerInput.isMain,
|
||||
agentType: process.env[EJCLAW_ENV.agentType] || 'claude-code',
|
||||
roomRole: runnerInput.roomRoleContext?.role || '',
|
||||
ipcDir: process.env[EJCLAW_ENV.ipcDir],
|
||||
hostIpcDir: process.env[EJCLAW_ENV.hostIpcDir],
|
||||
}),
|
||||
},
|
||||
hooks: {
|
||||
PreCompact: [
|
||||
{
|
||||
hooks: [
|
||||
createPreCompactHook({
|
||||
assistantName: runnerInput.assistantName,
|
||||
groupDir: GROUP_DIR,
|
||||
groupFolder: GROUP_FOLDER,
|
||||
hostTasksDir: HOST_TASKS_DIR,
|
||||
log,
|
||||
writeOutput,
|
||||
}),
|
||||
],
|
||||
},
|
||||
],
|
||||
PreToolUse: [
|
||||
{
|
||||
matcher: 'Bash',
|
||||
hooks: readonlyReviewerRuntime
|
||||
? [createReviewerBashGuardHook(), createSanitizeBashHook()]
|
||||
: [createSanitizeBashHook()],
|
||||
},
|
||||
],
|
||||
},
|
||||
agentProgressSummaries: true,
|
||||
},
|
||||
})) {
|
||||
messageCount++;
|
||||
const msgType =
|
||||
message.type === 'system'
|
||||
? `system/${(message as { subtype?: string }).subtype}`
|
||||
: message.type;
|
||||
log(`[msg #${messageCount}] type=${msgType}`);
|
||||
|
||||
// Flush pending intermediate text as a regular message on non-assistant events.
|
||||
if (message.type !== 'assistant' && pendingProgressText) {
|
||||
writeOutput({
|
||||
status: 'success',
|
||||
phase: 'intermediate',
|
||||
...normalizeStructuredOutput(pendingProgressText),
|
||||
newSessionId,
|
||||
});
|
||||
pendingProgressText = null;
|
||||
}
|
||||
|
||||
if (message.type === 'system' && message.subtype === 'init') {
|
||||
newSessionId = message.session_id;
|
||||
log(`Session initialized: ${newSessionId}`);
|
||||
}
|
||||
|
||||
compaction = compactBoundaryFromMessage(message) ?? compaction;
|
||||
|
||||
if (
|
||||
message.type === 'system' &&
|
||||
(message as { subtype?: string }).subtype === 'task_notification'
|
||||
) {
|
||||
const tn = message as {
|
||||
task_id: string;
|
||||
tool_use_id?: string;
|
||||
status: string;
|
||||
summary: string;
|
||||
};
|
||||
log(
|
||||
`Task notification: task=${tn.task_id} status=${tn.status} summary=${tn.summary}`,
|
||||
);
|
||||
const mapped = buildTaskNotificationOutput(
|
||||
trackedAgentTasks,
|
||||
tn,
|
||||
newSessionId,
|
||||
);
|
||||
if (mapped) {
|
||||
writeOutput(mapped);
|
||||
}
|
||||
}
|
||||
|
||||
if (
|
||||
message.type === 'system' &&
|
||||
(message as { subtype?: string }).subtype === 'task_progress'
|
||||
) {
|
||||
const tp = message as Record<string, unknown>;
|
||||
const description =
|
||||
typeof tp.description === 'string' ? tp.description : '';
|
||||
const mapped = buildTaskProgressOutput(
|
||||
trackedAgentTasks,
|
||||
tp,
|
||||
newSessionId,
|
||||
);
|
||||
if (mapped) {
|
||||
writeOutput(mapped);
|
||||
} else if (description) {
|
||||
// Long AI summary → skip (too long for progress sub-line)
|
||||
log(
|
||||
`Skipping long task_progress description (${description.length} chars)`,
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
if (
|
||||
message.type === 'system' &&
|
||||
(message as { subtype?: string }).subtype === 'task_started'
|
||||
) {
|
||||
const ts = message as { task_id: string; description?: string };
|
||||
const desc = ts.description || '';
|
||||
log(`Subagent started: task=${ts.task_id} desc=${desc.slice(0, 200)}`);
|
||||
const mapped = buildTaskStartedOutput(
|
||||
trackedAgentTasks,
|
||||
ts,
|
||||
newSessionId,
|
||||
);
|
||||
if (mapped) {
|
||||
writeOutput(mapped);
|
||||
}
|
||||
}
|
||||
|
||||
if (message.type === 'tool_progress') {
|
||||
const tp = message as {
|
||||
tool_name: string;
|
||||
elapsed_time_seconds: number;
|
||||
};
|
||||
const label = `${tp.tool_name} (${Math.round(tp.elapsed_time_seconds)}s)`;
|
||||
log(`Tool progress: ${label}`);
|
||||
const normalized = normalizeStructuredOutput(label);
|
||||
writeOutput({
|
||||
status: 'success',
|
||||
phase: 'progress',
|
||||
...normalized,
|
||||
newSessionId,
|
||||
});
|
||||
}
|
||||
|
||||
if (message.type === 'tool_use_summary') {
|
||||
const ts = message as { summary: string };
|
||||
log(`Tool use summary: ${ts.summary.slice(0, 200)}`);
|
||||
const normalized = normalizeStructuredOutput(ts.summary);
|
||||
writeOutput({
|
||||
status: 'success',
|
||||
phase: 'progress',
|
||||
...normalized,
|
||||
newSessionId,
|
||||
});
|
||||
}
|
||||
|
||||
if (message.type === 'result') {
|
||||
resultCount++;
|
||||
let textResult =
|
||||
'result' in message ? (message as { result?: string }).result : null;
|
||||
const isError = message.subtype?.startsWith('error');
|
||||
// Discard pending progress if it matches the final result (prevent duplicate)
|
||||
if (
|
||||
pendingProgressText &&
|
||||
textResult &&
|
||||
pendingProgressText === textResult
|
||||
) {
|
||||
log(`Discarding pending progress (matches result)`);
|
||||
pendingProgressText = null;
|
||||
} else if (pendingProgressText) {
|
||||
// If the result has no text, promote pending progress to the result
|
||||
// so it gets delivered as the final output instead of being lost.
|
||||
if (!textResult) {
|
||||
log(
|
||||
`Promoting pending progress text to result (${pendingProgressText.length} chars)`,
|
||||
);
|
||||
textResult = pendingProgressText;
|
||||
} else {
|
||||
writeOutput({
|
||||
status: 'success',
|
||||
phase: 'intermediate',
|
||||
result: pendingProgressText,
|
||||
newSessionId,
|
||||
});
|
||||
}
|
||||
pendingProgressText = null;
|
||||
}
|
||||
log(
|
||||
`Result #${resultCount}: subtype=${message.subtype}${textResult ? ` text=${textResult.slice(0, 200)}` : ''}`,
|
||||
);
|
||||
if (isError) {
|
||||
// Log full error details for debugging
|
||||
const msg = message as Record<string, unknown>;
|
||||
const sdkErrors = Array.isArray(msg.errors)
|
||||
? (msg.errors as string[])
|
||||
: [];
|
||||
const errorDetail = JSON.stringify({
|
||||
subtype: message.subtype,
|
||||
result: textResult?.slice(0, 500),
|
||||
errors: sdkErrors,
|
||||
stop_reason: msg.stop_reason,
|
||||
duration_ms: msg.duration_ms,
|
||||
duration_api_ms: msg.duration_api_ms,
|
||||
session_id: msg.session_id,
|
||||
});
|
||||
log(`Error result detail: ${errorDetail}`);
|
||||
// Pass SDK errors through so host can detect session issues
|
||||
const errorText =
|
||||
sdkErrors.length > 0 ? sdkErrors.join('; ') : undefined;
|
||||
writeOutput({
|
||||
status: 'error',
|
||||
result: textResult || null,
|
||||
newSessionId,
|
||||
error: errorText || `Agent error: ${message.subtype}`,
|
||||
...buildCompactionOutput(compaction),
|
||||
});
|
||||
} else {
|
||||
const normalized = normalizeStructuredOutput(textResult || null);
|
||||
writeOutput({
|
||||
status: 'success',
|
||||
...normalized,
|
||||
newSessionId,
|
||||
...buildCompactionOutput(compaction),
|
||||
});
|
||||
}
|
||||
|
||||
// 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;
|
||||
}
|
||||
|
||||
if (message.type === 'assistant') {
|
||||
trackedAgentTasks.rememberAssistantMessage(message);
|
||||
const stopReason = (message as { stop_reason?: string }).stop_reason;
|
||||
const textResult = extractAssistantText(message);
|
||||
// Only log when there's something interesting (text or terminal)
|
||||
if (textResult || stopReason === 'end_turn') {
|
||||
log(
|
||||
`Assistant: stop=${stopReason} text=${textResult ? textResult.length + ' chars' : 'null'}`,
|
||||
);
|
||||
}
|
||||
if (stopReason === 'end_turn' && textResult) {
|
||||
resultCount++;
|
||||
log(
|
||||
`Terminal assistant turn observed without result event (${textResult.length} chars), ending query stream`,
|
||||
);
|
||||
writeOutput({
|
||||
status: 'success',
|
||||
...normalizeStructuredOutput(textResult),
|
||||
newSessionId,
|
||||
...buildCompactionOutput(compaction),
|
||||
});
|
||||
terminalResultObserved = true;
|
||||
ipcPolling = false;
|
||||
stream.end();
|
||||
break;
|
||||
}
|
||||
// Intermediate assistant text between tool calls → buffer as pending progress.
|
||||
// Don't emit immediately — if the next message is a result with the same text,
|
||||
// this would cause a duplicate. The pending text is flushed when the next
|
||||
// non-result message arrives, or discarded if result matches.
|
||||
if (stopReason !== 'end_turn' && textResult) {
|
||||
// Flush previous pending as a regular message (not progress heading)
|
||||
if (pendingProgressText) {
|
||||
const normalized = normalizeStructuredOutput(pendingProgressText);
|
||||
writeOutput({
|
||||
status: 'success',
|
||||
phase: 'intermediate',
|
||||
...normalized,
|
||||
newSessionId,
|
||||
});
|
||||
}
|
||||
pendingProgressText = textResult;
|
||||
log(
|
||||
`Intermediate assistant text buffered (${textResult.length} chars, stop=${stopReason})`,
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Flush any remaining buffered text that was never followed by a result event.
|
||||
// This happens when the agent produces a short response without a formal end_turn.
|
||||
if (pendingProgressText && !terminalResultObserved) {
|
||||
log(
|
||||
`Flushing remaining pending progress text as final output (${pendingProgressText.length} chars)`,
|
||||
);
|
||||
writeOutput({
|
||||
status: 'success',
|
||||
...normalizeStructuredOutput(pendingProgressText),
|
||||
newSessionId,
|
||||
...buildCompactionOutput(compaction),
|
||||
});
|
||||
terminalResultObserved = true;
|
||||
resultCount++;
|
||||
}
|
||||
|
||||
ipcPolling = false;
|
||||
log(
|
||||
`Query done. Messages: ${messageCount}, results: ${resultCount}, closedDuringQuery: ${closedDuringQuery}`,
|
||||
);
|
||||
return {
|
||||
newSessionId,
|
||||
closedDuringQuery,
|
||||
terminalResultObserved,
|
||||
compaction,
|
||||
};
|
||||
}
|
||||
|
||||
async function main(): Promise<void> {
|
||||
let runnerInput: RunnerInput;
|
||||
|
||||
@@ -723,126 +151,18 @@ async function main(): Promise<void> {
|
||||
|
||||
if (isSessionSlashCommand) {
|
||||
log(`Handling session command: ${trimmedPrompt}`);
|
||||
let slashSessionId: string | undefined;
|
||||
let compactBoundarySeen = false;
|
||||
let slashCompaction: RunnerCompaction | undefined;
|
||||
let hadError = false;
|
||||
let resultEmitted = false;
|
||||
|
||||
try {
|
||||
for await (const message of query({
|
||||
prompt: trimmedPrompt,
|
||||
options: {
|
||||
// 번들 CLI binary를 명시해야 SDK가 musl/glibc 잘못 탐색하는 걸 우회함.
|
||||
pathToClaudeCodeExecutable: getClaudeCliPath(),
|
||||
cwd: mainEffectiveCwd,
|
||||
resume: sessionId,
|
||||
systemPrompt: undefined,
|
||||
allowedTools: [],
|
||||
env: guardedSdkEnv,
|
||||
permissionMode: 'bypassPermissions' as const,
|
||||
allowDangerouslySkipPermissions: true,
|
||||
settingSources: ['project', 'user'] as const,
|
||||
abortController: agentAbortController,
|
||||
hooks: {
|
||||
PreCompact: [
|
||||
{
|
||||
hooks: [
|
||||
createPreCompactHook({
|
||||
assistantName: runnerInput.assistantName,
|
||||
groupDir: GROUP_DIR,
|
||||
groupFolder: GROUP_FOLDER,
|
||||
hostTasksDir: HOST_TASKS_DIR,
|
||||
log,
|
||||
writeOutput,
|
||||
}),
|
||||
],
|
||||
},
|
||||
],
|
||||
},
|
||||
},
|
||||
})) {
|
||||
const msgType =
|
||||
message.type === 'system'
|
||||
? `system/${(message as { subtype?: string }).subtype}`
|
||||
: message.type;
|
||||
log(`[slash-cmd] type=${msgType}`);
|
||||
|
||||
if (message.type === 'system' && message.subtype === 'init') {
|
||||
slashSessionId = message.session_id;
|
||||
log(`Session after slash command: ${slashSessionId}`);
|
||||
}
|
||||
|
||||
const observedCompaction = compactBoundaryFromMessage(message);
|
||||
if (observedCompaction) {
|
||||
compactBoundarySeen = true;
|
||||
slashCompaction = observedCompaction;
|
||||
}
|
||||
|
||||
if (message.type === 'result') {
|
||||
const resultSubtype = (message as { subtype?: string }).subtype;
|
||||
const textResult =
|
||||
'result' in message
|
||||
? (message as { result?: string }).result
|
||||
: null;
|
||||
|
||||
if (resultSubtype?.startsWith('error')) {
|
||||
hadError = true;
|
||||
writeOutput({
|
||||
status: 'error',
|
||||
result: null,
|
||||
error: textResult || 'Session command failed.',
|
||||
newSessionId: slashSessionId,
|
||||
...buildCompactionOutput(slashCompaction),
|
||||
});
|
||||
} else {
|
||||
writeOutput({
|
||||
status: 'success',
|
||||
result: textResult || 'Conversation compacted.',
|
||||
newSessionId: slashSessionId,
|
||||
...buildCompactionOutput(slashCompaction),
|
||||
});
|
||||
}
|
||||
resultEmitted = true;
|
||||
}
|
||||
}
|
||||
} catch (err) {
|
||||
hadError = true;
|
||||
const errorMsg = err instanceof Error ? err.message : String(err);
|
||||
log(`Slash command error: ${errorMsg}`);
|
||||
writeOutput({ status: 'error', result: null, error: errorMsg });
|
||||
}
|
||||
|
||||
log(
|
||||
`Slash command done. compactBoundarySeen=${compactBoundarySeen}, hadError=${hadError}`,
|
||||
);
|
||||
|
||||
// Warn if compact_boundary was never observed — compaction may not have occurred
|
||||
if (!hadError && !compactBoundarySeen) {
|
||||
log(
|
||||
'WARNING: compact_boundary was not observed. Compaction may not have completed.',
|
||||
);
|
||||
}
|
||||
|
||||
// Only emit final session marker if no result was emitted yet and no error occurred
|
||||
if (!resultEmitted && !hadError) {
|
||||
writeOutput({
|
||||
status: 'success',
|
||||
result: compactBoundarySeen
|
||||
? 'Conversation compacted.'
|
||||
: 'Compaction requested but compact_boundary was not observed.',
|
||||
newSessionId: slashSessionId,
|
||||
...buildCompactionOutput(slashCompaction),
|
||||
});
|
||||
} else if (!hadError) {
|
||||
// Emit session-only marker so host updates session tracking
|
||||
writeOutput({
|
||||
status: 'success',
|
||||
result: null,
|
||||
newSessionId: slashSessionId,
|
||||
...buildCompactionOutput(slashCompaction),
|
||||
});
|
||||
}
|
||||
await runSessionCommand({
|
||||
prompt: trimmedPrompt,
|
||||
sessionId,
|
||||
cwd: mainEffectiveCwd,
|
||||
sdkEnv: guardedSdkEnv,
|
||||
abortController: agentAbortController,
|
||||
assistantName: runnerInput.assistantName,
|
||||
groupDir: GROUP_DIR,
|
||||
groupFolder: GROUP_FOLDER,
|
||||
hostTasksDir: HOST_TASKS_DIR,
|
||||
log,
|
||||
});
|
||||
return;
|
||||
}
|
||||
// --- End slash command handling ---
|
||||
@@ -850,16 +170,26 @@ async function main(): Promise<void> {
|
||||
try {
|
||||
log(`Starting query (session: ${sessionId || 'new'})...`);
|
||||
|
||||
const queryResult = await runQuery(
|
||||
const queryResult = await runClaudeQuery({
|
||||
prompt,
|
||||
sessionId,
|
||||
mcpServerPath,
|
||||
runnerInput,
|
||||
guardedSdkEnv,
|
||||
sdkEnv: guardedSdkEnv,
|
||||
reviewerRuntime,
|
||||
claudeReadonlyReviewerRuntime,
|
||||
claudeReadonlySandboxMode,
|
||||
);
|
||||
abortController: agentAbortController,
|
||||
paths: {
|
||||
groupDir: GROUP_DIR,
|
||||
groupFolder: GROUP_FOLDER,
|
||||
hostTasksDir: HOST_TASKS_DIR,
|
||||
ipcInputCloseSentinel: IPC_INPUT_CLOSE_SENTINEL,
|
||||
ipcInputDir: IPC_INPUT_DIR,
|
||||
workDir: WORK_DIR,
|
||||
},
|
||||
log,
|
||||
});
|
||||
if (queryResult.newSessionId) {
|
||||
sessionId = queryResult.newSessionId;
|
||||
}
|
||||
|
||||
13
runners/agent-runner/src/runner-input.ts
Normal file
13
runners/agent-runner/src/runner-input.ts
Normal file
@@ -0,0 +1,13 @@
|
||||
import type { RoomRoleContext } from './room-role-context.js';
|
||||
|
||||
export interface RunnerInput {
|
||||
prompt: string;
|
||||
sessionId?: string;
|
||||
groupFolder: string;
|
||||
chatJid: string;
|
||||
isMain: boolean;
|
||||
isScheduledTask?: boolean;
|
||||
assistantName?: string;
|
||||
secrets?: Record<string, string>;
|
||||
roomRoleContext?: RoomRoleContext;
|
||||
}
|
||||
151
runners/agent-runner/src/session-command.ts
Normal file
151
runners/agent-runner/src/session-command.ts
Normal file
@@ -0,0 +1,151 @@
|
||||
import { query } from '@anthropic-ai/claude-agent-sdk';
|
||||
|
||||
import { compactBoundaryFromMessage } from './compaction-boundary.js';
|
||||
import { getClaudeCliPath } from './claude-cli.js';
|
||||
import { createPreCompactHook } from './runner-hooks.js';
|
||||
import {
|
||||
buildCompactionOutput,
|
||||
type RunnerCompaction,
|
||||
writeOutput,
|
||||
} from './output-protocol.js';
|
||||
|
||||
export interface RunSessionCommandArgs {
|
||||
prompt: string;
|
||||
sessionId?: string;
|
||||
cwd: string;
|
||||
sdkEnv: Record<string, string | undefined>;
|
||||
abortController: AbortController;
|
||||
assistantName?: string;
|
||||
groupDir: string;
|
||||
groupFolder: string;
|
||||
hostTasksDir: string;
|
||||
log: (message: string) => void;
|
||||
}
|
||||
|
||||
export async function runSessionCommand({
|
||||
prompt,
|
||||
sessionId,
|
||||
cwd,
|
||||
sdkEnv,
|
||||
abortController,
|
||||
assistantName,
|
||||
groupDir,
|
||||
groupFolder,
|
||||
hostTasksDir,
|
||||
log,
|
||||
}: RunSessionCommandArgs): Promise<void> {
|
||||
let slashSessionId: string | undefined;
|
||||
let compactBoundarySeen = false;
|
||||
let slashCompaction: RunnerCompaction | undefined;
|
||||
let hadError = false;
|
||||
let resultEmitted = false;
|
||||
|
||||
try {
|
||||
for await (const message of query({
|
||||
prompt,
|
||||
options: {
|
||||
pathToClaudeCodeExecutable: getClaudeCliPath(log),
|
||||
cwd,
|
||||
resume: sessionId,
|
||||
systemPrompt: undefined,
|
||||
allowedTools: [],
|
||||
env: sdkEnv,
|
||||
permissionMode: 'bypassPermissions' as const,
|
||||
allowDangerouslySkipPermissions: true,
|
||||
settingSources: ['project', 'user'] as const,
|
||||
abortController,
|
||||
hooks: {
|
||||
PreCompact: [
|
||||
{
|
||||
hooks: [
|
||||
createPreCompactHook({
|
||||
assistantName,
|
||||
groupDir,
|
||||
groupFolder,
|
||||
hostTasksDir,
|
||||
log,
|
||||
writeOutput,
|
||||
}),
|
||||
],
|
||||
},
|
||||
],
|
||||
},
|
||||
},
|
||||
})) {
|
||||
const msgType =
|
||||
message.type === 'system'
|
||||
? `system/${(message as { subtype?: string }).subtype}`
|
||||
: message.type;
|
||||
log(`[slash-cmd] type=${msgType}`);
|
||||
|
||||
if (message.type === 'system' && message.subtype === 'init') {
|
||||
slashSessionId = message.session_id;
|
||||
log(`Session after slash command: ${slashSessionId}`);
|
||||
}
|
||||
|
||||
const observedCompaction = compactBoundaryFromMessage(message, log);
|
||||
if (observedCompaction) {
|
||||
compactBoundarySeen = true;
|
||||
slashCompaction = observedCompaction;
|
||||
}
|
||||
|
||||
if (message.type === 'result') {
|
||||
const resultSubtype = (message as { subtype?: string }).subtype;
|
||||
const textResult =
|
||||
'result' in message ? (message as { result?: string }).result : null;
|
||||
|
||||
if (resultSubtype?.startsWith('error')) {
|
||||
hadError = true;
|
||||
writeOutput({
|
||||
status: 'error',
|
||||
result: null,
|
||||
error: textResult || 'Session command failed.',
|
||||
newSessionId: slashSessionId,
|
||||
...buildCompactionOutput(slashCompaction),
|
||||
});
|
||||
} else {
|
||||
writeOutput({
|
||||
status: 'success',
|
||||
result: textResult || 'Conversation compacted.',
|
||||
newSessionId: slashSessionId,
|
||||
...buildCompactionOutput(slashCompaction),
|
||||
});
|
||||
}
|
||||
resultEmitted = true;
|
||||
}
|
||||
}
|
||||
} catch (err) {
|
||||
hadError = true;
|
||||
const errorMsg = err instanceof Error ? err.message : String(err);
|
||||
log(`Slash command error: ${errorMsg}`);
|
||||
writeOutput({ status: 'error', result: null, error: errorMsg });
|
||||
}
|
||||
|
||||
log(
|
||||
`Slash command done. compactBoundarySeen=${compactBoundarySeen}, hadError=${hadError}`,
|
||||
);
|
||||
|
||||
if (!hadError && !compactBoundarySeen) {
|
||||
log(
|
||||
'WARNING: compact_boundary was not observed. Compaction may not have completed.',
|
||||
);
|
||||
}
|
||||
|
||||
if (!resultEmitted && !hadError) {
|
||||
writeOutput({
|
||||
status: 'success',
|
||||
result: compactBoundarySeen
|
||||
? 'Conversation compacted.'
|
||||
: 'Compaction requested but compact_boundary was not observed.',
|
||||
newSessionId: slashSessionId,
|
||||
...buildCompactionOutput(slashCompaction),
|
||||
});
|
||||
} else if (!hadError) {
|
||||
writeOutput({
|
||||
status: 'success',
|
||||
result: null,
|
||||
newSessionId: slashSessionId,
|
||||
...buildCompactionOutput(slashCompaction),
|
||||
});
|
||||
}
|
||||
}
|
||||
35
runners/agent-runner/test/compaction-boundary.test.ts
Normal file
35
runners/agent-runner/test/compaction-boundary.test.ts
Normal file
@@ -0,0 +1,35 @@
|
||||
import { describe, expect, it, vi } from 'vitest';
|
||||
|
||||
import { compactBoundaryFromMessage } from '../src/compaction-boundary.js';
|
||||
|
||||
describe('compactBoundaryFromMessage', () => {
|
||||
it('extracts compact boundary metadata from SDK system messages', () => {
|
||||
const log = vi.fn();
|
||||
|
||||
const compaction = compactBoundaryFromMessage(
|
||||
{
|
||||
type: 'system',
|
||||
subtype: 'compact_boundary',
|
||||
compact_metadata: {
|
||||
trigger: 'manual',
|
||||
pre_tokens: 1234,
|
||||
},
|
||||
},
|
||||
log,
|
||||
);
|
||||
|
||||
expect(compaction).toEqual({
|
||||
completed: true,
|
||||
trigger: 'manual',
|
||||
});
|
||||
expect(log).toHaveBeenCalledWith(
|
||||
'Compact boundary — trigger=manual pre_tokens=1234',
|
||||
);
|
||||
});
|
||||
|
||||
it('ignores unrelated SDK messages', () => {
|
||||
expect(
|
||||
compactBoundaryFromMessage({ type: 'assistant' }, vi.fn()),
|
||||
).toBeUndefined();
|
||||
});
|
||||
});
|
||||
Reference in New Issue
Block a user