diff --git a/src/channels/discord.ts b/src/channels/discord.ts index e689e45..c84aad3 100644 --- a/src/channels/discord.ts +++ b/src/channels/discord.ts @@ -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> { const result = new Map(); 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 diff --git a/src/message-agent-executor.test.ts b/src/message-agent-executor.test.ts index 6a1f9c7..0bd37bb 100644 --- a/src/message-agent-executor.test.ts +++ b/src/message-agent-executor.test.ts @@ -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' }; diff --git a/src/message-agent-executor.ts b/src/message-agent-executor.ts index 0b924c3..970d004 100644 --- a/src/message-agent-executor.ts +++ b/src/message-agent-executor.ts @@ -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), diff --git a/src/message-runtime.ts b/src/message-runtime.ts index 8f35bcb..00b5f92 100644 --- a/src/message-runtime.ts +++ b/src/message-runtime.ts @@ -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') && diff --git a/src/message-turn-controller.test.ts b/src/message-turn-controller.test.ts new file mode 100644 index 0000000..2eafd85 --- /dev/null +++ b/src/message-turn-controller.test.ts @@ -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) => mockLogger, + }; + return { + logger: mockLogger, + createScopedLogger: (_bindings?: Record) => 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 { + await new Promise((resolve) => setTimeout(resolve, 0)); +} + +function getAuditEntries(): Array> { + return vi + .mocked(logger.info) + .mock.calls.map(([payload]) => payload) + .filter( + (payload): payload is Record => + !!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', + }), + ]), + ); + }); +}); diff --git a/src/message-turn-controller.ts b/src/message-turn-controller.ts index e396e47..7a5665c 100644 --- a/src/message-turn-controller.ts +++ b/src/message-turn-controller.ts @@ -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; + deliveryRole?: PairedRoomRole | null; + deliveryServiceId?: string | null; + pairedTurnIdentity?: PairedTurnIdentity | null; } export class MessageTurnController { @@ -66,6 +71,41 @@ export class MessageTurnController { }); } + private buildOutboundAuditContext( + extra: Record = {}, + ): Record { + 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 = {}, + ): 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 { 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'); } diff --git a/src/paired-execution-context.ts b/src/paired-execution-context.ts index 07e972f..cdef09e 100644 --- a/src/paired-execution-context.ts +++ b/src/paired-execution-context.ts @@ -171,6 +171,7 @@ function ensureActiveTask( export interface PreparedPairedExecutionContext { task: PairedTask; + claimedTaskUpdatedAt?: string; workspace: PairedWorkspace | null; envOverrides: Record; 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, diff --git a/src/room-role-context.test.ts b/src/room-role-context.test.ts index 717ca3d..5589586 100644 --- a/src/room-role-context.test.ts +++ b/src/room-role-context.test.ts @@ -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( diff --git a/src/room-role-context.ts b/src/room-role-context.ts index 49581d5..e242f45 100644 --- a/src/room-role-context.ts +++ b/src/room-role-context.ts @@ -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' diff --git a/src/types.ts b/src/types.ts index 1ae44fc..2a8ce0e 100644 --- a/src/types.ts +++ b/src/types.ts @@ -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; @@ -229,6 +235,8 @@ export interface Channel { getChannelMeta?(jids: string[]): Promise>; // Optional: delete all messages in a channel (used for dashboard cleanup). purgeChannel?(jid: string): Promise; + // Optional: expose runtime sender identity for outbound audit logs. + getOutboundAuditMeta?(): ChannelOutboundAuditMeta; } // Callback type that channels use to deliver inbound messages