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

refactor: extract pending paired turn helpers

Browse Source
This commit is contained in:
Eyejoker
2026-03-31 05:43:17 +09:00
parent 4fc7a88e86
commit 43f8341c17
2 changed files with 281 additions and 149 deletions
Download Patch File Download Diff File Expand all files Collapse all files

118
src/message-runtime.test.ts
View File

@@ -117,6 +117,7 @@ vi.mock('./db.js', () => {
getPendingServiceHandoffs: vi.fn(() => []),
getLatestOpenPairedTaskForChat: vi.fn(() => undefined),
getPairedTurnOutputs: vi.fn(() => []),
getRecentChatMessages: vi.fn(() => []),
createProducedWorkItem: vi.fn((input) => ({
id: 1,
group_folder: input.group_folder,
@@ -211,6 +212,7 @@ describe('createMessageRuntime', () => {
vi.resetAllMocks();
vi.mocked(db.getLastBotFinalMessage).mockReturnValue([]);
vi.mocked(db.isPairedRoomJid).mockReturnValue(false);
vi.mocked(db.getRecentChatMessages).mockReturnValue([]);
vi.mocked(config.isClaudeService).mockReturnValue(true);
vi.mocked(config.isReviewService).mockReturnValue(false);
});
@@ -748,6 +750,122 @@ describe('createMessageRuntime', () => {
expect(channel.sendMessage).toHaveBeenCalledWith(chatJid, '최종 정리 완료');
});
it('reuses the shared arbiter prompt builder for pending arbiter turns', async () => {
const chatJid = 'group@test';
const group = makeGroup('codex');
const channel = makeChannel(chatJid);
vi.mocked(db.isPairedRoomJid).mockReturnValue(true);
vi.mocked(db.getLatestOpenPairedTaskForChat).mockReturnValue({
id: 'task-arbiter',
chat_jid: chatJid,
group_folder: group.folder,
owner_service_id: 'claude',
reviewer_service_id: 'codex-main',
title: null,
source_ref: 'HEAD',
plan_notes: null,
review_requested_at: '2026-03-30T00:00:00.000Z',
round_trip_count: 3,
status: 'arbiter_requested',
arbiter_verdict: null,
arbiter_requested_at: '2026-03-30T00:00:10.000Z',
completion_reason: null,
created_at: '2026-03-30T00:00:00.000Z',
updated_at: '2026-03-30T00:00:10.000Z',
});
vi.mocked(db.getPairedTurnOutputs).mockReturnValue([
{
id: 1,
task_id: 'task-arbiter',
turn_number: 1,
role: 'owner',
output_text: 'owner 산출물',
created_at: '2026-03-30T00:00:01.000Z',
},
{
id: 2,
task_id: 'task-arbiter',
turn_number: 2,
role: 'reviewer',
output_text: 'reviewer 이견',
created_at: '2026-03-30T00:00:02.000Z',
},
]);
vi.mocked(db.getRecentChatMessages).mockReturnValue([
{
id: 'human-1',
chat_jid: chatJid,
sender: 'user@test',
sender_name: 'User',
content: '추가 맥락',
timestamp: '2026-03-30T00:00:00.500Z',
is_bot_message: false,
} as any,
{
id: 'bot-1',
chat_jid: chatJid,
sender: 'bot@test',
sender_name: 'Bot',
content: 'bot progress',
timestamp: '2026-03-30T00:00:03.000Z',
is_bot_message: true,
} as any,
]);
vi.mocked(agentRunner.runAgentProcess).mockImplementation(
async (_group, input, _onProcess, onOutput) => {
expect(input.prompt).toContain('<task-id>task-arbiter</task-id>');
expect(input.prompt).toContain('<round-trips>3</round-trips>');
expect(input.prompt).toContain('추가 맥락');
expect(input.prompt).toContain('owner 산출물');
expect(input.prompt).toContain('reviewer 이견');
expect(input.prompt).not.toContain('bot progress');
await onOutput?.({
status: 'success',
phase: 'final',
result: 'arbiter 확인 완료',
newSessionId: 'session-arbiter-pending',
});
return {
status: 'success',
result: 'arbiter 확인 완료',
newSessionId: 'session-arbiter-pending',
};
},
);
const runtime = createMessageRuntime({
assistantName: 'Andy',
idleTimeout: 1_000,
pollInterval: 1_000,
timezone: 'UTC',
triggerPattern: /^@Andy\b/i,
channels: [channel],
queue: {
registerProcess: vi.fn(),
closeStdin: vi.fn(),
notifyIdle: vi.fn(),
} as any,
getRegisteredGroups: () => ({ [chatJid]: group }),
getSessions: () => ({}),
getLastTimestamp: () => '',
setLastTimestamp: vi.fn(),
getLastAgentTimestamps: () => ({}),
saveState: vi.fn(),
persistSession: vi.fn(),
clearSession: vi.fn(),
});
const result = await runtime.processGroupMessages(chatJid, {
runId: 'run-arbiter-pending-shared-prompt',
reason: 'messages',
});
expect(result).toBe(true);
expect(agentRunner.runAgentProcess).toHaveBeenCalledTimes(1);
expect(channel.sendMessage).toHaveBeenCalledWith(chatJid, 'arbiter 확인 완료');
});
it('allows follow-up messages without a trigger after a visible reply in non-main groups', async () => {
const chatJid = 'group@test';
const group: RegisteredGroup = {

312
src/message-runtime.ts
View File

@@ -56,7 +56,13 @@ import {
isSessionCommandAllowed,
isSessionCommandControlMessage,
} from './session-commands.js';
import { Channel, NewMessage, RegisteredGroup } from './types.js';
import {
Channel,
NewMessage,
RegisteredGroup,
type PairedTask,
type PairedTurnOutput,
} from './types.js';
import { logger } from './logger.js';
import { resolveGroupIpcPath } from './group-folder.js';
import { getEffectiveChannelLease } from './service-routing.js';
@@ -163,7 +169,7 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): {
/** Convert paired turn outputs to NewMessage format for formatMessages(). */
const turnOutputsToMessages = (
outputs: import('./types.js').PairedTurnOutput[],
outputs: PairedTurnOutput[],
chatJid: string,
): NewMessage[] =>
outputs.map((t) => ({
@@ -177,6 +183,15 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): {
is_from_me: false as const,
}));
const mergeHumanAndTurnOutputMessages = (
chatJid: string,
humanMessages: NewMessage[],
turnOutputs: PairedTurnOutput[],
): NewMessage[] =>
[...humanMessages, ...turnOutputsToMessages(turnOutputs, chatJid)].sort(
(a, b) => a.timestamp.localeCompare(b.timestamp),
);
/**
* Build a prompt from paired_turn_outputs (Discord-independent) + human messages.
* Falls back to the legacy labelPairedSenders path when no turn outputs exist.
@@ -198,14 +213,72 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): {
// Human messages from the missed messages (exclude bot messages)
const humanMessages = missedMessages.filter((m) => !m.is_bot_message);
const agentMessages = turnOutputsToMessages(turnOutputs, chatJid);
// Merge human + agent messages chronologically
const allMessages = [...humanMessages, ...agentMessages].sort((a, b) =>
a.timestamp.localeCompare(b.timestamp),
return formatMessages(
mergeHumanAndTurnOutputMessages(chatJid, humanMessages, turnOutputs),
timezone,
);
};
return formatMessages(allMessages, timezone);
const buildReviewerPendingPrompt = (
task: PairedTask,
chatJid: string,
timezone: string,
): string => {
const turnOutputs = getPairedTurnOutputs(task.id);
if (turnOutputs.length > 0) {
const humanMessages = getRecentChatMessages(chatJid, 20).filter(
(message) => !message.is_bot_message,
);
return formatMessages(
mergeHumanAndTurnOutputMessages(chatJid, humanMessages, turnOutputs),
timezone,
);
}
const userMessage = getLastHumanMessageContent(chatJid);
if (!userMessage) {
return 'Review the latest owner changes in the workspace.';
}
return `User request:\n---\n${userMessage}\n---\n\nReview the latest owner changes in the workspace.`;
};
const buildArbiterPromptForTask = (
task: PairedTask,
chatJid: string,
timezone: string,
): string => {
const turnOutputs = getPairedTurnOutputs(task.id);
const recentMessages = getRecentChatMessages(chatJid, 20);
const arbiterMessages =
turnOutputs.length > 0
? mergeHumanAndTurnOutputMessages(
chatJid,
recentMessages.filter((message) => !message.is_bot_message),
turnOutputs,
)
: labelPairedSenders(chatJid, recentMessages);
return buildArbiterContextPrompt({
chatJid,
taskId: task.id,
roundTripCount: task.round_trip_count,
timezone,
messages: arbiterMessages,
});
};
const buildFinalizePendingPrompt = (task: PairedTask): string => {
const turnOutputs = getPairedTurnOutputs(task.id);
const lastReviewerOutput = [...turnOutputs]
.reverse()
.find((output) => output.role === 'reviewer');
const reviewerSummary = lastReviewerOutput?.output_text
? `\n\nReviewer's final assessment:\n${lastReviewerOutput.output_text.slice(0, 2000)}`
: '';
return `The reviewer approved your work (DONE). Finalize and report the result.${reviewerSummary}`;
};
const isBotOnlyPairedRoomTurn = (
@@ -591,6 +664,80 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): {
const resolveChannel = (taskStatus?: string | null): Channel =>
roleToChannel[resolveActiveRole(taskStatus)] ?? channel;
const buildPendingPairedTurn = (
task: PairedTask,
rawMissedMessages: NewMessage[],
):
| {
prompt: string;
channel: Channel;
cursor: string | number | null;
cursorKey?: string;
}
| null => {
const lastRaw = rawMissedMessages[rawMissedMessages.length - 1];
const cursor = lastRaw?.seq ?? lastRaw?.timestamp ?? null;
const taskStatus = task.status;
const pendingRole = resolveActiveRole(taskStatus);
if (pendingRole === 'reviewer') {
return {
prompt: buildReviewerPendingPrompt(task, chatJid, deps.timezone),
channel: resolveChannel(taskStatus),
cursor,
cursorKey: resolveCursorKey(chatJid, taskStatus),
};
}
if (pendingRole === 'arbiter') {
return {
prompt: buildArbiterPromptForTask(task, chatJid, deps.timezone),
channel: resolveChannel(taskStatus),
cursor,
cursorKey: resolveCursorKey(chatJid, taskStatus),
};
}
if (taskStatus === 'merge_ready') {
return {
prompt: buildFinalizePendingPrompt(task),
channel: resolveChannel(taskStatus),
cursor,
};
}
return null;
};
const executePendingPairedTurn = async (args: {
prompt: string;
channel: Channel;
cursor: string | number | null;
cursorKey?: string;
}): Promise<boolean> => {
if (args.cursor != null) {
advanceLastAgentCursor(
deps.getLastAgentTimestamps(),
deps.saveState,
chatJid,
args.cursor,
args.cursorKey,
);
}
const { deliverySucceeded } = await executeTurn({
group,
prompt: args.prompt,
chatJid,
runId,
channel: args.channel,
startSeq: null,
endSeq: null,
});
return deliverySucceeded;
};
if (isPairedRoomJid(chatJid)) {
logger.info(
{
@@ -649,132 +796,11 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): {
const pendingReviewTask = isPairedRoomJid(chatJid)
? getLatestOpenPairedTaskForChat(chatJid)
: null;
const pendingRole = resolveActiveRole(pendingReviewTask?.status);
if (pendingReviewTask && pendingRole === 'reviewer') {
// Build reviewer prompt from turn outputs (Discord-independent).
// Includes full owner/reviewer conversation history + human messages.
const turnOutputs = getPairedTurnOutputs(pendingReviewTask.id);
let reviewPrompt: string;
if (turnOutputs.length > 0) {
const humanMsgs = getRecentChatMessages(chatJid, 20).filter(
(m) => !m.is_bot_message,
);
const allMsgs = [
...humanMsgs,
...turnOutputsToMessages(turnOutputs, chatJid),
].sort((a, b) => a.timestamp.localeCompare(b.timestamp));
reviewPrompt = formatMessages(allMsgs, deps.timezone);
} else {
// Fallback: no turn outputs yet (pre-migration task)
const userMessage = getLastHumanMessageContent(chatJid);
const parts: string[] = [];
if (userMessage) {
parts.push(`User request:\n---\n${userMessage}\n---`);
}
reviewPrompt =
parts.length > 0
? `${parts.join('\n\n')}\n\nReview the latest owner changes in the workspace.`
: 'Review the latest owner changes in the workspace.';
}
// Advance reviewer cursor (not owner cursor) so the reviewer
// still sees the owner's messages on subsequent turns.
const lastRaw = rawMissedMessages[rawMissedMessages.length - 1];
const cursor = lastRaw?.seq ?? lastRaw?.timestamp;
if (cursor != null) {
advanceLastAgentCursor(
deps.getLastAgentTimestamps(),
deps.saveState,
chatJid,
cursor,
resolveCursorKey(chatJid, pendingReviewTask?.status),
);
}
const { deliverySucceeded } = await executeTurn({
group,
prompt: reviewPrompt,
chatJid,
runId,
channel: resolveChannel(pendingReviewTask?.status),
startSeq: null,
endSeq: null,
});
return deliverySucceeded;
}
if (pendingReviewTask && pendingRole === 'arbiter') {
const lastRaw = rawMissedMessages[rawMissedMessages.length - 1];
const cursor = lastRaw?.seq ?? lastRaw?.timestamp;
if (cursor != null) {
advanceLastAgentCursor(
deps.getLastAgentTimestamps(),
deps.saveState,
chatJid,
cursor,
resolveCursorKey(chatJid, pendingReviewTask?.status),
);
}
const arbiterTurnOutputs = getPairedTurnOutputs(pendingReviewTask.id);
const recentMsgs = getRecentChatMessages(chatJid, 20);
const arbiterMessages =
arbiterTurnOutputs.length > 0
? [
...recentMsgs.filter((m) => !m.is_bot_message),
...turnOutputsToMessages(arbiterTurnOutputs, chatJid),
].sort((a, b) => a.timestamp.localeCompare(b.timestamp))
: labelPairedSenders(chatJid, recentMsgs);
const arbiterPrompt = buildArbiterContextPrompt({
chatJid,
taskId: pendingReviewTask.id,
roundTripCount: pendingReviewTask.round_trip_count,
timezone: deps.timezone,
messages: arbiterMessages,
});
const { deliverySucceeded } = await executeTurn({
group,
prompt: arbiterPrompt,
chatJid,
runId,
channel: resolveChannel(pendingReviewTask?.status),
startSeq: null,
endSeq: null,
});
return deliverySucceeded;
}
// merge_ready: reviewer approved, owner gets final turn to finalize
if (pendingReviewTask && pendingReviewTask.status === 'merge_ready') {
const lastRaw = rawMissedMessages[rawMissedMessages.length - 1];
const cursor = lastRaw?.seq ?? lastRaw?.timestamp;
if (cursor != null) {
advanceLastAgentCursor(
deps.getLastAgentTimestamps(),
deps.saveState,
chatJid,
cursor,
);
}
const turnOutputs = getPairedTurnOutputs(pendingReviewTask.id);
const lastReviewerOutput = [...turnOutputs]
.reverse()
.find((output) => output.role === 'reviewer');
const reviewerSummary = lastReviewerOutput?.output_text
? `\n\nReviewer's final assessment:\n${lastReviewerOutput.output_text.slice(0, 2000)}`
: '';
const finalizePrompt = `The reviewer approved your work (DONE). Finalize and report the result.${reviewerSummary}`;
const { deliverySucceeded } = await executeTurn({
group,
prompt: finalizePrompt,
chatJid,
runId,
channel,
startSeq: null,
endSeq: null,
});
return deliverySucceeded;
const pendingTurn = pendingReviewTask
? buildPendingPairedTurn(pendingReviewTask, rawMissedMessages)
: null;
if (pendingTurn) {
return executePendingPairedTurn(pendingTurn);
}
const lastIgnored = rawMissedMessages[rawMissedMessages.length - 1];
@@ -893,23 +919,11 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): {
const turnRole = resolveActiveRole(taskStatus);
let prompt: string;
if (turnRole === 'arbiter' && pendingTaskForChannel) {
// Prefer direct turn outputs; fall back to Discord messages
const turnOutputs = getPairedTurnOutputs(pendingTaskForChannel.id);
const arbiterRecentMsgs = getRecentChatMessages(chatJid, 20);
const arbiterMsgs =
turnOutputs.length > 0
? [
...arbiterRecentMsgs.filter((m) => !m.is_bot_message),
...turnOutputsToMessages(turnOutputs, chatJid),
].sort((a, b) => a.timestamp.localeCompare(b.timestamp))
: labelPairedSenders(chatJid, arbiterRecentMsgs);
prompt = buildArbiterContextPrompt({
prompt = buildArbiterPromptForTask(
pendingTaskForChannel,
chatJid,
taskId: pendingTaskForChannel.id,
roundTripCount: pendingTaskForChannel.round_trip_count,
timezone: deps.timezone,
messages: arbiterMsgs,
});
deps.timezone,
);
} else if (pendingTaskForChannel) {
prompt = buildPairedTurnPrompt(
pendingTaskForChannel.id,