claude-bot/EJClaw
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'codex/owner/ejclaw'

Browse Source
This commit is contained in:
ejclaw
2026-04-09 16:31:41 +09:00
parent 27f7ee7911 0461e401e8
commit a48dd1cc20
10 changed files with 722 additions and 47 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
src/db.ts
View File

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

1
src/db/base-schema.ts
View File

@@ -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,

49
src/db/paired-state.ts
View File

@@ -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,18 +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,
);
.run(args.taskId, args.taskUpdatedAt, args.taskStatus);
if (claimedTask.changes === 0) {
throw new PairedTurnReservationClaimError();
@@ -593,10 +595,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 +618,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(),

148
src/db/schema.ts
View File

@@ -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,139 @@ 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 +286,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 +297,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)

48
src/message-runtime-queue.test.ts
View File

@@ -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();
});
});

56
src/message-runtime-queue.ts
View File

@@ -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) {

18
src/message-runtime-rules.test.ts
View File

@@ -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,

30
src/message-runtime-rules.ts
View File

@@ -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' }

397
src/paired-follow-up-scheduler.test.ts
View File

@@ -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();

8
src/paired-follow-up-scheduler.ts
View File

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