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

feat: recover paired executions after restart

Browse Source
This commit is contained in:
Eyejoker
2026-03-29 05:06:25 +09:00
parent bb30330c66
commit 29310ce67f
4 changed files with 537 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -38,6 +38,7 @@ vi.mock('./paired-execution-context.js', () => ({
completePairedExecutionContext: vi.fn(),
markRoomReviewReady: vi.fn(() => null),
formatRoomReviewReadyMessage: vi.fn(() => null),
planPairedExecutionRecovery: vi.fn(() => null),
setRoomTaskRiskLevel: vi.fn(() => null),
recordRoomPlan: vi.fn(() => null),
approveRoomPlan: vi.fn(() => null),
@@ -221,6 +222,9 @@ describe('createMessageRuntime', () => {
vi.mocked(db.isPairedRoomJid).mockReturnValue(false);
vi.mocked(config.isClaudeService).mockReturnValue(true);
vi.mocked(config.isReviewService).mockReturnValue(false);
vi.mocked(pairedExecutionContext.planPairedExecutionRecovery).mockReturnValue(
null,
);
});
it('surfaces the pending review message through the message-runtime /review path', async () => {
@@ -2947,6 +2951,32 @@ describe('createMessageRuntime', () => {
const chatJid = 'group@test';
const group = makeGroup('claude-code');
const enqueueMessageCheck = vi.fn();
const enqueueTask = vi.fn();
vi.mocked(pairedExecutionContext.planPairedExecutionRecovery).mockReturnValue(
{
task: {
id: 'task-1',
chat_jid: chatJid,
group_folder: group.folder,
owner_service_id: 'claude',
reviewer_service_id: 'codex-main',
title: null,
source_ref: 'HEAD',
task_policy: 'autonomous',
risk_level: 'low',
plan_status: 'not_requested',
review_requested_at: '2026-03-29T00:00:00.000Z',
status: 'review_pending',
created_at: '2026-03-29T00:00:00.000Z',
updated_at: '2026-03-29T00:00:00.000Z',
},
role: 'reviewer',
checkpointFingerprint: 'fingerprint-1',
recoveryKey: 'paired-recovery:task-1:reviewer:fingerprint-1',
prompt: 'resume reviewer',
},
);
vi.mocked(db.getOpenWorkItem).mockReturnValue({
id: 99,
@@ -2978,6 +3008,7 @@ describe('createMessageRuntime', () => {
closeStdin: vi.fn(),
notifyIdle: vi.fn(),
enqueueMessageCheck,
enqueueTask,
} as any,
getRegisteredGroups: () => ({ [chatJid]: group }),
getSessions: () => ({}),
@@ -2995,6 +3026,158 @@ describe('createMessageRuntime', () => {
chatJid,
resolveGroupIpcPath(group.folder),
);
expect(pairedExecutionContext.planPairedExecutionRecovery).not.toHaveBeenCalled();
expect(enqueueTask).not.toHaveBeenCalled();
expect(db.getMessagesSinceSeq).not.toHaveBeenCalled();
});
it('recovery queues paired execution resume when no messages are pending', () => {
const chatJid = 'group@test';
const group = {
...makeGroup('codex'),
workDir: '/repo/canonical',
};
const enqueueMessageCheck = vi.fn();
const enqueueTask = vi.fn();
vi.mocked(pairedExecutionContext.planPairedExecutionRecovery).mockReturnValue(
{
task: {
id: 'task-1',
chat_jid: chatJid,
group_folder: group.folder,
owner_service_id: 'claude',
reviewer_service_id: 'codex-main',
title: null,
source_ref: 'HEAD',
task_policy: 'autonomous',
risk_level: 'low',
plan_status: 'not_requested',
review_requested_at: '2026-03-29T00:00:00.000Z',
status: 'review_pending',
created_at: '2026-03-29T00:00:00.000Z',
updated_at: '2026-03-29T00:00:00.000Z',
},
role: 'reviewer',
checkpointFingerprint: 'fingerprint-1',
recoveryKey: 'paired-recovery:task-1:reviewer:fingerprint-1',
prompt: 'resume reviewer',
},
);
const runtime = createMessageRuntime({
assistantName: 'Andy',
idleTimeout: 1_000,
pollInterval: 1_000,
timezone: 'UTC',
triggerPattern: /^@Andy\b/i,
channels: [makeChannel(chatJid)],
queue: {
registerProcess: vi.fn(),
closeStdin: vi.fn(),
notifyIdle: vi.fn(),
enqueueMessageCheck,
enqueueTask,
} as any,
getRegisteredGroups: () => ({ [chatJid]: group }),
getSessions: () => ({}),
getLastTimestamp: () => '',
setLastTimestamp: vi.fn(),
getLastAgentTimestamps: () => ({}),
saveState: vi.fn(),
persistSession: vi.fn(),
clearSession: vi.fn(),
});
runtime.recoverPendingMessages();
expect(pairedExecutionContext.planPairedExecutionRecovery).toHaveBeenCalled();
expect(enqueueMessageCheck).not.toHaveBeenCalled();
expect(enqueueTask).toHaveBeenCalledWith(
chatJid,
'paired-recovery:task-1:reviewer:fingerprint-1',
expect.any(Function),
);
});
it('recovery does not queue paired execution resume when messages are already pending', () => {
const chatJid = 'group@test';
const group = {
...makeGroup('codex'),
workDir: '/repo/canonical',
};
const enqueueMessageCheck = vi.fn();
const enqueueTask = vi.fn();
vi.mocked(pairedExecutionContext.planPairedExecutionRecovery).mockReturnValue(
{
task: {
id: 'task-1',
chat_jid: chatJid,
group_folder: group.folder,
owner_service_id: 'claude',
reviewer_service_id: 'codex-main',
title: null,
source_ref: 'HEAD',
task_policy: 'autonomous',
risk_level: 'low',
plan_status: 'not_requested',
review_requested_at: '2026-03-29T00:00:00.000Z',
status: 'review_pending',
created_at: '2026-03-29T00:00:00.000Z',
updated_at: '2026-03-29T00:00:00.000Z',
},
role: 'reviewer',
checkpointFingerprint: 'fingerprint-1',
recoveryKey: 'paired-recovery:task-1:reviewer:fingerprint-1',
prompt: 'resume reviewer',
},
);
vi.mocked(db.getMessagesSinceSeq).mockReturnValue([
{
id: 'msg-1',
chat_jid: chatJid,
sender: 'user',
sender_name: 'User',
content: 'pending',
timestamp: '2026-03-29T00:00:00.000Z',
seq: 1,
is_bot_message: false,
is_from_me: false,
},
] as any);
const runtime = createMessageRuntime({
assistantName: 'Andy',
idleTimeout: 1_000,
pollInterval: 1_000,
timezone: 'UTC',
triggerPattern: /^@Andy\b/i,
channels: [makeChannel(chatJid)],
queue: {
registerProcess: vi.fn(),
closeStdin: vi.fn(),
notifyIdle: vi.fn(),
enqueueMessageCheck,
enqueueTask,
} as any,
getRegisteredGroups: () => ({ [chatJid]: group }),
getSessions: () => ({}),
getLastTimestamp: () => '',
setLastTimestamp: vi.fn(),
getLastAgentTimestamps: () => ({}),
saveState: vi.fn(),
persistSession: vi.fn(),
clearSession: vi.fn(),
});
runtime.recoverPendingMessages();
expect(enqueueMessageCheck).toHaveBeenCalledWith(
chatJid,
resolveGroupIpcPath(group.folder),
);
expect(pairedExecutionContext.planPairedExecutionRecovery).not.toHaveBeenCalled();
expect(enqueueTask).not.toHaveBeenCalled();
});
});

54
src/message-runtime.ts
View File

@@ -45,6 +45,7 @@ import {
approveRoomPlan,
formatRoomReviewReadyMessage,
markRoomReviewReady,
planPairedExecutionRecovery,
recordRoomPlan,
requestRoomPlanChanges,
setRoomTaskRiskLevel,
@@ -965,6 +966,7 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): {
chatJid,
resolveGroupIpcPath(group.folder),
);
continue;
} else if (rawPending.length > 0) {
const endSeq = rawPending[rawPending.length - 1].seq;
if (endSeq != null) {
@@ -976,6 +978,58 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): {
);
}
}
const lease = getEffectiveChannelLease(chatJid);
const roomRoleContext = buildRoomRoleContext(lease, SERVICE_ID);
const recoveryPlan = planPairedExecutionRecovery({
group,
chatJid,
roomRoleContext,
});
if (!recoveryPlan) {
continue;
}
deps.queue.enqueueTask(chatJid, recoveryPlan.recoveryKey, async () => {
const recoveryChannel = findChannel(deps.channels, chatJid);
if (!recoveryChannel) {
logger.warn(
{
chatJid,
taskId: recoveryPlan.task.id,
role: recoveryPlan.role,
recoveryKey: recoveryPlan.recoveryKey,
},
'Skipping paired execution recovery because no channel owns the chat',
);
return;
}
const runId = `recovery-${Date.now().toString(36)}-${Math.random()
.toString(36)
.slice(2, 8)}`;
logger.info(
{
chatJid,
group: group.name,
taskId: recoveryPlan.task.id,
role: recoveryPlan.role,
checkpointFingerprint: recoveryPlan.checkpointFingerprint,
recoveryKey: recoveryPlan.recoveryKey,
runId,
},
'Recovery: resuming paired execution after restart',
);
await executeTurn({
group,
prompt: recoveryPlan.prompt,
chatJid,
runId,
channel: recoveryChannel,
startSeq: null,
endSeq: null,
});
});
}
};

155
src/paired-execution-context.test.ts
View File

@@ -43,6 +43,7 @@ import {
completePairedExecutionContext,
formatRoomReviewReadyMessage,
markRoomReviewReady,
planPairedExecutionRecovery,
preparePairedExecutionContext,
recordRoomPlan,
requestRoomPlanChanges,
@@ -293,6 +294,160 @@ describe('paired execution context', () => {
});
});
it('plans reviewer recovery from the latest review checkpoint', () => {
vi.mocked(db.getLatestOpenPairedTaskForChat).mockReturnValue({
id: 'task-1',
chat_jid: 'dc:test',
group_folder: group.folder,
owner_service_id: 'codex-main',
reviewer_service_id: 'codex-review',
title: null,
source_ref: 'HEAD',
task_policy: 'autonomous',
risk_level: 'low',
plan_status: 'not_requested',
review_requested_at: '2026-03-29T00:00:00.000Z',
status: 'review_pending',
created_at: '2026-03-29T00:00:00.000Z',
updated_at: '2026-03-29T00:00:00.000Z',
});
vi.mocked(db.listPairedEventsForTask).mockReturnValue([
{
id: 1,
task_id: 'task-1',
event_type: 'request_review',
actor_role: 'owner',
source_service_id: 'codex-main',
source_fingerprint: 'fingerprint-2',
dedupe_key: 'request-review',
payload_json: null,
created_at: '2026-03-29T00:00:00.000Z',
},
]);
const result = planPairedExecutionRecovery({
group,
chatJid: 'dc:test',
roomRoleContext: reviewerContext,
});
expect(result).toMatchObject({
role: 'reviewer',
checkpointFingerprint: 'fingerprint-2',
recoveryKey: 'paired-recovery:task-1:reviewer:fingerprint-2',
});
expect(result?.prompt).toContain('formal paired review');
});
it('plans owner recovery for changes requested tasks', () => {
vi.mocked(db.getLatestOpenPairedTaskForChat).mockReturnValue({
id: 'task-1',
chat_jid: 'dc:test',
group_folder: group.folder,
owner_service_id: 'codex-main',
reviewer_service_id: 'codex-review',
title: null,
source_ref: 'HEAD',
task_policy: 'autonomous',
risk_level: 'low',
plan_status: 'not_requested',
review_requested_at: '2026-03-29T00:00:00.000Z',
status: 'changes_requested',
created_at: '2026-03-29T00:00:00.000Z',
updated_at: '2026-03-29T00:00:00.000Z',
});
vi.mocked(pairedWorkspaceManager.resolvePairedTaskSourceFingerprint).mockReturnValue(
'fingerprint-3',
);
const result = planPairedExecutionRecovery({
group,
chatJid: 'dc:test',
roomRoleContext: ownerContext,
});
expect(result).toMatchObject({
role: 'owner',
checkpointFingerprint: 'fingerprint-3',
recoveryKey:
'paired-recovery:task-1:owner:changes_requested:fingerprint-3',
});
expect(result?.prompt).toContain('owner rework');
});
it('plans owner recovery for an interrupted active owner execution only when the latest owner execution is still running', () => {
vi.mocked(db.getLatestOpenPairedTaskForChat).mockReturnValue({
id: 'task-1',
chat_jid: 'dc:test',
group_folder: group.folder,
owner_service_id: 'codex-main',
reviewer_service_id: 'codex-review',
title: null,
source_ref: 'HEAD',
task_policy: 'autonomous',
risk_level: 'low',
plan_status: 'not_requested',
review_requested_at: null,
status: 'active',
created_at: '2026-03-29T00:00:00.000Z',
updated_at: '2026-03-29T00:00:00.000Z',
});
vi.mocked(db.listPairedExecutionsForTask).mockReturnValue([
{
id: 'run-1:codex-main',
task_id: 'task-1',
service_id: 'codex-main',
role: 'owner',
workspace_id: 'task-1:owner',
checkpoint_fingerprint: 'fingerprint-1',
status: 'running',
summary: null,
created_at: '2026-03-29T00:00:00.000Z',
started_at: '2026-03-29T00:00:00.000Z',
completed_at: null,
},
]);
vi.mocked(pairedWorkspaceManager.resolvePairedTaskSourceFingerprint).mockReturnValue(
'fingerprint-2',
);
const result = planPairedExecutionRecovery({
group,
chatJid: 'dc:test',
roomRoleContext: ownerContext,
});
expect(result).toMatchObject({
role: 'owner',
checkpointFingerprint: 'fingerprint-2',
recoveryKey: 'paired-recovery:task-1:owner:active:fingerprint-2',
});
vi.mocked(db.listPairedExecutionsForTask).mockReturnValue([
{
id: 'run-1:codex-main',
task_id: 'task-1',
service_id: 'codex-main',
role: 'owner',
workspace_id: 'task-1:owner',
checkpoint_fingerprint: 'fingerprint-1',
status: 'succeeded',
summary: null,
created_at: '2026-03-29T00:00:00.000Z',
started_at: '2026-03-29T00:00:00.000Z',
completed_at: '2026-03-29T00:10:00.000Z',
},
]);
expect(
planPairedExecutionRecovery({
group,
chatJid: 'dc:test',
roomRoleContext: ownerContext,
}),
).toBeNull();
});
it('auto-requests review when a low-risk owner execution completes', () => {
vi.mocked(pairedWorkspaceManager.resolvePairedTaskSourceFingerprint).mockReturnValue(
'fingerprint-2',

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

@@ -261,6 +261,14 @@ export interface PreparedPairedExecutionContext {
blockMessage?: string;
}
export interface PairedExecutionRecoveryPlan {
task: PairedTask;
role: RoomRoleContext['role'];
checkpointFingerprint: string | null;
recoveryKey: string;
prompt: string;
}
export function preparePairedExecutionContext(args: {
group: RegisteredGroup;
chatJid: string;
@@ -488,6 +496,143 @@ function buildRoomRoleContextFromExecution(args: {
};
}
function getLatestExecutionForRole(
taskId: string,
role: PairedExecution['role'],
): PairedExecution | null {
const executions = listPairedExecutionsForTask(taskId).filter(
(execution) => execution.role === role,
);
return executions.at(-1) ?? null;
}
function buildPairedExecutionRecoveryPrompt(args: {
task: PairedTask;
role: RoomRoleContext['role'];
checkpointFingerprint: string | null;
mode: 'review' | 'owner-rework' | 'owner-resume';
}): string {
const lines = [
'Internal paired execution recovery.',
'- This run was started automatically after service restart.',
`- Task: ${args.task.id}`,
`- Role: ${args.role}`,
`- Status: ${args.task.status}`,
];
if (args.checkpointFingerprint) {
lines.push(`- Checkpoint: ${args.checkpointFingerprint}`);
}
if (args.mode === 'review') {
lines.push(
'Resume the formal paired review for the current checkpoint without waiting for a new user message.',
);
} else if (args.mode === 'owner-rework') {
lines.push(
'Resume the owner rework for the current task state without waiting for a new user message.',
);
} else {
lines.push(
'Resume the interrupted owner implementation for the current task without waiting for a new user message.',
);
}
lines.push('- Keep the current task scope unchanged.');
lines.push('- Follow the existing paired task guards and checkpoint rules.');
return lines.join('\n');
}
export function planPairedExecutionRecovery(args: {
group: RegisteredGroup;
chatJid: string;
roomRoleContext?: RoomRoleContext;
}): PairedExecutionRecoveryPlan | null {
const { group, chatJid, roomRoleContext } = args;
if (!roomRoleContext || !group.workDir) {
return null;
}
const task = getLatestOpenPairedTaskForChat(chatJid);
if (!task) {
return null;
}
if (roomRoleContext.role === 'reviewer') {
if (
task.status !== 'review_pending' &&
task.status !== 'review_ready' &&
task.status !== 'in_review'
) {
return null;
}
const checkpointFingerprint = getLatestReviewCheckpointFingerprint(task.id);
if (!checkpointFingerprint) {
return null;
}
return {
task,
role: 'reviewer',
checkpointFingerprint,
recoveryKey: `paired-recovery:${task.id}:reviewer:${checkpointFingerprint}`,
prompt: buildPairedExecutionRecoveryPrompt({
task,
role: 'reviewer',
checkpointFingerprint,
mode: 'review',
}),
};
}
if (task.status === 'changes_requested') {
const checkpointFingerprint =
resolvePairedTaskSourceFingerprint(task.id) ?? null;
return {
task,
role: 'owner',
checkpointFingerprint,
recoveryKey: `paired-recovery:${task.id}:owner:changes_requested:${
checkpointFingerprint ?? 'none'
}`,
prompt: buildPairedExecutionRecoveryPrompt({
task,
role: 'owner',
checkpointFingerprint,
mode: 'owner-rework',
}),
};
}
if (task.status !== 'active') {
return null;
}
const latestOwnerExecution = getLatestExecutionForRole(task.id, 'owner');
if (latestOwnerExecution?.status !== 'running') {
return null;
}
const checkpointFingerprint =
resolvePairedTaskSourceFingerprint(task.id) ??
latestOwnerExecution.checkpoint_fingerprint ??
null;
return {
task,
role: 'owner',
checkpointFingerprint,
recoveryKey: `paired-recovery:${task.id}:owner:active:${
checkpointFingerprint ?? 'none'
}`,
prompt: buildPairedExecutionRecoveryPrompt({
task,
role: 'owner',
checkpointFingerprint,
mode: 'owner-resume',
}),
};
}
function markTaskReviewPendingWithoutSnapshot(taskId: string): void {
const requestedAt = new Date().toISOString();
updatePairedTask(taskId, {