From 126ca85bdf09846a1f8dd2a6d20495809b476588 Mon Sep 17 00:00:00 2001 From: ejclaw Date: Fri, 10 Apr 2026 14:06:41 +0900 Subject: [PATCH] Repair startup FK recovery and paired turn runtime fixes --- src/db.test.ts | 59 ++++++++++++++++++++++++ src/db/bootstrap.ts | 21 +++++++++ src/message-agent-executor.test.ts | 62 +++++++++++++++++++++++++ src/message-agent-executor.ts | 10 ++-- src/message-runtime-handoffs.test.ts | 69 +++++++++++++--------------- src/message-runtime-handoffs.ts | 5 +- src/message-runtime-rules.test.ts | 7 ++- src/message-runtime-rules.ts | 18 +++++--- src/paired-execution-context.test.ts | 1 + src/paired-execution-context.ts | 3 ++ 10 files changed, 203 insertions(+), 52 deletions(-) diff --git a/src/db.test.ts b/src/db.test.ts index f0793b4..85942c7 100644 --- a/src/db.test.ts +++ b/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'; diff --git a/src/db/bootstrap.ts b/src/db/bootstrap.ts index d836f7e..f4f8c17 100644 --- a/src/db/bootstrap.ts +++ b/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 { diff --git a/src/message-agent-executor.test.ts b/src/message-agent-executor.test.ts index 6a1f9c7..0bd37bb 100644 --- a/src/message-agent-executor.test.ts +++ b/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' }; diff --git a/src/message-agent-executor.ts b/src/message-agent-executor.ts index 0b924c3..4b935a2 100644 --- a/src/message-agent-executor.ts +++ b/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 !== - pairedExecutionContext.task.updated_at + runtimePairedTurnIdentity.taskUpdatedAt !== preparedTurnTaskUpdatedAt ) { throw new Error( `Paired turn ${runtimePairedTurnIdentity.turnId} task_updated_at does not match the prepared execution context`, diff --git a/src/message-runtime-handoffs.test.ts b/src/message-runtime-handoffs.test.ts index fa28129..855aeac 100644 --- a/src/message-runtime-handoffs.test.ts +++ b/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,35 +53,30 @@ describe('message-runtime-handoffs', () => { vi.clearAllMocks(); }); - it('does not expose reviewer-targeted handoffs to the codex-main poller', () => { - vi.mocked(db.getPendingServiceHandoffs).mockImplementation( - (targetServiceId?: string) => - targetServiceId === 'codex-main' - ? [] - : ([ - { - id: 1, - chat_jid: 'group@test', - group_folder: 'test-group', - source_service_id: 'claude', - target_service_id: 'codex-review', - source_role: 'reviewer', - source_agent_type: 'claude-code', - target_role: 'reviewer', - target_agent_type: 'codex', - prompt: 'review retry', - status: 'pending', - start_seq: 1, - end_seq: 2, - reason: 'reviewer-auth-failure', - intended_role: 'reviewer', - created_at: '2026-04-10T00:00:00.000Z', - claimed_at: null, - completed_at: null, - last_error: null, - }, - ] as any), - ); + it('exposes reviewer-targeted handoffs to the unified runtime poller', () => { + vi.mocked(db.getAllPendingServiceHandoffs).mockReturnValue([ + { + id: 1, + chat_jid: 'group@test', + group_folder: 'test-group', + source_service_id: 'claude', + target_service_id: 'codex-review', + source_role: 'reviewer', + source_agent_type: 'claude-code', + target_role: 'reviewer', + target_agent_type: 'codex', + prompt: 'review retry', + status: 'pending', + start_seq: 1, + end_seq: 2, + reason: 'reviewer-auth-failure', + intended_role: 'reviewer', + created_at: '2026-04-10T00:00:00.000Z', + claimed_at: null, + completed_at: null, + last_error: null, + }, + ] 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.claimServiceHandoff).not.toHaveBeenCalled(); - expect(enqueueTask).not.toHaveBeenCalled(); + expect(db.getAllPendingServiceHandoffs).toHaveBeenCalled(); + expect(db.claimServiceHandoff).toHaveBeenCalledWith(1); + expect(enqueueTask).toHaveBeenCalledWith( + 'group@test', + 'handoff:1', + expect.any(Function), + ); }); it('fails a claimed handoff closed when its intended role cannot be resolved', async () => { diff --git a/src/message-runtime-handoffs.ts b/src/message-runtime-handoffs.ts index 990dea5..2a3cb07 100644 --- a/src/message-runtime-handoffs.ts +++ b/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 { - for (const handoff of getPendingServiceHandoffs(SERVICE_SESSION_SCOPE)) { + for (const handoff of getAllPendingServiceHandoffs()) { if (!claimServiceHandoff(handoff.id)) { continue; } diff --git a/src/message-runtime-rules.test.ts b/src/message-runtime-rules.test.ts index f31c406..5e64534 100644 --- a/src/message-runtime-rules.test.ts +++ b/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', () => { diff --git a/src/message-runtime-rules.ts b/src/message-runtime-rules.ts index 3e152da..96b9af0 100644 --- a/src/message-runtime-rules.ts +++ b/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, diff --git a/src/paired-execution-context.test.ts b/src/paired-execution-context.test.ts index af608ec..d6071dd 100644 --- a/src/paired-execution-context.test.ts +++ b/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', diff --git a/src/paired-execution-context.ts b/src/paired-execution-context.ts index 07e972f..cdef09e 100644 --- a/src/paired-execution-context.ts +++ b/src/paired-execution-context.ts @@ -171,6 +171,7 @@ function ensureActiveTask( export interface PreparedPairedExecutionContext { task: PairedTask; + claimedTaskUpdatedAt?: string; workspace: PairedWorkspace | null; envOverrides: Record; 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,