From 71650da5345616a60aa3aa4974374b0f58599c36 Mon Sep 17 00:00:00 2001 From: ejclaw Date: Wed, 27 May 2026 06:27:10 +0900 Subject: [PATCH] fix: recover local paired attempts after restart --- src/db-paired-turn-attempt-recovery.test.ts | 85 ++++++++++++++++++++- src/db/paired-turn-attempts.ts | 23 ++++-- src/db/runtime-paired.ts | 2 +- src/index.ts | 12 ++- 4 files changed, 109 insertions(+), 13 deletions(-) diff --git a/src/db-paired-turn-attempt-recovery.test.ts b/src/db-paired-turn-attempt-recovery.test.ts index ebbad83..9e87c6c 100644 --- a/src/db-paired-turn-attempt-recovery.test.ts +++ b/src/db-paired-turn-attempt-recovery.test.ts @@ -7,7 +7,8 @@ import { markPairedTurnRunning, recoverInterruptedPairedTurnAttemptsForService, } from './db.js'; -import { CODEX_MAIN_SERVICE_ID } from './config.js'; +import { CLAUDE_SERVICE_ID, CODEX_MAIN_SERVICE_ID } from './config.js'; +import { requireDatabase } from './db/runtime-database.js'; import { buildPairedTurnIdentity } from './paired-turn-identity.js'; import type { PairedTask } from './types.js'; @@ -46,7 +47,7 @@ describe('paired turn attempt restart recovery', () => { _initTestDatabase(); }); - it('fails same-service running attempts so restart recovery can retry the turn', () => { + it('fails local-service running attempts so restart recovery can retry the turn', () => { const task = makeTask(); createPairedTask(task); const turnIdentity = buildPairedTurnIdentity({ @@ -64,7 +65,7 @@ describe('paired turn attempt restart recovery', () => { }); const recovered = recoverInterruptedPairedTurnAttemptsForService({ - serviceId: CODEX_MAIN_SERVICE_ID, + serviceIds: [CLAUDE_SERVICE_ID, CODEX_MAIN_SERVICE_ID], now: '2026-05-27T00:11:00.000Z', }); @@ -130,7 +131,7 @@ describe('paired turn attempt restart recovery', () => { expect( recoverInterruptedPairedTurnAttemptsForService({ - serviceId: CODEX_MAIN_SERVICE_ID, + serviceIds: [CODEX_MAIN_SERVICE_ID], now: '2026-05-27T00:11:00.000Z', }), ).toEqual([]); @@ -141,4 +142,80 @@ describe('paired turn attempt restart recovery', () => { }), ]); }); + + it('recovers codex executor attempts even when the orchestration lease used the claude service id', () => { + const task = makeTask({ id: 'cross-service-lease-task' }); + createPairedTask(task); + const turnIdentity = buildPairedTurnIdentity({ + taskId: task.id, + taskUpdatedAt: task.updated_at, + intentKind: 'owner-turn', + role: 'owner', + }); + + markPairedTurnRunning({ + turnIdentity, + executorServiceId: CODEX_MAIN_SERVICE_ID, + executorAgentType: 'codex', + runId: 'codex-run-before-restart', + }); + requireDatabase() + .prepare( + ` + INSERT INTO paired_task_execution_leases ( + task_id, + chat_jid, + role, + turn_id, + turn_attempt_id, + turn_attempt_no, + intent_kind, + claimed_run_id, + claimed_service_id, + task_status, + task_updated_at, + claimed_at, + updated_at, + expires_at + ) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + `, + ) + .run( + task.id, + task.chat_jid, + 'owner', + turnIdentity.turnId, + `${turnIdentity.turnId}:attempt:1`, + 1, + 'owner-turn', + 'orchestrator-run-before-restart', + CLAUDE_SERVICE_ID, + task.status, + task.updated_at, + '2026-05-27T00:10:00.000Z', + '2026-05-27T00:10:00.000Z', + '2026-05-27T00:20:00.000Z', + ); + + expect( + recoverInterruptedPairedTurnAttemptsForService({ + serviceIds: [CLAUDE_SERVICE_ID, CODEX_MAIN_SERVICE_ID], + now: '2026-05-27T00:11:00.000Z', + }), + ).toEqual([ + expect.objectContaining({ + task_id: task.id, + turn_id: turnIdentity.turnId, + role: 'owner', + intent_kind: 'owner-turn', + }), + ]); + expect(getPairedTurnAttempts(turnIdentity.turnId)).toEqual([ + expect.objectContaining({ + state: 'failed', + active_run_id: null, + }), + ]); + }); }); diff --git a/src/db/paired-turn-attempts.ts b/src/db/paired-turn-attempts.ts index f6bccba..b9f3f33 100644 --- a/src/db/paired-turn-attempts.ts +++ b/src/db/paired-turn-attempts.ts @@ -446,12 +446,18 @@ export function getPairedTurnAttemptIdFromDatabase( export function recoverInterruptedPairedTurnAttemptsForServiceInDatabase( database: Database, args: { - serviceId: string; + serviceIds: string[]; now?: string; error?: string; }, ): InterruptedPairedTurnAttemptRecoveryCandidate[] { - const serviceId = normalizeServiceId(args.serviceId); + const serviceIds = Array.from( + new Set(args.serviceIds.map((serviceId) => normalizeServiceId(serviceId))), + ).filter((serviceId) => serviceId.length > 0); + if (serviceIds.length === 0) { + return []; + } + const servicePlaceholders = serviceIds.map(() => '?').join(', '); const now = args.now ?? new Date().toISOString(); const error = args.error ?? 'Interrupted by service restart before completion.'; @@ -473,19 +479,22 @@ export function recoverInterruptedPairedTurnAttemptsForServiceInDatabase( JOIN paired_tasks tasks ON tasks.id = attempts.task_id WHERE attempts.state = 'running' - AND attempts.executor_service_id = ? + AND attempts.executor_service_id IN (${servicePlaceholders}) AND tasks.status != 'completed' AND NOT EXISTS ( SELECT 1 FROM paired_task_execution_leases leases WHERE leases.task_id = attempts.task_id AND leases.turn_id = attempts.turn_id - AND leases.claimed_service_id = attempts.executor_service_id + AND leases.claimed_service_id NOT IN (${servicePlaceholders}) ) ORDER BY attempts.updated_at ASC, attempts.attempt_no ASC `, ) - .all(serviceId) as InterruptedPairedTurnAttemptRecoveryCandidate[]; + .all( + ...serviceIds, + ...serviceIds, + ) as InterruptedPairedTurnAttemptRecoveryCandidate[]; if (rows.length === 0) { return rows; @@ -501,11 +510,11 @@ export function recoverInterruptedPairedTurnAttemptsForServiceInDatabase( last_error = ? WHERE attempt_id = ? AND state = 'running' - AND executor_service_id = ? + AND executor_service_id IN (${servicePlaceholders}) `, ); for (const row of rows) { - update.run(now, now, error, row.attempt_id, serviceId); + update.run(now, now, error, row.attempt_id, ...serviceIds); } return rows; })(); diff --git a/src/db/runtime-paired.ts b/src/db/runtime-paired.ts index 1703d34..77f1bdb 100644 --- a/src/db/runtime-paired.ts +++ b/src/db/runtime-paired.ts @@ -242,7 +242,7 @@ export function getPairedTurnAttempts( } export function recoverInterruptedPairedTurnAttemptsForService(args: { - serviceId: string; + serviceIds: string[]; now?: string; error?: string; }): InterruptedPairedTurnAttemptRecoveryCandidate[] { diff --git a/src/index.ts b/src/index.ts index 43f5a5f..af887ff 100644 --- a/src/index.ts +++ b/src/index.ts @@ -1,5 +1,9 @@ import { ASSISTANT_NAME, + ARBITER_SERVICE_ID, + CLAUDE_SERVICE_ID, + CODEX_MAIN_SERVICE_ID, + CODEX_REVIEW_SERVICE_ID, IDLE_TIMEOUT, POLL_INTERVAL, SERVICE_ID, @@ -262,7 +266,13 @@ async function announceRestartRecovery( function queueInterruptedPairedTurnAttemptRecovery(): void { for (const candidate of recoverInterruptedPairedTurnAttemptsForService({ - serviceId: SERVICE_ID, + serviceIds: [ + SERVICE_ID, + CLAUDE_SERVICE_ID, + CODEX_MAIN_SERVICE_ID, + CODEX_REVIEW_SERVICE_ID, + ARBITER_SERVICE_ID, + ].filter((serviceId): serviceId is string => Boolean(serviceId)), })) { const binding = runtimeState.getRoomBindings()[candidate.chat_jid]; if (!binding) {