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

fix: block stale paired IPC and duplicate finalize turns

Browse Source
This commit is contained in:
ejclaw
2026-04-11 04:18:43 +09:00
parent 320fc58763
commit e88073f2f8
10 changed files with 435 additions and 61 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
src/group-queue.test.ts
View File

@@ -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<boolean>((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 () => {

54
src/group-queue.ts
View File

@@ -49,6 +49,7 @@ interface GroupState {
runningTaskId: string | null;
currentRunId: string | null;
directTerminalDeliveries: Map<string, string>;
recentDirectTerminalDeliveries: Map<string, Map<string, string>>;
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);

19
src/index.ts
View File

@@ -371,13 +371,30 @@ async function main(): Promise<void> {
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,

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

@@ -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,

4
src/ipc.ts
View File

@@ -33,6 +33,7 @@ export interface IpcDeps {
jid: string,
text: string,
senderRole?: string,
runId?: string,
) => Promise<void>;
nudgeScheduler?: () => void;
roomBindings: () => Record<string, RegisteredGroup>;
@@ -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,

76
src/message-runtime-flow.test.ts
View File

@@ -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', () => {

28
src/message-runtime-flow.ts
View File

@@ -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,