Repair startup FK recovery and paired turn runtime fixes

src/db.test.ts

@@ -698,6 +698,65 @@ describe('task CRUD', () => {
     expect(getTaskById('task-3')).toBeUndefined();
   });
 
+  it('prunes orphan task_run_logs rows during database initialization', () => {
+    const tempDir = fs.mkdtempSync('/tmp/ejclaw-task-log-fk-repair-');
+    const dbPath = path.join(tempDir, 'messages.db');
+    const legacyDb = new Database(dbPath);
+
+    try {
+      initializeDatabaseSchema(legacyDb);
+      legacyDb
+        .prepare(
+          `INSERT INTO scheduled_tasks (
+             id, group_folder, chat_jid, prompt, schedule_type, schedule_value, created_at
+           ) VALUES (?, ?, ?, ?, ?, ?, ?)`,
+        )
+        .run(
+          'task-legacy-orphan',
+          'infra-auto',
+          'dc:legacy',
+          'legacy task',
+          'once',
+          '2026-03-20T19:00:00.000Z',
+          '2026-03-20T18:59:00.000Z',
+        );
+      legacyDb
+        .prepare(
+          `INSERT INTO task_run_logs (task_id, run_at, duration_ms, status, result, error)
+           VALUES (?, ?, ?, ?, ?, ?)`,
+        )
+        .run(
+          'task-legacy-orphan',
+          '2026-03-20T19:00:58.882Z',
+          0,
+          'error',
+          null,
+          'Group not found: infra-auto',
+        );
+      legacyDb.exec('PRAGMA foreign_keys = OFF');
+      legacyDb
+        .prepare('DELETE FROM scheduled_tasks WHERE id = ?')
+        .run('task-legacy-orphan');
+    } finally {
+      legacyDb.close();
+    }
+
+    expect(() => _initTestDatabaseFromFile(dbPath)).not.toThrow();
+
+    const migratedDb = new Database(dbPath, { readonly: true });
+    try {
+      const remainingLogs = migratedDb
+        .prepare(`SELECT COUNT(*) AS count FROM task_run_logs WHERE task_id = ?`)
+        .get('task-legacy-orphan') as { count: number };
+      const fkViolations = migratedDb.prepare('PRAGMA foreign_key_check').all();
+
+      expect(remainingLogs.count).toBe(0);
+      expect(fkViolations).toHaveLength(0);
+    } finally {
+      migratedDb.close();
+    }
+  });
+
   it('deletes task-scoped IPC and session directories when removing a task', () => {
     const taskId = 'task-cleanup';
     const groupFolder = 'cleanup-group';

src/db/bootstrap.ts

@@ -3,6 +3,7 @@ import fs from 'fs';
 import path from 'path';
 
 import { ASSISTANT_NAME, STORE_DIR } from '../config.js';
+import { logger } from '../logger.js';
 import { applyBaseSchema } from './base-schema.js';
 import { applySchemaMigrations } from './schema.js';
 
@@ -10,6 +11,19 @@ function setForeignKeys(database: Database, enabled: boolean): void {
   database.exec(`PRAGMA foreign_keys = ${enabled ? 'ON' : 'OFF'}`);
 }
 
+function pruneOrphanTaskRunLogs(database: Database): number {
+  const result = database
+    .prepare(
+      `DELETE FROM task_run_logs
+       WHERE NOT EXISTS (
+         SELECT 1 FROM scheduled_tasks
+         WHERE scheduled_tasks.id = task_run_logs.task_id
+       )`,
+    )
+    .run();
+  return result.changes;
+}
+
 function assertNoForeignKeyViolations(database: Database): void {
   const violations = database
     .prepare('PRAGMA foreign_key_check')
@@ -35,8 +49,15 @@ export function initializeDatabaseSchema(database: Database): void {
   applySchemaMigrations(database, {
     assistantName: ASSISTANT_NAME,
   });
+  const prunedTaskRunLogs = pruneOrphanTaskRunLogs(database);
   setForeignKeys(database, true);
   assertNoForeignKeyViolations(database);
+  if (prunedTaskRunLogs > 0) {
+    logger.warn(
+      { count: prunedTaskRunLogs },
+      'Pruned orphan task_run_logs rows during database initialization',
+    );
+  }
 }
 
 export function openPersistentDatabase(): Database {

src/message-agent-executor.test.ts

@@ -743,6 +743,68 @@ describe('runAgentForGroup room memory', () => {
     expect(agentRunner.runAgentProcess).not.toHaveBeenCalled();
   });
 
+  it('allows a claimed paired turn revision when preparation advances the task before execution', async () => {
+    const group = { ...makeGroup(), folder: 'test-group' };
+
+    vi.mocked(
+      pairedExecutionContext.preparePairedExecutionContext,
+    ).mockReturnValue({
+      task: {
+        id: 'paired-task-prep-advance',
+        chat_jid: 'group@test',
+        group_folder: 'test-group',
+        owner_service_id: 'claude',
+        reviewer_service_id: 'codex-review',
+        title: null,
+        source_ref: 'HEAD',
+        plan_notes: null,
+        round_trip_count: 0,
+        review_requested_at: '2026-04-10T00:00:00.000Z',
+        status: 'in_review',
+        arbiter_verdict: null,
+        arbiter_requested_at: null,
+        completion_reason: null,
+        created_at: '2026-04-10T00:00:00.000Z',
+        updated_at: '2026-04-10T01:00:00.000Z',
+      },
+      claimedTaskUpdatedAt: '2026-04-10T00:00:00.000Z',
+      workspace: null,
+      envOverrides: {},
+    });
+
+    await expect(
+      runAgentForGroup(makeDeps(), {
+        group,
+        prompt: 'hello',
+        chatJid: 'group@test',
+        runId: 'run-claimed-paired-turn-revision',
+        forcedRole: 'reviewer',
+        pairedTurnIdentity: {
+          turnId:
+            'paired-task-prep-advance:2026-04-10T00:00:00.000Z:reviewer-turn',
+          taskId: 'paired-task-prep-advance',
+          taskUpdatedAt: '2026-04-10T00:00:00.000Z',
+          intentKind: 'reviewer-turn',
+          role: 'reviewer',
+        },
+      }),
+    ).resolves.toBe('success');
+
+    expect(agentRunner.runAgentProcess).toHaveBeenCalledWith(
+      group,
+      expect.any(Object),
+      expect.any(Function),
+      expect.any(Function),
+      expect.objectContaining({
+        EJCLAW_PAIRED_TURN_ID:
+          'paired-task-prep-advance:2026-04-10T00:00:00.000Z:reviewer-turn',
+        EJCLAW_PAIRED_TURN_ROLE: 'reviewer',
+        EJCLAW_PAIRED_TURN_INTENT: 'reviewer-turn',
+        EJCLAW_PAIRED_TASK_UPDATED_AT: '2026-04-10T00:00:00.000Z',
+      }),
+    );
+  });
+
   it('fails closed when a persisted paired turn revision mismatches the latest paired task fallback', async () => {
     const group = { ...makeGroup(), folder: 'test-group' };
 

src/message-agent-executor.ts

@@ -195,12 +195,17 @@ export async function runAgentForGroup(
     roomRoleContext,
     hasHumanMessage: args.hasHumanMessage,
   });
+  const preparedTurnTaskUpdatedAt = pairedExecutionContext
+    ? (pairedExecutionContext.claimedTaskUpdatedAt ??
+      pairedExecutionContext.task.updated_at)
+    : undefined;
   const runtimePairedTurnIdentity =
     args.pairedTurnIdentity ??
     (pairedExecutionContext
       ? resolveRuntimePairedTurnIdentity({
           taskId: pairedExecutionContext.task.id,
-          taskUpdatedAt: pairedExecutionContext.task.updated_at,
+          taskUpdatedAt:
+            preparedTurnTaskUpdatedAt ?? pairedExecutionContext.task.updated_at,
           role: activeRole,
           taskStatus: pairedExecutionContext.task.status,
           hasHumanMessage: args.hasHumanMessage,
@@ -230,8 +235,7 @@ export async function runAgentForGroup(
     }
     if (
       pairedExecutionContext &&
-      runtimePairedTurnIdentity.taskUpdatedAt !==
+      runtimePairedTurnIdentity.taskUpdatedAt !== preparedTurnTaskUpdatedAt
-        pairedExecutionContext.task.updated_at
     ) {
       throw new Error(
         `Paired turn ${runtimePairedTurnIdentity.turnId} task_updated_at does not match the prepared execution context`,

src/message-runtime-handoffs.test.ts

@@ -1,16 +1,12 @@
 import { beforeEach, describe, expect, it, vi } from 'vitest';
 
-vi.mock('./config.js', () => ({
-  SERVICE_SESSION_SCOPE: 'codex-main',
-}));
-
 vi.mock('./db.js', () => ({
   claimServiceHandoff: vi.fn(() => true),
   claimPairedTurnReservation: vi.fn(() => true),
   completeServiceHandoffAndAdvanceTargetCursor: vi.fn(() => null),
   failServiceHandoff: vi.fn(),
+  getAllPendingServiceHandoffs: vi.fn(() => []),
   getPairedTurnOutputs: vi.fn(() => []),
-  getPendingServiceHandoffs: vi.fn(() => []),
   reservePairedTurnReservation: vi.fn(() => true),
   _clearPairedTurnReservationsForTests: vi.fn(),
 }));
@@ -57,12 +53,8 @@ describe('message-runtime-handoffs', () => {
     vi.clearAllMocks();
   });
 
-  it('does not expose reviewer-targeted handoffs to the codex-main poller', () => {
+  it('exposes reviewer-targeted handoffs to the unified runtime poller', () => {
-    vi.mocked(db.getPendingServiceHandoffs).mockImplementation(
+    vi.mocked(db.getAllPendingServiceHandoffs).mockReturnValue([
-      (targetServiceId?: string) =>
-        targetServiceId === 'codex-main'
-          ? []
-          : ([
       {
         id: 1,
         chat_jid: 'group@test',
@@ -84,8 +76,7 @@ describe('message-runtime-handoffs', () => {
         completed_at: null,
         last_error: null,
       },
-            ] as any),
+    ] as any);
-    );
 
     const enqueueTask = vi.fn();
 
@@ -94,9 +85,13 @@ describe('message-runtime-handoffs', () => {
       processClaimedHandoff: vi.fn(),
     });
 
-    expect(db.getPendingServiceHandoffs).toHaveBeenCalledWith('codex-main');
+    expect(db.getAllPendingServiceHandoffs).toHaveBeenCalled();
-    expect(db.claimServiceHandoff).not.toHaveBeenCalled();
+    expect(db.claimServiceHandoff).toHaveBeenCalledWith(1);
-    expect(enqueueTask).not.toHaveBeenCalled();
+    expect(enqueueTask).toHaveBeenCalledWith(
+      'group@test',
+      'handoff:1',
+      expect.any(Function),
+    );
   });
 
   it('fails a claimed handoff closed when its intended role cannot be resolved', async () => {

src/message-runtime-handoffs.ts

@@ -1,10 +1,9 @@
-import { SERVICE_SESSION_SCOPE } from './config.js';
 import { getErrorMessage } from './utils.js';
 import {
   claimServiceHandoff,
   completeServiceHandoffAndAdvanceTargetCursor,
   failServiceHandoff,
-  getPendingServiceHandoffs,
+  getAllPendingServiceHandoffs,
   type ServiceHandoff,
 } from './db.js';
 import { findChannel, findChannelByName } from './router.js';
@@ -29,7 +28,7 @@ export function enqueuePendingHandoffs(args: {
   ) => void;
   processClaimedHandoff: (handoff: ServiceHandoff) => Promise<void>;
 }): void {
-  for (const handoff of getPendingServiceHandoffs(SERVICE_SESSION_SCOPE)) {
+  for (const handoff of getAllPendingServiceHandoffs()) {
     if (!claimServiceHandoff(handoff.id)) {
       continue;
     }

src/message-runtime-rules.test.ts

@@ -23,6 +23,7 @@ import {
   resolveLeaseServiceId,
   type EffectiveChannelLease,
 } from './service-routing.js';
+import { resolveRoleServiceShadow } from './role-service-shadow.js';
 
 const baseLease: EffectiveChannelLease = {
   chat_jid: 'chat-1',
@@ -354,7 +355,7 @@ describe('message-runtime-rules', () => {
     );
   });
 
-  it('applies forced agent type overrides without changing the configured role target', () => {
+  it('applies forced agent type overrides and switches to the forced role service shadow', () => {
     const resolution = resolveExecutionTarget({
       lease: baseLease,
       pairedTaskStatus: 'review_ready',
@@ -368,7 +369,9 @@ describe('message-runtime-rules', () => {
       configuredAgentType: 'claude-code',
       effectiveAgentType: 'codex',
     });
-    expect(resolution.effectiveServiceId).toBe(resolution.reviewerServiceId);
+    expect(resolution.effectiveServiceId).toBe(
+      resolveRoleServiceShadow('reviewer', 'codex'),
+    );
   });
 
   it('always gives arbiter a dedicated session folder', () => {

src/message-runtime-rules.ts

@@ -9,6 +9,7 @@ import {
   resolveLeaseServiceId,
   type EffectiveChannelLease,
 } from './service-routing.js';
+import { resolveRoleServiceShadow } from './role-service-shadow.js';
 import {
   resolveAgentTypeForRole,
   resolveRoleAgentPlan,
@@ -307,13 +308,6 @@ export function resolveExecutionTarget(args: {
     (args.forcedRole === 'arbiter' && args.lease.arbiter_agent_type),
   );
   const activeRole = canHonorForcedRole ? args.forcedRole! : inferredRole;
-  const effectiveServiceId = resolveLeaseServiceId(args.lease, activeRole);
-  if (!effectiveServiceId) {
-    throw new Error(`Missing runtime service id for ${activeRole} lease`);
-  }
-
-  const reviewerServiceId = resolveLeaseServiceId(args.lease, 'reviewer');
-  const arbiterServiceId = resolveLeaseServiceId(args.lease, 'arbiter');
   const roleAgentPlan = resolveConfiguredRoleAgentPlan(
     args.lease.reviewer_agent_type != null,
     args.groupAgentType,
@@ -323,6 +317,16 @@ export function resolveExecutionTarget(args: {
     args.groupAgentType,
   );
   const effectiveAgentType = args.forcedAgentType ?? configuredAgentType;
+  const effectiveServiceId =
+    (args.forcedAgentType
+      ? resolveRoleServiceShadow(activeRole, effectiveAgentType)
+      : null) ?? resolveLeaseServiceId(args.lease, activeRole);
+  if (!effectiveServiceId) {
+    throw new Error(`Missing runtime service id for ${activeRole} lease`);
+  }
+
+  const reviewerServiceId = resolveLeaseServiceId(args.lease, 'reviewer');
+  const arbiterServiceId = resolveLeaseServiceId(args.lease, 'arbiter');
 
   return {
     inferredRole,

src/paired-execution-context.test.ts

@@ -327,6 +327,7 @@ describe('paired execution context', () => {
       'task-1',
       expect.objectContaining({ status: 'in_review' }),
     );
+    expect(result?.claimedTaskUpdatedAt).toBe('2026-03-28T00:00:00.000Z');
     expect(result?.envOverrides).toMatchObject({
       EJCLAW_WORK_DIR: '/tmp/paired/task-1/reviewer',
       EJCLAW_REVIEWER_RUNTIME: '1',

src/paired-execution-context.ts

@@ -171,6 +171,7 @@ function ensureActiveTask(
 
 export interface PreparedPairedExecutionContext {
   task: PairedTask;
+  claimedTaskUpdatedAt?: string;
   workspace: PairedWorkspace | null;
   envOverrides: Record<string, string>;
   gateTurnKind?: string | null;
@@ -209,6 +210,7 @@ export function preparePairedExecutionContext(args: {
   }
 
   const latestTask = getPairedTaskById(task.id) ?? task;
+  const claimedTaskUpdatedAt = latestTask.updated_at;
   let workspace: PairedWorkspace | null = null;
   let blockMessage: string | undefined;
   const now = new Date().toISOString();
@@ -349,6 +351,7 @@ export function preparePairedExecutionContext(args: {
 
   return {
     task: getPairedTaskById(task.id) ?? task,
+    claimedTaskUpdatedAt,
     workspace,
     envOverrides,
     blockMessage,