From e88073f2f82e1f6ad58740833f2c2af431b4ef0a Mon Sep 17 00:00:00 2001 From: ejclaw Date: Sat, 11 Apr 2026 04:18:43 +0900 Subject: [PATCH] fix: block stale paired IPC and duplicate finalize turns --- runners/agent-runner/src/ipc-mcp-stdio.ts | 292 ++++++++++++++---- runners/agent-runner/src/ipc-message.ts | 2 + runners/agent-runner/test/ipc-message.test.ts | 2 + src/group-queue.test.ts | 16 +- src/group-queue.ts | 54 ++++ src/index.ts | 19 +- src/ipc-auth.test.ts | 3 + src/ipc.ts | 4 +- src/message-runtime-flow.test.ts | 76 ++++- src/message-runtime-flow.ts | 28 +- 10 files changed, 435 insertions(+), 61 deletions(-) diff --git a/runners/agent-runner/src/ipc-mcp-stdio.ts b/runners/agent-runner/src/ipc-mcp-stdio.ts index 2bff913..2840066 100644 --- a/runners/agent-runner/src/ipc-mcp-stdio.ts +++ b/runners/agent-runner/src/ipc-mcp-stdio.ts @@ -31,8 +31,9 @@ import { import { resolveIpcDirectories } from './ipc-paths.js'; import { buildSendMessageIpcPayload } from './ipc-message.js'; -const { ipcDir: IPC_DIR, hostIpcDir: HOST_IPC_DIR } = - resolveIpcDirectories(process.env); +const { ipcDir: IPC_DIR, hostIpcDir: HOST_IPC_DIR } = resolveIpcDirectories( + process.env, +); const MESSAGES_DIR = path.join(HOST_IPC_DIR, 'messages'); const TASKS_DIR = path.join(HOST_IPC_DIR, 'tasks'); const HOST_EVIDENCE_RESPONSES_DIR = @@ -71,7 +72,12 @@ server.tool( "Send a message to the user or group immediately while you're still running. Use this for progress updates or to send multiple messages. You can call this multiple times.", { text: z.string().describe('The message text to send'), - sender: z.string().optional().describe('Your role/identity name (e.g. "Researcher"). When set, messages appear from a dedicated bot in Telegram.'), + sender: z + .string() + .optional() + .describe( + 'Your role/identity name (e.g. "Researcher"). When set, messages appear from a dedicated bot in Telegram.', + ), }, async (args) => { const data = buildSendMessageIpcPayload({ @@ -79,6 +85,7 @@ server.tool( text: args.text, sender: args.sender || undefined, senderRole: process.env.EJCLAW_ROOM_ROLE || undefined, + runId: process.env.EJCLAW_RUN_ID || undefined, groupFolder, }); @@ -113,11 +120,33 @@ SCHEDULE VALUE FORMAT (all times are LOCAL timezone): \u2022 interval: Milliseconds between runs (e.g., "300000" for 5 minutes, "3600000" for 1 hour) \u2022 once: Local time WITHOUT "Z" suffix (e.g., "2026-02-01T15:30:00"). Do NOT use UTC/Z suffix.`, { - prompt: z.string().describe('What the agent should do when the task runs. For isolated mode, include all necessary context here.'), - schedule_type: z.enum(['cron', 'interval', 'once']).describe('cron=recurring at specific times, interval=recurring every N ms, once=run once at specific time'), - schedule_value: z.string().describe('cron: "*/5 * * * *" | interval: milliseconds like "300000" | once: local timestamp like "2026-02-01T15:30:00" (no Z suffix!)'), - context_mode: z.enum(['group', 'isolated']).default('group').describe('group=runs with chat history and memory, isolated=fresh session (include context in prompt)'), - target_group_jid: z.string().optional().describe('(Main group only) JID of the group to schedule the task for. Defaults to the current group.'), + prompt: z + .string() + .describe( + 'What the agent should do when the task runs. For isolated mode, include all necessary context here.', + ), + schedule_type: z + .enum(['cron', 'interval', 'once']) + .describe( + 'cron=recurring at specific times, interval=recurring every N ms, once=run once at specific time', + ), + schedule_value: z + .string() + .describe( + 'cron: "*/5 * * * *" | interval: milliseconds like "300000" | once: local timestamp like "2026-02-01T15:30:00" (no Z suffix!)', + ), + context_mode: z + .enum(['group', 'isolated']) + .default('group') + .describe( + 'group=runs with chat history and memory, isolated=fresh session (include context in prompt)', + ), + target_group_jid: z + .string() + .optional() + .describe( + '(Main group only) JID of the group to schedule the task for. Defaults to the current group.', + ), }, async (args) => { // Validate schedule_value before writing IPC @@ -126,7 +155,12 @@ SCHEDULE VALUE FORMAT (all times are LOCAL timezone): CronExpressionParser.parse(args.schedule_value); } catch { return { - content: [{ type: 'text' as const, text: `Invalid cron: "${args.schedule_value}". Use format like "0 9 * * *" (daily 9am) or "*/5 * * * *" (every 5 min).` }], + content: [ + { + type: 'text' as const, + text: `Invalid cron: "${args.schedule_value}". Use format like "0 9 * * *" (daily 9am) or "*/5 * * * *" (every 5 min).`, + }, + ], isError: true, }; } @@ -134,28 +168,47 @@ SCHEDULE VALUE FORMAT (all times are LOCAL timezone): const ms = parseInt(args.schedule_value, 10); if (isNaN(ms) || ms <= 0) { return { - content: [{ type: 'text' as const, text: `Invalid interval: "${args.schedule_value}". Must be positive milliseconds (e.g., "300000" for 5 min).` }], + content: [ + { + type: 'text' as const, + text: `Invalid interval: "${args.schedule_value}". Must be positive milliseconds (e.g., "300000" for 5 min).`, + }, + ], isError: true, }; } } else if (args.schedule_type === 'once') { - if (/[Zz]$/.test(args.schedule_value) || /[+-]\d{2}:\d{2}$/.test(args.schedule_value)) { + if ( + /[Zz]$/.test(args.schedule_value) || + /[+-]\d{2}:\d{2}$/.test(args.schedule_value) + ) { return { - content: [{ type: 'text' as const, text: `Timestamp must be local time without timezone suffix. Got "${args.schedule_value}" — use format like "2026-02-01T15:30:00".` }], + content: [ + { + type: 'text' as const, + text: `Timestamp must be local time without timezone suffix. Got "${args.schedule_value}" — use format like "2026-02-01T15:30:00".`, + }, + ], isError: true, }; } const date = new Date(args.schedule_value); if (isNaN(date.getTime())) { return { - content: [{ type: 'text' as const, text: `Invalid timestamp: "${args.schedule_value}". Use local time format like "2026-02-01T15:30:00".` }], + content: [ + { + type: 'text' as const, + text: `Invalid timestamp: "${args.schedule_value}". Use local time format like "2026-02-01T15:30:00".`, + }, + ], isError: true, }; } } // Non-main groups can only schedule for themselves - const targetJid = isMain && args.target_group_jid ? args.target_group_jid : chatJid; + const targetJid = + isMain && args.target_group_jid ? args.target_group_jid : chatJid; const taskId = `task-${Date.now()}-${Math.random().toString(36).slice(2, 8)}`; @@ -174,7 +227,12 @@ SCHEDULE VALUE FORMAT (all times are LOCAL timezone): writeIpcFile(TASKS_DIR, data); return { - content: [{ type: 'text' as const, text: `Task ${taskId} scheduled: ${args.schedule_type} - ${args.schedule_value}` }], + content: [ + { + type: 'text' as const, + text: `Task ${taskId} scheduled: ${args.schedule_type} - ${args.schedule_value}`, + }, + ], }; }, ); @@ -297,9 +355,12 @@ server.tool( let pollSeconds: number; try { - pollSeconds = normalizeWatchCiIntervalSeconds(args.poll_interval_seconds, { - ciProvider: args.ci_provider, - }); + pollSeconds = normalizeWatchCiIntervalSeconds( + args.poll_interval_seconds, + { + ciProvider: args.ci_provider, + }, + ); } catch (error) { return { content: [ @@ -407,8 +468,7 @@ server.tool( content: [ { type: 'text' as const, - text: - error instanceof Error ? error.message : String(error), + text: error instanceof Error ? error.message : String(error), }, ], isError: true, @@ -436,10 +496,7 @@ server.tool( content: [ { type: 'text' as const, - text: - error instanceof Error - ? error.message - : String(error), + text: error instanceof Error ? error.message : String(error), }, ], isError: true, @@ -470,8 +527,7 @@ server.tool( content: [ { type: 'text' as const, - text: - error instanceof Error ? error.message : String(error), + text: error instanceof Error ? error.message : String(error), }, ], isError: true, @@ -489,30 +545,56 @@ server.tool( try { if (!fs.existsSync(tasksFile)) { - return { content: [{ type: 'text' as const, text: 'No scheduled tasks found.' }] }; + return { + content: [ + { type: 'text' as const, text: 'No scheduled tasks found.' }, + ], + }; } const allTasks = JSON.parse(fs.readFileSync(tasksFile, 'utf-8')); const tasks = isMain ? allTasks - : allTasks.filter((t: { groupFolder: string }) => t.groupFolder === groupFolder); + : allTasks.filter( + (t: { groupFolder: string }) => t.groupFolder === groupFolder, + ); if (tasks.length === 0) { - return { content: [{ type: 'text' as const, text: 'No scheduled tasks found.' }] }; + return { + content: [ + { type: 'text' as const, text: 'No scheduled tasks found.' }, + ], + }; } const formatted = tasks .map( - (t: { id: string; prompt: string; schedule_type: string; schedule_value: string; status: string; next_run: string }) => + (t: { + id: string; + prompt: string; + schedule_type: string; + schedule_value: string; + status: string; + next_run: string; + }) => `- [${t.id}] ${t.prompt.slice(0, 50)}... (${t.schedule_type}: ${t.schedule_value}) - ${t.status}, next: ${t.next_run || 'N/A'}`, ) .join('\n'); - return { content: [{ type: 'text' as const, text: `Scheduled tasks:\n${formatted}` }] }; + return { + content: [ + { type: 'text' as const, text: `Scheduled tasks:\n${formatted}` }, + ], + }; } catch (err) { return { - content: [{ type: 'text' as const, text: `Error reading tasks: ${err instanceof Error ? err.message : String(err)}` }], + content: [ + { + type: 'text' as const, + text: `Error reading tasks: ${err instanceof Error ? err.message : String(err)}`, + }, + ], }; } }, @@ -533,7 +615,14 @@ server.tool( writeIpcFile(TASKS_DIR, data); - return { content: [{ type: 'text' as const, text: `Task ${args.task_id} pause requested.` }] }; + return { + content: [ + { + type: 'text' as const, + text: `Task ${args.task_id} pause requested.`, + }, + ], + }; }, ); @@ -552,19 +641,36 @@ server.tool( writeIpcFile(TASKS_DIR, data); - return { content: [{ type: 'text' as const, text: `Task ${args.task_id} resume requested.` }] }; + return { + content: [ + { + type: 'text' as const, + text: `Task ${args.task_id} resume requested.`, + }, + ], + }; }, ); server.tool( 'cancel_task', 'Cancel and delete a scheduled task. If no task_id is provided, cancels the current task (for background watchers).', - { task_id: z.string().optional().describe('The task ID to cancel. Omit to cancel the current task.') }, + { + task_id: z + .string() + .optional() + .describe('The task ID to cancel. Omit to cancel the current task.'), + }, async (args: { task_id?: string }) => { const resolvedId = args.task_id || runtimeTaskId; if (!resolvedId) { return { - content: [{ type: 'text' as const, text: 'No task_id provided and no current task context available.' }], + content: [ + { + type: 'text' as const, + text: 'No task_id provided and no current task context available.', + }, + ], isError: true, }; } @@ -579,7 +685,11 @@ server.tool( writeIpcFile(TASKS_DIR, data); - return { content: [{ type: 'text' as const, text: `Task cancellation requested.` }] }; + return { + content: [ + { type: 'text' as const, text: `Task cancellation requested.` }, + ], + }; }, ); @@ -590,18 +700,37 @@ if (allowGenericScheduling) { { task_id: z.string().describe('The task ID to update'), prompt: z.string().optional().describe('New prompt for the task'), - schedule_type: z.enum(['cron', 'interval', 'once']).optional().describe('New schedule type'), - schedule_value: z.string().optional().describe('New schedule value (see schedule_task for format)'), + schedule_type: z + .enum(['cron', 'interval', 'once']) + .optional() + .describe('New schedule type'), + schedule_value: z + .string() + .optional() + .describe('New schedule value (see schedule_task for format)'), }, - async (args: { task_id: string; prompt?: string; schedule_type?: 'cron' | 'interval' | 'once'; schedule_value?: string }) => { + async (args: { + task_id: string; + prompt?: string; + schedule_type?: 'cron' | 'interval' | 'once'; + schedule_value?: string; + }) => { // Validate schedule_value if provided - if (args.schedule_type === 'cron' || (!args.schedule_type && args.schedule_value)) { + if ( + args.schedule_type === 'cron' || + (!args.schedule_type && args.schedule_value) + ) { if (args.schedule_value) { try { CronExpressionParser.parse(args.schedule_value); } catch { return { - content: [{ type: 'text' as const, text: `Invalid cron: "${args.schedule_value}".` }], + content: [ + { + type: 'text' as const, + text: `Invalid cron: "${args.schedule_value}".`, + }, + ], isError: true, }; } @@ -611,7 +740,12 @@ if (allowGenericScheduling) { const ms = parseInt(args.schedule_value, 10); if (isNaN(ms) || ms <= 0) { return { - content: [{ type: 'text' as const, text: `Invalid interval: "${args.schedule_value}".` }], + content: [ + { + type: 'text' as const, + text: `Invalid interval: "${args.schedule_value}".`, + }, + ], isError: true, }; } @@ -625,12 +759,21 @@ if (allowGenericScheduling) { timestamp: new Date().toISOString(), }; if (args.prompt !== undefined) data.prompt = args.prompt; - if (args.schedule_type !== undefined) data.schedule_type = args.schedule_type; - if (args.schedule_value !== undefined) data.schedule_value = args.schedule_value; + if (args.schedule_type !== undefined) + data.schedule_type = args.schedule_type; + if (args.schedule_value !== undefined) + data.schedule_value = args.schedule_value; writeIpcFile(TASKS_DIR, data); - return { content: [{ type: 'text' as const, text: `Task ${args.task_id} update requested.` }] }; + return { + content: [ + { + type: 'text' as const, + text: `Task ${args.task_id} update requested.`, + }, + ], + }; }, ); } @@ -641,15 +784,42 @@ server.tool( If folder is omitted, the host generates one automatically.`, { - jid: z.string().describe('The chat JID (e.g., "120363336345536173@g.us", "tg:-1001234567890", "dc:1234567890123456")'), + jid: z + .string() + .describe( + 'The chat JID (e.g., "120363336345536173@g.us", "tg:-1001234567890", "dc:1234567890123456")', + ), name: z.string().describe('Display name for the room'), - room_mode: z.enum(['single', 'tribunal']).default('single').describe('single=owner only, tribunal=owner/reviewer roles enabled'), - owner_agent_type: z.enum(['claude-code', 'codex']).optional().describe('Preferred owner agent type for the room'), - folder: z.string().optional().describe('Optional folder override. If omitted, the host generates one automatically.'), - trigger: z.string().optional().describe('Optional canonical trigger label to store for the room'), - requires_trigger: z.boolean().optional().describe('Whether direct trigger mention is required'), - is_main: z.boolean().optional().describe('Whether this room is the privileged main control room'), - work_dir: z.string().optional().describe('Optional working directory override'), + room_mode: z + .enum(['single', 'tribunal']) + .default('single') + .describe('single=owner only, tribunal=owner/reviewer roles enabled'), + owner_agent_type: z + .enum(['claude-code', 'codex']) + .optional() + .describe('Preferred owner agent type for the room'), + folder: z + .string() + .optional() + .describe( + 'Optional folder override. If omitted, the host generates one automatically.', + ), + trigger: z + .string() + .optional() + .describe('Optional canonical trigger label to store for the room'), + requires_trigger: z + .boolean() + .optional() + .describe('Whether direct trigger mention is required'), + is_main: z + .boolean() + .optional() + .describe('Whether this room is the privileged main control room'), + work_dir: z + .string() + .optional() + .describe('Optional working directory override'), }, async (args: { jid: string; @@ -664,7 +834,12 @@ If folder is omitted, the host generates one automatically.`, }) => { if (!isMain) { return { - content: [{ type: 'text' as const, text: 'Only the main group can assign rooms.' }], + content: [ + { + type: 'text' as const, + text: 'Only the main group can assign rooms.', + }, + ], isError: true, }; } @@ -686,7 +861,12 @@ If folder is omitted, the host generates one automatically.`, writeIpcFile(TASKS_DIR, data); return { - content: [{ type: 'text' as const, text: `Room "${args.name}" assignment requested.` }], + content: [ + { + type: 'text' as const, + text: `Room "${args.name}" assignment requested.`, + }, + ], }; }, ); diff --git a/runners/agent-runner/src/ipc-message.ts b/runners/agent-runner/src/ipc-message.ts index 712b99f..0640212 100644 --- a/runners/agent-runner/src/ipc-message.ts +++ b/runners/agent-runner/src/ipc-message.ts @@ -3,6 +3,7 @@ export interface SendMessageIpcPayloadInput { text: string; sender?: string; senderRole?: string; + runId?: string; groupFolder: string; timestamp?: string; } @@ -16,6 +17,7 @@ export function buildSendMessageIpcPayload( text: input.text, sender: input.sender || undefined, senderRole: input.senderRole || undefined, + runId: input.runId || undefined, groupFolder: input.groupFolder, timestamp: input.timestamp || new Date().toISOString(), }; diff --git a/runners/agent-runner/test/ipc-message.test.ts b/runners/agent-runner/test/ipc-message.test.ts index e45a360..b2df47f 100644 --- a/runners/agent-runner/test/ipc-message.test.ts +++ b/runners/agent-runner/test/ipc-message.test.ts @@ -10,6 +10,7 @@ describe('agent runner IPC message payload', () => { text: 'hello', sender: 'Reviewer', senderRole: 'reviewer', + runId: 'run-reviewer-1', groupFolder: 'discord-review', timestamp: '2026-04-04T13:45:00.000Z', }), @@ -19,6 +20,7 @@ describe('agent runner IPC message payload', () => { text: 'hello', sender: 'Reviewer', senderRole: 'reviewer', + runId: 'run-reviewer-1', groupFolder: 'discord-review', timestamp: '2026-04-04T13:45:00.000Z', }); diff --git a/src/group-queue.test.ts b/src/group-queue.test.ts index 4c45386..4df83a9 100644 --- a/src/group-queue.test.ts +++ b/src/group-queue.test.ts @@ -124,7 +124,7 @@ describe('GroupQueue', () => { await vi.advanceTimersByTimeAsync(10); }); - it('stores direct terminal text for the active run and clears it when the run ends', async () => { + it('stores direct terminal text for the active run, keeps a recent run record, and clears the active slot when the run ends', async () => { let releaseRun!: (value: boolean) => void; let runId: string | undefined; const blocker = new Promise((resolve) => { @@ -155,6 +155,13 @@ describe('GroupQueue', () => { expect( queue.getDirectTerminalDeliveryForRun('group1@g.us', runId!, 'reviewer'), ).toBe('DONE_WITH_CONCERNS\n리뷰 완료'); + expect( + queue.hasRecordedDirectTerminalDeliveryForRun( + 'group1@g.us', + runId!, + 'reviewer', + ), + ).toBe(true); releaseRun(true); await vi.advanceTimersByTimeAsync(10); @@ -165,6 +172,13 @@ describe('GroupQueue', () => { expect( queue.getDirectTerminalDeliveryForRun('group1@g.us', runId!, 'reviewer'), ).toBeNull(); + expect( + queue.hasRecordedDirectTerminalDeliveryForRun( + 'group1@g.us', + runId!, + 'reviewer', + ), + ).toBe(true); }); it('force-terminates a lingering process after output was delivered', async () => { diff --git a/src/group-queue.ts b/src/group-queue.ts index 6a441e0..87969b3 100644 --- a/src/group-queue.ts +++ b/src/group-queue.ts @@ -49,6 +49,7 @@ interface GroupState { runningTaskId: string | null; currentRunId: string | null; directTerminalDeliveries: Map; + recentDirectTerminalDeliveries: Map>; pendingMessages: boolean; pendingTasks: QueuedTask[]; process: ChildProcess | null; @@ -62,6 +63,32 @@ interface GroupState { startedAt: number | null; } +const MAX_RECORDED_DIRECT_TERMINAL_RUNS = 16; + +function recordRecentDirectTerminalDelivery( + state: GroupState, + runId: string, + senderRole: string, + text: string, +): void { + const existing = state.recentDirectTerminalDeliveries.get(runId) ?? new Map(); + existing.set(senderRole, text); + state.recentDirectTerminalDeliveries.delete(runId); + state.recentDirectTerminalDeliveries.set(runId, existing); + + while ( + state.recentDirectTerminalDeliveries.size > + MAX_RECORDED_DIRECT_TERMINAL_RUNS + ) { + const oldestRunId = + state.recentDirectTerminalDeliveries.keys().next().value ?? null; + if (!oldestRunId) { + break; + } + state.recentDirectTerminalDeliveries.delete(oldestRunId); + } +} + function transitionRunPhase( state: GroupState, groupJid: string, @@ -200,6 +227,7 @@ export class GroupQueue { runningTaskId: null, currentRunId: null, directTerminalDeliveries: new Map(), + recentDirectTerminalDeliveries: new Map(), pendingMessages: false, pendingTasks: [], process: null, @@ -440,6 +468,12 @@ export class GroupQueue { return; } state.directTerminalDeliveries.set(senderRole, text); + recordRecentDirectTerminalDelivery( + state, + state.currentRunId, + senderRole, + text, + ); logger.info( { groupJid, @@ -475,6 +509,26 @@ export class GroupQueue { return state.directTerminalDeliveries.get(senderRole) ?? null; } + hasRecordedDirectTerminalDeliveryForRun( + groupJid: string, + runId: string, + senderRole?: string | null, + ): boolean { + const state = this.getGroup(groupJid); + if (!senderRole) { + return false; + } + if ( + state.currentRunId === runId && + state.directTerminalDeliveries.has(senderRole) + ) { + return true; + } + return ( + state.recentDirectTerminalDeliveries.get(runId)?.has(senderRole) ?? false + ); + } + private clearPostCloseTimers(state: GroupState): void { if (state.postCloseTermTimer) { clearTimeout(state.postCloseTermTimer); diff --git a/src/index.ts b/src/index.ts index 28e3fc4..d201805 100644 --- a/src/index.ts +++ b/src/index.ts @@ -371,13 +371,30 @@ async function main(): Promise { editFormattedTrackedChannelMessage(channels, jid, messageId, rawText), }); startIpcWatcher({ - sendMessage: async (jid, text, senderRole) => { + sendMessage: async (jid, text, senderRole, runId) => { + if ( + runId && + (senderRole === 'reviewer' || senderRole === 'arbiter') && + queue.hasRecordedDirectTerminalDeliveryForRun(jid, runId, senderRole) + ) { + logger.info( + { + transition: 'ipc:skip-post-terminal', + chatJid: jid, + senderRole, + runId, + }, + 'Skipped IPC relay message because the run already emitted a direct terminal verdict', + ); + return; + } const route = resolveChannelForDeliveryRole(channels, jid, senderRole); if (!route.channel) throw new Error(`No channel for JID: ${jid}`); logger.info( { transition: 'ipc:route', chatJid: jid, + runId: runId ?? null, senderRole: senderRole ?? null, requestedRoleChannel: route.requestedRoleChannelName, selectedChannel: route.selectedChannelName, diff --git a/src/ipc-auth.test.ts b/src/ipc-auth.test.ts index adf3ab4..bebf3db 100644 --- a/src/ipc-auth.test.ts +++ b/src/ipc-auth.test.ts @@ -543,6 +543,7 @@ describe('IPC message authorization', () => { chatJid: 'other@g.us', text: 'review text', senderRole: 'reviewer', + runId: 'run-reviewer-ipc', }, 'other-group', false, @@ -554,6 +555,7 @@ describe('IPC message authorization', () => { 'other@g.us', 'review text', 'reviewer', + 'run-reviewer-ipc', ); expect(result).toEqual( expect.objectContaining({ @@ -572,6 +574,7 @@ describe('IPC message authorization', () => { chatJid: 'third@g.us', text: 'review text', senderRole: 'reviewer', + runId: 'run-reviewer-ipc', }, 'other-group', false, diff --git a/src/ipc.ts b/src/ipc.ts index a26ff5b..3b89a06 100644 --- a/src/ipc.ts +++ b/src/ipc.ts @@ -33,6 +33,7 @@ export interface IpcDeps { jid: string, text: string, senderRole?: string, + runId?: string, ) => Promise; nudgeScheduler?: () => void; roomBindings: () => Record; @@ -51,6 +52,7 @@ export interface IpcMessagePayload { chatJid?: string; text?: string; senderRole?: string; + runId?: string; } export interface IpcMessageForwardResult { @@ -86,7 +88,7 @@ export async function forwardAuthorizedIpcMessage( }; } - await sendMessage(msg.chatJid, msg.text, msg.senderRole); + await sendMessage(msg.chatJid, msg.text, msg.senderRole, msg.runId); return { outcome: 'sent', chatJid: msg.chatJid, diff --git a/src/message-runtime-flow.test.ts b/src/message-runtime-flow.test.ts index f3eb117..559cd8a 100644 --- a/src/message-runtime-flow.test.ts +++ b/src/message-runtime-flow.test.ts @@ -1,11 +1,12 @@ import { beforeEach, describe, expect, it, vi } from 'vitest'; -import { _initTestDatabase } from './db.js'; +import { _initTestDatabase, createPairedTask } from './db.js'; import { executeBotOnlyPairedFollowUpAction, executePendingPairedTurn, } from './message-runtime-flow.js'; import { + claimPairedTurnExecution, resetPairedFollowUpScheduleState, schedulePairedFollowUpOnce, type ScheduledPairedFollowUpIntentKind, @@ -123,6 +124,79 @@ describe('executeBotOnlyPairedFollowUpAction', () => { 'Skipped duplicate paired pending turn requeue while task state was unchanged', ); }); + + it('skips inline finalize when the same finalize-owner turn revision was already claimed elsewhere', async () => { + const task: PairedTask = { + id: 'task-inline-finalize-dedup', + chat_jid: 'group@test', + group_folder: 'test-group', + 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: 1, + status: 'merge_ready', + arbiter_verdict: null, + arbiter_requested_at: null, + completion_reason: null, + created_at: '2026-03-30T00:00:00.000Z', + updated_at: '2026-03-30T00:05:00.000Z', + }; + const executeTurn = vi.fn(); + const log = { + info: vi.fn(), + } as any; + + createPairedTask(task as any); + + expect( + claimPairedTurnExecution({ + chatJid: 'group@test', + runId: 'run-existing-finalize-owner', + task, + intentKind: 'finalize-owner-turn', + }), + ).toBe(true); + + const result = await executeBotOnlyPairedFollowUpAction({ + action: { + kind: 'inline-finalize', + task, + cursor: 42, + }, + chatJid: 'group@test', + group: { + name: 'Test Group', + folder: 'test-group', + trigger: '@Andy', + added_at: '2026-03-30T00:00:00.000Z', + requiresTrigger: false, + agentType: 'codex', + }, + runId: 'run-inline-finalize', + channel: {} as any, + log, + saveState: vi.fn(), + lastAgentTimestamps: {}, + executeTurn, + schedulePairedFollowUp: vi.fn(() => true), + closeStdin: vi.fn(), + }); + + expect(result).toBe(true); + expect(executeTurn).not.toHaveBeenCalled(); + expect(log.info).toHaveBeenCalledWith( + expect.objectContaining({ + chatJid: 'group@test', + taskId: 'task-inline-finalize-dedup', + taskStatus: 'merge_ready', + handoffMode: 'inline-finalize', + }), + 'Skipped inline merge_ready finalize because the task revision was already claimed elsewhere', + ); + }); }); describe('executePendingPairedTurn', () => { diff --git a/src/message-runtime-flow.ts b/src/message-runtime-flow.ts index 3ec285f..9b52a95 100644 --- a/src/message-runtime-flow.ts +++ b/src/message-runtime-flow.ts @@ -22,7 +22,10 @@ import { } from './message-runtime-rules.js'; import type { ExecuteTurnFn } from './message-runtime-types.js'; import { buildPairedTurnIdentity } from './paired-turn-identity.js'; -import { type ScheduledPairedFollowUpIntentKind } from './paired-follow-up-scheduler.js'; +import { + claimPairedTurnExecution, + type ScheduledPairedFollowUpIntentKind, +} from './paired-follow-up-scheduler.js'; import { hasReviewerLease } from './service-routing.js'; import type { Channel, @@ -398,6 +401,29 @@ export async function executeBotOnlyPairedFollowUpAction(args: { } if (action.kind === 'inline-finalize') { + const claimed = claimPairedTurnExecution({ + chatJid, + runId, + task: action.task, + intentKind: 'finalize-owner-turn', + }); + if (!claimed) { + log.info( + { + chatJid, + group: group.name, + groupFolder: group.folder, + taskId: action.task.id, + taskStatus: action.task.status, + handoffMode: 'inline-finalize', + nextRole: 'owner', + cursor: action.cursor, + }, + 'Skipped inline merge_ready finalize because the task revision was already claimed elsewhere', + ); + return true; + } + log.info( { chatJid,