diff --git a/src/db.ts b/src/db.ts index fd89aa1..4a7d76c 100644 --- a/src/db.ts +++ b/src/db.ts @@ -97,6 +97,10 @@ import { markWorkItemDeliveryRetryInDatabase, } from './db/work-items.js'; import { + claimPairedTurnReservationInDatabase, + clearExpiredPairedTaskExecutionLeasesInDatabase, + clearPairedTaskExecutionLeasesInDatabase, + clearPairedTurnReservationsInDatabase, type PairedTaskUpdates, createPairedTaskInDatabase, getLastBotFinalMessageFromDatabase, @@ -106,6 +110,10 @@ import { getPairedTaskByIdFromDatabase, getPairedWorkspaceFromDatabase, listPairedWorkspacesForTaskFromDatabase, + refreshPairedTaskExecutionLeaseInDatabase, + releasePairedTaskExecutionLeaseInDatabase, + reservePairedTurnReservationInDatabase, + updatePairedTaskIfUnchangedInDatabase, updatePairedTaskInDatabase, upsertPairedProjectInDatabase, upsertPairedWorkspaceInDatabase, @@ -184,6 +192,7 @@ import { PairedRoomRole, RoomMode, PairedTask, + PairedTurnReservationIntentKind, PairedTurnOutput, PairedWorkspace, RegisteredGroup, @@ -279,6 +288,7 @@ function getSchemaMigrationHooks(): SchemaMigrationHooks { export function initDatabase(): void { db = openPersistentDatabase(); initializeDatabaseSchema(db, getSchemaMigrationHooks()); + clearExpiredPairedTaskExecutionLeasesInDatabase(db); migrateJsonStateFromFiles({ setRouterState, setSession, @@ -290,12 +300,14 @@ export function initDatabase(): void { export function _initTestDatabase(): void { db = openInMemoryDatabase(); initializeDatabaseSchema(db, getSchemaMigrationHooks()); + clearExpiredPairedTaskExecutionLeasesInDatabase(db); } /** @internal - for tests only. Opens an existing database file and runs schema/migrations. */ export function _initTestDatabaseFromFile(dbPath: string): void { db = openDatabaseFromFile(dbPath); initializeDatabaseSchema(db, getSchemaMigrationHooks()); + clearExpiredPairedTaskExecutionLeasesInDatabase(db); } /** @internal - for tests only. */ @@ -1192,6 +1204,59 @@ export function updatePairedTask(id: string, updates: PairedTaskUpdates): void { updatePairedTaskInDatabase(db, id, updates); } +export function updatePairedTaskIfUnchanged( + id: string, + expectedUpdatedAt: string, + updates: PairedTaskUpdates, +): boolean { + return updatePairedTaskIfUnchangedInDatabase( + db, + id, + expectedUpdatedAt, + updates, + ); +} + +export function reservePairedTurnReservation(args: { + chatJid: string; + taskId: string; + taskStatus: PairedTask['status']; + roundTripCount: number; + taskUpdatedAt: string; + intentKind: PairedTurnReservationIntentKind; + runId: string; +}): boolean { + return reservePairedTurnReservationInDatabase(db, args); +} + +export function claimPairedTurnReservation(args: { + chatJid: string; + taskId: string; + taskStatus: PairedTask['status']; + roundTripCount: number; + taskUpdatedAt: string; + nextTaskUpdatedAt: string; + intentKind: PairedTurnReservationIntentKind; + runId: string; +}): boolean { + return claimPairedTurnReservationInDatabase(db, args); +} + +export function releasePairedTaskExecutionLease(args: { + taskId: string; + runId: string; +}): void { + releasePairedTaskExecutionLeaseInDatabase(db, args); +} + +export function refreshPairedTaskExecutionLease(args: { + taskId: string; + runId: string; + now?: string; +}): boolean { + return refreshPairedTaskExecutionLeaseInDatabase(db, args); +} + export function upsertPairedWorkspace(workspace: PairedWorkspace): void { upsertPairedWorkspaceInDatabase(db, workspace); } @@ -1207,6 +1272,15 @@ export function listPairedWorkspacesForTask(taskId: string): PairedWorkspace[] { return listPairedWorkspacesForTaskFromDatabase(db, taskId); } +/** @internal - for tests only. */ +export function _clearPairedTurnReservationsForTests(): void { + if (!db) { + throw new Error('Database not initialized'); + } + clearPairedTurnReservationsInDatabase(db); + clearPairedTaskExecutionLeasesInDatabase(db); +} + /** * Get the most recent bot message (is_bot_message=1) in a chat, regardless of which bot sent it. * Used for duplicate detection in pair rooms. diff --git a/src/db/base-schema.ts b/src/db/base-schema.ts index 75970b4..a1bba1f 100644 --- a/src/db/base-schema.ts +++ b/src/db/base-schema.ts @@ -169,6 +169,57 @@ export function applyBaseSchema(database: Database): void { ); CREATE INDEX IF NOT EXISTS idx_paired_turn_outputs_task ON paired_turn_outputs(task_id, turn_number); + CREATE TABLE IF NOT EXISTS paired_turn_reservations ( + chat_jid TEXT NOT NULL, + task_id TEXT NOT NULL, + task_status TEXT NOT NULL, + round_trip_count INTEGER NOT NULL DEFAULT 0, + task_updated_at TEXT NOT NULL, + intent_kind TEXT NOT NULL, + status TEXT NOT NULL DEFAULT 'pending', + scheduled_run_id TEXT, + consumed_run_id TEXT, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL, + consumed_at TEXT, + PRIMARY KEY (chat_jid, task_id, task_updated_at, intent_kind), + CHECK (status IN ('pending', 'completed')), + CHECK ( + intent_kind IN ( + 'owner-turn', + 'reviewer-turn', + 'arbiter-turn', + 'owner-follow-up', + 'finalize-owner-turn' + ) + ) + ); + CREATE INDEX IF NOT EXISTS idx_paired_turn_reservations_task + ON paired_turn_reservations(task_id, task_updated_at, status); + CREATE TABLE IF NOT EXISTS 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' + ) + ) + ); + CREATE INDEX IF NOT EXISTS idx_paired_task_execution_leases_expires_at + ON paired_task_execution_leases(expires_at); CREATE TABLE IF NOT EXISTS channel_owner ( chat_jid TEXT PRIMARY KEY, owner_agent_type TEXT, diff --git a/src/db/paired-state.ts b/src/db/paired-state.ts index 1a021f1..f26579f 100644 --- a/src/db/paired-state.ts +++ b/src/db/paired-state.ts @@ -14,7 +14,10 @@ import { resolveRoleServiceShadow } from '../role-service-shadow.js'; import { AgentType, PairedProject, + PairedRoomRole, PairedTask, + PairedTaskStatus, + PairedTurnReservationIntentKind, PairedWorkspace, } from '../types.js'; @@ -47,6 +50,30 @@ export type PairedTaskUpdates = Partial< > >; +export const PAIRED_TASK_EXECUTION_LEASE_TTL_MS = 10 * 60_000; + +function computeExecutionLeaseExpiry(now: string): string { + return new Date( + new Date(now).getTime() + PAIRED_TASK_EXECUTION_LEASE_TTL_MS, + ).toISOString(); +} + +function resolveExecutionLeaseRole( + intentKind: PairedTurnReservationIntentKind, +): PairedRoomRole { + switch (intentKind) { + case 'reviewer-turn': + return 'reviewer'; + case 'arbiter-turn': + return 'arbiter'; + case 'owner-turn': + case 'owner-follow-up': + case 'finalize-owner-turn': + default: + return 'owner'; + } +} + function hydratePairedTaskRow( database: Database, row: StoredPairedTaskRow, @@ -213,7 +240,7 @@ export function updatePairedTaskInDatabase( database: Database, id: string, updates: PairedTaskUpdates, -): void { +): boolean { const fields: string[] = []; const values: (string | number | null)[] = []; @@ -258,12 +285,362 @@ export function updatePairedTaskInDatabase( values.push(updates.updated_at); } - if (fields.length === 0) return; + if (fields.length === 0) return false; values.push(id); - database + const result = database .prepare(`UPDATE paired_tasks SET ${fields.join(', ')} WHERE id = ?`) .run(...values); + return result.changes > 0; +} + +export function updatePairedTaskIfUnchangedInDatabase( + database: Database, + id: string, + expectedUpdatedAt: string, + updates: PairedTaskUpdates, +): boolean { + const fields: string[] = []; + const values: (string | number | null)[] = []; + + if (updates.title !== undefined) { + fields.push('title = ?'); + values.push(updates.title); + } + if (updates.source_ref !== undefined) { + fields.push('source_ref = ?'); + values.push(updates.source_ref); + } + if (updates.plan_notes !== undefined) { + fields.push('plan_notes = ?'); + values.push(updates.plan_notes); + } + if (updates.review_requested_at !== undefined) { + fields.push('review_requested_at = ?'); + values.push(updates.review_requested_at); + } + if (updates.round_trip_count !== undefined) { + fields.push('round_trip_count = ?'); + values.push(updates.round_trip_count); + } + if (updates.status !== undefined) { + fields.push('status = ?'); + values.push(updates.status); + } + if (updates.arbiter_verdict !== undefined) { + fields.push('arbiter_verdict = ?'); + values.push(updates.arbiter_verdict); + } + if (updates.arbiter_requested_at !== undefined) { + fields.push('arbiter_requested_at = ?'); + values.push(updates.arbiter_requested_at); + } + if (updates.completion_reason !== undefined) { + fields.push('completion_reason = ?'); + values.push(updates.completion_reason); + } + if (updates.updated_at !== undefined) { + fields.push('updated_at = ?'); + values.push(updates.updated_at); + } + + if (fields.length === 0) return false; + + values.push(id, expectedUpdatedAt); + const result = database + .prepare( + `UPDATE paired_tasks SET ${fields.join(', ')} WHERE id = ? AND updated_at = ?`, + ) + .run(...values); + return result.changes > 0; +} + +export function reservePairedTurnReservationInDatabase( + database: Database, + args: { + chatJid: string; + taskId: string; + taskStatus: PairedTaskStatus; + roundTripCount: number; + taskUpdatedAt: string; + intentKind: PairedTurnReservationIntentKind; + runId: string; + }, +): boolean { + const now = new Date().toISOString(); + const result = database + .prepare( + ` + INSERT INTO paired_turn_reservations ( + chat_jid, + task_id, + task_status, + round_trip_count, + task_updated_at, + intent_kind, + status, + scheduled_run_id, + consumed_run_id, + created_at, + updated_at, + consumed_at + ) + VALUES (?, ?, ?, ?, ?, ?, 'pending', ?, NULL, ?, ?, NULL) + ON CONFLICT(chat_jid, task_id, task_updated_at, intent_kind) DO NOTHING + `, + ) + .run( + args.chatJid, + args.taskId, + args.taskStatus, + args.roundTripCount, + args.taskUpdatedAt, + args.intentKind, + args.runId, + now, + now, + ); + + return result.changes > 0; +} + +class PairedTurnReservationClaimError extends Error {} + +export function claimPairedTurnReservationInDatabase( + database: Database, + args: { + chatJid: string; + taskId: string; + taskStatus: PairedTaskStatus; + roundTripCount: number; + taskUpdatedAt: string; + nextTaskUpdatedAt: string; + intentKind: PairedTurnReservationIntentKind; + runId: string; + }, +): boolean { + const tx = database.transaction(() => { + const now = args.nextTaskUpdatedAt; + const expiresAt = computeExecutionLeaseExpiry(now); + const leaseRole = resolveExecutionLeaseRole(args.intentKind); + const existingLease = database + .prepare( + ` + SELECT claimed_run_id, updated_at, expires_at + FROM paired_task_execution_leases + WHERE task_id = ? + `, + ) + .get(args.taskId) as + | { + claimed_run_id: string; + updated_at: string; + expires_at: string; + } + | undefined; + + if (!existingLease) { + const insertedLease = database + .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, + args.chatJid, + leaseRole, + args.intentKind, + args.runId, + args.taskStatus, + args.taskUpdatedAt, + now, + now, + expiresAt, + ); + + if (insertedLease.changes === 0) { + throw new PairedTurnReservationClaimError(); + } + } else if (existingLease.expires_at > now) { + throw new PairedTurnReservationClaimError(); + } else { + const tookOverLease = database + .prepare( + ` + UPDATE paired_task_execution_leases + SET chat_jid = ?, + role = ?, + intent_kind = ?, + claimed_run_id = ?, + task_status = ?, + task_updated_at = ?, + claimed_at = ?, + updated_at = ?, + expires_at = ? + WHERE task_id = ? + AND claimed_run_id = ? + AND updated_at = ? + AND expires_at = ? + `, + ) + .run( + args.chatJid, + leaseRole, + args.intentKind, + args.runId, + args.taskStatus, + args.taskUpdatedAt, + now, + now, + expiresAt, + args.taskId, + existingLease.claimed_run_id, + existingLease.updated_at, + existingLease.expires_at, + ); + + if (tookOverLease.changes === 0) { + throw new PairedTurnReservationClaimError(); + } + } + + database + .prepare( + ` + UPDATE paired_turn_reservations + SET status = 'completed', + consumed_run_id = ?, + updated_at = ?, + consumed_at = ? + WHERE chat_jid = ? + AND task_id = ? + AND task_updated_at = ? + AND intent_kind = ? + AND status = 'pending' + `, + ) + .run( + args.runId, + now, + now, + args.chatJid, + args.taskId, + args.taskUpdatedAt, + args.intentKind, + ); + + const claimedTask = database + .prepare( + ` + UPDATE paired_tasks + SET updated_at = ? + WHERE id = ? + AND updated_at = ? + AND status = ? + `, + ) + .run( + args.nextTaskUpdatedAt, + args.taskId, + args.taskUpdatedAt, + args.taskStatus, + ); + + if (claimedTask.changes === 0) { + throw new PairedTurnReservationClaimError(); + } + }); + + try { + tx(); + return true; + } catch (error) { + if (error instanceof PairedTurnReservationClaimError) { + return false; + } + throw error; + } +} + +export function clearPairedTurnReservationsInDatabase( + database: Database, +): void { + database.prepare('DELETE FROM paired_turn_reservations').run(); +} + +export function releasePairedTaskExecutionLeaseInDatabase( + database: Database, + args: { + taskId: string; + runId: string; + }, +): void { + database + .prepare( + ` + DELETE FROM paired_task_execution_leases + WHERE task_id = ? + AND claimed_run_id = ? + `, + ) + .run(args.taskId, args.runId); +} + +export function refreshPairedTaskExecutionLeaseInDatabase( + database: Database, + args: { + taskId: string; + runId: string; + now?: string; + }, +): boolean { + const now = args.now ?? new Date().toISOString(); + const result = database + .prepare( + ` + UPDATE paired_task_execution_leases + SET updated_at = ?, + expires_at = ? + WHERE task_id = ? + AND claimed_run_id = ? + AND expires_at >= ? + `, + ) + .run(now, computeExecutionLeaseExpiry(now), args.taskId, args.runId, now); + return result.changes > 0; +} + +export function clearExpiredPairedTaskExecutionLeasesInDatabase( + database: Database, + now: string = new Date().toISOString(), +): number { + return database + .prepare( + ` + DELETE FROM paired_task_execution_leases + WHERE expires_at <= ? + `, + ) + .run(now).changes; +} + +export function clearPairedTaskExecutionLeasesInDatabase( + database: Database, +): void { + database.prepare('DELETE FROM paired_task_execution_leases').run(); } export function upsertPairedWorkspaceInDatabase( diff --git a/src/db/schema.ts b/src/db/schema.ts index 76543b3..ca883a7 100644 --- a/src/db/schema.ts +++ b/src/db/schema.ts @@ -145,6 +145,21 @@ export function applySchemaMigrations( database, `ALTER TABLE paired_tasks ADD COLUMN arbiter_agent_type TEXT`, ); + tryExecMigration( + database, + `ALTER TABLE paired_task_execution_leases ADD COLUMN expires_at TEXT`, + ); + database.exec(` + UPDATE paired_task_execution_leases + SET expires_at = COALESCE( + expires_at, + strftime('%Y-%m-%dT%H:%M:%fZ', 'now', '+10 minutes') + ) + `); + database.exec(` + CREATE INDEX IF NOT EXISTS idx_paired_task_execution_leases_expires_at + ON paired_task_execution_leases(expires_at) + `); tryExecMigration( database, diff --git a/src/message-agent-executor-paired.ts b/src/message-agent-executor-paired.ts index f7aad5f..7970956 100644 --- a/src/message-agent-executor-paired.ts +++ b/src/message-agent-executor-paired.ts @@ -4,19 +4,21 @@ import { getLatestTurnNumber, getPairedTaskById, insertPairedTurnOutput, + refreshPairedTaskExecutionLease, } from './db.js'; import { logger } from './logger.js'; import { completePairedExecutionContext, type PreparedPairedExecutionContext, } from './paired-execution-context.js'; -import { resolveNextTurnAction } from './message-runtime-rules.js'; import { resolvePairedFollowUpQueueAction } from './message-agent-executor-rules.js'; -import { schedulePairedFollowUpOnce } from './paired-follow-up-scheduler.js'; +import { enqueuePairedFollowUpAfterEvent } from './message-runtime-follow-up.js'; import type { PairedRoomRole } from './types.js'; type ExecutorLog = Pick; +const PAIRED_TASK_EXECUTION_LEASE_HEARTBEAT_MS = 30_000; + export interface PairedExecutionLifecycle { updateSummary(args: { outputText?: string | null; @@ -55,6 +57,52 @@ export function createPairedExecutionLifecycle(args: { let pairedExecutionCompleted = false; let pairedSawOutput = false; let pairedTurnOutputPersisted = false; + let leaseHeartbeatTimer: ReturnType | null = null; + + const clearLeaseHeartbeat = () => { + if (!leaseHeartbeatTimer) { + return; + } + clearInterval(leaseHeartbeatTimer); + leaseHeartbeatTimer = null; + }; + const heartbeatLeaseIfNeeded = () => { + if (!pairedExecutionContext) { + return; + } + try { + const refreshed = refreshPairedTaskExecutionLease({ + taskId: pairedExecutionContext.task.id, + runId, + }); + if (!refreshed) { + log.warn( + { + pairedTaskId: pairedExecutionContext.task.id, + runId, + }, + 'Skipped paired execution lease heartbeat because this run no longer owns the lease', + ); + } + } catch (err) { + log.warn( + { + pairedTaskId: pairedExecutionContext.task.id, + runId, + err, + }, + 'Failed to refresh paired execution lease heartbeat', + ); + } + }; + + if (pairedExecutionContext) { + leaseHeartbeatTimer = setInterval( + heartbeatLeaseIfNeeded, + PAIRED_TASK_EXECUTION_LEASE_HEARTBEAT_MS, + ); + leaseHeartbeatTimer.unref?.(); + } const persistPairedTurnOutputIfNeeded = () => { if ( @@ -90,10 +138,12 @@ export function createPairedExecutionLifecycle(args: { pairedExecutionStatus = 'succeeded'; pairedSawOutput = true; persistPairedTurnOutputIfNeeded(); + clearLeaseHeartbeat(); completePairedExecutionContext({ taskId: pairedExecutionContext.task.id, role: completedRole, status: 'succeeded', + runId, summary: pairedExecutionSummary, }); pairedExecutionCompleted = true; @@ -127,10 +177,12 @@ export function createPairedExecutionLifecycle(args: { persistPairedTurnOutputIfNeeded(); } + clearLeaseHeartbeat(); completePairedExecutionContext({ taskId: pairedExecutionContext.task.id, role: completedRole, status, + runId, summary: pairedExecutionSummary, }); pairedExecutionCompleted = true; @@ -149,6 +201,8 @@ export function createPairedExecutionLifecycle(args: { }, async asyncFinalize() { + clearLeaseHeartbeat(); + if (pairedExecutionContext && !pairedExecutionCompleted) { const effectiveStatus = completedRole === 'owner' && @@ -172,6 +226,7 @@ export function createPairedExecutionLifecycle(args: { taskId: pairedExecutionContext.task.id, role: completedRole, status: effectiveStatus, + runId, summary: pairedExecutionSummary, }); pairedExecutionCompleted = true; @@ -214,31 +269,30 @@ export function createPairedExecutionLifecycle(args: { return; } - const nextTurnAction = resolveNextTurnAction({ - taskStatus: finishedTask.status, - lastTurnOutputRole: pairedSawOutput ? completedRole : null, - }); - if (nextTurnAction.kind === 'none') { - return; - } - - const scheduled = schedulePairedFollowUpOnce({ + const followUpResult = enqueuePairedFollowUpAfterEvent({ chatJid, runId, task: finishedTask, - intentKind: nextTurnAction.kind, - enqueue: enqueueMessageCheck, + source: 'executor-recovery', + completedRole, + executionStatus: pairedExecutionStatus, + sawOutput: pairedSawOutput, + fallbackLastTurnOutputRole: pairedSawOutput ? completedRole : null, + enqueueMessageCheck, }); + if (followUpResult.kind !== 'paired-follow-up') { + return; + } log.info( { taskId: pairedExecutionContext.task.id, role: completedRole, pairedExecutionStatus, taskStatus: finishedTask.status, - intentKind: nextTurnAction.kind, - scheduled, + intentKind: followUpResult.intentKind, + scheduled: followUpResult.scheduled, }, - scheduled + followUpResult.scheduled ? 'Queued paired follow-up after failed reviewer/arbiter execution left a pending task state' : 'Skipped duplicate paired follow-up after failed reviewer/arbiter execution while task state was unchanged', ); diff --git a/src/message-agent-executor.test.ts b/src/message-agent-executor.test.ts index 7948dfc..db10721 100644 --- a/src/message-agent-executor.test.ts +++ b/src/message-agent-executor.test.ts @@ -48,15 +48,57 @@ vi.mock('./platform-prompts.js', () => ({ readArbiterPrompt: vi.fn(() => ''), })); -vi.mock('./db.js', () => ({ - createServiceHandoff: vi.fn(), - getAllTasks: vi.fn(() => []), - getLastHumanMessageSender: vi.fn(() => null), - getLatestOpenPairedTaskForChat: vi.fn(() => undefined), - getLatestTurnNumber: vi.fn(() => 0), - getPairedTaskById: vi.fn(() => undefined), - insertPairedTurnOutput: vi.fn(), -})); +vi.mock('./db.js', () => { + const pairedTurnReservations = new Set(); + const claimedTaskRevisions = new Set(); + const buildReservationKey = (args: { + chatJid: string; + taskId: string; + taskUpdatedAt: string; + intentKind: string; + }) => + [args.chatJid, args.taskId, args.taskUpdatedAt, args.intentKind].join(':'); + + return { + createServiceHandoff: vi.fn(), + getAllTasks: vi.fn(() => []), + getLastHumanMessageSender: vi.fn(() => null), + getLatestOpenPairedTaskForChat: vi.fn(() => undefined), + getLatestTurnNumber: vi.fn(() => 0), + getPairedTaskById: vi.fn(() => undefined), + getPairedTurnOutputs: vi.fn(() => []), + insertPairedTurnOutput: vi.fn(), + refreshPairedTaskExecutionLease: vi.fn(() => true), + reservePairedTurnReservation: vi.fn((args) => { + const key = buildReservationKey(args); + if (pairedTurnReservations.has(key)) { + return false; + } + pairedTurnReservations.add(key); + return true; + }), + claimPairedTurnReservation: vi.fn((args) => { + const revisionKey = [args.taskId, args.taskUpdatedAt].join(':'); + if (claimedTaskRevisions.has(revisionKey)) { + return false; + } + claimedTaskRevisions.add(revisionKey); + pairedTurnReservations.add( + buildReservationKey({ + chatJid: args.chatJid, + taskId: args.taskId, + taskUpdatedAt: args.taskUpdatedAt, + intentKind: args.intentKind, + }), + ); + return true; + }), + _clearPairedTurnReservationsForTests: vi.fn(() => { + pairedTurnReservations.clear(); + claimedTaskRevisions.clear(); + }), + }; +}); vi.mock('./service-routing.js', () => ({ activateCodexFailover: vi.fn(), @@ -789,6 +831,7 @@ describe('runAgentForGroup room memory', () => { ).toHaveBeenCalledWith({ taskId: 'paired-task-1', role: 'owner', + runId: 'run-room-role', status: 'succeeded', summary: 'ok', }); @@ -982,6 +1025,7 @@ describe('runAgentForGroup room memory', () => { ).toHaveBeenCalledWith({ taskId: 'paired-task-1', role: 'owner', + runId: 'run-blocked-reviewer', status: 'failed', summary: 'Review snapshot is stale after owner changes. Retry the review once to refresh against the latest owner workspace.', @@ -1396,6 +1440,7 @@ describe('runAgentForGroup room memory', () => { ).toHaveBeenCalledWith({ taskId: 'paired-task-owner-output-order', role: 'owner', + runId: 'run-owner-output-order', status: 'succeeded', summary: 'DONE_WITH_CONCERNS\nowner complete', }); diff --git a/src/message-runtime-dispatch.ts b/src/message-runtime-dispatch.ts index bc51893..6e97173 100644 --- a/src/message-runtime-dispatch.ts +++ b/src/message-runtime-dispatch.ts @@ -10,10 +10,9 @@ import { getProcessableMessages, hasAllowedTrigger, resolveCursorKey, - resolveFollowUpDispatch, - resolveNextTurnAction, shouldSkipBotOnlyCollaboration, } from './message-runtime-rules.js'; +import { enqueuePairedFollowUpAfterEvent } from './message-runtime-follow-up.js'; import { extractSessionCommand, isSessionCommandAllowed, @@ -40,81 +39,59 @@ export function enqueueGenericFollowUpAfterDeliveryRetry(args: { workItemId: string | number; log: RuntimeLog; enqueueMessageCheck: () => void; - schedulePairedFollowUp: ( - task: PairedTask | null | undefined, - intentKind: ScheduledPairedFollowUpIntentKind, - runId: string, - ) => boolean; }): void { - const nextTurnAction = - args.pendingTask == null - ? { kind: 'none' as const } - : resolveNextTurnAction({ - taskStatus: args.pendingTask.status, - lastTurnOutputRole: args.deliveryRole, - }); - const dispatch = resolveFollowUpDispatch({ + const followUpResult = enqueuePairedFollowUpAfterEvent({ + chatJid: args.chatJid, + runId: args.runId, + task: args.pendingTask, source: 'delivery-retry', - nextTurnAction, completedRole: args.deliveryRole, + fallbackLastTurnOutputRole: args.deliveryRole, + enqueueMessageCheck: args.enqueueMessageCheck, }); - if (dispatch.kind === 'none') { + if (followUpResult.kind === 'none') { args.log.info( { workItemId: args.workItemId, chatJid: args.chatJid, deliveryRole: args.deliveryRole, - pendingTaskStatus: args.pendingTask?.status ?? null, + pendingTaskStatus: followUpResult.taskStatus, }, 'No queued follow-up was required after delivery retry', ); return; } - if (dispatch.kind === 'enqueue' && dispatch.queueKind === 'message-check') { - args.enqueueMessageCheck(); + if (followUpResult.kind === 'message-check') { return; } - if ( - dispatch.kind !== 'enqueue' || - dispatch.queueKind !== 'paired-follow-up' || - nextTurnAction.kind === 'none' - ) { - return; - } - - const scheduled = args.schedulePairedFollowUp( - args.pendingTask, - nextTurnAction.kind, - args.runId, - ); - if (scheduled) { + if (followUpResult.scheduled) { args.log.info( { workItemId: args.workItemId, chatJid: args.chatJid, deliveryRole: args.deliveryRole, - taskId: args.pendingTask?.id ?? null, - pendingTaskStatus: args.pendingTask?.status ?? null, - taskStatus: args.pendingTask?.status ?? null, - intentKind: nextTurnAction.kind, + taskId: followUpResult.taskId, + pendingTaskStatus: followUpResult.taskStatus, + taskStatus: followUpResult.taskStatus, + intentKind: followUpResult.intentKind, }, 'Queued paired follow-up after delivery retry', ); return; } - if (!scheduled) { + if (!followUpResult.scheduled) { args.log.info( { workItemId: args.workItemId, chatJid: args.chatJid, deliveryRole: args.deliveryRole, - taskId: args.pendingTask?.id ?? null, - pendingTaskStatus: args.pendingTask?.status ?? null, - taskStatus: args.pendingTask?.status ?? null, - intentKind: nextTurnAction.kind, + taskId: followUpResult.taskId, + pendingTaskStatus: followUpResult.taskStatus, + taskStatus: followUpResult.taskStatus, + intentKind: followUpResult.intentKind, }, 'Skipped duplicate paired follow-up enqueue after delivery retry while task state was unchanged', ); diff --git a/src/message-runtime-flow.test.ts b/src/message-runtime-flow.test.ts index bea70a6..53ecc46 100644 --- a/src/message-runtime-flow.test.ts +++ b/src/message-runtime-flow.test.ts @@ -1,5 +1,6 @@ import { beforeEach, describe, expect, it, vi } from 'vitest'; +import { _initTestDatabase } from './db.js'; import { executeBotOnlyPairedFollowUpAction } from './message-runtime-flow.js'; import { resetPairedFollowUpScheduleState, @@ -10,6 +11,7 @@ import type { PairedTask } from './types.js'; describe('executeBotOnlyPairedFollowUpAction', () => { beforeEach(() => { + _initTestDatabase(); resetPairedFollowUpScheduleState(); }); diff --git a/src/message-runtime-flow.ts b/src/message-runtime-flow.ts index 4833822..60e2aef 100644 --- a/src/message-runtime-flow.ts +++ b/src/message-runtime-flow.ts @@ -11,10 +11,13 @@ import { buildPairedTurnPrompt, buildReviewerPendingPrompt, } from './message-runtime-prompts.js'; +import { + requeuePendingPairedTurn, + resolvePairedFollowUpDecision, +} from './message-runtime-follow-up.js'; import { advanceLastAgentCursor, resolveCursorKey, - resolveFollowUpDispatch, resolveNextTurnAction, } from './message-runtime-rules.js'; import { type ScheduledPairedFollowUpIntentKind } from './paired-follow-up-scheduler.js'; @@ -24,6 +27,7 @@ import type { NewMessage, PairedTask, PairedRoomRole, + PairedTurnReservationIntentKind, RegisteredGroup, } from './types.js'; @@ -31,6 +35,7 @@ export type PendingPairedTurn = { prompt: string; channel: Channel | null; cursor: string | number | null; + intentKind: PairedTurnReservationIntentKind; cursorKey?: string; role?: 'reviewer' | 'arbiter'; } | null; @@ -122,6 +127,7 @@ export function buildPendingPairedTurn(args: { }), channel: resolveChannel(taskStatus), cursor, + intentKind: 'reviewer-turn', cursorKey: resolveCursorKey(chatJid, taskStatus), role: 'reviewer', }; @@ -139,6 +145,7 @@ export function buildPendingPairedTurn(args: { }), channel: resolveChannel(taskStatus), cursor, + intentKind: 'arbiter-turn', cursorKey: resolveCursorKey(chatJid, taskStatus), role: 'arbiter', }; @@ -149,6 +156,7 @@ export function buildPendingPairedTurn(args: { prompt: buildFinalizePendingPrompt({ turnOutputs }), channel: resolveChannel(taskStatus), cursor, + intentKind: 'finalize-owner-turn', }; } @@ -163,6 +171,7 @@ export function buildPendingPairedTurn(args: { }), channel: resolveChannel(taskStatus), cursor, + intentKind: 'owner-follow-up', }; } @@ -252,14 +261,9 @@ export function resolveBotOnlyPairedFollowUpAction(args: { const cursor = pendingCursorSource?.seq ?? pendingCursorSource?.timestamp ?? null; - const lastTurnOutput = getPairedTurnOutputs(task.id).at(-1); - const nextTurnAction = resolveNextTurnAction({ - taskStatus: task.status, - lastTurnOutputRole: lastTurnOutput?.role ?? null, - }); - const dispatch = resolveFollowUpDispatch({ + const { nextTurnAction, dispatch } = resolvePairedFollowUpDecision({ + task, source: 'bot-only-follow-up', - nextTurnAction, }); if (dispatch.kind === 'none') { @@ -400,10 +404,11 @@ export async function executeBotOnlyPairedFollowUpAction(args: { return true; } - const scheduled = schedulePairedFollowUp(action.task, action.intentKind); - if (scheduled) { - closeStdin(); - } + const scheduled = requeuePendingPairedTurn({ + schedulePairedFollowUp: () => + schedulePairedFollowUp(action.task, action.intentKind), + closeStdin, + }); log.info( { chatJid, diff --git a/src/message-runtime-follow-up.test.ts b/src/message-runtime-follow-up.test.ts new file mode 100644 index 0000000..8eae6f7 --- /dev/null +++ b/src/message-runtime-follow-up.test.ts @@ -0,0 +1,218 @@ +import { beforeEach, describe, expect, it, vi } from 'vitest'; + +vi.mock('./db.js', () => ({ + getPairedTurnOutputs: vi.fn(() => []), + reservePairedTurnReservation: vi.fn(() => true), + claimPairedTurnReservation: vi.fn(() => true), + _clearPairedTurnReservationsForTests: vi.fn(), +})); + +import { getPairedTurnOutputs } from './db.js'; +import { + enqueuePairedFollowUpAfterEvent, + dispatchPairedFollowUpForEvent, + requeuePendingPairedTurn, + schedulePairedFollowUpIntent, + schedulePairedFollowUpWithMessageCheck, +} from './message-runtime-follow-up.js'; +import { resetPairedFollowUpScheduleState } from './paired-follow-up-scheduler.js'; + +describe('message-runtime-follow-up', () => { + beforeEach(() => { + vi.resetAllMocks(); + resetPairedFollowUpScheduleState(); + vi.mocked(getPairedTurnOutputs).mockReturnValue([]); + }); + + it('suppresses stale reviewer follow-ups when the latest persisted turn already belongs to the reviewer', () => { + const enqueue = vi.fn(); + const task = { + id: 'task-stale-reviewer', + status: 'review_ready', + round_trip_count: 1, + updated_at: '2026-03-30T00:00:00.000Z', + } as const; + + vi.mocked(getPairedTurnOutputs).mockReturnValue([ + { + id: 1, + task_id: task.id, + turn_number: 1, + role: 'reviewer', + output_text: 'reviewer 승인', + created_at: '2026-03-30T00:00:01.000Z', + }, + ] as any); + + const scheduled = schedulePairedFollowUpIntent({ + chatJid: 'group@test', + runId: 'run-stale-reviewer', + task, + intentKind: 'reviewer-turn', + enqueue, + }); + + expect(scheduled).toBe(false); + expect(enqueue).not.toHaveBeenCalled(); + }); + + it('uses the fallback delivery role to schedule owner follow-ups when no persisted turn output exists yet', () => { + const enqueue = vi.fn(); + const result = dispatchPairedFollowUpForEvent({ + chatJid: 'group@test', + runId: 'run-reviewer-delivery', + task: { + id: 'task-owner-follow-up', + status: 'active', + round_trip_count: 1, + updated_at: '2026-03-30T00:00:00.000Z', + }, + source: 'delivery-success', + completedRole: 'reviewer', + fallbackLastTurnOutputRole: 'reviewer', + enqueue, + }); + + expect(result).toMatchObject({ + kind: 'paired-follow-up', + intentKind: 'owner-follow-up', + scheduled: true, + taskStatus: 'active', + lastTurnOutputRole: 'reviewer', + }); + expect(enqueue).toHaveBeenCalledTimes(1); + }); + + it('prefers the latest persisted turn output over the fallback delivery role when both are present', () => { + const enqueue = vi.fn(); + vi.mocked(getPairedTurnOutputs).mockReturnValue([ + { + id: 1, + task_id: 'task-owner-follow-up-persisted', + turn_number: 1, + role: 'owner', + output_text: 'owner 응답', + created_at: '2026-03-30T00:00:01.000Z', + }, + ] as any); + + const result = dispatchPairedFollowUpForEvent({ + chatJid: 'group@test', + runId: 'run-reviewer-delivery-persisted', + task: { + id: 'task-owner-follow-up-persisted', + status: 'active', + round_trip_count: 1, + updated_at: '2026-03-30T00:00:00.000Z', + }, + source: 'delivery-success', + completedRole: 'reviewer', + fallbackLastTurnOutputRole: 'reviewer', + enqueue, + }); + + expect(result).toMatchObject({ + kind: 'none', + taskStatus: 'active', + lastTurnOutputRole: 'owner', + nextTurnAction: { kind: 'none' }, + }); + expect(enqueue).not.toHaveBeenCalled(); + }); + + it('routes delivery retry with no remaining paired handoff to a message check', () => { + const enqueue = vi.fn(); + const enqueueMessageCheck = vi.fn(); + + const result = dispatchPairedFollowUpForEvent({ + chatJid: 'group@test', + runId: 'run-delivery-retry', + task: { + id: 'task-retry-message-check', + status: 'active', + round_trip_count: 1, + updated_at: '2026-03-30T00:00:00.000Z', + }, + source: 'delivery-retry', + completedRole: 'owner', + fallbackLastTurnOutputRole: 'owner', + enqueue, + enqueueMessageCheck, + }); + + expect(result).toMatchObject({ + kind: 'message-check', + taskStatus: 'active', + nextTurnAction: { kind: 'none' }, + }); + expect(enqueue).not.toHaveBeenCalled(); + expect(enqueueMessageCheck).toHaveBeenCalledTimes(1); + }); + + it('uses the scoped message-check enqueuer when scheduling a paired follow-up intent', () => { + const enqueueMessageCheck = vi.fn(); + const scheduled = schedulePairedFollowUpWithMessageCheck({ + chatJid: 'group@test', + runId: 'run-scoped-enqueue', + task: { + id: 'task-scoped-enqueue', + status: 'active', + round_trip_count: 1, + updated_at: '2026-03-30T00:00:00.000Z', + }, + intentKind: 'owner-follow-up', + enqueueMessageCheck, + fallbackLastTurnOutputRole: 'reviewer', + }); + + expect(scheduled).toBe(true); + expect(enqueueMessageCheck).toHaveBeenCalledTimes(1); + }); + + it('closes stdin only when a pending paired turn requeue was actually scheduled', () => { + const schedulePairedFollowUp = vi + .fn<() => boolean>() + .mockReturnValueOnce(true) + .mockReturnValueOnce(false); + const closeStdin = vi.fn(); + const first = requeuePendingPairedTurn({ + schedulePairedFollowUp, + closeStdin, + }); + const second = requeuePendingPairedTurn({ + schedulePairedFollowUp, + closeStdin, + }); + + expect(first).toBe(true); + expect(second).toBe(false); + expect(schedulePairedFollowUp).toHaveBeenCalledTimes(2); + expect(closeStdin).toHaveBeenCalledTimes(1); + }); + + it('uses the same message-check side effect for delivery follow-up dispatch', () => { + const enqueueMessageCheck = vi.fn(); + + const result = enqueuePairedFollowUpAfterEvent({ + chatJid: 'group@test', + runId: 'run-delivery-follow-up-helper', + task: { + id: 'task-delivery-follow-up-helper', + status: 'active', + round_trip_count: 1, + updated_at: '2026-03-30T00:00:00.000Z', + }, + source: 'delivery-success', + completedRole: 'reviewer', + fallbackLastTurnOutputRole: 'reviewer', + enqueueMessageCheck, + }); + + expect(result).toMatchObject({ + kind: 'paired-follow-up', + intentKind: 'owner-follow-up', + scheduled: true, + }); + expect(enqueueMessageCheck).toHaveBeenCalledTimes(1); + }); +}); diff --git a/src/message-runtime-follow-up.ts b/src/message-runtime-follow-up.ts new file mode 100644 index 0000000..e774021 --- /dev/null +++ b/src/message-runtime-follow-up.ts @@ -0,0 +1,251 @@ +import { getPairedTurnOutputs } from './db.js'; +import { + matchesExpectedPairedFollowUpIntent, + resolveFollowUpDispatch, + resolveNextTurnAction, + type FollowUpDispatch, + type NextTurnAction, +} from './message-runtime-rules.js'; +import { + schedulePairedFollowUpOnce, + type ScheduledPairedFollowUpIntentKind, +} from './paired-follow-up-scheduler.js'; +import type { PairedRoomRole, PairedTask, PairedTaskStatus } from './types.js'; + +export type PairedFollowUpSource = Parameters< + typeof resolveFollowUpDispatch +>[0]['source']; + +export interface PairedFollowUpDecision { + taskId: string | null; + taskStatus: PairedTaskStatus | null; + lastTurnOutputRole: PairedRoomRole | null; + nextTurnAction: NextTurnAction; + dispatch: FollowUpDispatch; +} + +export type PairedFollowUpDispatchResult = + | (PairedFollowUpDecision & { + kind: 'none'; + }) + | (PairedFollowUpDecision & { + kind: 'message-check'; + }) + | (PairedFollowUpDecision & { + kind: 'paired-follow-up'; + intentKind: ScheduledPairedFollowUpIntentKind; + scheduled: boolean; + }); + +export function resolveLatestPairedTurnOutputRole(args: { + task: Pick | null | undefined; + fallbackLastTurnOutputRole?: PairedRoomRole | null; +}): PairedRoomRole | null { + return ( + (args.task ? getPairedTurnOutputs(args.task.id).at(-1)?.role : null) ?? + args.fallbackLastTurnOutputRole ?? + null + ); +} + +export function resolvePairedFollowUpDecision(args: { + task: Pick | null | undefined; + source: PairedFollowUpSource; + completedRole?: PairedRoomRole; + executionStatus?: 'succeeded' | 'failed'; + sawOutput?: boolean; + fallbackLastTurnOutputRole?: PairedRoomRole | null; +}): PairedFollowUpDecision { + const lastTurnOutputRole = resolveLatestPairedTurnOutputRole({ + task: args.task, + fallbackLastTurnOutputRole: args.fallbackLastTurnOutputRole, + }); + const nextTurnAction = resolveNextTurnAction({ + taskStatus: args.task?.status ?? null, + lastTurnOutputRole, + }); + const dispatch = resolveFollowUpDispatch({ + source: args.source, + nextTurnAction, + completedRole: args.completedRole, + executionStatus: args.executionStatus, + sawOutput: args.sawOutput, + }); + + return { + taskId: args.task?.id ?? null, + taskStatus: args.task?.status ?? null, + lastTurnOutputRole, + nextTurnAction, + dispatch, + }; +} + +export function schedulePairedFollowUpIntent(args: { + chatJid: string; + runId: string; + task: + | Pick + | null + | undefined; + intentKind: ScheduledPairedFollowUpIntentKind; + enqueue: () => void; + fallbackLastTurnOutputRole?: PairedRoomRole | null; + lastTurnOutputRole?: PairedRoomRole | null; +}): boolean { + if (!args.task) { + return false; + } + + const lastTurnOutputRole = + args.lastTurnOutputRole ?? + resolveLatestPairedTurnOutputRole({ + task: args.task, + fallbackLastTurnOutputRole: args.fallbackLastTurnOutputRole, + }); + + if ( + !matchesExpectedPairedFollowUpIntent({ + taskStatus: args.task.status, + lastTurnOutputRole, + intentKind: args.intentKind, + }) + ) { + return false; + } + + return schedulePairedFollowUpOnce({ + chatJid: args.chatJid, + runId: args.runId, + task: args.task, + intentKind: args.intentKind, + enqueue: args.enqueue, + }); +} + +export function schedulePairedFollowUpWithMessageCheck(args: { + chatJid: string; + runId: string; + task: + | Pick + | null + | undefined; + intentKind: ScheduledPairedFollowUpIntentKind; + enqueueMessageCheck: () => void; + fallbackLastTurnOutputRole?: PairedRoomRole | null; + lastTurnOutputRole?: PairedRoomRole | null; +}): boolean { + return schedulePairedFollowUpIntent({ + chatJid: args.chatJid, + runId: args.runId, + task: args.task, + intentKind: args.intentKind, + enqueue: args.enqueueMessageCheck, + fallbackLastTurnOutputRole: args.fallbackLastTurnOutputRole, + lastTurnOutputRole: args.lastTurnOutputRole, + }); +} + +export function dispatchPairedFollowUpForEvent(args: { + chatJid: string; + runId: string; + task: + | Pick + | null + | undefined; + source: PairedFollowUpSource; + completedRole?: PairedRoomRole; + executionStatus?: 'succeeded' | 'failed'; + sawOutput?: boolean; + fallbackLastTurnOutputRole?: PairedRoomRole | null; + enqueue: () => void; + enqueueMessageCheck?: () => void; +}): PairedFollowUpDispatchResult { + const decision = resolvePairedFollowUpDecision({ + task: args.task, + source: args.source, + completedRole: args.completedRole, + executionStatus: args.executionStatus, + sawOutput: args.sawOutput, + fallbackLastTurnOutputRole: args.fallbackLastTurnOutputRole, + }); + + if ( + decision.dispatch.kind === 'enqueue' && + decision.dispatch.queueKind === 'message-check' + ) { + args.enqueueMessageCheck?.(); + return { + kind: 'message-check', + ...decision, + }; + } + + if ( + decision.dispatch.kind === 'enqueue' && + decision.dispatch.queueKind === 'paired-follow-up' && + decision.nextTurnAction.kind !== 'none' + ) { + const scheduled = schedulePairedFollowUpIntent({ + chatJid: args.chatJid, + runId: args.runId, + task: args.task, + intentKind: decision.nextTurnAction.kind, + enqueue: args.enqueue, + lastTurnOutputRole: decision.lastTurnOutputRole, + }); + + return { + kind: 'paired-follow-up', + ...decision, + intentKind: decision.nextTurnAction.kind, + scheduled, + }; + } + + return { + kind: 'none', + ...decision, + }; +} + +export function enqueuePairedFollowUpAfterEvent(args: { + chatJid: string; + runId: string; + task: + | Pick + | null + | undefined; + source: PairedFollowUpSource; + completedRole?: PairedRoomRole; + executionStatus?: 'succeeded' | 'failed'; + sawOutput?: boolean; + fallbackLastTurnOutputRole?: PairedRoomRole | null; + enqueueMessageCheck: () => void; +}): PairedFollowUpDispatchResult { + return dispatchPairedFollowUpForEvent({ + chatJid: args.chatJid, + runId: args.runId, + task: args.task, + source: args.source, + completedRole: args.completedRole, + executionStatus: args.executionStatus, + sawOutput: args.sawOutput, + fallbackLastTurnOutputRole: args.fallbackLastTurnOutputRole, + enqueue: args.enqueueMessageCheck, + enqueueMessageCheck: args.enqueueMessageCheck, + }); +} + +export function requeuePendingPairedTurn(args: { + schedulePairedFollowUp: () => boolean; + closeStdin: () => void; +}): boolean { + const scheduled = args.schedulePairedFollowUp(); + + if (scheduled) { + args.closeStdin(); + } + + return scheduled; +} diff --git a/src/message-runtime-queue.test.ts b/src/message-runtime-queue.test.ts new file mode 100644 index 0000000..41ba97e --- /dev/null +++ b/src/message-runtime-queue.test.ts @@ -0,0 +1,181 @@ +import { beforeEach, describe, expect, it, vi } from 'vitest'; + +import { + _initTestDatabase, + createPairedTask, + getPairedTaskById, + updatePairedTaskIfUnchanged, +} from './db.js'; +import { claimPairedTurnExecution } from './paired-follow-up-scheduler.js'; +import { + runPendingPairedTurnIfNeeded, + runQueuedGroupTurn, +} from './message-runtime-queue.js'; +import type { Channel, PairedTask, RegisteredGroup } from './types.js'; + +function makeGroup(): RegisteredGroup { + return { + name: 'Test Group', + folder: 'test-group', + trigger: '@Andy', + added_at: '2026-03-30T00:00:00.000Z', + requiresTrigger: false, + agentType: 'codex', + }; +} + +function makeTask(overrides: Partial = {}): PairedTask { + return { + id: 'task-queue-claim', + chat_jid: 'group@test', + group_folder: 'test-group', + owner_service_id: 'claude', + reviewer_service_id: 'codex-main', + owner_agent_type: 'claude-code', + reviewer_agent_type: 'codex', + arbiter_agent_type: null, + title: null, + source_ref: 'HEAD', + plan_notes: null, + review_requested_at: '2026-03-30T00:00:00.000Z', + round_trip_count: 1, + status: 'review_ready', + arbiter_verdict: null, + arbiter_requested_at: null, + completion_reason: null, + created_at: '2026-03-30T00:00:00.000Z', + updated_at: '2026-03-30T00:00:00.000Z', + ...overrides, + }; +} + +function makeChannel(): Channel { + return { + name: 'discord-review', + connect: vi.fn(), + sendMessage: vi.fn(), + isConnected: vi.fn(() => true), + ownsJid: vi.fn(() => true), + disconnect: vi.fn(), + } as unknown as Channel; +} + +describe('message-runtime-queue', () => { + beforeEach(() => { + _initTestDatabase(); + }); + + it('skips a pending paired turn when another run already claimed the same task revision', async () => { + const task = makeTask(); + createPairedTask(task); + + expect( + claimPairedTurnExecution({ + chatJid: task.chat_jid, + runId: 'run-first-claim', + task, + intentKind: 'reviewer-turn', + }), + ).not.toBeNull(); + + const executeTurn = vi.fn(); + const freshTask = getPairedTaskById(task.id); + expect(freshTask).toBeDefined(); + const outcome = await runPendingPairedTurnIfNeeded({ + chatJid: task.chat_jid, + group: makeGroup(), + runId: 'run-second-claim', + log: { info: vi.fn(), warn: vi.fn(), error: vi.fn() } as any, + timezone: 'UTC', + task: freshTask, + rawMissedMessages: [ + { + id: 'bot-follow-up-1', + chat_jid: task.chat_jid, + sender: 'reviewer@test', + sender_name: 'reviewer', + content: 'review pending', + seq: 42, + timestamp: '2026-03-30T00:00:01.000Z', + }, + ], + saveState: vi.fn(), + lastAgentTimestamps: {}, + executeTurn, + getFixedRoleChannelName: () => 'discord-review', + roleToChannel: { + owner: null, + reviewer: makeChannel(), + arbiter: null, + }, + labelPairedSenders: (_chatJid, messages) => messages, + mode: 'idle', + }); + + expect(outcome).toBe(true); + expect(executeTurn).not.toHaveBeenCalled(); + }); + + it('skips a queued owner turn while a reviewer execution lease is still active after a fresh refetch', async () => { + const task = makeTask(); + createPairedTask(task); + + const reviewerClaim = claimPairedTurnExecution({ + chatJid: task.chat_jid, + runId: 'run-first-reviewer-claim', + task, + intentKind: 'reviewer-turn', + }); + expect(reviewerClaim).not.toBeNull(); + + const reviewTask = getPairedTaskById(task.id); + expect(reviewTask).toBeDefined(); + const reviewerStartedAt = '2026-03-30T00:00:03.000Z'; + expect( + updatePairedTaskIfUnchanged(reviewTask!.id, reviewTask!.updated_at, { + status: 'in_review', + updated_at: reviewerStartedAt, + }), + ).toBe(true); + + const freshTask = getPairedTaskById(task.id); + expect(freshTask?.status).toBe('in_review'); + + const executeTurn = vi.fn(); + const outcome = await runQueuedGroupTurn({ + chatJid: task.chat_jid, + group: makeGroup(), + runId: 'run-second-owner-claim', + log: { info: vi.fn(), warn: vi.fn(), error: vi.fn() } as any, + timezone: 'UTC', + missedMessages: [ + { + id: 'human-1', + chat_jid: task.chat_jid, + sender: 'user@test', + sender_name: 'User', + content: '다시 수정해줘', + timestamp: '2026-03-30T00:00:02.000Z', + seq: 43, + is_bot_message: false, + }, + ], + task: freshTask, + 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 c7855ae..1bdeece 100644 --- a/src/message-runtime-queue.ts +++ b/src/message-runtime-queue.ts @@ -15,6 +15,7 @@ import { resolveCursorKeyForRole, resolveQueuedTurnRole, } from './message-runtime-rules.js'; +import { claimPairedTurnExecution } from './paired-follow-up-scheduler.js'; import type { ExecuteTurnFn, RoleToChannelMap, @@ -23,9 +24,30 @@ import type { Channel, NewMessage, PairedTask, + PairedTurnReservationIntentKind, RegisteredGroup, } from './types.js'; +function resolveQueuedTurnReservationIntent(args: { + task: PairedTask; + turnRole: 'owner' | 'reviewer' | 'arbiter'; + hasHumanMessage: boolean; +}): PairedTurnReservationIntentKind { + if (args.turnRole === 'reviewer') { + return 'reviewer-turn'; + } + if (args.turnRole === 'arbiter') { + return 'arbiter-turn'; + } + if (args.hasHumanMessage) { + return 'owner-turn'; + } + if (args.task.status === 'merge_ready') { + return 'finalize-owner-turn'; + } + return 'owner-follow-up'; +} + export async function runPendingPairedTurnIfNeeded(args: { chatJid: string; group: RegisteredGroup; @@ -79,6 +101,29 @@ export async function runPendingPairedTurnIfNeeded(args: { return null; } + if (pendingTurn.channel) { + const claimedUpdatedAt = claimPairedTurnExecution({ + chatJid, + runId: args.runId, + task, + intentKind: pendingTurn.intentKind, + }); + if (!claimedUpdatedAt) { + args.log.info( + { + chatJid, + taskId: task.id, + taskStatus: task.status, + taskUpdatedAt: task.updated_at, + intentKind: pendingTurn.intentKind, + }, + 'Skipped paired pending turn because the task revision was already claimed elsewhere', + ); + return true; + } + task.updated_at = claimedUpdatedAt; + } + return executePendingPairedTurn({ pendingTurn, chatJid, @@ -176,6 +221,34 @@ export async function runQueuedGroupTurn(args: { return false; } + if (task) { + const intentKind = resolveQueuedTurnReservationIntent({ + task, + turnRole, + hasHumanMessage: hasHumanMsg, + }); + const claimedUpdatedAt = claimPairedTurnExecution({ + chatJid, + runId, + task, + intentKind, + }); + if (!claimedUpdatedAt) { + log.info( + { + taskId: task.id, + taskStatus, + taskUpdatedAt: task.updated_at, + intentKind, + turnRole, + }, + 'Skipped queued paired turn because the task revision was already claimed elsewhere', + ); + return true; + } + task.updated_at = claimedUpdatedAt; + } + if (endSeq !== null) { advanceLastAgentCursor( args.lastAgentTimestamps, diff --git a/src/message-runtime-rules.test.ts b/src/message-runtime-rules.test.ts index f5a434e..a338720 100644 --- a/src/message-runtime-rules.test.ts +++ b/src/message-runtime-rules.test.ts @@ -11,6 +11,7 @@ vi.mock('./config.js', async () => { }); import { + matchesExpectedPairedFollowUpIntent, resolveFollowUpDispatch, resolveExecutionTarget, resolveNextTurnAction, @@ -79,6 +80,40 @@ describe('message-runtime-rules', () => { ).toEqual({ kind: 'none' }); }); + it('rejects stale reviewer and finalize follow-ups once the latest persisted turn already closed that handoff', () => { + expect( + matchesExpectedPairedFollowUpIntent({ + taskStatus: 'review_ready', + lastTurnOutputRole: 'reviewer', + intentKind: 'reviewer-turn', + }), + ).toBe(false); + expect( + matchesExpectedPairedFollowUpIntent({ + taskStatus: 'merge_ready', + lastTurnOutputRole: 'owner', + intentKind: 'finalize-owner-turn', + }), + ).toBe(false); + }); + + it('keeps owner follow-ups schedulable when reviewer or arbiter is still the latest persisted turn', () => { + expect( + matchesExpectedPairedFollowUpIntent({ + taskStatus: 'active', + lastTurnOutputRole: 'reviewer', + intentKind: 'owner-follow-up', + }), + ).toBe(true); + expect( + matchesExpectedPairedFollowUpIntent({ + taskStatus: 'active', + lastTurnOutputRole: 'arbiter', + intentKind: 'owner-follow-up', + }), + ).toBe(true); + }); + it('maps active tasks with reviewer output to an owner follow-up', () => { expect( resolveNextTurnAction({ diff --git a/src/message-runtime-rules.ts b/src/message-runtime-rules.ts index 1a495e0..a60ada6 100644 --- a/src/message-runtime-rules.ts +++ b/src/message-runtime-rules.ts @@ -82,6 +82,11 @@ export type NextTurnAction = | { kind: 'owner-follow-up' } | { kind: 'finalize-owner-turn' }; +export type ScheduledNextTurnActionKind = Exclude< + NextTurnAction['kind'], + 'none' +>; + export type FollowUpDispatch = | { kind: 'none' } | { kind: 'inline' } @@ -116,6 +121,19 @@ export function resolveNextTurnAction(args: { } } +export function matchesExpectedPairedFollowUpIntent(args: { + taskStatus?: PairedTaskStatus | null; + lastTurnOutputRole?: PairedRoomRole | null; + intentKind: ScheduledNextTurnActionKind; +}): boolean { + return ( + resolveNextTurnAction({ + taskStatus: args.taskStatus, + lastTurnOutputRole: args.lastTurnOutputRole, + }).kind === args.intentKind + ); +} + export function resolveFollowUpDispatch(args: { source: | 'delivery-success' diff --git a/src/message-runtime.test.ts b/src/message-runtime.test.ts index b381bef..5436745 100644 --- a/src/message-runtime.test.ts +++ b/src/message-runtime.test.ts @@ -65,6 +65,15 @@ vi.mock('./db.js', () => { ...message, seq: typeof message.seq === 'number' ? message.seq : index + 1, })); + const pairedTurnReservations = new Set(); + const claimedTaskRevisions = new Set(); + const buildReservationKey = (args: { + chatJid: string; + taskId: string; + taskUpdatedAt: string; + intentKind: string; + }) => + [args.chatJid, args.taskId, args.taskUpdatedAt, args.intentKind].join(':'); return { claimServiceHandoff: vi.fn(() => true), @@ -149,6 +158,34 @@ vi.mock('./db.js', () => { markWorkItemDelivered: vi.fn(), markWorkItemDeliveryRetry: vi.fn(), getLastBotFinalMessage: vi.fn(() => []), + reservePairedTurnReservation: vi.fn((args) => { + const key = buildReservationKey(args); + if (pairedTurnReservations.has(key)) { + return false; + } + pairedTurnReservations.add(key); + return true; + }), + claimPairedTurnReservation: vi.fn((args) => { + const revisionKey = [args.taskId, args.taskUpdatedAt].join(':'); + if (claimedTaskRevisions.has(revisionKey)) { + return false; + } + claimedTaskRevisions.add(revisionKey); + pairedTurnReservations.add( + buildReservationKey({ + chatJid: args.chatJid, + taskId: args.taskId, + taskUpdatedAt: args.taskUpdatedAt, + intentKind: args.intentKind, + }), + ); + return true; + }), + _clearPairedTurnReservationsForTests: vi.fn(() => { + pairedTurnReservations.clear(); + claimedTaskRevisions.clear(); + }), }; }); @@ -573,6 +610,7 @@ describe('createMessageRuntime', () => { const group = makeGroup('codex'); const channel = makeChannel(chatJid); const enqueueMessageCheck = vi.fn(); + const lastAgentTimestamps: Record = {}; vi.mocked(db.getOpenWorkItem).mockReturnValue({ id: 99, @@ -619,7 +657,7 @@ describe('createMessageRuntime', () => { getSessions: () => ({}), getLastTimestamp: () => '', setLastTimestamp: vi.fn(), - getLastAgentTimestamps: () => ({}), + getLastAgentTimestamps: () => lastAgentTimestamps, saveState: vi.fn(), persistSession: vi.fn(), clearSession: vi.fn(), @@ -1154,6 +1192,7 @@ describe('createMessageRuntime', () => { }; }, ); + const lastAgentTimestamps: Record = {}; const runtime = createMessageRuntime({ assistantName: 'Andy', @@ -1178,7 +1217,7 @@ describe('createMessageRuntime', () => { getSessions: () => ({}), getLastTimestamp: () => '', setLastTimestamp: vi.fn(), - getLastAgentTimestamps: () => ({}), + getLastAgentTimestamps: () => lastAgentTimestamps, saveState: vi.fn(), persistSession: vi.fn(), clearSession: vi.fn(), @@ -1818,6 +1857,124 @@ If your first line is DONE_WITH_CONCERNS, the system will reopen review instead ); }); + it('skips a stale reviewer follow-up after owner delivery when the latest persisted turn already belongs to the reviewer', async () => { + const chatJid = 'group@test'; + const group = makeGroup('codex'); + const ownerChannel: Channel = { + ...makeChannel(chatJid), + isOwnMessage: vi.fn((msg) => msg.sender === 'owner-bot@test'), + }; + const reviewerChannel = makeChannel(chatJid, 'discord-review', false); + const enqueueMessageCheck = vi.fn(); + const pairedTask = { + id: 'task-owner-delivery-stale-reviewer-follow-up', + chat_jid: chatJid, + group_folder: group.folder, + owner_service_id: 'claude', + reviewer_service_id: 'codex-main', + title: null, + source_ref: 'HEAD', + plan_notes: null, + review_requested_at: null, + round_trip_count: 0, + status: 'active', + arbiter_verdict: null, + arbiter_requested_at: null, + completion_reason: null, + created_at: '2026-03-30T00:00:00.000Z', + updated_at: '2026-03-30T00:00:00.000Z', + } as any; + let persistedTurnOutputs: any[] = []; + + vi.mocked(serviceRouting.hasReviewerLease).mockReturnValue(true); + vi.mocked(db.getLatestOpenPairedTaskForChat).mockImplementation( + () => pairedTask, + ); + vi.mocked(db.getPairedTurnOutputs).mockImplementation((taskId: string) => + taskId === pairedTask.id ? persistedTurnOutputs : [], + ); + vi.mocked(db.getMessagesSince).mockReturnValue([ + { + id: 'human-stale-reviewer-1', + chat_jid: chatJid, + sender: 'user@test', + sender_name: 'User', + content: '이 구현 진행해줘', + timestamp: '2026-03-30T00:00:00.000Z', + seq: 1, + is_bot_message: false, + } as any, + ]); + vi.mocked(agentRunner.runAgentProcess).mockImplementation( + async (_group, _input, _onProcess, onOutput) => { + pairedTask.status = 'review_ready'; + pairedTask.review_requested_at = '2026-03-30T00:00:01.000Z'; + pairedTask.round_trip_count = 1; + pairedTask.updated_at = '2026-03-30T00:00:02.000Z'; + persistedTurnOutputs = [ + { + id: 1, + task_id: pairedTask.id, + turn_number: 1, + role: 'owner', + output_text: 'owner 응답', + created_at: '2026-03-30T00:00:01.000Z', + }, + { + id: 2, + task_id: pairedTask.id, + turn_number: 2, + role: 'reviewer', + output_text: 'reviewer 승인', + created_at: '2026-03-30T00:00:02.000Z', + }, + ]; + await onOutput?.({ + status: 'success', + phase: 'final', + result: 'DONE_WITH_CONCERNS\nowner complete', + newSessionId: 'session-owner-delivery-stale-reviewer-follow-up', + }); + return { + status: 'success', + result: 'DONE_WITH_CONCERNS\nowner complete', + newSessionId: 'session-owner-delivery-stale-reviewer-follow-up', + }; + }, + ); + + const runtime = createMessageRuntime({ + assistantName: 'Andy', + idleTimeout: 1_000, + pollInterval: 1_000, + timezone: 'UTC', + triggerPattern: /^@Andy\b/i, + channels: [ownerChannel, reviewerChannel], + queue: { + registerProcess: vi.fn(), + closeStdin: vi.fn(), + notifyIdle: vi.fn(), + enqueueMessageCheck, + } as any, + getRegisteredGroups: () => ({ [chatJid]: group }), + getSessions: () => ({}), + getLastTimestamp: () => '', + setLastTimestamp: vi.fn(), + getLastAgentTimestamps: () => ({}), + saveState: vi.fn(), + persistSession: vi.fn(), + clearSession: vi.fn(), + }); + + const result = await runtime.processGroupMessages(chatJid, { + runId: 'run-owner-delivery-stale-reviewer-follow-up', + reason: 'messages', + }); + + expect(result).toBe(true); + expect(enqueueMessageCheck).not.toHaveBeenCalled(); + }); + it('does not enqueue the same reviewer follow-up twice within the same run', async () => { const chatJid = 'group@test'; const group = makeGroup('codex'); @@ -1850,18 +2007,23 @@ If your first line is DONE_WITH_CONCERNS, the system will reopen review instead vi.mocked(db.getLatestOpenPairedTaskForChat).mockImplementation( () => pairedTask, ); - vi.mocked(db.getMessagesSince).mockReturnValue([ - { - id: 'human-dedup-1', - chat_jid: chatJid, - sender: 'user@test', - sender_name: 'User', - content: '이 구현 진행해줘', - timestamp: '2026-03-30T00:00:00.000Z', - seq: 1, - is_bot_message: false, - } as any, - ]); + vi.mocked(db.getMessagesSince).mockImplementation( + (_chatJid, sinceCursor) => + sinceCursor && sinceCursor !== '0' + ? [] + : ([ + { + id: 'human-dedup-1', + chat_jid: chatJid, + sender: 'user@test', + sender_name: 'User', + content: '이 구현 진행해줘', + timestamp: '2026-03-30T00:00:00.000Z', + seq: 1, + is_bot_message: false, + }, + ] as any), + ); vi.mocked(agentRunner.runAgentProcess).mockImplementation( async (_group, _input, _onProcess, onOutput) => { pairedTask.status = 'review_ready'; @@ -1908,23 +2070,16 @@ If your first line is DONE_WITH_CONCERNS, the system will reopen review instead runId: 'run-owner-delivery-dedup', reason: 'messages', }); - await runtime.processGroupMessages(chatJid, { + const secondScheduled = schedulePairedFollowUpOnce({ + chatJid, runId: 'run-owner-delivery-dedup', - reason: 'messages', + task: pairedTask, + intentKind: 'reviewer-turn', + enqueue: enqueueMessageCheck, }); expect(enqueueMessageCheck).toHaveBeenCalledTimes(1); - expect(logger.info).toHaveBeenCalledWith( - expect.objectContaining({ - chatJid, - runId: 'run-owner-delivery-dedup', - completedRole: 'owner', - taskId: 'task-owner-delivery-dedup', - taskStatus: 'review_ready', - scheduled: false, - }), - 'Skipped duplicate paired follow-up after successful owner delivery while task state was unchanged', - ); + expect(secondScheduled).toBe(false); }); it('re-enqueues owner after a successful reviewer delivery moves the task back to active', async () => { @@ -2354,18 +2509,23 @@ If your first line is DONE_WITH_CONCERNS, the system will reopen review instead vi.mocked(db.getLatestOpenPairedTaskForChat).mockImplementation( () => pairedTask, ); - vi.mocked(db.getMessagesSince).mockReturnValue([ - { - id: 'human-dedup-across-runs-1', - chat_jid: chatJid, - sender: 'user@test', - sender_name: 'User', - content: '이 구현 진행해줘', - timestamp: '2026-03-30T00:00:00.000Z', - seq: 1, - is_bot_message: false, - } as any, - ]); + vi.mocked(db.getMessagesSince).mockImplementation( + (_chatJid, sinceCursor) => + sinceCursor && sinceCursor !== '0' + ? [] + : ([ + { + id: 'human-dedup-across-runs-1', + chat_jid: chatJid, + sender: 'user@test', + sender_name: 'User', + content: '이 구현 진행해줘', + timestamp: '2026-03-30T00:00:00.000Z', + seq: 1, + is_bot_message: false, + }, + ] as any), + ); vi.mocked(agentRunner.runAgentProcess).mockImplementation( async (_group, _input, _onProcess, onOutput) => { pairedTask.status = 'review_ready'; @@ -2412,23 +2572,16 @@ If your first line is DONE_WITH_CONCERNS, the system will reopen review instead runId: 'run-owner-delivery-dedup-1', reason: 'messages', }); - await runtime.processGroupMessages(chatJid, { + const secondScheduled = schedulePairedFollowUpOnce({ + chatJid, runId: 'run-owner-delivery-dedup-2', - reason: 'drain', + task: pairedTask, + intentKind: 'reviewer-turn', + enqueue: enqueueMessageCheck, }); expect(enqueueMessageCheck).toHaveBeenCalledTimes(1); - expect(logger.info).toHaveBeenCalledWith( - expect.objectContaining({ - chatJid, - runId: 'run-owner-delivery-dedup-2', - completedRole: 'owner', - taskId: 'task-owner-delivery-dedup-across-runs', - taskStatus: 'review_ready', - scheduled: false, - }), - 'Skipped duplicate paired follow-up after successful owner delivery while task state was unchanged', - ); + expect(secondScheduled).toBe(false); }); it('routes fresh human input to owner even when the latest task is review_ready', async () => { diff --git a/src/message-runtime.ts b/src/message-runtime.ts index 9ee7937..0060ea5 100644 --- a/src/message-runtime.ts +++ b/src/message-runtime.ts @@ -38,18 +38,17 @@ import { import { advanceLastAgentCursor, createImplicitContinuationTracker, - resolveNextTurnAction, - resolveFollowUpDispatch, filterLoopingPairedBotMessages, getProcessableMessages, shouldSkipBotOnlyCollaboration, } from './message-runtime-rules.js'; +import { + enqueuePairedFollowUpAfterEvent, + schedulePairedFollowUpWithMessageCheck, +} from './message-runtime-follow-up.js'; import { runAgentForGroup } from './message-agent-executor.js'; import { MessageTurnController } from './message-turn-controller.js'; -import { - schedulePairedFollowUpOnce, - type ScheduledPairedFollowUpIntentKind, -} from './paired-follow-up-scheduler.js'; +import { type ScheduledPairedFollowUpIntentKind } from './paired-follow-up-scheduler.js'; import { transitionPairedTaskStatus } from './paired-execution-context-shared.js'; import { Channel, @@ -168,18 +167,16 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { task: PairedTask | null | undefined; intentKind: ScheduledPairedFollowUpIntentKind; enqueue: () => void; - }): boolean => { - if (!args.task) { - return false; - } - return schedulePairedFollowUpOnce({ + fallbackLastTurnOutputRole?: PairedRoomRole | null; + }): boolean => + schedulePairedFollowUpWithMessageCheck({ chatJid: args.chatJid, runId: args.runId, task: args.task, intentKind: args.intentKind, - enqueue: args.enqueue, + enqueueMessageCheck: args.enqueue, + fallbackLastTurnOutputRole: args.fallbackLastTurnOutputRole, }); - }; /** * Check if a message is a duplicate of the last bot final message in a paired room. @@ -346,37 +343,27 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { if (deliverySucceeded && pairedRoom && resolvedDeliveryRole) { const pendingTaskAfterDelivery = getLatestOpenPairedTaskForChat(chatJid); - const nextTurnAction = resolveNextTurnAction({ - taskStatus: pendingTaskAfterDelivery?.status, - lastTurnOutputRole: resolvedDeliveryRole, - }); - const dispatch = resolveFollowUpDispatch({ + const followUpResult = enqueuePairedFollowUpAfterEvent({ + chatJid, + runId, + task: pendingTaskAfterDelivery, source: 'delivery-success', - nextTurnAction, completedRole: resolvedDeliveryRole, + fallbackLastTurnOutputRole: resolvedDeliveryRole, + enqueueMessageCheck: () => deps.queue.enqueueMessageCheck(chatJid), }); - if ( - dispatch.kind === 'enqueue' && - dispatch.queueKind === 'paired-follow-up' && - nextTurnAction.kind !== 'none' - ) { - const scheduled = scheduleQueuedPairedFollowUp({ - chatJid, - runId, - task: pendingTaskAfterDelivery, - intentKind: nextTurnAction.kind, - enqueue: () => deps.queue.enqueueMessageCheck(chatJid), - }); + if (followUpResult.kind === 'paired-follow-up') { logger.info( { chatJid, runId, completedRole: resolvedDeliveryRole, - taskId: pendingTaskAfterDelivery?.id ?? null, - taskStatus: pendingTaskAfterDelivery?.status ?? null, - scheduled, + taskId: followUpResult.taskId, + taskStatus: followUpResult.taskStatus, + intentKind: followUpResult.intentKind, + scheduled: followUpResult.scheduled, }, - scheduled + followUpResult.scheduled ? resolvedDeliveryRole === 'owner' ? 'Queued paired follow-up after successful owner delivery' : 'Queued paired follow-up after successful reviewer/arbiter delivery' @@ -513,14 +500,6 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { workItemId, log, enqueueMessageCheck: () => deps.queue.enqueueMessageCheck(chatJid), - schedulePairedFollowUp: (task, intentKind, followUpRunId) => - scheduleQueuedPairedFollowUp({ - chatJid, - runId: followUpRunId, - task, - intentKind, - enqueue: () => deps.queue.enqueueMessageCheck(chatJid), - }), }), }); if (openWorkItemOutcome === 'failed') { @@ -668,6 +647,7 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { taskId: task.id, currentStatus: task.status, nextStatus: 'completed', + expectedUpdatedAt: task.updated_at, updatedAt: now, patch: { completion_reason: 'stopped', diff --git a/src/paired-execution-context-arbiter.ts b/src/paired-execution-context-arbiter.ts index 04c33f1..9f52b22 100644 --- a/src/paired-execution-context-arbiter.ts +++ b/src/paired-execution-context-arbiter.ts @@ -21,6 +21,7 @@ export function handleFailedArbiterExecution(args: { taskId, currentStatus: task.status, nextStatus: fallbackStatus, + expectedUpdatedAt: task.updated_at, updatedAt: now, }); logger.warn( @@ -36,10 +37,11 @@ export function handleFailedArbiterExecution(args: { } export function handleArbiterCompletion(args: { + task: PairedTask; taskId: string; summary?: string | null; }): void { - const { taskId, summary } = args; + const { task, taskId, summary } = args; const now = new Date().toISOString(); const arbiterVerdict = classifyArbiterVerdict(summary); @@ -56,6 +58,7 @@ export function handleArbiterCompletion(args: { taskId, currentStatus: 'in_arbitration', nextStatus: 'active', + expectedUpdatedAt: task.updated_at, updatedAt: now, patch: { round_trip_count: Math.max(0, ARBITER_DEADLOCK_THRESHOLD - 1), @@ -72,6 +75,7 @@ export function handleArbiterCompletion(args: { taskId, currentStatus: 'in_arbitration', nextStatus: 'completed', + expectedUpdatedAt: task.updated_at, updatedAt: now, patch: { arbiter_verdict: 'escalate', @@ -85,6 +89,7 @@ export function handleArbiterCompletion(args: { taskId, currentStatus: 'in_arbitration', nextStatus: 'active', + expectedUpdatedAt: task.updated_at, updatedAt: now, patch: { round_trip_count: Math.max(0, ARBITER_DEADLOCK_THRESHOLD - 1), diff --git a/src/paired-execution-context-owner.ts b/src/paired-execution-context-owner.ts index f26eedd..d132320 100644 --- a/src/paired-execution-context-owner.ts +++ b/src/paired-execution-context-owner.ts @@ -2,7 +2,11 @@ import { ARBITER_DEADLOCK_THRESHOLD, PAIRED_MAX_ROUND_TRIPS, } from './config.js'; -import { getPairedWorkspace, hasActiveCiWatcherForChat } from './db.js'; +import { + getPairedTaskById, + getPairedWorkspace, + hasActiveCiWatcherForChat, +} from './db.js'; import { logger } from './logger.js'; import { markPairedTaskReviewReady } from './paired-workspace-manager.js'; import { @@ -29,6 +33,7 @@ export function handleFailedOwnerExecution(args: { taskId, currentStatus: task.status, nextStatus: 'active', + expectedUpdatedAt: task.updated_at, updatedAt: now, }); logger.info( @@ -74,6 +79,7 @@ function handleOwnerFinalizeCompletion(args: { requestArbiterOrEscalate({ taskId, currentStatus: task.status, + expectedUpdatedAt: task.updated_at, now, arbiterLogMessage, escalateLogMessage, @@ -94,6 +100,7 @@ function handleOwnerFinalizeCompletion(args: { taskId, currentStatus: task.status, nextStatus: 'active', + expectedUpdatedAt: task.updated_at, updatedAt: now, }); } @@ -115,6 +122,7 @@ function handleOwnerFinalizeCompletion(args: { taskId, currentStatus: task.status, nextStatus: 'completed', + expectedUpdatedAt: task.updated_at, updatedAt: now, patch: { completion_reason: 'done', @@ -157,8 +165,13 @@ function maybeAutoTriggerReviewerAfterOwnerCompletion(args: { const result = markPairedTaskReviewReady(taskId); if (result) { + const reviewReadyTask = getPairedTaskById(taskId); + if (!reviewReadyTask) { + return; + } applyPairedTaskPatch({ taskId, + expectedUpdatedAt: reviewReadyTask.updated_at, updatedAt: now, patch: { round_trip_count: task.round_trip_count + 1, @@ -205,6 +218,7 @@ export function handleOwnerCompletion(args: { requestArbiterOrEscalate({ taskId, currentStatus: task.status, + expectedUpdatedAt: task.updated_at, now, arbiterLogMessage: 'Owner blocked/needs_context — requesting arbiter', escalateLogMessage: 'Owner blocked/needs_context — escalating to user', diff --git a/src/paired-execution-context-reviewer.ts b/src/paired-execution-context-reviewer.ts index 39b5602..939b639 100644 --- a/src/paired-execution-context-reviewer.ts +++ b/src/paired-execution-context-reviewer.ts @@ -1,5 +1,5 @@ import { ARBITER_DEADLOCK_THRESHOLD } from './config.js'; -import { getPairedWorkspace, updatePairedTask } from './db.js'; +import { getPairedWorkspace } from './db.js'; import { logger } from './logger.js'; import { parseVisibleVerdict, @@ -43,6 +43,7 @@ export function handleFailedReviewerExecution(args: { signal.kind === 'request_owner_finalize' ? 'merge_ready' : 'completed', + expectedUpdatedAt: task.updated_at, updatedAt: now, patch: { ...(signal.kind === 'request_owner_finalize' @@ -75,6 +76,7 @@ export function handleFailedReviewerExecution(args: { taskId, currentStatus: task.status, nextStatus: fallbackStatus, + expectedUpdatedAt: task.updated_at, updatedAt: now, }); logger.warn( @@ -113,6 +115,7 @@ export function handleReviewerCompletion(args: { taskId, currentStatus: task.status, nextStatus: 'merge_ready', + expectedUpdatedAt: task.updated_at, updatedAt: now, patch: { source_ref: approvedSourceRef, @@ -142,6 +145,7 @@ export function handleReviewerCompletion(args: { requestArbiterOrEscalate({ taskId, currentStatus: task.status, + expectedUpdatedAt: task.updated_at, now, arbiterLogMessage, escalateLogMessage, @@ -154,6 +158,7 @@ export function handleReviewerCompletion(args: { taskId, currentStatus: task.status, nextStatus: 'active', + expectedUpdatedAt: task.updated_at, updatedAt: now, }); logger.info( diff --git a/src/paired-execution-context-shared.ts b/src/paired-execution-context-shared.ts index aa0e87b..58d2f95 100644 --- a/src/paired-execution-context-shared.ts +++ b/src/paired-execution-context-shared.ts @@ -1,7 +1,7 @@ import { execFileSync } from 'child_process'; import { isArbiterEnabled } from './config.js'; -import { updatePairedTask } from './db.js'; +import { updatePairedTaskIfUnchanged } from './db.js'; import { logger } from './logger.js'; import type { PairedTaskStatus } from './types.js'; @@ -239,70 +239,134 @@ export function transitionPairedTaskStatus(args: { taskId: string; currentStatus: PairedTaskStatus; nextStatus: PairedTaskStatus; + expectedUpdatedAt: string; updatedAt: string; - patch?: Omit[1], 'status' | 'updated_at'>; -}): void { + patch?: { + title?: string | null; + source_ref?: string | null; + plan_notes?: string | null; + review_requested_at?: string | null; + round_trip_count?: number; + arbiter_verdict?: string | null; + arbiter_requested_at?: string | null; + completion_reason?: string | null; + }; +}): boolean { assertPairedTaskStatusTransition({ currentStatus: args.currentStatus, nextStatus: args.nextStatus, }); - updatePairedTask(args.taskId, { - ...args.patch, - status: args.nextStatus, - updated_at: args.updatedAt, - }); + const updated = updatePairedTaskIfUnchanged( + args.taskId, + args.expectedUpdatedAt, + { + ...args.patch, + status: args.nextStatus, + updated_at: args.updatedAt, + }, + ); + if (!updated) { + logger.warn( + { + taskId: args.taskId, + currentStatus: args.currentStatus, + nextStatus: args.nextStatus, + expectedUpdatedAt: args.expectedUpdatedAt, + }, + 'Skipped stale paired task status transition because the task revision changed', + ); + } + return updated; } export function applyPairedTaskPatch(args: { taskId: string; + expectedUpdatedAt: string; updatedAt: string; - patch: Omit[1], 'updated_at'>; -}): void { - updatePairedTask(args.taskId, { - ...args.patch, - updated_at: args.updatedAt, - }); + patch: { + title?: string | null; + source_ref?: string | null; + plan_notes?: string | null; + review_requested_at?: string | null; + round_trip_count?: number; + status?: PairedTaskStatus; + arbiter_verdict?: string | null; + arbiter_requested_at?: string | null; + completion_reason?: string | null; + }; +}): boolean { + const updated = updatePairedTaskIfUnchanged( + args.taskId, + args.expectedUpdatedAt, + { + ...args.patch, + updated_at: args.updatedAt, + }, + ); + if (!updated) { + logger.warn( + { + taskId: args.taskId, + expectedUpdatedAt: args.expectedUpdatedAt, + }, + 'Skipped stale paired task patch because the task revision changed', + ); + } + return updated; } export function requestArbiterOrEscalate(args: { taskId: string; currentStatus: PairedTaskStatus; + expectedUpdatedAt: string; now: string; arbiterLogMessage: string; escalateLogMessage: string; logContext?: Record; -}): void { +}): boolean { const { taskId, currentStatus, + expectedUpdatedAt, now, arbiterLogMessage, escalateLogMessage, logContext, } = args; if (isArbiterEnabled()) { - transitionPairedTaskStatus({ + const transitioned = transitionPairedTaskStatus({ taskId, currentStatus, nextStatus: 'arbiter_requested', + expectedUpdatedAt, updatedAt: now, patch: { arbiter_requested_at: now, }, }); - logger.info(logContext ?? { taskId }, arbiterLogMessage); - return; + if (transitioned) { + logger.info(logContext ?? { taskId }, arbiterLogMessage); + } + return transitioned; } - transitionPairedTaskStatus({ + const transitioned = transitionPairedTaskStatus({ taskId, currentStatus, nextStatus: 'completed', + expectedUpdatedAt, updatedAt: now, patch: { completion_reason: 'escalated', }, }); - logger.info(logContext ?? { taskId }, escalateLogMessage); + if (transitioned) { + logger.info(logContext ?? { taskId }, escalateLogMessage); + } + return transitioned; } + +export type TransitionPairedTaskStatusFn = typeof transitionPairedTaskStatus; +export type ApplyPairedTaskPatchFn = typeof applyPairedTaskPatch; +export type RequestArbiterOrEscalateFn = typeof requestArbiterOrEscalate; diff --git a/src/paired-execution-context.test.ts b/src/paired-execution-context.test.ts index 6014fa9..af608ec 100644 --- a/src/paired-execution-context.test.ts +++ b/src/paired-execution-context.test.ts @@ -5,16 +5,24 @@ import path from 'path'; import { afterAll, beforeEach, describe, expect, it, vi } from 'vitest'; -vi.mock('./db.js', () => ({ - createPairedTask: vi.fn(), - getLatestPairedTaskForChat: vi.fn(), - getLatestOpenPairedTaskForChat: vi.fn(), - getPairedTaskById: vi.fn(), - getPairedWorkspace: vi.fn(), - updatePairedTask: vi.fn(), - upsertPairedProject: vi.fn(), - hasActiveCiWatcherForChat: vi.fn(() => false), -})); +vi.mock('./db.js', () => { + const updatePairedTask = vi.fn(); + return { + createPairedTask: vi.fn(), + getLatestPairedTaskForChat: vi.fn(), + getLatestOpenPairedTaskForChat: vi.fn(), + getPairedTaskById: vi.fn(), + getPairedWorkspace: vi.fn(), + updatePairedTask, + updatePairedTaskIfUnchanged: vi.fn((id, _expectedUpdatedAt, updates) => { + updatePairedTask(id, updates); + return true; + }), + upsertPairedProject: vi.fn(), + hasActiveCiWatcherForChat: vi.fn(() => false), + releasePairedTaskExecutionLease: vi.fn(), + }; +}); vi.mock('./paired-workspace-manager.js', () => ({ markPairedTaskReviewReady: vi.fn(), @@ -857,4 +865,30 @@ describe('paired execution context', () => { }), ); }); + + it('releases the execution lease even when a completion handler throws', () => { + const transitionError = new Error('transition failed'); + vi.mocked(db.getPairedTaskById).mockReturnValue( + buildPairedTask({ + status: 'merge_ready', + }), + ); + vi.mocked(db.updatePairedTaskIfUnchanged).mockImplementationOnce(() => { + throw transitionError; + }); + + expect(() => + completePairedExecutionContext({ + taskId: 'task-1', + role: 'owner', + status: 'failed', + runId: 'run-lease-release', + summary: 'push failed', + }), + ).toThrow('transition failed'); + expect(db.releasePairedTaskExecutionLease).toHaveBeenCalledWith({ + taskId: 'task-1', + runId: 'run-lease-release', + }); + }); }); diff --git a/src/paired-execution-context.ts b/src/paired-execution-context.ts index 3f2a3be..07e972f 100644 --- a/src/paired-execution-context.ts +++ b/src/paired-execution-context.ts @@ -19,6 +19,7 @@ import { getPairedTaskById, getPairedWorkspace, hasActiveCiWatcherForChat, + releasePairedTaskExecutionLease, upsertPairedProject, } from './db.js'; import { logger } from './logger.js'; @@ -232,6 +233,7 @@ export function preparePairedExecutionContext(args: { taskId: latestTask.id, currentStatus: latestTask.status, nextStatus: 'active', + expectedUpdatedAt: latestTask.updated_at, updatedAt: now, patch: { ...(hasHuman ? { round_trip_count: 0 } : {}), @@ -240,6 +242,7 @@ export function preparePairedExecutionContext(args: { } else { applyPairedTaskPatch({ taskId: latestTask.id, + expectedUpdatedAt: latestTask.updated_at, updatedAt: now, patch: { ...(hasHuman ? { round_trip_count: 0 } : {}), @@ -258,6 +261,7 @@ export function preparePairedExecutionContext(args: { if (wsRef !== latestTask.source_ref) { applyPairedTaskPatch({ taskId: latestTask.id, + expectedUpdatedAt: latestTask.updated_at, updatedAt: now, patch: { source_ref: wsRef, @@ -275,6 +279,7 @@ export function preparePairedExecutionContext(args: { taskId: latestTask.id, currentStatus: refreshedTask.status, nextStatus: 'in_review', + expectedUpdatedAt: refreshedTask.updated_at, updatedAt: now, }); } @@ -289,6 +294,7 @@ export function preparePairedExecutionContext(args: { taskId: latestTask.id, currentStatus: refreshedTask.status, nextStatus: 'in_arbitration', + expectedUpdatedAt: refreshedTask.updated_at, updatedAt: now, }); } @@ -353,9 +359,11 @@ export function completePairedExecutionContext(args: { taskId: string; role: PairedRoomRole; status: 'succeeded' | 'failed'; + runId?: string; summary?: string | null; }): void { const { taskId, role, status } = args; + let completionError: unknown; logger.info( { taskId, @@ -366,49 +374,76 @@ export function completePairedExecutionContext(args: { 'Paired execution completed', ); - const task = getPairedTaskById(taskId); - if (!task) return; - if (task.status === 'completed') { - logger.info( - { - taskId, - role, - status, - completionReason: task.completion_reason ?? null, - }, - 'Ignoring late paired execution completion for an already completed task', - ); - return; - } + try { + const task = getPairedTaskById(taskId); + if (!task) { + return; + } + if (task.status === 'completed') { + logger.info( + { + taskId, + role, + status, + completionReason: task.completion_reason ?? null, + }, + 'Ignoring late paired execution completion for an already completed task', + ); + return; + } + + if (status !== 'succeeded') { + if (role === 'reviewer') { + handleFailedReviewerExecution({ + task, + taskId, + summary: args.summary, + }); + return; + } + if (role === 'arbiter') { + handleFailedArbiterExecution({ task, taskId }); + return; + } + handleFailedOwnerExecution({ task, taskId }); + return; + } + + if (role === 'owner') { + handleOwnerCompletion({ task, taskId, summary: args.summary }); + return; + } - if (status !== 'succeeded') { if (role === 'reviewer') { - handleFailedReviewerExecution({ - task, - taskId, - summary: args.summary, - }); + handleReviewerCompletion({ task, taskId, summary: args.summary }); return; } + if (role === 'arbiter') { - handleFailedArbiterExecution({ task, taskId }); + handleArbiterCompletion({ task, taskId, summary: args.summary }); + } + } catch (error) { + completionError = error; + throw error; + } finally { + if (!args.runId) { return; } - handleFailedOwnerExecution({ task, taskId }); - return; - } - - if (role === 'owner') { - handleOwnerCompletion({ task, taskId, summary: args.summary }); - return; - } - - if (role === 'reviewer') { - handleReviewerCompletion({ task, taskId, summary: args.summary }); - return; - } - - if (role === 'arbiter') { - handleArbiterCompletion({ taskId, summary: args.summary }); + try { + releasePairedTaskExecutionLease({ taskId, runId: args.runId }); + } catch (releaseError) { + logger.error( + { + taskId, + role, + runId: args.runId, + releaseError, + }, + 'Failed to release paired task execution lease after completion', + ); + if (!completionError) { + throw releaseError; + } + } } } diff --git a/src/paired-follow-up-scheduler.test.ts b/src/paired-follow-up-scheduler.test.ts index e20f6df..e762b51 100644 --- a/src/paired-follow-up-scheduler.test.ts +++ b/src/paired-follow-up-scheduler.test.ts @@ -1,14 +1,26 @@ +import { Database } from 'bun:sqlite'; +import fs from 'fs'; +import os from 'os'; +import path from 'path'; + import { beforeEach, describe, expect, it, vi } from 'vitest'; import { - SCHEDULED_PAIRED_FOLLOW_UP_TTL_MS, + _initTestDatabase, + _initTestDatabaseFromFile, + createPairedTask, + getPairedTaskById, +} from './db.js'; +import { buildPairedFollowUpKey, + claimPairedTurnExecution, resetPairedFollowUpScheduleState, schedulePairedFollowUpOnce, } from './paired-follow-up-scheduler.js'; describe('paired follow-up scheduler', () => { beforeEach(() => { + _initTestDatabase(); resetPairedFollowUpScheduleState(); vi.useRealTimers(); }); @@ -148,38 +160,132 @@ describe('paired follow-up scheduler', () => { expect(enqueue).toHaveBeenCalledTimes(2); }); - it('allows the same follow-up again after the TTL expires', () => { - vi.useFakeTimers(); - vi.setSystemTime(new Date('2026-03-30T00:00:00.000Z')); - - const enqueue = vi.fn(); + it('blocks a fresh-refetched task revision from reclaiming the same turn while the execution lease is active', () => { const task = { id: 'task-1', + 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); - const first = schedulePairedFollowUpOnce({ + const first = claimPairedTurnExecution({ chatJid: 'group@test', runId: 'run-1', task, intentKind: 'reviewer-turn', - enqueue, }); - - vi.advanceTimersByTime(SCHEDULED_PAIRED_FOLLOW_UP_TTL_MS + 1); - - const second = schedulePairedFollowUpOnce({ + const freshRefetch = getPairedTaskById(task.id); + const second = claimPairedTurnExecution({ chatJid: 'group@test', - runId: 'run-1', - task, + runId: 'run-2', + task: freshRefetch ?? task, intentKind: 'reviewer-turn', - enqueue, }); - expect(first).toBe(true); - expect(second).toBe(true); - expect(enqueue).toHaveBeenCalledTimes(2); + 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); + }); + + it('reclaims an expired execution lease after restart while blocking fresh claims before expiry', () => { + const tempDir = fs.mkdtempSync( + path.join(os.tmpdir(), 'ejclaw-paired-lease-'), + ); + const dbPath = path.join(tempDir, 'paired-state.db'); + + try { + _initTestDatabaseFromFile(dbPath); + resetPairedFollowUpScheduleState(); + + const task = { + id: 'task-restart-lease', + 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); + + const first = claimPairedTurnExecution({ + chatJid: task.chat_jid, + runId: 'run-first-reviewer', + task, + intentKind: 'reviewer-turn', + }); + expect(first).toMatch(/^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d{3}Z$/); + + _initTestDatabaseFromFile(dbPath); + const freshTask = getPairedTaskById(task.id); + expect(freshTask).toBeDefined(); + expect( + claimPairedTurnExecution({ + chatJid: task.chat_jid, + runId: 'run-before-expiry-reviewer', + task: freshTask ?? task, + intentKind: 'reviewer-turn', + }), + ).toBeNull(); + expect( + claimPairedTurnExecution({ + chatJid: task.chat_jid, + runId: 'run-before-expiry-owner', + task: freshTask ?? task, + intentKind: 'owner-turn', + }), + ).toBeNull(); + + const rawDatabase = new Database(dbPath); + rawDatabase + .prepare( + ` + UPDATE paired_task_execution_leases + SET updated_at = ?, + expires_at = ? + WHERE task_id = ? + `, + ) + .run('2020-01-01T00:00:00.000Z', '2020-01-01T00:00:00.000Z', task.id); + rawDatabase.close(); + + _initTestDatabaseFromFile(dbPath); + const recoveredTask = getPairedTaskById(task.id); + expect(recoveredTask).toBeDefined(); + expect( + claimPairedTurnExecution({ + chatJid: task.chat_jid, + runId: 'run-after-expiry-reviewer', + task: recoveredTask ?? task, + intentKind: 'reviewer-turn', + }), + ).toMatch(/^\d{4}-\d{2}-\d{2}T/); + } finally { + fs.rmSync(tempDir, { recursive: true, force: true }); + _initTestDatabase(); + resetPairedFollowUpScheduleState(); + } }); }); diff --git a/src/paired-follow-up-scheduler.ts b/src/paired-follow-up-scheduler.ts index 9a91f4b..6f0faf1 100644 --- a/src/paired-follow-up-scheduler.ts +++ b/src/paired-follow-up-scheduler.ts @@ -1,4 +1,13 @@ -import type { PairedTask, PairedTaskStatus } from './types.js'; +import { + _clearPairedTurnReservationsForTests, + claimPairedTurnReservation, + reservePairedTurnReservation, +} from './db.js'; +import type { + PairedTask, + PairedTaskStatus, + PairedTurnReservationIntentKind, +} from './types.js'; export type ScheduledPairedFollowUpIntentKind = | 'reviewer-turn' @@ -11,17 +20,6 @@ type ScheduledPairedFollowUpTask = Pick< 'id' | 'status' | 'round_trip_count' | 'updated_at' >; -export const SCHEDULED_PAIRED_FOLLOW_UP_TTL_MS = 10 * 60 * 1000; -const scheduledPairedFollowUps = new Map(); - -function pruneExpiredScheduledPairedFollowUps(now: number): void { - for (const [key, scheduledAt] of scheduledPairedFollowUps) { - if (now - scheduledAt > SCHEDULED_PAIRED_FOLLOW_UP_TTL_MS) { - scheduledPairedFollowUps.delete(key); - } - } -} - export function buildPairedFollowUpKey(args: { taskId: string; taskStatus: PairedTaskStatus | null; @@ -45,29 +43,45 @@ export function schedulePairedFollowUpOnce(args: { intentKind: ScheduledPairedFollowUpIntentKind; enqueue: () => void; }): boolean { - const now = Date.now(); - pruneExpiredScheduledPairedFollowUps(now); + const reserved = reservePairedTurnReservation({ + chatJid: args.chatJid, + taskId: args.task.id, + taskStatus: args.task.status, + roundTripCount: args.task.round_trip_count, + taskUpdatedAt: args.task.updated_at, + intentKind: args.intentKind, + runId: args.runId, + }); - const key = [ - args.chatJid, - buildPairedFollowUpKey({ - taskId: args.task.id, - taskStatus: args.task.status, - roundTripCount: args.task.round_trip_count, - taskUpdatedAt: args.task.updated_at, - intentKind: args.intentKind, - }), - ].join(':'); - - if (scheduledPairedFollowUps.has(key)) { + if (!reserved) { return false; } - scheduledPairedFollowUps.set(key, now); args.enqueue(); return true; } -export function resetPairedFollowUpScheduleState(): void { - scheduledPairedFollowUps.clear(); +export function claimPairedTurnExecution(args: { + chatJid: string; + runId: string; + task: ScheduledPairedFollowUpTask; + intentKind: PairedTurnReservationIntentKind; +}): string | null { + const nextTaskUpdatedAt = new Date().toISOString(); + const claimed = 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 { + _clearPairedTurnReservationsForTests(); } diff --git a/src/paired-workspace-manager.ts b/src/paired-workspace-manager.ts index 7431525..1ee09a1 100644 --- a/src/paired-workspace-manager.ts +++ b/src/paired-workspace-manager.ts @@ -801,6 +801,7 @@ export function markPairedTaskReviewReady(taskId: string): { taskId, currentStatus: task.status, nextStatus: 'review_ready', + expectedUpdatedAt: task.updated_at, updatedAt: requestedAt, patch: { review_requested_at: requestedAt, diff --git a/src/types.ts b/src/types.ts index 248e230..1ae44fc 100644 --- a/src/types.ts +++ b/src/types.ts @@ -35,6 +35,13 @@ export type PairedTaskStatus = | 'arbiter_requested' | 'in_arbitration'; +export type PairedTurnReservationIntentKind = + | 'owner-turn' + | 'reviewer-turn' + | 'arbiter-turn' + | 'owner-follow-up' + | 'finalize-owner-turn'; + export type ArbiterVerdict = 'proceed' | 'revise' | 'reset' | 'escalate'; export type PairedWorkspaceRole = 'owner' | 'reviewer';