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

Audit outbound delivery and fix reviewer turn routing

Browse Source
This commit is contained in:
ejclaw
2026-04-10 15:22:45 +09:00
parent c67bd1c622
commit 2e62e125ae
10 changed files with 457 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

33
src/channels/discord.ts
View File

@@ -601,6 +601,8 @@ export class DiscordChannel implements Channel {
attachmentCount: files.length,
messageId: sentMessageIds[0] ?? null,
messageIds: sentMessageIds,
botUserId: this.client.user?.id ?? null,
botUsername: this.client.user?.username ?? null,
},
'Discord message sent',
);
@@ -689,6 +691,18 @@ export class DiscordChannel implements Channel {
const channel = await this.client.channels.fetch(channelId);
if (!channel || !('send' in channel)) return null;
const msg = await (channel as TextChannel).send(text);
logger.info(
{
jid,
channelName: this.name,
deliveryMode: 'tracked-send',
messageId: msg.id,
botUserId: this.client.user?.id ?? null,
botUsername: this.client.user?.username ?? null,
length: text.length,
},
'Discord tracked message sent',
);
return msg.id;
} catch (err) {
logger.error({ jid, err }, 'Failed to send tracked Discord message');
@@ -696,6 +710,14 @@ export class DiscordChannel implements Channel {
}
}
getOutboundAuditMeta() {
return {
channelName: this.name,
botUserId: this.client?.user?.id ?? null,
botUsername: this.client?.user?.username ?? null,
};
}
async getChannelMeta(jids: string[]): Promise<Map<string, ChannelMeta>> {
const result = new Map<string, ChannelMeta>();
if (!this.client) return result;
@@ -813,12 +835,21 @@ export class DiscordChannel implements Channel {
deliveryMode: 'edit',
messageId,
length: text.length,
botUserId: this.client.user?.id ?? null,
botUsername: this.client.user?.username ?? null,
},
'Discord message edited',
);
} catch (err) {
logger.debug(
{ jid, channelName: this.name, messageId, err },
{
jid,
channelName: this.name,
messageId,
botUserId: this.client?.user?.id ?? null,
botUsername: this.client?.user?.username ?? null,
err,
},
'Failed to edit Discord message',
);
throw err; // Re-throw so callers (e.g. dashboard) can reset message ID

62
src/message-agent-executor.test.ts
View File

@@ -743,6 +743,68 @@ describe('runAgentForGroup room memory', () => {
expect(agentRunner.runAgentProcess).not.toHaveBeenCalled();
});
it('allows a claimed paired turn revision when preparation advances the task before execution', async () => {
const group = { ...makeGroup(), folder: 'test-group' };
vi.mocked(
pairedExecutionContext.preparePairedExecutionContext,
).mockReturnValue({
task: {
id: 'paired-task-prep-advance',
chat_jid: 'group@test',
group_folder: 'test-group',
owner_service_id: 'claude',
reviewer_service_id: 'codex-review',
title: null,
source_ref: 'HEAD',
plan_notes: null,
round_trip_count: 0,
review_requested_at: '2026-04-10T00:00:00.000Z',
status: 'in_review',
arbiter_verdict: null,
arbiter_requested_at: null,
completion_reason: null,
created_at: '2026-04-10T00:00:00.000Z',
updated_at: '2026-04-10T01:00:00.000Z',
},
claimedTaskUpdatedAt: '2026-04-10T00:00:00.000Z',
workspace: null,
envOverrides: {},
});
await expect(
runAgentForGroup(makeDeps(), {
group,
prompt: 'hello',
chatJid: 'group@test',
runId: 'run-claimed-paired-turn-revision',
forcedRole: 'reviewer',
pairedTurnIdentity: {
turnId:
'paired-task-prep-advance:2026-04-10T00:00:00.000Z:reviewer-turn',
taskId: 'paired-task-prep-advance',
taskUpdatedAt: '2026-04-10T00:00:00.000Z',
intentKind: 'reviewer-turn',
role: 'reviewer',
},
}),
).resolves.toBe('success');
expect(agentRunner.runAgentProcess).toHaveBeenCalledWith(
group,
expect.any(Object),
expect.any(Function),
expect.any(Function),
expect.objectContaining({
EJCLAW_PAIRED_TURN_ID:
'paired-task-prep-advance:2026-04-10T00:00:00.000Z:reviewer-turn',
EJCLAW_PAIRED_TURN_ROLE: 'reviewer',
EJCLAW_PAIRED_TURN_INTENT: 'reviewer-turn',
EJCLAW_PAIRED_TASK_UPDATED_AT: '2026-04-10T00:00:00.000Z',
}),
);
});
it('fails closed when a persisted paired turn revision mismatches the latest paired task fallback', async () => {
const group = { ...makeGroup(), folder: 'test-group' };

12
src/message-agent-executor.ts
View File

@@ -195,12 +195,17 @@ export async function runAgentForGroup(
roomRoleContext,
hasHumanMessage: args.hasHumanMessage,
});
const preparedTurnTaskUpdatedAt = pairedExecutionContext
? (pairedExecutionContext.claimedTaskUpdatedAt ??
pairedExecutionContext.task.updated_at)
: undefined;
const runtimePairedTurnIdentity =
args.pairedTurnIdentity ??
(pairedExecutionContext
? resolveRuntimePairedTurnIdentity({
taskId: pairedExecutionContext.task.id,
taskUpdatedAt: pairedExecutionContext.task.updated_at,
taskUpdatedAt:
preparedTurnTaskUpdatedAt ?? pairedExecutionContext.task.updated_at,
role: activeRole,
taskStatus: pairedExecutionContext.task.status,
hasHumanMessage: args.hasHumanMessage,
@@ -230,8 +235,7 @@ export async function runAgentForGroup(
}
if (
pairedExecutionContext &&
runtimePairedTurnIdentity.taskUpdatedAt !==
pairedExecutionContext.task.updated_at
runtimePairedTurnIdentity.taskUpdatedAt !== preparedTurnTaskUpdatedAt
) {
throw new Error(
`Paired turn ${runtimePairedTurnIdentity.turnId} task_updated_at does not match the prepared execution context`,
@@ -393,7 +397,7 @@ export async function runAgentForGroup(
const pairedExecutionLifecycle = createPairedExecutionLifecycle({
pairedExecutionContext,
pairedTurnIdentity: runtimePairedTurnIdentity,
completedRole: roomRoleContext?.role ?? 'owner',
completedRole: runtimePairedTurnIdentity?.role ?? activeRole,
chatJid,
runId,
enqueueMessageCheck: () => deps.queue.enqueueMessageCheck(chatJid),

12
src/message-runtime.ts
View File

@@ -268,6 +268,10 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): {
const pairedRoom = hasReviewerLease(chatJid);
const resolvedDeliveryRole =
args.deliveryRole ?? args.forcedRole ?? (pairedRoom ? 'owner' : null);
const resolvedDeliveryServiceId = resolveLeaseServiceId(
getEffectiveChannelLease(chatJid),
resolvedDeliveryRole ?? 'owner',
);
const turnController = new MessageTurnController({
chatJid,
group,
@@ -279,13 +283,13 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): {
clearSession: () => deps.clearSession(group.folder),
requestClose: (reason) =>
deps.queue.closeStdin(chatJid, { runId, reason }),
deliveryRole: resolvedDeliveryRole,
deliveryServiceId: resolvedDeliveryServiceId,
pairedTurnIdentity: args.pairedTurnIdentity ?? null,
deliverFinalText: async (text) => {
try {
const persistedDeliveryRole = resolvedDeliveryRole;
const persistedDeliveryServiceId = resolveLeaseServiceId(
getEffectiveChannelLease(chatJid),
persistedDeliveryRole ?? 'owner',
);
const persistedDeliveryServiceId = resolvedDeliveryServiceId;
if (
(persistedDeliveryRole === 'reviewer' ||
persistedDeliveryRole === 'arbiter') &&

224
src/message-turn-controller.test.ts Normal file
View File

@@ -0,0 +1,224 @@
import { beforeEach, describe, expect, it, vi } from 'vitest';
vi.mock('./logger.js', () => {
const mockLogger = {
debug: vi.fn(),
info: vi.fn(),
warn: vi.fn(),
error: vi.fn(),
child: (_bindings?: Record<string, unknown>) => mockLogger,
};
return {
logger: mockLogger,
createScopedLogger: (_bindings?: Record<string, unknown>) => mockLogger,
};
});
import { MessageTurnController } from './message-turn-controller.js';
import { logger } from './logger.js';
import type { Channel, RegisteredGroup } from './types.js';
import type { PairedTurnIdentity } from './paired-turn-identity.js';
function makeGroup(): RegisteredGroup {
return {
name: 'Test Group',
folder: 'test-group',
trigger: '@Andy',
added_at: new Date().toISOString(),
requiresTrigger: false,
agentType: 'claude-code',
};
}
function makeChannel(): Channel {
return {
name: 'discord-review',
connect: vi.fn().mockResolvedValue(undefined),
disconnect: vi.fn().mockResolvedValue(undefined),
isConnected: vi.fn(() => true),
ownsJid: vi.fn(() => true),
sendMessage: vi.fn().mockResolvedValue(undefined),
sendAndTrack: vi.fn().mockResolvedValue('progress-1'),
editMessage: vi.fn().mockResolvedValue(undefined),
setTyping: vi.fn().mockResolvedValue(undefined),
getOutboundAuditMeta: vi.fn(() => ({
channelName: 'discord-review',
botUserId: 'bot-review',
botUsername: 'reviewer-bot',
})),
};
}
function makeTurnIdentity(): PairedTurnIdentity {
return {
turnId: 'task-1:2026-04-10T14:22:00.000Z:reviewer-turn',
taskId: 'task-1',
taskUpdatedAt: '2026-04-10T14:22:00.000Z',
intentKind: 'reviewer-turn',
role: 'reviewer',
};
}
async function flushAsync(): Promise<void> {
await new Promise((resolve) => setTimeout(resolve, 0));
}
function getAuditEntries(): Array<Record<string, unknown>> {
return vi
.mocked(logger.info)
.mock.calls.map(([payload]) => payload)
.filter(
(payload): payload is Record<string, unknown> =>
!!payload && typeof payload === 'object' && 'auditEvent' in payload,
);
}
describe('MessageTurnController outbound audit logging', () => {
beforeEach(() => {
vi.resetAllMocks();
});
it('logs outbound progress and final audit context with role/service/turn metadata', async () => {
const channel = makeChannel();
const deliverFinalText = vi.fn().mockResolvedValue(true);
const controller = new MessageTurnController({
chatJid: 'dc:test-room',
group: makeGroup(),
runId: 'run-review-1',
channel,
idleTimeout: 1_000,
failureFinalText: '실패',
isClaudeCodeAgent: true,
clearSession: vi.fn(),
requestClose: vi.fn(),
deliverFinalText,
deliveryRole: 'reviewer',
deliveryServiceId: 'codex-review',
pairedTurnIdentity: makeTurnIdentity(),
});
await controller.start();
await controller.handleOutput({
status: 'success',
phase: 'progress',
result: '첫 진행 상황',
} as any);
await controller.handleOutput({
status: 'success',
phase: 'progress',
result: '둘째 진행 상황',
} as any);
await flushAsync();
await controller.handleOutput({
status: 'success',
phase: 'final',
result: '최종 답변',
} as any);
await controller.finish('success');
expect(channel.sendAndTrack).toHaveBeenCalledTimes(1);
expect(channel.editMessage).toHaveBeenCalledWith(
'dc:test-room',
'progress-1',
expect.stringContaining('둘째 진행 상황'),
);
expect(deliverFinalText).toHaveBeenCalledWith('최종 답변');
expect(getAuditEntries()).toEqual(
expect.arrayContaining([
expect.objectContaining({
auditEvent: 'progress-create',
chatJid: 'dc:test-room',
runId: 'run-review-1',
deliveryRole: 'reviewer',
serviceId: 'codex-review',
turnId: 'task-1:2026-04-10T14:22:00.000Z:reviewer-turn',
turnRole: 'reviewer',
channelName: 'discord-review',
botUserId: 'bot-review',
botUsername: 'reviewer-bot',
messageId: 'progress-1',
}),
expect.objectContaining({
auditEvent: 'progress-edit',
deliveryRole: 'reviewer',
serviceId: 'codex-review',
turnId: 'task-1:2026-04-10T14:22:00.000Z:reviewer-turn',
messageId: 'progress-1',
}),
expect.objectContaining({
auditEvent: 'final-delivery-attempt',
deliveryRole: 'reviewer',
serviceId: 'codex-review',
turnId: 'task-1:2026-04-10T14:22:00.000Z:reviewer-turn',
}),
expect.objectContaining({
auditEvent: 'final-delivery-result',
deliveryRole: 'reviewer',
serviceId: 'codex-review',
turnId: 'task-1:2026-04-10T14:22:00.000Z:reviewer-turn',
delivered: true,
}),
]),
);
});
it('logs fallback progress audit when tracked message creation fails', async () => {
const channel = makeChannel();
const deliverFinalText = vi.fn().mockResolvedValue(true);
vi.mocked(channel.sendAndTrack!).mockRejectedValueOnce(
new Error('send failed'),
);
const controller = new MessageTurnController({
chatJid: 'dc:test-room',
group: makeGroup(),
runId: 'run-review-fallback',
channel,
idleTimeout: 1_000,
failureFinalText: '실패',
isClaudeCodeAgent: true,
clearSession: vi.fn(),
requestClose: vi.fn(),
deliverFinalText,
deliveryRole: 'reviewer',
deliveryServiceId: 'codex-review',
pairedTurnIdentity: makeTurnIdentity(),
});
await controller.start();
await controller.handleOutput({
status: 'success',
phase: 'progress',
result: '첫 진행 상황',
} as any);
await controller.handleOutput({
status: 'success',
phase: 'progress',
result: '둘째 진행 상황',
} as any);
await flushAsync();
await controller.finish('success');
expect(channel.sendMessage).toHaveBeenCalledWith(
'dc:test-room',
expect.stringContaining('첫 진행 상황'),
);
expect(getAuditEntries()).toEqual(
expect.arrayContaining([
expect.objectContaining({
auditEvent: 'progress-fallback-send',
chatJid: 'dc:test-room',
runId: 'run-review-fallback',
deliveryRole: 'reviewer',
serviceId: 'codex-review',
turnId: 'task-1:2026-04-10T14:22:00.000Z:reviewer-turn',
channelName: 'discord-review',
botUserId: 'bot-review',
messageId: null,
fallbackReason: 'tracked-send-error',
}),
]),
);
});
});

80
src/message-turn-controller.ts
View File

@@ -5,11 +5,13 @@ import { formatOutbound } from './router.js';
import { shouldResetSessionOnAgentFailure } from './session-recovery.js';
import { TASK_STATUS_MESSAGE_PREFIX } from './task-watch-status.js';
import { formatElapsedKorean } from './utils.js';
import type { PairedTurnIdentity } from './paired-turn-identity.js';
import {
normalizeAgentOutputPhase,
toVisiblePhase,
type AgentOutputPhase,
type Channel,
type PairedRoomRole,
type RegisteredGroup,
type VisiblePhase,
} from './types.js';
@@ -32,6 +34,9 @@ interface MessageTurnControllerOptions {
clearSession: () => void;
requestClose: (reason: string) => void;
deliverFinalText: (text: string) => Promise<boolean>;
deliveryRole?: PairedRoomRole | null;
deliveryServiceId?: string | null;
pairedTurnIdentity?: PairedTurnIdentity | null;
}
export class MessageTurnController {
@@ -66,6 +71,41 @@ export class MessageTurnController {
});
}
private buildOutboundAuditContext(
extra: Record<string, unknown> = {},
): Record<string, unknown> {
const outboundMeta = this.options.channel.getOutboundAuditMeta?.();
return {
chatJid: this.options.chatJid,
runId: this.options.runId,
deliveryRole:
this.options.deliveryRole ??
this.options.pairedTurnIdentity?.role ??
null,
serviceId: this.options.deliveryServiceId ?? null,
turnId: this.options.pairedTurnIdentity?.turnId ?? null,
turnRole: this.options.pairedTurnIdentity?.role ?? null,
intentKind: this.options.pairedTurnIdentity?.intentKind ?? null,
channelName: outboundMeta?.channelName ?? this.options.channel.name,
botUserId: outboundMeta?.botUserId ?? null,
botUsername: outboundMeta?.botUsername ?? null,
...extra,
};
}
private logOutboundAudit(
auditEvent: string,
extra: Record<string, unknown> = {},
): void {
this.log.info(
this.buildOutboundAuditContext({
auditEvent,
...extra,
}),
'Outbound message audit',
);
}
private async setTyping(
isTyping: boolean,
source: string,
@@ -481,6 +521,11 @@ export class MessageTurnController {
);
this.latestProgressRendered = rendered;
this.progressEditFailCount = 0;
this.logOutboundAudit('progress-edit', {
messageId: this.progressMessageId,
textLength: this.latestProgressText.length,
renderedLength: rendered.length,
});
} catch (err) {
this.progressEditFailCount++;
this.log.warn(
@@ -529,7 +574,16 @@ export class MessageTurnController {
private async deliverFinalText(text: string): Promise<void> {
await this.activateTyping('turn:deliver-final');
this.visiblePhase = toVisiblePhase('final');
this.logOutboundAudit('final-delivery-attempt', {
messageId: this.progressMessageId,
textLength: text.length,
});
const delivered = await this.options.deliverFinalText(text);
this.logOutboundAudit('final-delivery-result', {
messageId: this.progressMessageId,
textLength: text.length,
delivered,
});
if (!delivered) {
this.producedDeliverySucceeded = false;
}
@@ -581,6 +635,12 @@ export class MessageTurnController {
if (!this.options.channel.sendAndTrack) {
this.latestProgressRendered = rendered;
await this.options.channel.sendMessage(this.options.chatJid, rendered);
this.logOutboundAudit('progress-fallback-send', {
messageId: null,
tracked: false,
textLength: text.length,
renderedLength: rendered.length,
});
this.visiblePhase = toVisiblePhase('progress');
return;
}
@@ -594,6 +654,13 @@ export class MessageTurnController {
this.log.warn({ err }, 'Failed to send tracked progress message');
this.latestProgressRendered = rendered;
await this.options.channel.sendMessage(this.options.chatJid, rendered);
this.logOutboundAudit('progress-fallback-send', {
messageId: null,
tracked: false,
fallbackReason: 'tracked-send-error',
textLength: text.length,
renderedLength: rendered.length,
});
this.visiblePhase = toVisiblePhase('progress');
return;
}
@@ -607,6 +674,12 @@ export class MessageTurnController {
'Created tracked progress message',
);
this.latestProgressRendered = rendered;
this.logOutboundAudit('progress-create', {
messageId: this.progressMessageId,
tracked: true,
textLength: text.length,
renderedLength: rendered.length,
});
this.ensureProgressTicker();
this.visiblePhase = toVisiblePhase('progress');
return;
@@ -614,6 +687,13 @@ export class MessageTurnController {
this.latestProgressRendered = rendered;
await this.options.channel.sendMessage(this.options.chatJid, rendered);
this.logOutboundAudit('progress-fallback-send', {
messageId: null,
tracked: false,
fallbackReason: 'tracked-send-returned-null',
textLength: text.length,
renderedLength: rendered.length,
});
this.visiblePhase = toVisiblePhase('progress');
}

3
src/paired-execution-context.ts
View File

@@ -171,6 +171,7 @@ function ensureActiveTask(
export interface PreparedPairedExecutionContext {
task: PairedTask;
claimedTaskUpdatedAt?: string;
workspace: PairedWorkspace | null;
envOverrides: Record<string, string>;
gateTurnKind?: string | null;
@@ -209,6 +210,7 @@ export function preparePairedExecutionContext(args: {
}
const latestTask = getPairedTaskById(task.id) ?? task;
const claimedTaskUpdatedAt = latestTask.updated_at;
let workspace: PairedWorkspace | null = null;
let blockMessage: string | undefined;
const now = new Date().toISOString();
@@ -349,6 +351,7 @@ export function preparePairedExecutionContext(args: {
return {
task: getPairedTaskById(task.id) ?? task,
claimedTaskUpdatedAt,
workspace,
envOverrides,
blockMessage,

26
src/room-role-context.test.ts
View File

@@ -137,6 +137,32 @@ describe('buildRoomRoleContext', () => {
});
});
it('keeps the preferred reviewer role when fallback execution shares the arbiter service shadow', () => {
expect(
buildRoomRoleContext(
{
chat_jid: 'group@test',
owner_service_id: 'codex-main',
reviewer_service_id: 'claude',
arbiter_service_id: 'codex-review',
owner_failover_active: false,
activated_at: null,
reason: null,
explicit: false,
},
'codex-review',
'reviewer',
),
).toEqual({
serviceId: 'codex-review',
role: 'reviewer',
ownerServiceId: 'codex-main',
reviewerServiceId: 'claude',
failoverOwner: false,
arbiterServiceId: 'codex-review',
});
});
it('returns undefined for a non-paired room', () => {
expect(
buildRoomRoleContext(

7
src/room-role-context.ts
View File

@@ -26,8 +26,13 @@ export function buildRoomRoleContext(
arbiter: arbiterServiceId === normalizedServiceId,
};
const canHonorPreferredRole =
preferredRole === 'owner' ||
(preferredRole === 'reviewer' && reviewerServiceId !== null) ||
(preferredRole === 'arbiter' && arbiterServiceId !== undefined);
const role =
preferredRole && matches[preferredRole]
preferredRole && canHonorPreferredRole
? preferredRole
: matches.arbiter
? 'arbiter'

8
src/types.ts
View File

@@ -209,6 +209,12 @@ export interface ChannelMeta {
categoryPosition: number;
}
export interface ChannelOutboundAuditMeta {
channelName: string;
botUserId: string | null;
botUsername: string | null;
}
export interface Channel {
name: string;
connect(): Promise<void>;
@@ -229,6 +235,8 @@ export interface Channel {
getChannelMeta?(jids: string[]): Promise<Map<string, ChannelMeta>>;
// Optional: delete all messages in a channel (used for dashboard cleanup).
purgeChannel?(jid: string): Promise<number>;
// Optional: expose runtime sender identity for outbound audit logs.
getOutboundAuditMeta?(): ChannelOutboundAuditMeta;
}
// Callback type that channels use to deliver inbound messages