diff --git a/src/db-paired-turn-attempt-recovery.test.ts b/src/db-paired-turn-attempt-recovery.test.ts new file mode 100644 index 0000000..ebbad83 --- /dev/null +++ b/src/db-paired-turn-attempt-recovery.test.ts @@ -0,0 +1,144 @@ +import { beforeEach, describe, expect, it } from 'vitest'; + +import { + _initTestDatabase, + createPairedTask, + getPairedTurnAttempts, + markPairedTurnRunning, + recoverInterruptedPairedTurnAttemptsForService, +} from './db.js'; +import { CODEX_MAIN_SERVICE_ID } from './config.js'; +import { buildPairedTurnIdentity } from './paired-turn-identity.js'; +import type { PairedTask } from './types.js'; + +function makeTask(overrides: Partial = {}): PairedTask { + return { + id: 'restart-running-attempt-task', + chat_jid: 'dc:restart-room', + group_folder: 'restart-room', + owner_service_id: CODEX_MAIN_SERVICE_ID, + reviewer_service_id: 'claude', + owner_agent_type: 'codex', + reviewer_agent_type: 'claude-code', + arbiter_agent_type: null, + title: null, + source_ref: 'HEAD', + plan_notes: null, + review_requested_at: null, + round_trip_count: 1, + owner_failure_count: 0, + owner_step_done_streak: 0, + finalize_step_done_count: 0, + task_done_then_user_reopen_count: 0, + empty_step_done_streak: 0, + status: 'active', + arbiter_verdict: null, + arbiter_requested_at: null, + completion_reason: null, + created_at: '2026-05-27T00:00:00.000Z', + updated_at: '2026-05-27T00:10:00.000Z', + ...overrides, + }; +} + +describe('paired turn attempt restart recovery', () => { + beforeEach(() => { + _initTestDatabase(); + }); + + it('fails same-service running attempts so restart recovery can retry the turn', () => { + const task = makeTask(); + 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: 'run-before-restart', + }); + + const recovered = recoverInterruptedPairedTurnAttemptsForService({ + serviceId: CODEX_MAIN_SERVICE_ID, + now: '2026-05-27T00:11:00.000Z', + }); + + expect(recovered).toEqual([ + expect.objectContaining({ + chat_jid: task.chat_jid, + group_folder: task.group_folder, + task_id: task.id, + task_status: 'active', + turn_id: turnIdentity.turnId, + attempt_no: 1, + role: 'owner', + intent_kind: 'owner-turn', + }), + ]); + expect(getPairedTurnAttempts(turnIdentity.turnId)).toEqual([ + expect.objectContaining({ + attempt_no: 1, + state: 'failed', + active_run_id: null, + completed_at: '2026-05-27T00:11:00.000Z', + last_error: 'Interrupted by service restart before completion.', + }), + ]); + + markPairedTurnRunning({ + turnIdentity, + executorServiceId: CODEX_MAIN_SERVICE_ID, + executorAgentType: 'codex', + runId: 'run-after-restart', + }); + + expect(getPairedTurnAttempts(turnIdentity.turnId)).toEqual([ + expect.objectContaining({ + attempt_no: 1, + state: 'failed', + }), + expect.objectContaining({ + attempt_no: 2, + parent_attempt_id: `${turnIdentity.turnId}:attempt:1`, + state: 'running', + active_run_id: 'run-after-restart', + }), + ]); + }); + + it('does not touch running attempts owned by another service', () => { + const task = makeTask({ id: 'other-service-task' }); + createPairedTask(task); + const turnIdentity = buildPairedTurnIdentity({ + taskId: task.id, + taskUpdatedAt: task.updated_at, + intentKind: 'reviewer-turn', + role: 'reviewer', + }); + + markPairedTurnRunning({ + turnIdentity, + executorServiceId: 'claude', + executorAgentType: 'claude-code', + runId: 'reviewer-run', + }); + + expect( + recoverInterruptedPairedTurnAttemptsForService({ + serviceId: CODEX_MAIN_SERVICE_ID, + now: '2026-05-27T00:11:00.000Z', + }), + ).toEqual([]); + expect(getPairedTurnAttempts(turnIdentity.turnId)).toEqual([ + expect.objectContaining({ + state: 'running', + active_run_id: 'reviewer-run', + }), + ]); + }); +}); diff --git a/src/db.ts b/src/db.ts index 90f346c..f547eb0 100644 --- a/src/db.ts +++ b/src/db.ts @@ -117,6 +117,7 @@ export { listPairedWorkspacesForTask, markPairedTurnRunning, refreshPairedTaskExecutionLease, + recoverInterruptedPairedTurnAttemptsForService, releasePairedTaskExecutionLease, reservePairedTurnReservation, setChannelOwnerLease, @@ -133,7 +134,10 @@ export type { MemorySourceKind, RecallMemoryQuery, } from './db/memories.js'; -export type { PairedTurnAttemptRecord } from './db/paired-turn-attempts.js'; +export type { + InterruptedPairedTurnAttemptRecoveryCandidate, + PairedTurnAttemptRecord, +} from './db/paired-turn-attempts.js'; export type { PairedTurnRecord } from './db/paired-turns.js'; export type { ChatInfo } from './db/messages.js'; export type { WorkItem } from './db/work-items.js'; diff --git a/src/db/paired-turn-attempts.ts b/src/db/paired-turn-attempts.ts index d32a4eb..f6bccba 100644 --- a/src/db/paired-turn-attempts.ts +++ b/src/db/paired-turn-attempts.ts @@ -34,6 +34,18 @@ export interface PairedTurnAttemptRecord { last_error: string | null; } +export interface InterruptedPairedTurnAttemptRecoveryCandidate { + chat_jid: string; + group_folder: string; + task_id: string; + task_status: string; + turn_id: string; + attempt_id: string; + attempt_no: number; + role: PairedRoomRole; + intent_kind: PairedTurnIdentity['intentKind']; +} + export interface OwnerCodexBadRequestFailureSummary { taskId: string; failures: number; @@ -431,6 +443,74 @@ export function getPairedTurnAttemptIdFromDatabase( return row?.attempt_id ?? null; } +export function recoverInterruptedPairedTurnAttemptsForServiceInDatabase( + database: Database, + args: { + serviceId: string; + now?: string; + error?: string; + }, +): InterruptedPairedTurnAttemptRecoveryCandidate[] { + const serviceId = normalizeServiceId(args.serviceId); + const now = args.now ?? new Date().toISOString(); + const error = + args.error ?? 'Interrupted by service restart before completion.'; + return database.transaction(() => { + const rows = database + .prepare( + ` + SELECT + tasks.chat_jid, + tasks.group_folder, + attempts.task_id, + tasks.status AS task_status, + attempts.turn_id, + attempts.attempt_id, + attempts.attempt_no, + attempts.role, + attempts.intent_kind + FROM paired_turn_attempts attempts + JOIN paired_tasks tasks + ON tasks.id = attempts.task_id + WHERE attempts.state = 'running' + AND attempts.executor_service_id = ? + 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 + ) + ORDER BY attempts.updated_at ASC, attempts.attempt_no ASC + `, + ) + .all(serviceId) as InterruptedPairedTurnAttemptRecoveryCandidate[]; + + if (rows.length === 0) { + return rows; + } + + const update = database.prepare( + ` + UPDATE paired_turn_attempts + SET state = 'failed', + active_run_id = NULL, + updated_at = ?, + completed_at = ?, + last_error = ? + WHERE attempt_id = ? + AND state = 'running' + AND executor_service_id = ? + `, + ); + for (const row of rows) { + update.run(now, now, error, row.attempt_id, serviceId); + } + return rows; + })(); +} + export function getOwnerCodexBadRequestFailureSummaryForTaskFromDatabase( database: Database, args: { diff --git a/src/db/runtime-paired.ts b/src/db/runtime-paired.ts index 30049e0..1703d34 100644 --- a/src/db/runtime-paired.ts +++ b/src/db/runtime-paired.ts @@ -45,6 +45,8 @@ import { clearPairedTurnAttemptsInDatabase, getOwnerCodexBadRequestFailureSummaryForTaskFromDatabase, getPairedTurnAttemptsForTurnFromDatabase, + recoverInterruptedPairedTurnAttemptsForServiceInDatabase, + type InterruptedPairedTurnAttemptRecoveryCandidate, type OwnerCodexBadRequestFailureSummary, type PairedTurnAttemptRecord, } from './paired-turn-attempts.js'; @@ -239,6 +241,17 @@ export function getPairedTurnAttempts( return getPairedTurnAttemptsForTurnFromDatabase(requireDatabase(), turnId); } +export function recoverInterruptedPairedTurnAttemptsForService(args: { + serviceId: string; + now?: string; + error?: string; +}): InterruptedPairedTurnAttemptRecoveryCandidate[] { + return recoverInterruptedPairedTurnAttemptsForServiceInDatabase( + requireDatabase(), + args, + ); +} + export function getOwnerCodexBadRequestFailureSummaryForTask(args: { taskId: string; threshold: number; diff --git a/src/index.ts b/src/index.ts index 27d2cdf..43f5a5f 100644 --- a/src/index.ts +++ b/src/index.ts @@ -19,6 +19,7 @@ import { writeGroupsSnapshot } from './agent-runner.js'; import { hasRecentRestartAnnouncement, initDatabase, + recoverInterruptedPairedTurnAttemptsForService, storeChatMetadata, storeMessage, } from './db.js'; @@ -259,6 +260,45 @@ async function announceRestartRecovery( return null; } +function queueInterruptedPairedTurnAttemptRecovery(): void { + for (const candidate of recoverInterruptedPairedTurnAttemptsForService({ + serviceId: SERVICE_ID, + })) { + const binding = runtimeState.getRoomBindings()[candidate.chat_jid]; + if (!binding) { + logger.warn( + { + chatJid: candidate.chat_jid, + groupFolder: candidate.group_folder, + taskId: candidate.task_id, + turnId: candidate.turn_id, + attemptId: candidate.attempt_id, + }, + 'Skipped interrupted paired turn recovery because room binding is missing', + ); + continue; + } + queue.enqueueMessageCheck( + candidate.chat_jid, + resolveGroupIpcPath(binding.folder), + ); + logger.info( + { + chatJid: candidate.chat_jid, + groupFolder: binding.folder, + taskId: candidate.task_id, + taskStatus: candidate.task_status, + turnId: candidate.turn_id, + attemptId: candidate.attempt_id, + attemptNo: candidate.attempt_no, + role: candidate.role, + intentKind: candidate.intent_kind, + }, + 'Queued interrupted paired turn attempt for restart recovery', + ); + } +} + async function main(): Promise { const processStartedAtMs = Date.now(); initDatabase(); @@ -477,6 +517,7 @@ async function main(): Promise { }); queue.setProcessMessagesFn(runtime.processGroupMessages); queue.enterRecoveryMode(); + queueInterruptedPairedTurnAttemptRecovery(); runtime.recoverPendingMessages(); const restartContext = await announceRestartRecovery(processStartedAtMs); const roomBindings = runtimeState.getRoomBindings();