From 29310ce67fbcf55558ceb26529498707f3e84b13 Mon Sep 17 00:00:00 2001 From: Eyejoker Date: Sun, 29 Mar 2026 05:06:25 +0900 Subject: [PATCH] feat: recover paired executions after restart --- src/message-runtime.test.ts | 183 +++++++++++++++++++++++++++ src/message-runtime.ts | 54 ++++++++ src/paired-execution-context.test.ts | 155 +++++++++++++++++++++++ src/paired-execution-context.ts | 145 +++++++++++++++++++++ 4 files changed, 537 insertions(+) diff --git a/src/message-runtime.test.ts b/src/message-runtime.test.ts index aeaa923..50d75ed 100644 --- a/src/message-runtime.test.ts +++ b/src/message-runtime.test.ts @@ -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(); + }); }); diff --git a/src/message-runtime.ts b/src/message-runtime.ts index 572902f..d617975 100644 --- a/src/message-runtime.ts +++ b/src/message-runtime.ts @@ -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, + }); + }); } }; diff --git a/src/paired-execution-context.test.ts b/src/paired-execution-context.test.ts index 87a9f2b..216979a 100644 --- a/src/paired-execution-context.test.ts +++ b/src/paired-execution-context.test.ts @@ -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', diff --git a/src/paired-execution-context.ts b/src/paired-execution-context.ts index bc8c2a5..46a0ef7 100644 --- a/src/paired-execution-context.ts +++ b/src/paired-execution-context.ts @@ -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, {