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

Add paired delivery diagnostic logging

Browse Source
This commit is contained in:
Eyejoker
2026-04-04 23:58:57 +09:00
parent 80567de4ed
commit ac225f2b1e
9 changed files with 370 additions and 62 deletions
Download Patch File Download Diff File Expand all files Collapse all files

27
src/channels/discord.test.ts
View File

@@ -113,6 +113,7 @@ vi.mock('discord.js', () => {
});
import { DiscordChannel, DiscordChannelOpts } from './discord.js';
import { logger } from '../logger.js';
// --- Test helpers ---
@@ -815,6 +816,32 @@ describe('DiscordChannel', () => {
flags: 1 << 2,
});
});
it('logs channel name and Discord message ids after sending', async () => {
const opts = createTestOpts();
const channel = new DiscordChannel('test-token', opts);
await channel.connect();
const mockChannel = {
send: vi.fn().mockResolvedValue({ id: 'discord-message-1' }),
sendTyping: vi.fn(),
};
currentClient().channels.fetch.mockResolvedValue(mockChannel);
await channel.sendMessage('dc:1234567890123456', 'Hello');
expect(logger.info).toHaveBeenCalledWith(
expect.objectContaining({
jid: 'dc:1234567890123456',
channelName: 'discord',
deliveryMode: 'send',
chunkCount: 1,
messageId: 'discord-message-1',
messageIds: ['discord-message-1'],
}),
'Discord message sent',
);
});
});
// --- ownsJid ---

62
src/channels/discord.ts
View File

@@ -533,9 +533,19 @@ export class DiscordChannel implements Channel {
attachment: f,
name: path.basename(f),
}));
const sentMessageIds: string[] = [];
let chunkCount = 0;
const recordSentMessage = (message: Message | null | undefined) => {
chunkCount += 1;
if (message?.id) sentMessageIds.push(message.id);
};
if (!cleaned && files.length === 0) {
logger.debug({ jid }, 'Skipping empty Discord outbound message');
logger.debug(
{ jid, channelName: this.name },
'Skipping empty Discord outbound message',
);
return;
}
@@ -547,17 +557,21 @@ export class DiscordChannel implements Channel {
if (cleaned.length <= MAX_LENGTH) {
// Send text with first batch of files
recordSentMessage(
await textChannel.send({
content: cleaned || undefined,
files: fileBatches[0]?.length ? fileBatches[0] : undefined,
flags: MessageFlags.SuppressEmbeds,
});
}),
);
// Send remaining file batches as follow-up messages
for (let b = 1; b < fileBatches.length; b++) {
recordSentMessage(
await textChannel.send({
files: fileBatches[b],
flags: MessageFlags.SuppressEmbeds,
});
}),
);
}
} else {
// Send text in chunks, attach first batch to the first chunk
@@ -565,24 +579,43 @@ export class DiscordChannel implements Channel {
for (let i = 0; i < cleaned.length; i += MAX_LENGTH) {
const chunk = cleaned.slice(i, i + MAX_LENGTH);
const batch = fileBatches[fileBatchIndex];
recordSentMessage(
await textChannel.send({
content: chunk,
files: batch?.length ? batch : undefined,
flags: MessageFlags.SuppressEmbeds,
});
}),
);
if (batch?.length) fileBatchIndex++;
}
// Send any remaining file batches
for (let b = fileBatchIndex; b < fileBatches.length; b++) {
recordSentMessage(
await textChannel.send({
files: fileBatches[b],
flags: MessageFlags.SuppressEmbeds,
});
}),
);
}
}
logger.info({ jid, length: text.length }, 'Discord message sent');
logger.info(
{
jid,
channelName: this.name,
length: text.length,
deliveryMode: 'send',
chunkCount,
attachmentCount: files.length,
messageId: sentMessageIds[0] ?? null,
messageIds: sentMessageIds,
},
'Discord message sent',
);
} catch (err) {
logger.error({ jid, err }, 'Failed to send Discord message');
logger.error(
{ jid, channelName: this.name, err },
'Failed to send Discord message',
);
throw err;
}
}
@@ -780,8 +813,21 @@ export class DiscordChannel implements Channel {
if (!channel || !('messages' in channel)) return;
const msg = await (channel as TextChannel).messages.fetch(messageId);
await msg.edit(text);
logger.info(
{
jid,
channelName: this.name,
deliveryMode: 'edit',
messageId,
length: text.length,
},
'Discord message edited',
);
} catch (err) {
logger.debug({ jid, messageId, err }, 'Failed to edit Discord message');
logger.debug(
{ jid, channelName: this.name, messageId, err },
'Failed to edit Discord message',
);
throw err; // Re-throw so callers (e.g. dashboard) can reset message ID
}
}

