From 1e46b05b03bf7dce2b85e7313439468a10a44c96 Mon Sep 17 00:00:00 2001 From: ejclaw Date: Thu, 9 Apr 2026 16:31:01 +0900 Subject: [PATCH] fix: stabilize paired runtime turn identity --- src/db.ts | 14 +- src/db/base-schema.ts | 1 + src/db/paired-state.ts | 43 ++- src/db/schema.ts | 143 ++++++++- src/message-runtime-queue.test.ts | 48 ++- src/message-runtime-queue.ts | 56 +++- src/message-runtime-rules.test.ts | 18 ++ src/message-runtime-rules.ts | 30 ++ src/paired-follow-up-scheduler.test.ts | 397 ++++++++++++++++++++++++- src/paired-follow-up-scheduler.ts | 8 +- 10 files changed, 716 insertions(+), 42 deletions(-) diff --git a/src/db.ts b/src/db.ts index 4a7d76c..0836e36 100644 --- a/src/db.ts +++ b/src/db.ts @@ -98,6 +98,7 @@ import { } from './db/work-items.js'; import { claimPairedTurnReservationInDatabase, + clearPairedTaskExecutionLeasesForServiceInDatabase, clearExpiredPairedTaskExecutionLeasesInDatabase, clearPairedTaskExecutionLeasesInDatabase, clearPairedTurnReservationsInDatabase, @@ -288,6 +289,10 @@ function getSchemaMigrationHooks(): SchemaMigrationHooks { export function initDatabase(): void { db = openPersistentDatabase(); initializeDatabaseSchema(db, getSchemaMigrationHooks()); + clearPairedTaskExecutionLeasesForServiceInDatabase( + db, + normalizeServiceId(SERVICE_ID), + ); clearExpiredPairedTaskExecutionLeasesInDatabase(db); migrateJsonStateFromFiles({ setRouterState, @@ -300,6 +305,10 @@ export function initDatabase(): void { export function _initTestDatabase(): void { db = openInMemoryDatabase(); initializeDatabaseSchema(db, getSchemaMigrationHooks()); + clearPairedTaskExecutionLeasesForServiceInDatabase( + db, + normalizeServiceId(SERVICE_ID), + ); clearExpiredPairedTaskExecutionLeasesInDatabase(db); } @@ -307,6 +316,10 @@ export function _initTestDatabase(): void { export function _initTestDatabaseFromFile(dbPath: string): void { db = openDatabaseFromFile(dbPath); initializeDatabaseSchema(db, getSchemaMigrationHooks()); + clearPairedTaskExecutionLeasesForServiceInDatabase( + db, + normalizeServiceId(SERVICE_ID), + ); clearExpiredPairedTaskExecutionLeasesInDatabase(db); } @@ -1235,7 +1248,6 @@ export function claimPairedTurnReservation(args: { taskStatus: PairedTask['status']; roundTripCount: number; taskUpdatedAt: string; - nextTaskUpdatedAt: string; intentKind: PairedTurnReservationIntentKind; runId: string; }): boolean { diff --git a/src/db/base-schema.ts b/src/db/base-schema.ts index a1bba1f..c647e28 100644 --- a/src/db/base-schema.ts +++ b/src/db/base-schema.ts @@ -202,6 +202,7 @@ export function applyBaseSchema(database: Database): void { role TEXT NOT NULL, intent_kind TEXT NOT NULL, claimed_run_id TEXT NOT NULL, + claimed_service_id TEXT NOT NULL, task_status TEXT NOT NULL, task_updated_at TEXT NOT NULL, claimed_at TEXT NOT NULL, diff --git a/src/db/paired-state.ts b/src/db/paired-state.ts index f26579f..5cf742b 100644 --- a/src/db/paired-state.ts +++ b/src/db/paired-state.ts @@ -4,6 +4,8 @@ import { ARBITER_AGENT_TYPE, CODEX_MAIN_SERVICE_ID, CODEX_REVIEW_SERVICE_ID, + SERVICE_ID, + normalizeServiceId, } from '../config.js'; import { resolveStablePairedTaskOwnerAgentType, @@ -51,6 +53,7 @@ export type PairedTaskUpdates = Partial< >; export const PAIRED_TASK_EXECUTION_LEASE_TTL_MS = 10 * 60_000; +const CURRENT_SERVICE_ID = normalizeServiceId(SERVICE_ID); function computeExecutionLeaseExpiry(now: string): string { return new Date( @@ -414,13 +417,12 @@ export function claimPairedTurnReservationInDatabase( taskStatus: PairedTaskStatus; roundTripCount: number; taskUpdatedAt: string; - nextTaskUpdatedAt: string; intentKind: PairedTurnReservationIntentKind; runId: string; }, ): boolean { const tx = database.transaction(() => { - const now = args.nextTaskUpdatedAt; + const now = new Date().toISOString(); const expiresAt = computeExecutionLeaseExpiry(now); const leaseRole = resolveExecutionLeaseRole(args.intentKind); const existingLease = database @@ -434,6 +436,7 @@ export function claimPairedTurnReservationInDatabase( .get(args.taskId) as | { claimed_run_id: string; + claimed_service_id: string; updated_at: string; expires_at: string; } @@ -449,13 +452,14 @@ export function claimPairedTurnReservationInDatabase( role, intent_kind, claimed_run_id, + claimed_service_id, task_status, task_updated_at, claimed_at, updated_at, expires_at ) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) `, ) .run( @@ -464,6 +468,7 @@ export function claimPairedTurnReservationInDatabase( leaseRole, args.intentKind, args.runId, + CURRENT_SERVICE_ID, args.taskStatus, args.taskUpdatedAt, now, @@ -485,6 +490,7 @@ export function claimPairedTurnReservationInDatabase( role = ?, intent_kind = ?, claimed_run_id = ?, + claimed_service_id = ?, task_status = ?, task_updated_at = ?, claimed_at = ?, @@ -501,6 +507,7 @@ export function claimPairedTurnReservationInDatabase( leaseRole, args.intentKind, args.runId, + CURRENT_SERVICE_ID, args.taskStatus, args.taskUpdatedAt, now, @@ -546,14 +553,13 @@ export function claimPairedTurnReservationInDatabase( .prepare( ` UPDATE paired_tasks - SET updated_at = ? + SET updated_at = updated_at WHERE id = ? AND updated_at = ? AND status = ? `, ) .run( - args.nextTaskUpdatedAt, args.taskId, args.taskUpdatedAt, args.taskStatus, @@ -593,10 +599,11 @@ export function releasePairedTaskExecutionLeaseInDatabase( ` DELETE FROM paired_task_execution_leases WHERE task_id = ? + AND claimed_service_id = ? AND claimed_run_id = ? `, ) - .run(args.taskId, args.runId); + .run(args.taskId, CURRENT_SERVICE_ID, args.runId); } export function refreshPairedTaskExecutionLeaseInDatabase( @@ -615,14 +622,36 @@ export function refreshPairedTaskExecutionLeaseInDatabase( SET updated_at = ?, expires_at = ? WHERE task_id = ? + AND claimed_service_id = ? AND claimed_run_id = ? AND expires_at >= ? `, ) - .run(now, computeExecutionLeaseExpiry(now), args.taskId, args.runId, now); + .run( + now, + computeExecutionLeaseExpiry(now), + args.taskId, + CURRENT_SERVICE_ID, + args.runId, + now, + ); return result.changes > 0; } +export function clearPairedTaskExecutionLeasesForServiceInDatabase( + database: Database, + serviceId: string = CURRENT_SERVICE_ID, +): number { + return database + .prepare( + ` + DELETE FROM paired_task_execution_leases + WHERE claimed_service_id = ? + `, + ) + .run(serviceId).changes; +} + export function clearExpiredPairedTaskExecutionLeasesInDatabase( database: Database, now: string = new Date().toISOString(), diff --git a/src/db/schema.ts b/src/db/schema.ts index ca883a7..bace81a 100644 --- a/src/db/schema.ts +++ b/src/db/schema.ts @@ -1,6 +1,10 @@ import { Database } from 'bun:sqlite'; -import { inferAgentTypeFromServiceShadow } from '../role-service-shadow.js'; +import { normalizeServiceId } from '../config.js'; +import { + inferAgentTypeFromServiceShadow, + resolveRoleServiceShadow, +} from '../role-service-shadow.js'; import { backfillLegacyServiceSessions, dropLegacyServiceSessionsTable, @@ -44,6 +48,134 @@ function tryExecMigration(database: Database, sql: string): void { } } +interface LegacyExecutionLeaseServiceRow { + rowid: number; + role: string; + owner_service_id?: string | null; + reviewer_service_id?: string | null; + arbiter_service_id?: string | null; + owner_agent_type?: string | null; + reviewer_agent_type?: string | null; + arbiter_agent_type?: string | null; +} + +function normalizePairedAgentType( + agentType: string | null | undefined, +): 'codex' | 'claude-code' | null { + if (agentType === 'codex' || agentType === 'claude-code') { + return agentType; + } + return null; +} + +function inferLegacyExecutionLeaseServiceId( + row: LegacyExecutionLeaseServiceRow, +): string | null { + switch (row.role) { + case 'owner': { + const ownerAgentType = normalizePairedAgentType(row.owner_agent_type); + return ( + (row.owner_service_id ? normalizeServiceId(row.owner_service_id) : null) ?? + resolveRoleServiceShadow('owner', ownerAgentType) + ); + } + case 'reviewer': { + const reviewerAgentType = normalizePairedAgentType( + row.reviewer_agent_type, + ); + return ( + (row.reviewer_service_id + ? normalizeServiceId(row.reviewer_service_id) + : null) ?? + resolveRoleServiceShadow('reviewer', reviewerAgentType) + ); + } + case 'arbiter': { + const arbiterAgentType = normalizePairedAgentType(row.arbiter_agent_type); + return ( + (row.arbiter_service_id + ? normalizeServiceId(row.arbiter_service_id) + : null) ?? + resolveRoleServiceShadow('arbiter', arbiterAgentType) + ); + } + default: + return null; + } +} + +function backfillLegacyExecutionLeaseServiceShadows(database: Database): void { + if ( + !tableHasColumn(database, 'paired_task_execution_leases', 'claimed_service_id') + ) { + return; + } + + const pairedTasksTable = database + .prepare( + ` + SELECT 1 + FROM sqlite_master + WHERE type = 'table' + AND name = 'paired_tasks' + `, + ) + .get(); + if (!pairedTasksTable) { + return; + } + + const pairedTaskColumns = getTableColumns(database, 'paired_tasks'); + const selectPairedTaskColumn = (columnName: string): string => + pairedTaskColumns.includes(columnName) + ? `paired_tasks.${columnName} AS ${columnName}` + : `NULL AS ${columnName}`; + + const rows = database + .prepare( + ` + SELECT + paired_task_execution_leases.rowid AS rowid, + paired_task_execution_leases.role AS role, + ${selectPairedTaskColumn('owner_service_id')}, + ${selectPairedTaskColumn('reviewer_service_id')}, + ${selectPairedTaskColumn('arbiter_service_id')}, + ${selectPairedTaskColumn('owner_agent_type')}, + ${selectPairedTaskColumn('reviewer_agent_type')}, + ${selectPairedTaskColumn('arbiter_agent_type')} + FROM paired_task_execution_leases + LEFT JOIN paired_tasks + ON paired_tasks.id = paired_task_execution_leases.task_id + WHERE paired_task_execution_leases.claimed_service_id IS NULL + `, + ) + .all() as LegacyExecutionLeaseServiceRow[]; + + if (rows.length === 0) { + return; + } + + const update = database.prepare( + ` + UPDATE paired_task_execution_leases + SET claimed_service_id = ? + WHERE rowid = ? + AND claimed_service_id IS NULL + `, + ); + const tx = database.transaction((leaseRows: LegacyExecutionLeaseServiceRow[]) => { + for (const row of leaseRows) { + const claimedServiceId = inferLegacyExecutionLeaseServiceId(row); + if (!claimedServiceId) { + continue; + } + update.run(claimedServiceId, row.rowid); + } + }); + + tx(rows); +} + export function applySchemaMigrations( database: Database, args: { @@ -149,6 +281,10 @@ export function applySchemaMigrations( database, `ALTER TABLE paired_task_execution_leases ADD COLUMN expires_at TEXT`, ); + tryExecMigration( + database, + `ALTER TABLE paired_task_execution_leases ADD COLUMN claimed_service_id TEXT`, + ); database.exec(` UPDATE paired_task_execution_leases SET expires_at = COALESCE( @@ -156,6 +292,11 @@ export function applySchemaMigrations( strftime('%Y-%m-%dT%H:%M:%fZ', 'now', '+10 minutes') ) `); + // Legacy lease rows predate claimed_service_id. Infer the original service + // from paired-task role metadata when possible instead of blanket-marking + // everything as the current service, which would let startup cleanup delete + // leases that belong to another runtime. + backfillLegacyExecutionLeaseServiceShadows(database); database.exec(` CREATE INDEX IF NOT EXISTS idx_paired_task_execution_leases_expires_at ON paired_task_execution_leases(expires_at) diff --git a/src/message-runtime-queue.test.ts b/src/message-runtime-queue.test.ts index 41ba97e..effbbec 100644 --- a/src/message-runtime-queue.test.ts +++ b/src/message-runtime-queue.test.ts @@ -4,6 +4,7 @@ import { _initTestDatabase, createPairedTask, getPairedTaskById, + insertPairedTurnOutput, updatePairedTaskIfUnchanged, } from './db.js'; import { claimPairedTurnExecution } from './paired-follow-up-scheduler.js'; @@ -76,7 +77,7 @@ describe('message-runtime-queue', () => { task, intentKind: 'reviewer-turn', }), - ).not.toBeNull(); + ).toBe(true); const executeTurn = vi.fn(); const freshTask = getPairedTaskById(task.id); @@ -126,7 +127,7 @@ describe('message-runtime-queue', () => { task, intentKind: 'reviewer-turn', }); - expect(reviewerClaim).not.toBeNull(); + expect(reviewerClaim).toBe(true); const reviewTask = getPairedTaskById(task.id); expect(reviewTask).toBeDefined(); @@ -178,4 +179,47 @@ describe('message-runtime-queue', () => { expect(outcome).toBe(true); expect(executeTurn).not.toHaveBeenCalled(); }); + + it('skips a queued reviewer rerun when the latest persisted turn already belongs to the reviewer', async () => { + const task = makeTask(); + createPairedTask(task); + insertPairedTurnOutput(task.id, 1, 'reviewer', 'DONE\nreview finished'); + + const executeTurn = vi.fn(); + const outcome = await runQueuedGroupTurn({ + chatJid: task.chat_jid, + group: makeGroup(), + runId: 'run-stale-reviewer-rerun', + log: { info: vi.fn(), warn: vi.fn(), error: vi.fn() } as any, + timezone: 'UTC', + missedMessages: [ + { + id: 'bot-1', + chat_jid: task.chat_jid, + sender: 'reviewer-bot@test', + sender_name: 'reviewer', + content: '추가 상태 메시지', + timestamp: '2026-03-30T00:00:02.000Z', + seq: 44, + is_bot_message: true, + }, + ], + task, + roleToChannel: { + owner: null, + reviewer: makeChannel(), + arbiter: null, + }, + ownerChannel: makeChannel(), + lastAgentTimestamps: {}, + saveState: vi.fn(), + executeTurn, + getFixedRoleChannelName: () => 'discord-review', + labelPairedSenders: (_chatJid, messages) => messages, + formatMessages: () => 'formatted prompt', + }); + + expect(outcome).toBe(true); + expect(executeTurn).not.toHaveBeenCalled(); + }); }); diff --git a/src/message-runtime-queue.ts b/src/message-runtime-queue.ts index 1bdeece..355d1d3 100644 --- a/src/message-runtime-queue.ts +++ b/src/message-runtime-queue.ts @@ -13,6 +13,7 @@ import { advanceLastAgentCursor, resolveActiveRole, resolveCursorKeyForRole, + resolveQueuedPairedTurnRole, resolveQueuedTurnRole, } from './message-runtime-rules.js'; import { claimPairedTurnExecution } from './paired-follow-up-scheduler.js'; @@ -102,13 +103,13 @@ export async function runPendingPairedTurnIfNeeded(args: { } if (pendingTurn.channel) { - const claimedUpdatedAt = claimPairedTurnExecution({ + const claimed = claimPairedTurnExecution({ chatJid, runId: args.runId, task, intentKind: pendingTurn.intentKind, }); - if (!claimedUpdatedAt) { + if (!claimed) { args.log.info( { chatJid, @@ -121,7 +122,6 @@ export async function runPendingPairedTurnIfNeeded(args: { ); return true; } - task.updated_at = claimedUpdatedAt; } return executePendingPairedTurn({ @@ -157,13 +157,48 @@ export async function runQueuedGroupTurn(args: { const { chatJid, group, runId, log, missedMessages, task, roleToChannel } = args; const taskStatus = task?.status; - const hasHumanMsg = !isBotOnlyPairedRoomTurn(chatJid, missedMessages); + const hasHumanMsg = task + ? !missedMessages.every( + (message) => message.is_from_me === true || !!message.is_bot_message, + ) + : !isBotOnlyPairedRoomTurn(chatJid, missedMessages); + const lastTurnOutputRole = task + ? (getPairedTurnOutputs(task.id).at(-1)?.role ?? null) + : null; const turnRole = task - ? resolveQueuedTurnRole({ - taskStatus, - hasHumanMessage: hasHumanMsg, - }) + ? hasHumanMsg + ? resolveQueuedTurnRole({ + taskStatus, + hasHumanMessage: true, + }) + : resolveQueuedPairedTurnRole({ + taskStatus, + hasHumanMessage: false, + lastTurnOutputRole, + }) : 'owner'; + if (!turnRole) { + const endSeq = missedMessages[missedMessages.length - 1]?.seq ?? null; + if (endSeq !== null) { + advanceLastAgentCursor( + args.lastAgentTimestamps, + args.saveState, + chatJid, + endSeq, + resolveCursorKeyForRole(chatJid, resolveActiveRole(taskStatus)), + ); + } + log.info( + { + chatJid, + taskId: task?.id ?? null, + taskStatus, + lastTurnOutputRole, + }, + 'Skipped queued paired turn because the latest persisted turn already closed the pending handoff', + ); + return true; + } const turnChannel = turnRole === 'owner' ? args.ownerChannel : roleToChannel[turnRole]; const cursorKey = resolveCursorKeyForRole(chatJid, turnRole); @@ -227,13 +262,13 @@ export async function runQueuedGroupTurn(args: { turnRole, hasHumanMessage: hasHumanMsg, }); - const claimedUpdatedAt = claimPairedTurnExecution({ + const claimed = claimPairedTurnExecution({ chatJid, runId, task, intentKind, }); - if (!claimedUpdatedAt) { + if (!claimed) { log.info( { taskId: task.id, @@ -246,7 +281,6 @@ export async function runQueuedGroupTurn(args: { ); return true; } - task.updated_at = claimedUpdatedAt; } if (endSeq !== null) { diff --git a/src/message-runtime-rules.test.ts b/src/message-runtime-rules.test.ts index a338720..f31c406 100644 --- a/src/message-runtime-rules.test.ts +++ b/src/message-runtime-rules.test.ts @@ -14,6 +14,7 @@ import { matchesExpectedPairedFollowUpIntent, resolveFollowUpDispatch, resolveExecutionTarget, + resolveQueuedPairedTurnRole, resolveNextTurnAction, resolveQueuedTurnRole, resolveSessionFolder, @@ -255,6 +256,23 @@ describe('message-runtime-rules', () => { ).toBe('reviewer'); }); + it('suppresses queued reviewer reruns when the latest persisted turn already belongs to the reviewer', () => { + expect( + resolveQueuedPairedTurnRole({ + taskStatus: 'review_ready', + hasHumanMessage: false, + lastTurnOutputRole: 'reviewer', + }), + ).toBeNull(); + expect( + resolveQueuedPairedTurnRole({ + taskStatus: 'review_ready', + hasHumanMessage: false, + lastTurnOutputRole: 'owner', + }), + ).toBe('reviewer'); + }); + it('resolves reviewer execution target from review_ready task status', () => { const resolution = resolveExecutionTarget({ lease: baseLease, diff --git a/src/message-runtime-rules.ts b/src/message-runtime-rules.ts index a60ada6..3e152da 100644 --- a/src/message-runtime-rules.ts +++ b/src/message-runtime-rules.ts @@ -75,6 +75,36 @@ export function resolveQueuedTurnRole(args: { return resolveActiveRole(args.taskStatus); } +export function resolveQueuedPairedTurnRole(args: { + taskStatus?: PairedTaskStatus | null; + hasHumanMessage: boolean; + lastTurnOutputRole?: PairedRoomRole | null; +}): 'owner' | 'reviewer' | 'arbiter' | null { + if (args.hasHumanMessage) { + return resolveQueuedTurnRole({ + taskStatus: args.taskStatus, + hasHumanMessage: true, + }); + } + + const nextTurnAction = resolveNextTurnAction({ + taskStatus: args.taskStatus, + lastTurnOutputRole: args.lastTurnOutputRole, + }); + + switch (nextTurnAction.kind) { + case 'reviewer-turn': + return 'reviewer'; + case 'arbiter-turn': + return 'arbiter'; + case 'owner-follow-up': + case 'finalize-owner-turn': + return 'owner'; + default: + return null; + } +} + export type NextTurnAction = | { kind: 'none' } | { kind: 'reviewer-turn' } diff --git a/src/paired-follow-up-scheduler.test.ts b/src/paired-follow-up-scheduler.test.ts index e762b51..e9ede8e 100644 --- a/src/paired-follow-up-scheduler.test.ts +++ b/src/paired-follow-up-scheduler.test.ts @@ -5,6 +5,13 @@ import path from 'path'; import { beforeEach, describe, expect, it, vi } from 'vitest'; +import { + CLAUDE_SERVICE_ID, + CODEX_MAIN_SERVICE_ID, + CODEX_REVIEW_SERVICE_ID, + SERVICE_ID, + normalizeServiceId, +} from './config.js'; import { _initTestDatabase, _initTestDatabaseFromFile, @@ -18,6 +25,159 @@ import { schedulePairedFollowUpOnce, } from './paired-follow-up-scheduler.js'; +const CURRENT_SERVICE_ID = normalizeServiceId(SERVICE_ID); +const CURRENT_AGENT_TYPE = + CURRENT_SERVICE_ID === CLAUDE_SERVICE_ID ? 'claude-code' : 'codex'; +const OTHER_SERVICE_ID = + CURRENT_SERVICE_ID === CLAUDE_SERVICE_ID + ? CODEX_MAIN_SERVICE_ID + : CLAUDE_SERVICE_ID; +const OTHER_AGENT_TYPE = + OTHER_SERVICE_ID === CLAUDE_SERVICE_ID ? 'claude-code' : 'codex'; + +function createLegacyLeaseDatabase(args: { + dbPath: string; + taskId: string; + role?: 'owner' | 'reviewer' | 'arbiter'; + intentKind?: + | 'owner-turn' + | 'reviewer-turn' + | 'arbiter-turn' + | 'owner-follow-up' + | 'finalize-owner-turn'; + claimedRunId?: string; + taskStatus?: string; + reviewerServiceId: string; + reviewerAgentType: 'codex' | 'claude-code'; + ownerServiceId?: string; + ownerAgentType?: 'codex' | 'claude-code'; + arbiterServiceId?: string; + arbiterAgentType?: 'codex' | 'claude-code'; +}): void { + const legacyDb = new Database(args.dbPath); + legacyDb.exec(` + CREATE TABLE paired_tasks ( + id TEXT PRIMARY KEY, + chat_jid TEXT NOT NULL, + group_folder TEXT NOT NULL, + owner_service_id TEXT NOT NULL, + reviewer_service_id TEXT NOT NULL, + arbiter_service_id TEXT, + owner_agent_type TEXT, + reviewer_agent_type TEXT, + arbiter_agent_type TEXT, + title TEXT, + source_ref TEXT, + plan_notes TEXT, + review_requested_at TEXT, + round_trip_count INTEGER NOT NULL DEFAULT 0, + status TEXT NOT NULL DEFAULT 'active', + arbiter_verdict TEXT, + arbiter_requested_at TEXT, + completion_reason TEXT, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL + ); + CREATE TABLE paired_task_execution_leases ( + task_id TEXT PRIMARY KEY, + chat_jid TEXT NOT NULL, + role TEXT NOT NULL, + intent_kind TEXT NOT NULL, + claimed_run_id TEXT NOT NULL, + task_status TEXT NOT NULL, + task_updated_at TEXT NOT NULL, + claimed_at TEXT NOT NULL, + updated_at TEXT NOT NULL, + expires_at TEXT NOT NULL, + CHECK (role IN ('owner', 'reviewer', 'arbiter')), + CHECK ( + intent_kind IN ( + 'owner-turn', + 'reviewer-turn', + 'arbiter-turn', + 'owner-follow-up', + 'finalize-owner-turn' + ) + ) + ); + `); + legacyDb + .prepare( + `INSERT INTO paired_tasks ( + id, + chat_jid, + group_folder, + owner_service_id, + reviewer_service_id, + arbiter_service_id, + owner_agent_type, + reviewer_agent_type, + arbiter_agent_type, + title, + source_ref, + plan_notes, + review_requested_at, + round_trip_count, + status, + arbiter_verdict, + arbiter_requested_at, + completion_reason, + created_at, + updated_at + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, + ) + .run( + args.taskId, + 'group@test', + 'test-group', + args.ownerServiceId ?? OTHER_SERVICE_ID, + args.reviewerServiceId, + args.arbiterServiceId ?? null, + args.ownerAgentType ?? OTHER_AGENT_TYPE, + args.reviewerAgentType, + args.arbiterAgentType ?? null, + null, + 'HEAD', + null, + null, + 1, + 'review_ready', + null, + null, + null, + '2026-03-30T00:00:00.000Z', + '2026-03-30T00:00:00.000Z', + ); + legacyDb + .prepare( + `INSERT INTO paired_task_execution_leases ( + task_id, + chat_jid, + role, + intent_kind, + claimed_run_id, + task_status, + task_updated_at, + claimed_at, + updated_at, + expires_at + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, + ) + .run( + args.taskId, + 'group@test', + args.role ?? 'reviewer', + args.intentKind ?? 'reviewer-turn', + args.claimedRunId ?? `legacy-run-${args.role ?? 'reviewer'}`, + args.taskStatus ?? 'review_ready', + '2026-03-30T00:00:00.000Z', + '2026-03-30T00:00:00.000Z', + '2026-03-30T00:00:00.000Z', + '2099-03-30T00:10:00.000Z', + ); + legacyDb.close(); +} + describe('paired follow-up scheduler', () => { beforeEach(() => { _initTestDatabase(); @@ -160,6 +320,60 @@ describe('paired follow-up scheduler', () => { expect(enqueue).toHaveBeenCalledTimes(2); }); + it('does not create a fresh reviewer handoff identity after a pure claim leaves the semantic task revision unchanged', () => { + const enqueue = vi.fn(); + const task = { + id: 'task-semantic-reviewer-handoff', + chat_jid: 'group@test', + group_folder: 'test-group', + owner_service_id: 'claude', + reviewer_service_id: 'codex-main', + title: null, + source_ref: 'HEAD', + plan_notes: null, + review_requested_at: null, + arbiter_verdict: null, + arbiter_requested_at: null, + completion_reason: null, + created_at: '2026-03-30T00:00:00.000Z', + status: 'review_ready', + round_trip_count: 1, + updated_at: '2026-03-30T00:00:00.000Z', + } as const; + createPairedTask(task as any); + + expect( + schedulePairedFollowUpOnce({ + chatJid: task.chat_jid, + runId: 'run-reviewer-schedule-1', + task, + intentKind: 'reviewer-turn', + enqueue, + }), + ).toBe(true); + expect( + claimPairedTurnExecution({ + chatJid: task.chat_jid, + runId: 'run-reviewer-claim-1', + task, + intentKind: 'reviewer-turn', + }), + ).toBe(true); + + const freshRefetch = getPairedTaskById(task.id); + expect(freshRefetch?.updated_at).toBe(task.updated_at); + expect( + schedulePairedFollowUpOnce({ + chatJid: task.chat_jid, + runId: 'run-reviewer-schedule-2', + task: freshRefetch ?? task, + intentKind: 'reviewer-turn', + enqueue, + }), + ).toBe(false); + expect(enqueue).toHaveBeenCalledTimes(1); + }); + it('blocks a fresh-refetched task revision from reclaiming the same turn while the execution lease is active', () => { const task = { id: 'task-1', @@ -195,9 +409,9 @@ describe('paired follow-up scheduler', () => { intentKind: 'reviewer-turn', }); - expect(first).toMatch(/^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d{3}Z$/); - expect(second).toBeNull(); - expect(getPairedTaskById(task.id)?.updated_at).toBe(first); + expect(first).toBe(true); + expect(second).toBe(false); + expect(getPairedTaskById(task.id)?.updated_at).toBe(task.updated_at); }); it('reclaims an expired execution lease after restart while blocking fresh claims before expiry', () => { @@ -236,30 +450,42 @@ describe('paired follow-up scheduler', () => { task, intentKind: 'reviewer-turn', }); - expect(first).toMatch(/^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d{3}Z$/); + expect(first).toBe(true); + + const rawDatabase = new Database(dbPath); + rawDatabase + .prepare( + ` + UPDATE paired_task_execution_leases + SET claimed_service_id = ? + WHERE task_id = ? + `, + ) + .run('other-service', task.id); + rawDatabase.close(); _initTestDatabaseFromFile(dbPath); - const freshTask = getPairedTaskById(task.id); - expect(freshTask).toBeDefined(); + const crossServiceTask = getPairedTaskById(task.id); + expect(crossServiceTask).toBeDefined(); expect( claimPairedTurnExecution({ chatJid: task.chat_jid, runId: 'run-before-expiry-reviewer', - task: freshTask ?? task, + task: crossServiceTask ?? task, intentKind: 'reviewer-turn', }), - ).toBeNull(); + ).toBe(false); expect( claimPairedTurnExecution({ chatJid: task.chat_jid, runId: 'run-before-expiry-owner', - task: freshTask ?? task, + task: crossServiceTask ?? task, intentKind: 'owner-turn', }), - ).toBeNull(); + ).toBe(false); - const rawDatabase = new Database(dbPath); - rawDatabase + const expiryDatabase = new Database(dbPath); + expiryDatabase .prepare( ` UPDATE paired_task_execution_leases @@ -269,7 +495,7 @@ describe('paired follow-up scheduler', () => { `, ) .run('2020-01-01T00:00:00.000Z', '2020-01-01T00:00:00.000Z', task.id); - rawDatabase.close(); + expiryDatabase.close(); _initTestDatabaseFromFile(dbPath); const recoveredTask = getPairedTaskById(task.id); @@ -281,7 +507,150 @@ describe('paired follow-up scheduler', () => { task: recoveredTask ?? task, intentKind: 'reviewer-turn', }), - ).toMatch(/^\d{4}-\d{2}-\d{2}T/); + ).toBe(true); + } finally { + fs.rmSync(tempDir, { recursive: true, force: true }); + _initTestDatabase(); + resetPairedFollowUpScheduleState(); + } + }); + + it('clears same-service legacy execution leases during startup so an interrupted turn can be reclaimed immediately after restart', () => { + const tempDir = fs.mkdtempSync( + path.join(os.tmpdir(), 'ejclaw-paired-lease-startup-'), + ); + const dbPath = path.join(tempDir, 'paired-state.db'); + + try { + createLegacyLeaseDatabase({ + dbPath, + taskId: 'task-restart-cleanup', + reviewerServiceId: CURRENT_SERVICE_ID, + reviewerAgentType: CURRENT_AGENT_TYPE, + }); + resetPairedFollowUpScheduleState(); + + _initTestDatabaseFromFile(dbPath); + const migratedDb = new Database(dbPath, { readonly: true }); + const leaseCount = ( + migratedDb + .prepare( + 'SELECT COUNT(*) AS count FROM paired_task_execution_leases WHERE task_id = ?', + ) + .get('task-restart-cleanup') as { count: number } + ).count; + migratedDb.close(); + + expect(leaseCount).toBe(0); + + const recoveredTask = getPairedTaskById('task-restart-cleanup'); + expect(recoveredTask).toBeDefined(); + expect( + claimPairedTurnExecution({ + chatJid: 'group@test', + runId: 'run-after-restart', + task: recoveredTask!, + intentKind: 'reviewer-turn', + }), + ).toBe(true); + } finally { + fs.rmSync(tempDir, { recursive: true, force: true }); + _initTestDatabase(); + resetPairedFollowUpScheduleState(); + } + }); + + it('preserves legacy execution leases that belong to another service during startup cleanup', () => { + const tempDir = fs.mkdtempSync( + path.join(os.tmpdir(), 'ejclaw-paired-lease-preserve-'), + ); + const dbPath = path.join(tempDir, 'paired-state.db'); + + try { + createLegacyLeaseDatabase({ + dbPath, + taskId: 'task-cross-service-legacy-lease', + reviewerServiceId: OTHER_SERVICE_ID, + reviewerAgentType: OTHER_AGENT_TYPE, + }); + resetPairedFollowUpScheduleState(); + + _initTestDatabaseFromFile(dbPath); + + const migratedDb = new Database(dbPath, { readonly: true }); + const leaseRow = migratedDb + .prepare( + ` + SELECT claimed_service_id + FROM paired_task_execution_leases + WHERE task_id = ? + `, + ) + .get('task-cross-service-legacy-lease') as + | { claimed_service_id: string | null } + | undefined; + migratedDb.close(); + + expect(leaseRow).toEqual({ + claimed_service_id: OTHER_SERVICE_ID, + }); + + const task = getPairedTaskById('task-cross-service-legacy-lease'); + expect(task).toBeDefined(); + expect( + claimPairedTurnExecution({ + chatJid: 'group@test', + runId: 'run-cross-service-reclaim', + task: task!, + intentKind: 'reviewer-turn', + }), + ).toBe(false); + } finally { + fs.rmSync(tempDir, { recursive: true, force: true }); + _initTestDatabase(); + resetPairedFollowUpScheduleState(); + } + }); + + it('preserves the raw runtime service id for legacy owner leases instead of canonicalizing to the owner shadow', () => { + const tempDir = fs.mkdtempSync( + path.join(os.tmpdir(), 'ejclaw-paired-lease-owner-failover-'), + ); + const dbPath = path.join(tempDir, 'paired-state.db'); + + try { + createLegacyLeaseDatabase({ + dbPath, + taskId: 'task-owner-failover-legacy-lease', + role: 'owner', + intentKind: 'owner-turn', + taskStatus: 'active', + ownerServiceId: CODEX_REVIEW_SERVICE_ID, + ownerAgentType: 'codex', + reviewerServiceId: CLAUDE_SERVICE_ID, + reviewerAgentType: 'claude-code', + }); + resetPairedFollowUpScheduleState(); + + _initTestDatabaseFromFile(dbPath); + + const migratedDb = new Database(dbPath, { readonly: true }); + const leaseRow = migratedDb + .prepare( + ` + SELECT claimed_service_id + FROM paired_task_execution_leases + WHERE task_id = ? + `, + ) + .get('task-owner-failover-legacy-lease') as + | { claimed_service_id: string | null } + | undefined; + migratedDb.close(); + + expect(leaseRow).toEqual({ + claimed_service_id: CODEX_REVIEW_SERVICE_ID, + }); } finally { fs.rmSync(tempDir, { recursive: true, force: true }); _initTestDatabase(); diff --git a/src/paired-follow-up-scheduler.ts b/src/paired-follow-up-scheduler.ts index 6f0faf1..02475b5 100644 --- a/src/paired-follow-up-scheduler.ts +++ b/src/paired-follow-up-scheduler.ts @@ -66,20 +66,16 @@ export function claimPairedTurnExecution(args: { runId: string; task: ScheduledPairedFollowUpTask; intentKind: PairedTurnReservationIntentKind; -}): string | null { - const nextTaskUpdatedAt = new Date().toISOString(); - const claimed = claimPairedTurnReservation({ +}): boolean { + return claimPairedTurnReservation({ chatJid: args.chatJid, taskId: args.task.id, taskStatus: args.task.status, roundTripCount: args.task.round_trip_count, taskUpdatedAt: args.task.updated_at, - nextTaskUpdatedAt, intentKind: args.intentKind, runId: args.runId, }); - - return claimed ? nextTaskUpdatedAt : null; } export function resetPairedFollowUpScheduleState(): void {