Merge pull request #174 from phj1081/codex/restart-attempt-recovery

fix: recover interrupted paired turn attempts

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> = {}): 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',
+      }),
+    ]);
+  });
+});

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';

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: {

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;

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<void> {
   const processStartedAtMs = Date.now();
   initDatabase();
@@ -477,6 +517,7 @@ async function main(): Promise<void> {
   });
   queue.setProcessMessagesFn(runtime.processGroupMessages);
   queue.enterRecoveryMode();
+  queueInterruptedPairedTurnAttemptRecovery();
   runtime.recoverPendingMessages();
   const restartContext = await announceRestartRecovery(processStartedAtMs);
   const roomBindings = runtimeState.getRoomBindings();