20
src/index.ts
View File

@@ -46,9 +46,9 @@ import { startIpcWatcher } from './ipc.js';
import {
findChannel,
findChannelByName,
findChannelForDeliveryRole,
formatOutbound,
normalizeMessageForDedupe,
resolveChannelForDeliveryRole,
} from './router.js';
import {
buildRestartAnnouncement,
@@ -473,9 +473,21 @@ async function main(): Promise<void> {
});
startIpcWatcher({
sendMessage: (jid, text, senderRole) => {
const channel = findChannelForDeliveryRole(channels, jid, senderRole);
if (!channel) throw new Error(`No channel for JID: ${jid}`);
return channel.sendMessage(jid, text);
const route = resolveChannelForDeliveryRole(channels, jid, senderRole);
if (!route.channel) throw new Error(`No channel for JID: ${jid}`);
logger.info(
{
transition: 'ipc:route',
chatJid: jid,
senderRole: senderRole ?? null,
requestedRoleChannel: route.requestedRoleChannelName,
selectedChannel: route.selectedChannelName,
usedRoleChannel: route.usedRoleChannel,
fallbackUsed: route.fallbackUsed,
},
'IPC relay routed message to channel',
);
return route.channel.sendMessage(jid, text);
},
nudgeScheduler: nudgeSchedulerLoop,
registeredGroups: () => registeredGroups,

9
src/ipc-auth.test.ts
View File

@@ -540,7 +540,7 @@ describe('IPC message authorization', () => {
it('forwards senderRole through authorized IPC messages', async () => {
const sendMessage = vi.fn(async () => {});
await forwardAuthorizedIpcMessage(
const result = await forwardAuthorizedIpcMessage(
{
type: 'message',
chatJid: 'other@g.us',
@@ -558,6 +558,12 @@ describe('IPC message authorization', () => {
'review text',
'reviewer',
);
expect(result).toEqual(
expect.objectContaining({
outcome: 'sent',
senderRole: 'reviewer',
}),
);
});
it('does not forward unauthorized IPC messages even when senderRole exists', async () => {
@@ -577,6 +583,7 @@ describe('IPC message authorization', () => {
);
expect(result.outcome).toBe('blocked');
expect(result.senderRole).toBe('reviewer');
expect(sendMessage).not.toHaveBeenCalled();
});
});

7
src/ipc.ts
View File

@@ -63,6 +63,7 @@ export interface IpcMessageForwardResult {
chatJid?: string;
targetGroup?: string | null;
isMainOverride?: boolean;
senderRole?: string | null;
}
export async function forwardAuthorizedIpcMessage(
@@ -73,7 +74,7 @@ export async function forwardAuthorizedIpcMessage(
sendMessage: IpcDeps['sendMessage'],
): Promise<IpcMessageForwardResult> {
if (!(msg.type === 'message' && msg.chatJid && msg.text)) {
return { outcome: 'ignored' };
return { outcome: 'ignored', senderRole: msg.senderRole ?? null };
}
const targetGroup = registeredGroups[msg.chatJid];
@@ -86,6 +87,7 @@ export async function forwardAuthorizedIpcMessage(
chatJid: msg.chatJid,
targetGroup: targetGroup?.folder ?? null,
isMainOverride,
senderRole: msg.senderRole ?? null,
};
}
@@ -95,6 +97,7 @@ export async function forwardAuthorizedIpcMessage(
chatJid: msg.chatJid,
targetGroup: targetGroup?.folder ?? null,
isMainOverride,
senderRole: msg.senderRole ?? null,
};
}
@@ -253,6 +256,7 @@ export function startIpcWatcher(deps: IpcDeps): void {
sourceGroup,
targetGroup: forwardResult.targetGroup ?? null,
isMainOverride: forwardResult.isMainOverride,
senderRole: forwardResult.senderRole ?? null,
},
'IPC message sent',
);
@@ -264,6 +268,7 @@ export function startIpcWatcher(deps: IpcDeps): void {
sourceGroup,
targetGroup: forwardResult.targetGroup ?? null,
isMainOverride: forwardResult.isMainOverride,
senderRole: forwardResult.senderRole ?? null,
},
'Unauthorized IPC message attempt blocked',
);

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

@@ -214,6 +214,7 @@ import {
resolveHandoffRoleOverride,
} from './message-runtime.js';
import * as config from './config.js';
import { logger } from './logger.js';
import * as serviceRouting from './service-routing.js';
import type { Channel, RegisteredGroup } from './types.js';
@@ -628,6 +629,90 @@ describe('createMessageRuntime', () => {
expect(enqueueMessageCheck).toHaveBeenCalledWith(chatJid);
});
it('suppresses duplicate stale work item delivery and logs the suppression reason', async () => {
const chatJid = 'group@test';
const group = makeGroup('codex');
const channel = makeChannel(chatJid);
const enqueueMessageCheck = vi.fn();
vi.mocked(serviceRouting.hasReviewerLease).mockReturnValue(true);
vi.mocked(db.getOpenWorkItem).mockReturnValue({
id: 199,
group_folder: group.folder,
chat_jid: chatJid,
agent_type: 'codex',
service_id: 'claude',
delivery_role: 'owner',
status: 'delivery_retry',
start_seq: 1,
end_seq: 1,
result_payload: '같은 최종 답변입니다.',
delivery_attempts: 1,
created_at: new Date().toISOString(),
updated_at: new Date().toISOString(),
delivered_at: null,
delivery_message_id: null,
last_error: 'discord send failed',
});
vi.mocked(db.getLastBotFinalMessage).mockReturnValue([
{
id: 'last-final',
chat_jid: chatJid,
sender: 'reviewer-bot@test',
sender_name: '리뷰어',
content: '같은 최종 답변입니다.',
timestamp: new Date().toISOString(),
is_from_me: false,
is_bot_message: true,
} as any,
]);
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(),
enqueueMessageCheck,
} 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-open-work-item-duplicate-suppressed',
reason: 'messages',
});
expect(result).toBe(true);
expect(channel.sendMessage).not.toHaveBeenCalled();
expect(db.markWorkItemDelivered).toHaveBeenCalledWith(199, null);
expect(agentRunner.runAgentProcess).not.toHaveBeenCalled();
expect(enqueueMessageCheck).toHaveBeenCalledWith(chatJid);
expect(logger.info).toHaveBeenCalledWith(
expect.objectContaining({
chatJid,
channelName: 'discord',
workItemId: 199,
deliveryRole: 'owner',
suppressionReason: 'paired-final-duplicate',
preview: '같은 최종 답변입니다.',
}),
'Suppressed duplicate final message in paired room (marked as delivered)',
);
});
it('retries a stale reviewer work item when the reviewer channel is missing', async () => {
const chatJid = 'group@test';
const group = makeGroup('codex');
@@ -1325,6 +1410,17 @@ describe('createMessageRuntime', () => {
chatJid,
resolveGroupIpcPath(group.folder),
);
expect(logger.info).toHaveBeenCalledWith(
expect.objectContaining({
chatJid,
taskId: 'task-merge-ready-inline-loop',
taskStatus: 'merge_ready',
handoffMode: 'inline-finalize',
nextRole: 'owner',
cursor: 42,
}),
'Executing merge_ready finalize turn inline after bot-only reviewer follow-up',
);
});
it('reuses the shared arbiter prompt builder for pending arbiter turns', async () => {

75
src/message-runtime.ts
View File

@@ -319,6 +319,18 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): {
return isDuplicateOfLastBotFinal(chatJid, text);
};
const buildDeliveryLogContext = (
channel: Channel,
item: WorkItem,
extra: Record<string, unknown> = {},
): Record<string, unknown> => ({
chatJid: item.chat_jid,
channelName: channel.name,
workItemId: item.id,
deliveryRole: item.delivery_role ?? null,
...extra,
});
const deliverOpenWorkItem = async (
channel: Channel,
item: WorkItem,
@@ -338,11 +350,10 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): {
// Mark as delivered without sending, and don't open continuation
markWorkItemDelivered(item.id, null);
logger.info(
{
chatJid: item.chat_jid,
workItemId: item.id,
buildDeliveryLogContext(channel, item, {
preview: item.result_payload.slice(0, 100),
},
suppressionReason: 'paired-final-duplicate',
}),
'Suppressed duplicate final message in paired room (marked as delivered)',
);
return true;
@@ -350,6 +361,14 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): {
try {
if (replaceMessageId && channel.editMessage) {
logger.info(
buildDeliveryLogContext(channel, item, {
deliveryAttempts: item.delivery_attempts + 1,
deliveryMode: 'edit',
replacedMessageId: replaceMessageId,
}),
'Attempting to deliver produced work item by replacing tracked progress message',
);
await channel.editMessage(
item.chat_jid,
replaceMessageId,
@@ -358,39 +377,43 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): {
markWorkItemDelivered(item.id, replaceMessageId);
continuationTracker.open(item.chat_jid);
logger.info(
{
chatJid: item.chat_jid,
workItemId: item.id,
buildDeliveryLogContext(channel, item, {
deliveryAttempts: item.delivery_attempts + 1,
deliveryMode: 'edit',
replacedMessageId: replaceMessageId,
},
}),
'Delivered produced work item by replacing tracked progress message',
);
return true;
}
} catch (err) {
logger.warn(
{
chatJid: item.chat_jid,
workItemId: item.id,
buildDeliveryLogContext(channel, item, {
deliveryAttempts: item.delivery_attempts + 1,
deliveryMode: 'edit',
replacedMessageId: replaceMessageId,
err,
},
}),
'Failed to replace tracked progress message; falling back to a new message',
);
}
try {
logger.info(
buildDeliveryLogContext(channel, item, {
deliveryAttempts: item.delivery_attempts + 1,
deliveryMode: 'send',
}),
'Attempting to deliver produced work item as a new message',
);
await channel.sendMessage(item.chat_jid, item.result_payload);
markWorkItemDelivered(item.id);
continuationTracker.open(item.chat_jid);
logger.info(
{
chatJid: item.chat_jid,
workItemId: item.id,
buildDeliveryLogContext(channel, item, {
deliveryAttempts: item.delivery_attempts + 1,
},
deliveryMode: 'send',
}),
'Delivered produced work item',
);
return true;
@@ -398,12 +421,11 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): {
const errorMessage = getErrorMessage(err);
markWorkItemDeliveryRetry(item.id, errorMessage);
logger.warn(
{
chatJid: item.chat_jid,
workItemId: item.id,
buildDeliveryLogContext(channel, item, {
deliveryAttempts: item.delivery_attempts + 1,
deliveryMode: 'send',
err,
},
}),
'Failed to deliver produced work item',
);
return false;
@@ -1311,6 +1333,9 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): {
groupFolder: group.folder,
taskId: loopPendingTask?.id ?? null,
taskStatus: loopPendingTask?.status ?? null,
handoffMode: 'inline-finalize',
nextRole: 'owner',
cursor: mergeReadyCursor,
},
'Executing merge_ready finalize turn inline after bot-only reviewer follow-up',
);
@@ -1374,6 +1399,16 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): {
groupFolder: group.folder,
taskId: loopPendingTask?.id ?? null,
taskStatus: loopPendingTask?.status ?? null,
handoffMode: 'requeue',
nextRole:
loopPendingTask?.status === 'review_ready'
? 'reviewer'
: loopPendingTask?.status === 'arbiter_requested' ||
loopPendingTask?.status === 'in_arbitration'
? 'arbiter'
: 'reviewer',
cursor: botOnlyPendingTurn.cursor,
cursorKey: botOnlyPendingTurn.cursorKey,
},
'Queued fresh paired pending turn instead of piping bot-only follow-up into the active agent',
);

42
src/router.test.ts
View File

@@ -1,6 +1,9 @@
import { describe, expect, it } from 'vitest';
import { findChannelForDeliveryRole } from './router.js';
import {
findChannelForDeliveryRole,
resolveChannelForDeliveryRole,
} from './router.js';
import { type Channel } from './types.js';
function createChannel(name: string, ownedJids: string[]): Channel {
@@ -51,4 +54,41 @@ describe('findChannelForDeliveryRole', () => {
ownerChannel,
);
});
it('reports fallback metadata when a role-specific channel is unavailable', () => {
const ownerChannel = createChannel('discord-main', [jid]);
expect(
resolveChannelForDeliveryRole([ownerChannel], jid, 'reviewer'),
).toEqual(
expect.objectContaining({
channel: ownerChannel,
requestedRoleChannelName: 'discord-review',
selectedChannelName: 'discord-main',
usedRoleChannel: false,
fallbackUsed: true,
}),
);
});
it('reports direct role routing metadata when the role channel exists', () => {
const ownerChannel = createChannel('discord-main', [jid]);
const reviewerChannel = createChannel('discord-review', [jid]);
expect(
resolveChannelForDeliveryRole(
[ownerChannel, reviewerChannel],
jid,
'reviewer',
),
).toEqual(
expect.objectContaining({
channel: reviewerChannel,
requestedRoleChannelName: 'discord-review',
selectedChannelName: 'discord-review',
usedRoleChannel: true,
fallbackUsed: false,
}),
);
});
});

66
src/router.ts
View File

@@ -93,24 +93,64 @@ export function findChannel(
return channels.find((c) => c.ownsJid(jid));
}
export interface DeliveryRouteResolution {
channel?: Channel;
requestedRoleChannelName: string | null;
selectedChannelName: string | null;
usedRoleChannel: boolean;
fallbackUsed: boolean;
}
function resolveRequestedRoleChannelName(senderRole?: string): string | null {
if (senderRole === 'reviewer') return 'discord-review';
if (senderRole === 'arbiter') return 'discord-arbiter';
return null;
}
export function resolveChannelForDeliveryRole(
channels: Channel[],
jid: string,
senderRole?: string,
): DeliveryRouteResolution {
const requestedRoleChannelName = resolveRequestedRoleChannelName(senderRole);
if (!requestedRoleChannelName) {
const channel = findChannel(channels, jid);
return {
channel,
requestedRoleChannelName,
selectedChannelName: channel?.name ?? null,
usedRoleChannel: false,
fallbackUsed: false,
};
}
const roleChannel = findChannelByName(channels, requestedRoleChannelName);
if (roleChannel) {
return {
channel: roleChannel,
requestedRoleChannelName,
selectedChannelName: roleChannel.name,
usedRoleChannel: true,
fallbackUsed: false,
};
}
const fallbackChannel = findChannel(channels, jid);
return {
channel: fallbackChannel,
requestedRoleChannelName,
selectedChannelName: fallbackChannel?.name ?? null,
usedRoleChannel: false,
fallbackUsed: true,
};
}
export function findChannelForDeliveryRole(
channels: Channel[],
jid: string,
senderRole?: string,
): Channel | undefined {
if (senderRole === 'reviewer') {
return (
findChannelByName(channels, 'discord-review') ||
findChannel(channels, jid)
);
}
if (senderRole === 'arbiter') {
return (
findChannelByName(channels, 'discord-arbiter') ||
findChannel(channels, jid)
);
}
return findChannel(channels, jid);
return resolveChannelForDeliveryRole(channels, jid, senderRole).channel;
}
export function findChannelByName(