From f9d9c2af6d485724372a26cfb9eb04ac6210f844 Mon Sep 17 00:00:00 2001 From: ejclaw Date: Tue, 7 Apr 2026 23:45:53 +0900 Subject: [PATCH] Refactor paired runtime orchestration --- src/message-agent-executor-paired.ts | 244 ++++ src/message-agent-executor-rules.ts | 27 +- src/message-agent-executor.test.ts | 102 ++ src/message-agent-executor.ts | 212 +--- src/message-runtime-delivery.ts | 199 ++++ src/message-runtime-dispatch.ts | 351 ++++++ src/message-runtime-flow.test.ts | 120 ++ src/message-runtime-flow.ts | 49 +- src/message-runtime-gating.ts | 49 + src/message-runtime-handoffs.ts | 173 +++ src/message-runtime-loop.ts | 201 ++++ src/message-runtime-queue.ts | 244 ++++ src/message-runtime-rules.test.ts | 67 ++ src/message-runtime-rules.ts | 61 + src/message-runtime-shared.ts | 40 + src/message-runtime.test.ts | 112 ++ src/message-runtime.ts | 1109 ++++--------------- src/paired-execution-context-owner.ts | 113 +- src/paired-execution-context-reviewer.ts | 83 +- src/paired-execution-context-shared.test.ts | 113 ++ src/paired-execution-context-shared.ts | 103 +- src/paired-follow-up-scheduler.test.ts | 119 ++ src/paired-follow-up-scheduler.ts | 71 ++ 23 files changed, 2770 insertions(+), 1192 deletions(-) create mode 100644 src/message-agent-executor-paired.ts create mode 100644 src/message-runtime-delivery.ts create mode 100644 src/message-runtime-dispatch.ts create mode 100644 src/message-runtime-flow.test.ts create mode 100644 src/message-runtime-gating.ts create mode 100644 src/message-runtime-handoffs.ts create mode 100644 src/message-runtime-loop.ts create mode 100644 src/message-runtime-queue.ts create mode 100644 src/message-runtime-shared.ts create mode 100644 src/paired-execution-context-shared.test.ts create mode 100644 src/paired-follow-up-scheduler.test.ts create mode 100644 src/paired-follow-up-scheduler.ts diff --git a/src/message-agent-executor-paired.ts b/src/message-agent-executor-paired.ts new file mode 100644 index 0000000..d2c30f0 --- /dev/null +++ b/src/message-agent-executor-paired.ts @@ -0,0 +1,244 @@ +import type { AgentOutput } from './agent-runner.js'; +import { + getLastHumanMessageSender, + getLatestTurnNumber, + getPairedTaskById, + insertPairedTurnOutput, +} 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 type { PairedRoomRole } from './types.js'; + +type ExecutorLog = Pick; + +export interface PairedExecutionLifecycle { + updateSummary(args: { + outputText?: string | null; + errorText?: string | null; + }): void; + recordFinalOutputBeforeDelivery(outputText: string): void; + completeImmediately(args: { status: 'succeeded' | 'failed' }): void; + markStatus(status: 'succeeded' | 'failed'): void; + markSawOutput(sawOutput: boolean): void; + getSummary(): string | null; + asyncFinalize(): Promise; +} + +export function createPairedExecutionLifecycle(args: { + pairedExecutionContext?: PreparedPairedExecutionContext; + completedRole: PairedRoomRole; + chatJid: string; + runId: string; + enqueueMessageCheck: () => void; + onOutput?: (output: AgentOutput) => Promise; + log: ExecutorLog; +}): PairedExecutionLifecycle { + const { + pairedExecutionContext, + completedRole, + chatJid, + runId, + enqueueMessageCheck, + onOutput, + log, + } = args; + + let pairedExecutionStatus: 'succeeded' | 'failed' = 'failed'; + let pairedExecutionSummary: string | null = null; + let pairedFinalOutput: string | null = null; + let pairedExecutionCompleted = false; + let pairedSawOutput = false; + let pairedTurnOutputPersisted = false; + + const persistPairedTurnOutputIfNeeded = () => { + if ( + !pairedExecutionContext || + pairedTurnOutputPersisted || + !pairedFinalOutput || + pairedFinalOutput.length === 0 + ) { + return; + } + + const turnNumber = getLatestTurnNumber(pairedExecutionContext.task.id) + 1; + insertPairedTurnOutput( + pairedExecutionContext.task.id, + turnNumber, + completedRole, + pairedFinalOutput, + ); + pairedTurnOutputPersisted = true; + }; + + const completeSuccessfulOwnerTurnBeforeDeliveryIfNeeded = () => { + if ( + completedRole !== 'owner' || + !pairedExecutionContext || + pairedExecutionCompleted || + !pairedFinalOutput || + pairedFinalOutput.length === 0 + ) { + return; + } + + pairedExecutionStatus = 'succeeded'; + pairedSawOutput = true; + persistPairedTurnOutputIfNeeded(); + completePairedExecutionContext({ + taskId: pairedExecutionContext.task.id, + role: completedRole, + status: 'succeeded', + summary: pairedExecutionSummary, + }); + pairedExecutionCompleted = true; + }; + + return { + updateSummary({ outputText, errorText }) { + if (outputText && outputText.length > 0) { + pairedExecutionSummary = outputText.slice(0, 500); + return; + } + + if (errorText && errorText.length > 0) { + pairedExecutionSummary = errorText.slice(0, 500); + } + }, + + recordFinalOutputBeforeDelivery(outputText) { + pairedFinalOutput = outputText; + completeSuccessfulOwnerTurnBeforeDeliveryIfNeeded(); + persistPairedTurnOutputIfNeeded(); + }, + + completeImmediately({ status }) { + if (!pairedExecutionContext || pairedExecutionCompleted) { + return; + } + + pairedExecutionStatus = status; + if (status === 'succeeded') { + persistPairedTurnOutputIfNeeded(); + } + + completePairedExecutionContext({ + taskId: pairedExecutionContext.task.id, + role: completedRole, + status, + summary: pairedExecutionSummary, + }); + pairedExecutionCompleted = true; + }, + + markStatus(status) { + pairedExecutionStatus = status; + }, + + markSawOutput(sawOutput) { + pairedSawOutput = sawOutput; + }, + + getSummary() { + return pairedExecutionSummary; + }, + + async asyncFinalize() { + if (pairedExecutionContext && !pairedExecutionCompleted) { + const effectiveStatus = + completedRole === 'owner' && + pairedExecutionStatus === 'succeeded' && + !pairedSawOutput + ? 'failed' + : pairedExecutionStatus; + + if (effectiveStatus === 'succeeded') { + try { + persistPairedTurnOutputIfNeeded(); + } catch (err) { + log.warn( + { pairedTaskId: pairedExecutionContext.task.id, err }, + 'Failed to store paired turn output', + ); + } + } + + completePairedExecutionContext({ + taskId: pairedExecutionContext.task.id, + role: completedRole, + status: effectiveStatus, + summary: pairedExecutionSummary, + }); + pairedExecutionCompleted = true; + } + + if (!pairedExecutionContext) { + return; + } + + const finishedTask = getPairedTaskById(pairedExecutionContext.task.id); + if (finishedTask?.status === 'completed' && finishedTask.completion_reason) { + const sender = getLastHumanMessageSender(chatJid); + const mention = sender ? `<@${sender}>` : ''; + const notifications: Record = { + done: `${mention} ✅ 작업 완료.`, + escalated: `${mention} ⚠️ 자동 해결 불가 — 확인이 필요합니다.`, + arbiter_escalated: `${mention} ⚠️ 중재자 판단: 사람 개입이 필요합니다.`, + }; + const message = notifications[finishedTask.completion_reason]; + if (message) { + await onOutput?.({ + status: 'success', + result: message, + output: { visibility: 'public', text: message }, + phase: 'final', + }); + } + } + + const queueAction = resolvePairedFollowUpQueueAction({ + completedRole, + executionStatus: pairedExecutionStatus, + sawOutput: pairedSawOutput, + taskStatus: finishedTask?.status ?? null, + }); + if (queueAction !== 'pending' || !finishedTask) { + return; + } + + const nextTurnAction = resolveNextTurnAction({ + taskStatus: finishedTask.status, + lastTurnOutputRole: null, + }); + if (nextTurnAction.kind === 'none') { + return; + } + + const scheduled = schedulePairedFollowUpOnce({ + chatJid, + runId, + task: finishedTask, + intentKind: nextTurnAction.kind, + enqueue: enqueueMessageCheck, + }); + log.info( + { + taskId: pairedExecutionContext.task.id, + role: completedRole, + pairedExecutionStatus, + taskStatus: finishedTask.status, + intentKind: nextTurnAction.kind, + scheduled, + }, + 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 in the same run', + ); + }, + }; +} diff --git a/src/message-agent-executor-rules.ts b/src/message-agent-executor-rules.ts index 9e31971..2f2b96a 100644 --- a/src/message-agent-executor-rules.ts +++ b/src/message-agent-executor-rules.ts @@ -1,4 +1,7 @@ -import { resolveNextTurnAction } from './message-runtime-rules.js'; +import { + resolveFollowUpDispatch, + resolveNextTurnAction, +} from './message-runtime-rules.js'; import type { PairedRoomRole, PairedTaskStatus } from './types.js'; export { isRetryableClaudeSessionFailureAttempt, @@ -25,15 +28,15 @@ export function resolvePairedFollowUpQueueAction(args: { ? args.completedRole : null, }); - - if (args.executionStatus === 'succeeded' && args.sawOutput) { - return 'none'; - } - - const shouldRequeuePendingPairedTurn = - (args.completedRole === 'reviewer' || args.completedRole === 'arbiter') && - (nextTurnAction.kind === 'reviewer-turn' || - nextTurnAction.kind === 'arbiter-turn' || - nextTurnAction.kind === 'finalize-owner-turn'); - return shouldRequeuePendingPairedTurn ? 'pending' : 'none'; + const dispatch = resolveFollowUpDispatch({ + source: 'executor-recovery', + nextTurnAction, + completedRole: args.completedRole, + executionStatus: args.executionStatus, + sawOutput: args.sawOutput, + }); + return dispatch.kind === 'enqueue' && + dispatch.queueKind === 'paired-follow-up' + ? 'pending' + : 'none'; } diff --git a/src/message-agent-executor.test.ts b/src/message-agent-executor.test.ts index e2c3921..8b5fab3 100644 --- a/src/message-agent-executor.test.ts +++ b/src/message-agent-executor.test.ts @@ -200,6 +200,7 @@ import * as db from './db.js'; import { logger } from './logger.js'; import { buildRoomMemoryBriefing } from './sqlite-memory-store.js'; import { runAgentForGroup } from './message-agent-executor.js'; +import { resetPairedFollowUpScheduleState } from './paired-follow-up-scheduler.js'; import * as pairedExecutionContext from './paired-execution-context.js'; import * as sessionRecovery from './session-recovery.js'; import * as serviceRouting from './service-routing.js'; @@ -238,6 +239,7 @@ const ORIGINAL_UNSAFE_HOST_PAIRED_MODE = describe('runAgentForGroup room memory', () => { beforeEach(() => { vi.resetAllMocks(); + resetPairedFollowUpScheduleState(); delete process.env.EJCLAW_UNSAFE_HOST_PAIRED_MODE; vi.mocked(agentRunner.runAgentProcess).mockImplementation( async (_group, _input, _onProcess, onOutput) => { @@ -984,6 +986,9 @@ describe('runAgentForGroup room memory', () => { summary: 'Review snapshot is stale after owner changes. Retry the review once to refresh against the latest owner workspace.', }); + expect( + pairedExecutionContext.completePairedExecutionContext, + ).toHaveBeenCalledTimes(1); }); it('uses the role plan for reviewer execution while keeping task snapshots on the owner agent', async () => { @@ -1769,11 +1774,107 @@ describe('runAgentForGroup room memory', () => { expect(result).toBe('error'); expect(deps.queue.enqueueMessageCheck).toHaveBeenCalledWith('group@test'); }); + + it('does not re-enqueue the same review_ready follow-up twice in one run', async () => { + const group = { ...makeGroup(), folder: 'test-group' }; + const deps = makeDeps(); + + vi.mocked(serviceRouting.getEffectiveChannelLease).mockReturnValue({ + chat_jid: 'group@test', + owner_agent_type: 'claude-code', + reviewer_agent_type: 'claude-code', + arbiter_agent_type: null, + owner_service_id: 'claude', + reviewer_service_id: 'claude', + arbiter_service_id: null, + activated_at: null, + reason: null, + explicit: false, + }); + vi.mocked( + pairedExecutionContext.preparePairedExecutionContext, + ).mockReturnValue({ + task: { + id: 'paired-task-review-ready-dedup', + chat_jid: 'group@test', + group_folder: 'test-group', + owner_service_id: 'claude', + reviewer_service_id: 'claude', + title: null, + source_ref: 'HEAD', + plan_notes: null, + round_trip_count: 1, + review_requested_at: '2026-03-31T00:00:00.000Z', + status: 'in_review', + arbiter_verdict: null, + arbiter_requested_at: null, + completion_reason: null, + created_at: '2026-03-31T00:00:00.000Z', + updated_at: '2026-03-31T00:00:00.000Z', + }, + workspace: null, + envOverrides: {}, + }); + vi.mocked(db.getPairedTaskById).mockReturnValue({ + id: 'paired-task-review-ready-dedup', + chat_jid: 'group@test', + group_folder: 'test-group', + owner_service_id: 'claude', + reviewer_service_id: 'claude', + title: null, + source_ref: 'HEAD', + plan_notes: null, + round_trip_count: 1, + review_requested_at: '2026-03-31T00:00:00.000Z', + status: 'review_ready', + arbiter_verdict: null, + arbiter_requested_at: null, + completion_reason: null, + created_at: '2026-03-31T00:00:00.000Z', + updated_at: '2026-03-31T00:00:00.000Z', + }); + vi.mocked(agentRunner.runAgentProcess).mockResolvedValue({ + status: 'error', + result: null, + error: 'SDK crashed with exit code 1', + }); + + await runAgentForGroup(deps, { + group, + prompt: 'please review', + chatJid: 'group@test', + runId: 'run-review-ready-requeue-dedup', + forcedRole: 'reviewer', + onOutput: async () => {}, + }); + await runAgentForGroup(deps, { + group, + prompt: 'please review', + chatJid: 'group@test', + runId: 'run-review-ready-requeue-dedup', + forcedRole: 'reviewer', + onOutput: async () => {}, + }); + + expect(deps.queue.enqueueMessageCheck).toHaveBeenCalledTimes(1); + expect(logger.info).toHaveBeenCalledWith( + expect.objectContaining({ + taskId: 'paired-task-review-ready-dedup', + role: 'reviewer', + pairedExecutionStatus: 'failed', + taskStatus: 'review_ready', + intentKind: 'reviewer-turn', + scheduled: false, + }), + 'Skipped duplicate paired follow-up after failed reviewer/arbiter execution in the same run', + ); + }); }); describe('runAgentForGroup Claude rotation', () => { beforeEach(() => { vi.resetAllMocks(); + resetPairedFollowUpScheduleState(); vi.mocked(buildRoomMemoryBriefing).mockResolvedValue(undefined); vi.mocked(tokenRotation.getTokenCount).mockReturnValue(1); vi.mocked(tokenRotation.getCurrentTokenIndex).mockReturnValue(0); @@ -2585,6 +2686,7 @@ describe('runAgentForGroup Claude rotation', () => { describe('runAgentForGroup Codex rotation', () => { beforeEach(() => { vi.resetAllMocks(); + resetPairedFollowUpScheduleState(); vi.mocked(buildRoomMemoryBriefing).mockResolvedValue(undefined); vi.mocked(codexTokenRotation.getCodexAccountCount).mockReturnValue(2); vi.mocked(codexTokenRotation.rotateCodexToken).mockReturnValueOnce(true); diff --git a/src/message-agent-executor.ts b/src/message-agent-executor.ts index 8479cd5..8c9e1e0 100644 --- a/src/message-agent-executor.ts +++ b/src/message-agent-executor.ts @@ -21,22 +21,19 @@ import { listAvailableGroups } from './available-groups.js'; import { createServiceHandoff, getAllTasks, - getLastHumanMessageSender, getLatestOpenPairedTaskForChat, - getLatestTurnNumber, - getPairedTaskById, - insertPairedTurnOutput, } from './db.js'; import { GroupQueue } from './group-queue.js'; import { createScopedLogger } from './logger.js'; import { buildRoomMemoryBriefing } from './sqlite-memory-store.js'; import { - completePairedExecutionContext, preparePairedExecutionContext, } from './paired-execution-context.js'; import { resolveCodexFallbackHandoff } from './paired-turn-fallback.js'; -import { resolveExecutionTarget } from './message-runtime-rules.js'; -import { resolvePairedFollowUpQueueAction } from './message-agent-executor-rules.js'; +import { createPairedExecutionLifecycle } from './message-agent-executor-paired.js'; +import { + resolveExecutionTarget, +} from './message-runtime-rules.js'; import { buildRoomRoleContext } from './room-role-context.js'; import { type AgentTriggerReason } from './agent-error-detection.js'; import { @@ -311,55 +308,15 @@ export async function runAgentForGroup( } const effectivePrompt = moaEnrichedPrompt; - let pairedExecutionStatus: 'succeeded' | 'failed' = 'failed'; - let pairedExecutionSummary: string | null = null; - let pairedFinalOutput: string | null = null; - let pairedExecutionCompleted = false; - let pairedSawOutput = false; - let pairedTurnOutputPersisted = false; - - const persistPairedTurnOutputIfNeeded = (completedRole: PairedRoomRole) => { - if ( - !pairedExecutionContext || - pairedTurnOutputPersisted || - !pairedFinalOutput || - pairedFinalOutput.length === 0 - ) { - return; - } - const turnNumber = getLatestTurnNumber(pairedExecutionContext.task.id) + 1; - insertPairedTurnOutput( - pairedExecutionContext.task.id, - turnNumber, - completedRole, - pairedFinalOutput, - ); - pairedTurnOutputPersisted = true; - }; - - const completeSuccessfulOwnerTurnBeforeDeliveryIfNeeded = () => { - const completedRole = roomRoleContext?.role ?? 'owner'; - if ( - completedRole !== 'owner' || - !pairedExecutionContext || - pairedExecutionCompleted || - !pairedFinalOutput || - pairedFinalOutput.length === 0 - ) { - return; - } - - pairedExecutionStatus = 'succeeded'; - pairedSawOutput = true; - persistPairedTurnOutputIfNeeded(completedRole); - completePairedExecutionContext({ - taskId: pairedExecutionContext.task.id, - role: completedRole, - status: 'succeeded', - summary: pairedExecutionSummary, - }); - pairedExecutionCompleted = true; - }; + const pairedExecutionLifecycle = createPairedExecutionLifecycle({ + pairedExecutionContext, + completedRole: roomRoleContext?.role ?? 'owner', + chatJid, + runId, + enqueueMessageCheck: () => deps.queue.enqueueMessageCheck(chatJid), + onOutput, + log, + }); const maybeHandoffToCodex = ( reason: AgentTriggerReason, @@ -415,7 +372,9 @@ export async function runAgentForGroup( }; if (pairedExecutionContext?.blockMessage) { - pairedExecutionSummary = pairedExecutionContext.blockMessage.slice(0, 500); + pairedExecutionLifecycle.updateSummary({ + outputText: pairedExecutionContext.blockMessage, + }); log.warn( { roomRoleServiceId: roomRoleContext?.serviceId, @@ -432,13 +391,7 @@ export async function runAgentForGroup( }, phase: 'final', }); - completePairedExecutionContext({ - taskId: pairedExecutionContext.task.id, - role: roomRoleContext?.role ?? 'owner', - status: pairedExecutionStatus, - summary: pairedExecutionSummary, - }); - pairedExecutionCompleted = true; + pairedExecutionLifecycle.completeImmediately({ status: 'failed' }); return 'success'; } @@ -523,14 +476,10 @@ export async function runAgentForGroup( currentSessionId = output.newSessionId; } - if (outputText && outputText.length > 0) { - pairedExecutionSummary = outputText.slice(0, 500); - } else if ( - typeof output.error === 'string' && - output.error.length > 0 - ) { - pairedExecutionSummary = output.error.slice(0, 500); - } + pairedExecutionLifecycle.updateSummary({ + outputText, + errorText: typeof output.error === 'string' ? output.error : null, + }); if ( evaluation.newTrigger && outputText && @@ -589,10 +538,8 @@ export async function runAgentForGroup( outputText && outputText.length > 0 ) { - pairedFinalOutput = outputText; try { - completeSuccessfulOwnerTurnBeforeDeliveryIfNeeded(); - persistPairedTurnOutputIfNeeded(roomRoleContext?.role ?? 'owner'); + pairedExecutionLifecycle.recordFinalOutputBeforeDelivery(outputText); } catch (err) { log.warn( { pairedTaskId: pairedExecutionContext?.task.id ?? null, err }, @@ -701,13 +648,13 @@ export async function runAgentForGroup( runId, }; - return runClaudeAttemptWithRotation({ + return runClaudeAttemptWithRotation({ initialTrigger, runAttempt: () => runAttempt('claude'), logContext: logCtx, rotationMessage, onSuccess: ({ sawOutput }) => { - pairedSawOutput = sawOutput; + pairedExecutionLifecycle.markSawOutput(sawOutput); }, }); }; @@ -733,7 +680,7 @@ export async function runAgentForGroup( return maybeHandoffAfterError(retryAction.trigger.reason, attempt); } - pairedExecutionStatus = 'succeeded'; + pairedExecutionLifecycle.markStatus('succeeded'); return retryAction.result; }; @@ -755,7 +702,7 @@ export async function runAgentForGroup( } if (retryAction.result === 'success') { - pairedExecutionStatus = 'succeeded'; + pairedExecutionLifecycle.markStatus('succeeded'); } return retryAction.result; }; @@ -770,9 +717,6 @@ export async function runAgentForGroup( return 'error'; }; - const getPairedExecutionStatus = (): 'succeeded' | 'failed' => - pairedExecutionStatus; - const isRetryableClaudeSessionFailure = (attempt: AgentAttempt): boolean => isRetryableClaudeSessionFailureAttempt({ attempt, @@ -863,15 +807,18 @@ export async function runAgentForGroup( return 'error'; } - if (!pairedExecutionSummary) { + if (!pairedExecutionLifecycle.getSummary()) { const finalOutputText = getAgentOutputText(output); - pairedExecutionSummary = + pairedExecutionLifecycle.updateSummary({ + outputText: (typeof finalOutputText === 'string' && finalOutputText.length > 0 - ? finalOutputText.slice(0, 500) - : null) ?? - (typeof output.error === 'string' && output.error.length > 0 - ? output.error.slice(0, 500) - : null); + ? finalOutputText + : null), + errorText: + typeof output.error === 'string' && output.error.length > 0 + ? output.error + : null, + }); } if (!attempt.sawOutput && output.status !== 'error') { @@ -937,8 +884,8 @@ export async function runAgentForGroup( return 'error'; } - pairedExecutionStatus = 'succeeded'; - pairedSawOutput = attempt.sawOutput; + pairedExecutionLifecycle.markStatus('succeeded'); + pairedExecutionLifecycle.markSawOutput(attempt.sawOutput); return 'success'; }; @@ -960,87 +907,6 @@ export async function runAgentForGroup( return await finalizePrimaryAttempt(primaryAttempt); } finally { - if (pairedExecutionContext && !pairedExecutionCompleted) { - const completedRole = roomRoleContext?.role ?? 'owner'; - const completionStatus = getPairedExecutionStatus(); - // Owner was interrupted without producing output (e.g. /stop) — - // treat as failed so reviewer is not auto-triggered. - const effectiveStatus = - completedRole === 'owner' && - completionStatus === 'succeeded' && - !pairedSawOutput - ? 'failed' - : completionStatus; - if (effectiveStatus === 'succeeded') { - try { - persistPairedTurnOutputIfNeeded(completedRole); - } catch (err) { - log.warn( - { pairedTaskId: pairedExecutionContext.task.id, err }, - 'Failed to store paired turn output', - ); - } - } - - completePairedExecutionContext({ - taskId: pairedExecutionContext.task.id, - role: completedRole, - status: effectiveStatus, - summary: pairedExecutionSummary, - }); - } - - // Notify user when paired task reaches a terminal state that requires attention. - if (pairedExecutionContext) { - const finishedTask = getPairedTaskById(pairedExecutionContext.task.id); - if ( - finishedTask?.status === 'completed' && - finishedTask.completion_reason - ) { - const sender = getLastHumanMessageSender(chatJid); - const mention = sender ? `<@${sender}>` : ''; - const notifications: Record = { - done: `${mention} ✅ 작업 완료.`, - escalated: `${mention} ⚠️ 자동 해결 불가 — 확인이 필요합니다.`, - arbiter_escalated: `${mention} ⚠️ 중재자 판단: 사람 개입이 필요합니다.`, - }; - const message = notifications[finishedTask.completion_reason]; - if (message) { - await args.onOutput?.({ - status: 'success', - result: message, - output: { visibility: 'public', text: message }, - phase: 'final', - }); - } - } - } - - // Failed reviewer/arbiter runs can leave a pending paired task without a - // visible bot message. Requeue only those recovery cases here. Successful - // owner -> reviewer handoffs are queued by message-runtime after final - // delivery succeeds, which avoids waking reviewer off a failed send. - if (pairedExecutionContext) { - const completedRole = roomRoleContext?.role ?? 'owner'; - const finishedCheck = getPairedTaskById(pairedExecutionContext.task.id); - const queueAction = resolvePairedFollowUpQueueAction({ - completedRole, - executionStatus: pairedExecutionStatus, - sawOutput: pairedSawOutput, - taskStatus: finishedCheck?.status ?? null, - }); - if (queueAction === 'pending') { - deps.queue.enqueueMessageCheck(chatJid); - log.info( - { - taskId: pairedExecutionContext.task.id, - role: completedRole, - pairedExecutionStatus, - taskStatus: finishedCheck?.status ?? null, - }, - 'Queued paired follow-up after failed reviewer/arbiter execution left a pending task state', - ); - } - } + await pairedExecutionLifecycle.asyncFinalize(); } } diff --git a/src/message-runtime-delivery.ts b/src/message-runtime-delivery.ts new file mode 100644 index 0000000..81533eb --- /dev/null +++ b/src/message-runtime-delivery.ts @@ -0,0 +1,199 @@ +import { + markWorkItemDelivered, + markWorkItemDeliveryRetry, + type WorkItem, +} from './db.js'; +import { logger } from './logger.js'; +import { resolveActiveRole } from './message-runtime-rules.js'; +import { getErrorMessage } from './utils.js'; +import type { Channel, PairedTask, PairedRoomRole } from './types.js'; + +type RuntimeDeliveryLog = Pick; + +function buildDeliveryLogContext( + channel: Channel, + item: WorkItem, + extra: Record = {}, +): Record { + return { + chatJid: item.chat_jid, + channelName: channel.name, + workItemId: item.id, + deliveryRole: item.delivery_role ?? null, + ...extra, + }; +} + +export async function deliverOpenWorkItem(args: { + channel: Channel; + item: WorkItem; + log: RuntimeDeliveryLog; + replaceMessageId?: string | null; + isDuplicateOfLastBotFinal: (chatJid: string, text: string) => boolean; + openContinuation: (chatJid: string) => void; +}): Promise { + const replaceMessageId = args.replaceMessageId ?? null; + + const isDuplicate = args.isDuplicateOfLastBotFinal( + args.item.chat_jid, + args.item.result_payload, + ); + + if (isDuplicate) { + markWorkItemDelivered(args.item.id, null); + args.log.info( + buildDeliveryLogContext(args.channel, args.item, { + preview: args.item.result_payload.slice(0, 100), + suppressionReason: 'paired-final-duplicate', + }), + 'Suppressed duplicate final message in paired room (marked as delivered)', + ); + return true; + } + + try { + if (replaceMessageId && args.channel.editMessage) { + args.log.info( + buildDeliveryLogContext(args.channel, args.item, { + deliveryAttempts: args.item.delivery_attempts + 1, + deliveryMode: 'edit', + replacedMessageId: replaceMessageId, + }), + 'Attempting to deliver produced work item by replacing tracked progress message', + ); + await args.channel.editMessage( + args.item.chat_jid, + replaceMessageId, + args.item.result_payload, + ); + markWorkItemDelivered(args.item.id, replaceMessageId); + args.openContinuation(args.item.chat_jid); + args.log.info( + buildDeliveryLogContext(args.channel, args.item, { + deliveryAttempts: args.item.delivery_attempts + 1, + deliveryMode: 'edit', + replacedMessageId: replaceMessageId, + }), + 'Delivered produced work item by replacing tracked progress message', + ); + return true; + } + } catch (err) { + args.log.warn( + buildDeliveryLogContext(args.channel, args.item, { + deliveryAttempts: args.item.delivery_attempts + 1, + deliveryMode: 'edit', + replacedMessageId: replaceMessageId, + err, + }), + 'Failed to replace tracked progress message; falling back to a new message', + ); + } + + try { + args.log.info( + buildDeliveryLogContext(args.channel, args.item, { + deliveryAttempts: args.item.delivery_attempts + 1, + deliveryMode: 'send', + }), + 'Attempting to deliver produced work item as a new message', + ); + await args.channel.sendMessage(args.item.chat_jid, args.item.result_payload); + markWorkItemDelivered(args.item.id); + args.openContinuation(args.item.chat_jid); + args.log.info( + buildDeliveryLogContext(args.channel, args.item, { + deliveryAttempts: args.item.delivery_attempts + 1, + deliveryMode: 'send', + }), + 'Delivered produced work item', + ); + return true; + } catch (err) { + const errorMessage = getErrorMessage(err); + markWorkItemDeliveryRetry(args.item.id, errorMessage); + args.log.warn( + buildDeliveryLogContext(args.channel, args.item, { + deliveryAttempts: args.item.delivery_attempts + 1, + deliveryMode: 'send', + err, + }), + 'Failed to deliver produced work item', + ); + return false; + } +} + +export async function processOpenWorkItemDelivery(args: { + chatJid: string; + runId: string; + openWorkItem: WorkItem | null | undefined; + pendingTask: PairedTask | null | undefined; + channel: Channel; + roleToChannel: Record<'owner' | 'reviewer' | 'arbiter', Channel | null>; + log: RuntimeDeliveryLog; + isPairedRoom: boolean; + getMissingRoleChannelMessage: (role: 'reviewer' | 'arbiter') => string; + isDuplicateOfLastBotFinal: (chatJid: string, text: string) => boolean; + openContinuation: (chatJid: string) => void; + enqueueFollowUpAfterDeliveryRetry: (args: { + deliveryRole: PairedRoomRole; + pendingTask: PairedTask | null | undefined; + workItemId: string | number; + }) => void; +}): Promise<'not-found' | 'delivered' | 'failed'> { + const { openWorkItem } = args; + if (!openWorkItem) { + return 'not-found'; + } + + if (args.isPairedRoom && openWorkItem.delivery_role == null) { + args.log.warn( + { + workItemId: openWorkItem.id, + chatJid: args.chatJid, + pendingTaskStatus: args.pendingTask?.status ?? null, + }, + 'Paired-room delivery retry is missing a persisted delivery role; falling back to inferred routing', + ); + } + + const deliveryRole = + openWorkItem.delivery_role ?? + (args.pendingTask ? resolveActiveRole(args.pendingTask.status) : 'owner'); + const deliveryChannel = + deliveryRole === 'owner' ? args.channel : args.roleToChannel[deliveryRole]; + if (!deliveryChannel) { + const missingRole = deliveryRole === 'arbiter' ? 'arbiter' : 'reviewer'; + const errorMessage = args.getMissingRoleChannelMessage(missingRole); + markWorkItemDeliveryRetry(openWorkItem.id, errorMessage); + args.log.error( + { + workItemId: openWorkItem.id, + role: deliveryRole, + requiredChannel: + missingRole === 'arbiter' ? 'discord-arbiter' : 'discord-review', + }, + 'Unable to deliver paired-room work item because the dedicated Discord role channel is not configured', + ); + return 'failed'; + } + + const delivered = await deliverOpenWorkItem({ + channel: deliveryChannel, + item: openWorkItem, + log: args.log, + isDuplicateOfLastBotFinal: args.isDuplicateOfLastBotFinal, + openContinuation: args.openContinuation, + }); + if (!delivered) { + return 'failed'; + } + + args.enqueueFollowUpAfterDeliveryRetry({ + deliveryRole, + pendingTask: args.pendingTask, + workItemId: openWorkItem.id, + }); + return 'delivered'; +} diff --git a/src/message-runtime-dispatch.ts b/src/message-runtime-dispatch.ts new file mode 100644 index 0000000..03693ad --- /dev/null +++ b/src/message-runtime-dispatch.ts @@ -0,0 +1,351 @@ +import { logger } from './logger.js'; +import { + getLatestOpenPairedTaskForChat, + getMessagesSinceSeq, +} from './db.js'; +import { + buildQueuedTurnDispatch, + executeBotOnlyPairedFollowUpAction, +} from './message-runtime-flow.js'; +import { + advanceLastAgentCursor, + filterLoopingPairedBotMessages, + getProcessableMessages, + hasAllowedTrigger, + resolveCursorKey, + resolveFollowUpDispatch, + resolveNextTurnAction, + shouldSkipBotOnlyCollaboration, +} from './message-runtime-rules.js'; +import { extractSessionCommand, isSessionCommandAllowed } from './session-commands.js'; +import { isSessionCommandSenderAllowed } from './config.js'; +import { hasReviewerLease } from './service-routing.js'; +import type { ScheduledPairedFollowUpIntentKind } from './paired-follow-up-scheduler.js'; +import type { + AgentType, + Channel, + NewMessage, + PairedTask, + PairedRoomRole, + RegisteredGroup, +} from './types.js'; + +type RuntimeLog = Pick; + +type ExecuteTurnFn = (args: { + group: RegisteredGroup; + prompt: string; + chatJid: string; + runId: string; + channel: Channel; + startSeq: number | null; + endSeq: number | null; + deliveryRole?: PairedRoomRole; + hasHumanMessage?: boolean; + forcedRole?: PairedRoomRole; + forcedAgentType?: AgentType; +}) => Promise<{ + outputStatus: 'success' | 'error'; + deliverySucceeded: boolean; + visiblePhase: unknown; +}>; + +export function enqueueGenericFollowUpAfterDeliveryRetry(args: { + chatJid: string; + runId: string; + deliveryRole: PairedRoomRole; + pendingTask: PairedTask | null | undefined; + 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({ + source: 'delivery-retry', + nextTurnAction, + completedRole: args.deliveryRole, + }); + + if (dispatch.kind === 'none') { + args.log.info( + { + workItemId: args.workItemId, + chatJid: args.chatJid, + deliveryRole: args.deliveryRole, + pendingTaskStatus: args.pendingTask?.status ?? null, + }, + 'Skipping queued follow-up after reviewer merge_ready delivery because inline finalize will handle the handoff', + ); + return; + } + + if (dispatch.kind === 'enqueue' && dispatch.queueKind === 'message-check') { + args.enqueueMessageCheck(); + 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) { + args.log.info( + { + workItemId: args.workItemId, + chatJid: args.chatJid, + deliveryRole: args.deliveryRole, + taskId: args.pendingTask?.id ?? null, + taskStatus: args.pendingTask?.status ?? null, + intentKind: nextTurnAction.kind, + }, + 'Skipped duplicate paired follow-up enqueue after delivery retry in the same run', + ); + } +} + +export async function processQueuedGroupDispatch(args: { + chatJid: string; + group: RegisteredGroup; + channel: Channel; + processableGroupMessages: NewMessage[]; + assistantName: string; + failureFinalText: string; + timezone: string; + lastAgentTimestamps: Record; + saveState: () => void; + executeTurn: ExecuteTurnFn; + schedulePairedFollowUp: ( + task: PairedTask | null | undefined, + intentKind: ScheduledPairedFollowUpIntentKind, + runId: string, + ) => boolean; + enqueueMessageCheck: () => void; + sendQueuedMessage: (chatJid: string, text: string) => boolean; + closeStdin: (reason: string) => void; + labelPairedSenders: (chatJid: string, messages: NewMessage[]) => NewMessage[]; + formatMessages: (messages: NewMessage[], timezone: string) => string; +}): Promise { + const { chatJid, group, channel, processableGroupMessages } = args; + const loopPendingTask = hasReviewerLease(chatJid) + ? getLatestOpenPairedTaskForChat(chatJid) + : null; + const loopCursorKey = resolveCursorKey(chatJid, loopPendingTask?.status); + const rawPendingMessages = getMessagesSinceSeq( + chatJid, + args.lastAgentTimestamps[loopCursorKey] || '0', + args.assistantName, + ); + const pendingMessages = filterLoopingPairedBotMessages( + chatJid, + getProcessableMessages(chatJid, rawPendingMessages, channel), + args.failureFinalText, + ); + const messagesToSend = + pendingMessages.length > 0 ? pendingMessages : processableGroupMessages; + const labeledMessagesToSend = args.labelPairedSenders(chatJid, messagesToSend); + const { + formatted, + botOnlyFollowUpAction, + isBotOnlyPairedFollowUp, + loopCursorKey: dispatchCursorKey, + endSeq, + } = buildQueuedTurnDispatch({ + chatJid, + timezone: args.timezone, + loopPendingTask, + rawPendingMessages, + messagesToSend, + labeledMessagesToSend, + formatMessages: args.formatMessages, + }); + + const botOnlyFollowUpRunId = `loop-merge-ready-${Date.now().toString(36)}`; + if ( + await executeBotOnlyPairedFollowUpAction({ + action: botOnlyFollowUpAction, + chatJid, + group, + runId: botOnlyFollowUpRunId, + channel, + log: logger, + saveState: args.saveState, + lastAgentTimestamps: args.lastAgentTimestamps, + executeTurn: args.executeTurn, + schedulePairedFollowUp: (task, intentKind) => + args.schedulePairedFollowUp(task, intentKind, botOnlyFollowUpRunId), + closeStdin: () => args.closeStdin('paired-pending-turn-follow-up'), + }) + ) { + return; + } + + if (args.sendQueuedMessage(chatJid, formatted)) { + if (endSeq != null) { + advanceLastAgentCursor( + args.lastAgentTimestamps, + args.saveState, + chatJid, + endSeq, + dispatchCursorKey, + ); + } + logger.debug( + { + transition: 'typing:on', + source: 'follow-up-queued', + chatJid, + group: group.name, + groupFolder: group.folder, + endSeq: endSeq ?? null, + suppressed: isBotOnlyPairedFollowUp, + }, + 'Typing indicator transition', + ); + if (!isBotOnlyPairedFollowUp) { + await channel + .setTyping?.(chatJid, true) + ?.catch((err) => + logger.warn({ chatJid, err }, 'Failed to set typing indicator'), + ); + } + return; + } + + args.enqueueMessageCheck(); +} + +export async function processLoopGroupMessages(args: { + chatJid: string; + group: RegisteredGroup; + groupMessages: NewMessage[]; + channel: Channel; + assistantName: string; + failureFinalText: string; + triggerPattern: RegExp; + hasImplicitContinuationWindow: ( + chatJid: string, + messages: NewMessage[], + ) => boolean; + lastAgentTimestamps: Record; + saveState: () => void; + timezone: string; + executeTurn: ExecuteTurnFn; + schedulePairedFollowUp: ( + task: PairedTask | null | undefined, + intentKind: ScheduledPairedFollowUpIntentKind, + runId: string, + ) => boolean; + enqueueMessageCheck: () => void; + sendQueuedMessage: (chatJid: string, text: string) => boolean; + closeStdin: (reason: string) => void; + labelPairedSenders: (chatJid: string, messages: NewMessage[]) => NewMessage[]; + formatMessages: (messages: NewMessage[], timezone: string) => string; +}): Promise { + const { chatJid, group, groupMessages, channel } = args; + const isMainGroup = group.isMain === true; + const processableGroupMessages = getProcessableMessages( + chatJid, + groupMessages, + channel, + ); + + if (processableGroupMessages.length === 0) { + const lastIgnored = groupMessages[groupMessages.length - 1]; + if (lastIgnored?.seq != null) { + advanceLastAgentCursor( + args.lastAgentTimestamps, + args.saveState, + chatJid, + lastIgnored.seq, + ); + } + return; + } + + if (shouldSkipBotOnlyCollaboration(chatJid, processableGroupMessages)) { + const lastIgnored = + processableGroupMessages[processableGroupMessages.length - 1]; + if (lastIgnored?.seq != null) { + advanceLastAgentCursor( + args.lastAgentTimestamps, + args.saveState, + chatJid, + lastIgnored.seq, + ); + } + logger.info( + { chatJid, group: group.name, groupFolder: group.folder }, + 'Bot-collaboration timeout: no recent human message, skipping', + ); + return; + } + + const loopCmdMsg = groupMessages.find( + (msg) => extractSessionCommand(msg.content, args.triggerPattern) !== null, + ); + + if (loopCmdMsg) { + if ( + isSessionCommandAllowed( + isMainGroup, + loopCmdMsg.is_from_me === true, + isSessionCommandSenderAllowed(loopCmdMsg.sender), + ) + ) { + args.closeStdin('session-command-detected'); + } + args.enqueueMessageCheck(); + return; + } + + if ( + !hasAllowedTrigger({ + chatJid, + messages: processableGroupMessages, + group, + triggerPattern: args.triggerPattern, + hasImplicitContinuationWindow: args.hasImplicitContinuationWindow, + }) + ) { + return; + } + + await processQueuedGroupDispatch({ + chatJid, + group, + channel, + processableGroupMessages, + assistantName: args.assistantName, + failureFinalText: args.failureFinalText, + timezone: args.timezone, + lastAgentTimestamps: args.lastAgentTimestamps, + saveState: args.saveState, + executeTurn: args.executeTurn, + schedulePairedFollowUp: args.schedulePairedFollowUp, + enqueueMessageCheck: args.enqueueMessageCheck, + sendQueuedMessage: args.sendQueuedMessage, + closeStdin: args.closeStdin, + labelPairedSenders: args.labelPairedSenders, + formatMessages: args.formatMessages, + }); +} diff --git a/src/message-runtime-flow.test.ts b/src/message-runtime-flow.test.ts new file mode 100644 index 0000000..2d1b55c --- /dev/null +++ b/src/message-runtime-flow.test.ts @@ -0,0 +1,120 @@ +import { beforeEach, describe, expect, it, vi } from 'vitest'; + +import { executeBotOnlyPairedFollowUpAction } from './message-runtime-flow.js'; +import { + resetPairedFollowUpScheduleState, + schedulePairedFollowUpOnce, + type ScheduledPairedFollowUpIntentKind, +} from './paired-follow-up-scheduler.js'; +import type { PairedTask } from './types.js'; + +describe('executeBotOnlyPairedFollowUpAction', () => { + beforeEach(() => { + resetPairedFollowUpScheduleState(); + }); + + it('deduplicates bot-only requeue follow-ups within the same run', async () => { + const task: PairedTask = { + id: 'task-bot-only-dedup', + 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: '2026-03-30T00:00:00.000Z', + round_trip_count: 1, + 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', + }; + const enqueue = vi.fn(); + const closeStdin = vi.fn(); + const log = { + info: vi.fn(), + } as any; + const schedulePairedFollowUp = ( + scheduledTask: PairedTask, + intentKind: ScheduledPairedFollowUpIntentKind, + ) => + schedulePairedFollowUpOnce({ + chatJid: 'group@test', + runId: 'run-bot-only-dedup', + task: scheduledTask, + intentKind, + enqueue, + }); + + const action = { + kind: 'requeue-pending-turn' as const, + task, + cursor: 42, + cursorKey: 'group@test', + intentKind: 'owner-follow-up' as const, + nextRole: 'owner' as const, + }; + + const first = await executeBotOnlyPairedFollowUpAction({ + action, + chatJid: 'group@test', + group: { + name: 'Test Group', + folder: 'test-group', + trigger: '@Andy', + added_at: '2026-03-30T00:00:00.000Z', + requiresTrigger: false, + agentType: 'codex', + }, + runId: 'run-bot-only-dedup', + channel: {} as any, + log, + saveState: vi.fn(), + lastAgentTimestamps: {}, + executeTurn: vi.fn(), + schedulePairedFollowUp, + closeStdin, + }); + + const second = await executeBotOnlyPairedFollowUpAction({ + action, + chatJid: 'group@test', + group: { + name: 'Test Group', + folder: 'test-group', + trigger: '@Andy', + added_at: '2026-03-30T00:00:00.000Z', + requiresTrigger: false, + agentType: 'codex', + }, + runId: 'run-bot-only-dedup', + channel: {} as any, + log, + saveState: vi.fn(), + lastAgentTimestamps: {}, + executeTurn: vi.fn(), + schedulePairedFollowUp, + closeStdin, + }); + + expect(first).toBe(true); + expect(second).toBe(true); + expect(closeStdin).toHaveBeenCalledTimes(2); + expect(enqueue).toHaveBeenCalledTimes(1); + expect(log.info).toHaveBeenCalledWith( + expect.objectContaining({ + chatJid: 'group@test', + taskId: 'task-bot-only-dedup', + taskStatus: 'active', + handoffMode: 'requeue', + nextRole: 'owner', + intentKind: 'owner-follow-up', + scheduled: false, + }), + 'Skipped duplicate paired pending turn requeue in the same run', + ); + }); +}); diff --git a/src/message-runtime-flow.ts b/src/message-runtime-flow.ts index ab49fab..d766caf 100644 --- a/src/message-runtime-flow.ts +++ b/src/message-runtime-flow.ts @@ -14,8 +14,10 @@ import { import { advanceLastAgentCursor, resolveCursorKey, + resolveFollowUpDispatch, resolveNextTurnAction, } from './message-runtime-rules.js'; +import { type ScheduledPairedFollowUpIntentKind } from './paired-follow-up-scheduler.js'; import { hasReviewerLease } from './service-routing.js'; import type { Channel, @@ -51,6 +53,7 @@ export type BotOnlyPairedFollowUpAction = task: PairedTask; cursor: string | number | null; cursorKey: string; + intentKind: ScheduledPairedFollowUpIntentKind; nextRole: 'owner' | 'reviewer' | 'arbiter'; }; @@ -254,8 +257,12 @@ export function resolveBotOnlyPairedFollowUpAction(args: { taskStatus: task.status, lastTurnOutputRole: lastTurnOutput?.role ?? null, }); + const dispatch = resolveFollowUpDispatch({ + source: 'bot-only-follow-up', + nextTurnAction, + }); - if (nextTurnAction.kind === 'none') { + if (dispatch.kind === 'none') { return { kind: 'consume-stale-bot-message', task, @@ -264,7 +271,7 @@ export function resolveBotOnlyPairedFollowUpAction(args: { }; } - if (nextTurnAction.kind === 'finalize-owner-turn') { + if (dispatch.kind === 'inline') { return { kind: 'inline-finalize', task, @@ -273,15 +280,16 @@ export function resolveBotOnlyPairedFollowUpAction(args: { } if ( - nextTurnAction.kind === 'owner-follow-up' || - nextTurnAction.kind === 'reviewer-turn' || - nextTurnAction.kind === 'arbiter-turn' + dispatch.kind === 'enqueue' && + dispatch.queueKind === 'paired-follow-up' && + nextTurnAction.kind !== 'none' ) { return { kind: 'requeue-pending-turn', task, cursor, cursorKey: resolveCursorKey(chatJid, task.status), + intentKind: nextTurnAction.kind, nextRole: nextTurnAction.kind === 'owner-follow-up' ? 'owner' @@ -312,7 +320,10 @@ export async function executeBotOnlyPairedFollowUpAction(args: { startSeq: number | null; endSeq: number | null; }) => Promise<{ deliverySucceeded: boolean }>; - enqueueGroupMessageCheck: () => void; + schedulePairedFollowUp: ( + task: PairedTask, + intentKind: ScheduledPairedFollowUpIntentKind, + ) => boolean; closeStdin: () => void; }): Promise { const { @@ -325,7 +336,7 @@ export async function executeBotOnlyPairedFollowUpAction(args: { saveState, lastAgentTimestamps, executeTurn, - enqueueGroupMessageCheck, + schedulePairedFollowUp, closeStdin, } = args; @@ -384,13 +395,13 @@ export async function executeBotOnlyPairedFollowUpAction(args: { endSeq: null, }); if (!deliverySucceeded) { - enqueueGroupMessageCheck(); + schedulePairedFollowUp(action.task, 'finalize-owner-turn'); } return true; } closeStdin(); - enqueueGroupMessageCheck(); + const scheduled = schedulePairedFollowUp(action.task, action.intentKind); log.info( { chatJid, @@ -400,10 +411,14 @@ export async function executeBotOnlyPairedFollowUpAction(args: { taskStatus: action.task.status, handoffMode: 'requeue', nextRole: action.nextRole, + intentKind: action.intentKind, cursor: action.cursor, cursorKey: action.cursorKey, + scheduled, }, - 'Queued fresh paired pending turn instead of piping bot-only follow-up into the active agent', + scheduled + ? 'Queued fresh paired pending turn instead of piping bot-only follow-up into the active agent' + : 'Skipped duplicate paired pending turn requeue in the same run', ); return true; } @@ -454,17 +469,3 @@ export function buildQueuedTurnDispatch(args: { endSeq: args.messagesToSend[args.messagesToSend.length - 1]?.seq ?? null, }; } - -export function shouldSkipGenericFollowUpAfterDeliveryRetry(args: { - chatJid: string; - deliveryRole: PairedRoomRole; - pendingTask: PairedTask | null | undefined; -}): boolean { - return ( - hasReviewerLease(args.chatJid) && - args.deliveryRole === 'reviewer' && - resolveNextTurnAction({ - taskStatus: args.pendingTask?.status ?? null, - }).kind === 'finalize-owner-turn' - ); -} diff --git a/src/message-runtime-gating.ts b/src/message-runtime-gating.ts new file mode 100644 index 0000000..387dea4 --- /dev/null +++ b/src/message-runtime-gating.ts @@ -0,0 +1,49 @@ +import { logger } from './logger.js'; +import { hasAllowedTrigger } from './message-runtime-rules.js'; +import { handleSessionCommand, type SessionCommandDeps } from './session-commands.js'; +import type { NewMessage, RegisteredGroup } from './types.js'; + +export async function handleQueuedRunGates(args: { + chatJid: string; + group: RegisteredGroup; + runId: string; + missedMessages: NewMessage[]; + triggerPattern: RegExp; + timezone: string; + hasImplicitContinuationWindow: ( + chatJid: string, + messages: NewMessage[], + ) => boolean; + sessionCommandDeps: SessionCommandDeps; +}): Promise<{ handled: true; success: boolean } | { handled: false }> { + const cmdResult = await handleSessionCommand({ + missedMessages: args.missedMessages, + isMainGroup: args.group.isMain === true, + groupName: args.group.name, + runId: args.runId, + triggerPattern: args.triggerPattern, + timezone: args.timezone, + deps: args.sessionCommandDeps, + }); + if (cmdResult.handled) { + return cmdResult; + } + + if ( + !hasAllowedTrigger({ + chatJid: args.chatJid, + messages: args.missedMessages, + group: args.group, + triggerPattern: args.triggerPattern, + hasImplicitContinuationWindow: args.hasImplicitContinuationWindow, + }) + ) { + logger.info( + { chatJid: args.chatJid, group: args.group.name, runId: args.runId }, + 'Skipping queued run because no allowed trigger was found', + ); + return { handled: true, success: true }; + } + + return { handled: false }; +} diff --git a/src/message-runtime-handoffs.ts b/src/message-runtime-handoffs.ts new file mode 100644 index 0000000..3f6535f --- /dev/null +++ b/src/message-runtime-handoffs.ts @@ -0,0 +1,173 @@ +import { getErrorMessage } from './utils.js'; +import { + claimServiceHandoff, + completeServiceHandoffAndAdvanceTargetCursor, + failServiceHandoff, + getAllPendingServiceHandoffs, + type ServiceHandoff, +} from './db.js'; +import { findChannel, findChannelByName } from './router.js'; +import { logger } from './logger.js'; +import { + getFixedRoleChannelName, + getMissingRoleChannelMessage, + resolveHandoffCursorKey, + resolveHandoffRoleOverride, +} from './message-runtime-shared.js'; +import type { + AgentType, + Channel, + PairedRoomRole, + RegisteredGroup, +} from './types.js'; + +type ExecuteTurnFn = (args: { + group: RegisteredGroup; + prompt: string; + chatJid: string; + runId: string; + channel: Channel; + startSeq: number | null; + endSeq: number | null; + deliveryRole?: PairedRoomRole; + forcedRole?: PairedRoomRole; + forcedAgentType?: AgentType; +}) => Promise<{ + outputStatus: 'success' | 'error'; + deliverySucceeded: boolean; + visiblePhase: unknown; +}>; + +export function enqueuePendingHandoffs(args: { + enqueueTask: ( + chatJid: string, + taskId: string, + task: () => Promise, + ) => void; + processClaimedHandoff: (handoff: ServiceHandoff) => Promise; +}): void { + for (const handoff of getAllPendingServiceHandoffs()) { + if (!claimServiceHandoff(handoff.id)) { + continue; + } + + args.enqueueTask(handoff.chat_jid, `handoff:${handoff.id}`, async () => { + await args.processClaimedHandoff(handoff); + }); + } +} + +export async function processClaimedHandoff(args: { + handoff: ServiceHandoff; + getRegisteredGroups: () => Record; + channels: Channel[]; + executeTurn: ExecuteTurnFn; + lastAgentTimestamps: Record; + saveState: () => void; +}): Promise { + const { handoff } = args; + const group = args.getRegisteredGroups()[handoff.chat_jid]; + if (!group) { + failServiceHandoff(handoff.id, 'Group not registered on target service'); + return; + } + + const channel = findChannel(args.channels, handoff.chat_jid); + if (!channel) { + failServiceHandoff(handoff.id, 'No channel owns handoff jid'); + return; + } + + const handoffRole = resolveHandoffRoleOverride(handoff); + let handoffChannel = channel; + if (handoffRole === 'reviewer') { + const reviewerChannel = findChannelByName( + args.channels, + getFixedRoleChannelName('reviewer'), + ); + if (!reviewerChannel) { + failServiceHandoff(handoff.id, getMissingRoleChannelMessage('reviewer')); + return; + } + handoffChannel = reviewerChannel; + } else if (handoffRole === 'arbiter') { + const arbiterChannel = findChannelByName( + args.channels, + getFixedRoleChannelName('arbiter'), + ); + if (!arbiterChannel) { + failServiceHandoff(handoff.id, getMissingRoleChannelMessage('arbiter')); + return; + } + handoffChannel = arbiterChannel; + } + + const runId = `handoff-${handoff.id}`; + try { + logger.info( + { + chatJid: handoff.chat_jid, + handoffId: handoff.id, + runId, + handoffRole, + targetRole: handoff.target_role ?? null, + targetServiceId: handoff.target_service_id, + targetAgentType: handoff.target_agent_type, + reason: handoff.reason, + intendedRole: handoff.intended_role ?? null, + channelName: handoffChannel.name, + }, + 'Dispatching claimed service handoff', + ); + const result = await args.executeTurn({ + group, + prompt: handoff.prompt, + chatJid: handoff.chat_jid, + runId, + channel: handoffChannel, + startSeq: handoff.start_seq, + endSeq: handoff.end_seq, + forcedRole: handoffRole, + forcedAgentType: handoff.target_agent_type, + }); + + if (!result.deliverySucceeded) { + failServiceHandoff(handoff.id, 'Handoff delivery failed'); + return; + } + + const cursorKey = resolveHandoffCursorKey(handoff.chat_jid, handoffRole); + const appliedCursor = completeServiceHandoffAndAdvanceTargetCursor({ + id: handoff.id, + chat_jid: handoff.chat_jid, + cursor_key: cursorKey, + end_seq: handoff.end_seq, + }); + if (appliedCursor) { + args.lastAgentTimestamps[cursorKey] = appliedCursor; + args.saveState(); + } + logger.info( + { + chatJid: handoff.chat_jid, + handoffId: handoff.id, + runId, + outputStatus: result.outputStatus, + visiblePhase: result.visiblePhase, + appliedCursor, + cursorKey: + appliedCursor != null + ? resolveHandoffCursorKey(handoff.chat_jid, handoffRole) + : null, + }, + 'Completed claimed service handoff', + ); + } catch (err) { + const errorMessage = getErrorMessage(err); + failServiceHandoff(handoff.id, errorMessage); + logger.error( + { chatJid: handoff.chat_jid, handoffId: handoff.id, err }, + 'Claimed service handoff failed', + ); + } +} diff --git a/src/message-runtime-loop.ts b/src/message-runtime-loop.ts new file mode 100644 index 0000000..064cb0f --- /dev/null +++ b/src/message-runtime-loop.ts @@ -0,0 +1,201 @@ +import { + getMessagesSinceSeq, + getNewMessagesBySeq, + getOpenWorkItemForChat, +} from './db.js'; +import { resolveGroupIpcPath } from './group-folder.js'; +import { logger } from './logger.js'; +import { processLoopGroupMessages } from './message-runtime-dispatch.js'; +import { + advanceLastAgentCursor, + getProcessableMessages, +} from './message-runtime-rules.js'; +import type { ScheduledPairedFollowUpIntentKind } from './paired-follow-up-scheduler.js'; +import { findChannel, formatMessages } from './router.js'; +import type { + AgentType, + Channel, + NewMessage, + PairedRoomRole, + PairedTask, + RegisteredGroup, +} from './types.js'; + +type ExecuteTurnFn = (args: { + group: RegisteredGroup; + prompt: string; + chatJid: string; + runId: string; + channel: Channel; + startSeq: number | null; + endSeq: number | null; + deliveryRole?: PairedRoomRole; + hasHumanMessage?: boolean; + forcedRole?: PairedRoomRole; + forcedAgentType?: AgentType; +}) => Promise<{ + outputStatus: 'success' | 'error'; + deliverySucceeded: boolean; + visiblePhase: unknown; +}>; + +export async function processMessageLoopTick(args: { + assistantName: string; + failureFinalText: string; + triggerPattern: RegExp; + timezone: string; + channels: Channel[]; + getRegisteredGroups: () => Record; + getLastTimestamp: () => string; + setLastTimestamp: (timestamp: string) => void; + lastAgentTimestamps: Record; + saveState: () => void; + hasImplicitContinuationWindow: ( + chatJid: string, + messages: NewMessage[], + ) => boolean; + executeTurn: ExecuteTurnFn; + enqueuePendingHandoffs: () => void; + scheduleQueuedPairedFollowUp: (args: { + chatJid: string; + runId: string; + task: PairedTask | null | undefined; + intentKind: ScheduledPairedFollowUpIntentKind; + enqueue: () => void; + }) => boolean; + enqueueScopedGroupMessageCheck: (chatJid: string, groupFolder: string) => void; + sendQueuedMessage: (chatJid: string, text: string) => boolean; + closeStdin: (chatJid: string, reason: string) => void; + labelPairedSenders: (chatJid: string, messages: NewMessage[]) => NewMessage[]; +}): Promise { + args.enqueuePendingHandoffs(); + const registeredGroups = args.getRegisteredGroups(); + const jids = Object.keys(registeredGroups); + const { messages, newSeqCursor } = getNewMessagesBySeq( + jids, + args.getLastTimestamp(), + args.assistantName, + ); + + if (messages.length === 0) { + return; + } + + logger.info({ count: messages.length }, 'New messages'); + args.setLastTimestamp(newSeqCursor); + args.saveState(); + + const messagesByGroup = new Map(); + for (const msg of messages) { + const existing = messagesByGroup.get(msg.chat_jid); + if (existing) { + existing.push(msg); + } else { + messagesByGroup.set(msg.chat_jid, [msg]); + } + } + + for (const [chatJid, groupMessages] of messagesByGroup) { + const group = registeredGroups[chatJid]; + if (!group) continue; + + const channel = findChannel(args.channels, chatJid); + if (!channel) { + logger.warn({ chatJid }, 'No channel owns JID, skipping messages'); + continue; + } + + await processLoopGroupMessages({ + chatJid, + group, + groupMessages, + channel, + assistantName: args.assistantName, + failureFinalText: args.failureFinalText, + triggerPattern: args.triggerPattern, + hasImplicitContinuationWindow: args.hasImplicitContinuationWindow, + lastAgentTimestamps: args.lastAgentTimestamps, + saveState: args.saveState, + timezone: args.timezone, + executeTurn: args.executeTurn, + schedulePairedFollowUp: (task, intentKind, followUpRunId) => + args.scheduleQueuedPairedFollowUp({ + chatJid, + runId: followUpRunId, + task, + intentKind, + enqueue: () => + args.enqueueScopedGroupMessageCheck(chatJid, group.folder), + }), + enqueueMessageCheck: () => + args.enqueueScopedGroupMessageCheck(chatJid, group.folder), + sendQueuedMessage: args.sendQueuedMessage, + closeStdin: (reason) => args.closeStdin(chatJid, reason), + labelPairedSenders: args.labelPairedSenders, + formatMessages, + }); + } +} + +export function recoverPendingMessages(args: { + assistantName: string; + channels: Channel[]; + getRegisteredGroups: () => Record; + lastAgentTimestamps: Record; + saveState: () => void; + enqueueScopedGroupMessageCheck: (chatJid: string, groupFolder: string) => void; +}): void { + const registeredGroups = args.getRegisteredGroups(); + for (const [chatJid, group] of Object.entries(registeredGroups)) { + const openWorkItem = getOpenWorkItemForChat(chatJid); + if (openWorkItem) { + logger.info( + { chatJid, group: group.name, workItemId: openWorkItem.id }, + 'Recovery: found open work item awaiting delivery', + ); + args.enqueueScopedGroupMessageCheck(chatJid, group.folder); + continue; + } + + const sinceSeqCursor = args.lastAgentTimestamps[chatJid] || ''; + const rawPending = getMessagesSinceSeq( + chatJid, + sinceSeqCursor, + args.assistantName, + ); + const recoveryChannel = findChannel(args.channels, chatJid); + const pending = getProcessableMessages( + chatJid, + rawPending, + recoveryChannel ?? undefined, + ); + if (pending.length > 0) { + logger.info( + { group: group.name, pendingCount: pending.length }, + 'Recovery: found unprocessed messages', + ); + args.enqueueScopedGroupMessageCheck(chatJid, group.folder); + continue; + } + + if (rawPending.length > 0) { + const endSeq = rawPending[rawPending.length - 1].seq; + if (endSeq != null) { + advanceLastAgentCursor( + args.lastAgentTimestamps, + args.saveState, + chatJid, + endSeq, + ); + } + } + } +} + +export function buildScopedMessageCheckEnqueuer(queue: { + enqueueMessageCheck: (chatJid: string, ipcDir?: string) => void; +}): (chatJid: string, groupFolder: string) => void { + return (chatJid: string, groupFolder: string): void => { + queue.enqueueMessageCheck(chatJid, resolveGroupIpcPath(groupFolder)); + }; +} diff --git a/src/message-runtime-queue.ts b/src/message-runtime-queue.ts new file mode 100644 index 0000000..8336d11 --- /dev/null +++ b/src/message-runtime-queue.ts @@ -0,0 +1,244 @@ +import { + getPairedTurnOutputs, + getRecentChatMessages, +} from './db.js'; +import { logger } from './logger.js'; +import { + buildArbiterPromptForTask, + buildPairedTurnPrompt, +} from './message-runtime-prompts.js'; +import { + buildPendingPairedTurn, + executePendingPairedTurn, + isBotOnlyPairedRoomTurn, +} from './message-runtime-flow.js'; +import { + advanceLastAgentCursor, + resolveActiveRole, + resolveCursorKeyForRole, + resolveQueuedTurnRole, +} from './message-runtime-rules.js'; +import type { + AgentType, + Channel, + NewMessage, + PairedRoomRole, + PairedTask, + RegisteredGroup, +} from './types.js'; + +type ExecuteTurnFn = (args: { + group: RegisteredGroup; + prompt: string; + chatJid: string; + runId: string; + channel: Channel; + startSeq: number | null; + endSeq: number | null; + deliveryRole?: PairedRoomRole; + hasHumanMessage?: boolean; + forcedRole?: PairedRoomRole; + forcedAgentType?: AgentType; +}) => Promise<{ + outputStatus: 'success' | 'error'; + deliverySucceeded: boolean; + visiblePhase: unknown; +}>; + +type RoleToChannelMap = Record<'owner' | 'reviewer' | 'arbiter', Channel | null>; + +export async function runPendingPairedTurnIfNeeded(args: { + chatJid: string; + group: RegisteredGroup; + runId: string; + log: typeof logger; + timezone: string; + task: PairedTask | null | undefined; + rawMissedMessages: NewMessage[]; + saveState: () => void; + lastAgentTimestamps: Record; + executeTurn: ExecuteTurnFn; + getFixedRoleChannelName: (role: 'reviewer' | 'arbiter') => string; + roleToChannel: RoleToChannelMap; + labelPairedSenders: (chatJid: string, messages: NewMessage[]) => NewMessage[]; + mode: 'idle' | 'bot-only'; + missedMessages?: NewMessage[]; +}): Promise { + const { chatJid, task, roleToChannel } = args; + if (!task) { + return null; + } + + if ( + args.mode === 'bot-only' && + (!args.missedMessages || + !isBotOnlyPairedRoomTurn(chatJid, args.missedMessages)) + ) { + return null; + } + + const recentMessages = getRecentChatMessages(chatJid, 20); + const recentHumanMessages = recentMessages.filter( + (message) => !message.is_bot_message, + ); + const labeledRecentMessages = args.labelPairedSenders( + chatJid, + recentMessages, + ); + const pendingTurn = buildPendingPairedTurn({ + chatJid, + timezone: args.timezone, + task, + rawMissedMessages: args.rawMissedMessages, + recentHumanMessages, + labeledRecentMessages, + resolveChannel: (taskStatus) => + roleToChannel[resolveActiveRole(taskStatus)] ?? null, + }); + + if (!pendingTurn) { + return null; + } + + return executePendingPairedTurn({ + pendingTurn, + chatJid, + group: args.group, + runId: args.runId, + log: args.log, + saveState: args.saveState, + lastAgentTimestamps: args.lastAgentTimestamps, + executeTurn: args.executeTurn, + getFixedRoleChannelName: args.getFixedRoleChannelName, + }); +} + +export async function runQueuedGroupTurn(args: { + chatJid: string; + group: RegisteredGroup; + runId: string; + log: typeof logger; + timezone: string; + missedMessages: NewMessage[]; + task: PairedTask | null | undefined; + roleToChannel: RoleToChannelMap; + ownerChannel: Channel; + lastAgentTimestamps: Record; + saveState: () => void; + executeTurn: ExecuteTurnFn; + getFixedRoleChannelName: (role: 'reviewer' | 'arbiter') => string; + labelPairedSenders: (chatJid: string, messages: NewMessage[]) => NewMessage[]; + formatMessages: (messages: NewMessage[], timezone: string) => string; +}): Promise { + const { chatJid, group, runId, log, missedMessages, task, roleToChannel } = + args; + const taskStatus = task?.status; + const hasHumanMsg = !isBotOnlyPairedRoomTurn(chatJid, missedMessages); + const turnRole = task + ? resolveQueuedTurnRole({ + taskStatus, + hasHumanMessage: hasHumanMsg, + }) + : 'owner'; + const turnChannel = + turnRole === 'owner' ? args.ownerChannel : roleToChannel[turnRole]; + const cursorKey = resolveCursorKeyForRole(chatJid, turnRole); + const forcedRole = + task && turnRole !== resolveActiveRole(taskStatus) ? turnRole : undefined; + + let prompt: string; + if (turnRole === 'arbiter' && task) { + const recentMessages = getRecentChatMessages(chatJid, 20); + prompt = buildArbiterPromptForTask({ + task, + chatJid, + timezone: args.timezone, + turnOutputs: getPairedTurnOutputs(task.id), + recentMessages, + labeledRecentMessages: args.labelPairedSenders(chatJid, recentMessages), + }); + } else if (task) { + prompt = buildPairedTurnPrompt({ + taskId: task.id, + chatJid, + timezone: args.timezone, + missedMessages, + labeledFallbackMessages: args.labelPairedSenders(chatJid, missedMessages), + turnOutputs: getPairedTurnOutputs(task.id), + }); + } else { + prompt = args.formatMessages( + args.labelPairedSenders(chatJid, missedMessages), + args.timezone, + ); + } + + const startSeq = missedMessages[0].seq ?? null; + const endSeq = missedMessages[missedMessages.length - 1].seq ?? null; + log.info( + { + messageCount: missedMessages.length, + messageSeqStart: startSeq, + messageSeqEnd: endSeq, + }, + 'Dispatching queued messages to agent', + ); + + if (!turnChannel) { + const missingRole = turnRole === 'arbiter' ? 'arbiter' : 'reviewer'; + log.error( + { + taskStatus, + role: turnRole, + requiredChannel: args.getFixedRoleChannelName(missingRole), + }, + 'Skipping paired-room run because the dedicated Discord role channel is not configured', + ); + return false; + } + + if (endSeq !== null) { + advanceLastAgentCursor( + args.lastAgentTimestamps, + args.saveState, + chatJid, + endSeq, + cursorKey, + ); + } + + const { deliverySucceeded, visiblePhase } = await args.executeTurn({ + group, + prompt, + chatJid, + runId, + channel: turnChannel, + deliveryRole: task ? turnRole : undefined, + startSeq, + endSeq, + hasHumanMessage: hasHumanMsg, + forcedRole, + }); + + if (!deliverySucceeded) { + log.warn( + { + messageSeqStart: startSeq, + messageSeqEnd: endSeq, + }, + 'Persisted produced output for delivery retry without rerunning agent', + ); + return false; + } + + log.info( + { + visiblePhase, + messageSeqStart: startSeq, + messageSeqEnd: endSeq, + }, + 'Queued run completed successfully', + ); + + return true; +} diff --git a/src/message-runtime-rules.test.ts b/src/message-runtime-rules.test.ts index 21e21d7..79e4a32 100644 --- a/src/message-runtime-rules.test.ts +++ b/src/message-runtime-rules.test.ts @@ -1,6 +1,7 @@ import { describe, expect, it } from 'vitest'; import { + resolveFollowUpDispatch, resolveExecutionTarget, resolveNextTurnAction, resolveQueuedTurnRole, @@ -86,6 +87,72 @@ describe('message-runtime-rules', () => { ).toEqual({ kind: 'none' }); }); + it('dispatches owner delivery success through a paired follow-up enqueue only for reviewer turns', () => { + expect( + resolveFollowUpDispatch({ + source: 'owner-delivery-success', + nextTurnAction: { kind: 'reviewer-turn' }, + }), + ).toEqual({ + kind: 'enqueue', + queueKind: 'paired-follow-up', + }); + expect( + resolveFollowUpDispatch({ + source: 'owner-delivery-success', + nextTurnAction: { kind: 'none' }, + }), + ).toEqual({ kind: 'none' }); + }); + + it('dispatches delivery retry follow-ups through either generic message checks or paired follow-up enqueue', () => { + expect( + resolveFollowUpDispatch({ + source: 'delivery-retry', + nextTurnAction: { kind: 'none' }, + completedRole: 'owner', + }), + ).toEqual({ + kind: 'enqueue', + queueKind: 'message-check', + }); + expect( + resolveFollowUpDispatch({ + source: 'delivery-retry', + nextTurnAction: { kind: 'reviewer-turn' }, + completedRole: 'owner', + }), + ).toEqual({ + kind: 'enqueue', + queueKind: 'paired-follow-up', + }); + expect( + resolveFollowUpDispatch({ + source: 'delivery-retry', + nextTurnAction: { kind: 'finalize-owner-turn' }, + completedRole: 'reviewer', + }), + ).toEqual({ kind: 'none' }); + }); + + it('dispatches bot-only follow-ups through inline finalize or paired enqueue', () => { + expect( + resolveFollowUpDispatch({ + source: 'bot-only-follow-up', + nextTurnAction: { kind: 'finalize-owner-turn' }, + }), + ).toEqual({ kind: 'inline' }); + expect( + resolveFollowUpDispatch({ + source: 'bot-only-follow-up', + nextTurnAction: { kind: 'owner-follow-up' }, + }), + ).toEqual({ + kind: 'enqueue', + queueKind: 'paired-follow-up', + }); + }); + it('routes fresh human input to owner even while review is pending', () => { expect( resolveQueuedTurnRole({ diff --git a/src/message-runtime-rules.ts b/src/message-runtime-rules.ts index b1653c9..1369d62 100644 --- a/src/message-runtime-rules.ts +++ b/src/message-runtime-rules.ts @@ -79,6 +79,11 @@ export type NextTurnAction = | { kind: 'owner-follow-up' } | { kind: 'finalize-owner-turn' }; +export type FollowUpDispatch = + | { kind: 'none' } + | { kind: 'inline' } + | { kind: 'enqueue'; queueKind: 'paired-follow-up' | 'message-check' }; + export function resolveNextTurnAction(args: { taskStatus?: PairedTaskStatus | null; lastTurnOutputRole?: PairedRoomRole | null; @@ -108,6 +113,62 @@ export function resolveNextTurnAction(args: { } } +export function resolveFollowUpDispatch(args: { + source: + | 'owner-delivery-success' + | 'delivery-retry' + | 'bot-only-follow-up' + | 'executor-recovery'; + nextTurnAction: NextTurnAction; + completedRole?: PairedRoomRole; + executionStatus?: 'succeeded' | 'failed'; + sawOutput?: boolean; +}): FollowUpDispatch { + switch (args.source) { + case 'owner-delivery-success': + return args.nextTurnAction.kind === 'reviewer-turn' + ? { kind: 'enqueue', queueKind: 'paired-follow-up' } + : { kind: 'none' }; + + case 'delivery-retry': + if ( + args.completedRole === 'reviewer' && + args.nextTurnAction.kind === 'finalize-owner-turn' + ) { + return { kind: 'none' }; + } + if (args.nextTurnAction.kind === 'none') { + return { kind: 'enqueue', queueKind: 'message-check' }; + } + return { kind: 'enqueue', queueKind: 'paired-follow-up' }; + + case 'bot-only-follow-up': + if (args.nextTurnAction.kind === 'none') { + return { kind: 'none' }; + } + if (args.nextTurnAction.kind === 'finalize-owner-turn') { + return { kind: 'inline' }; + } + return { kind: 'enqueue', queueKind: 'paired-follow-up' }; + + case 'executor-recovery': + if (args.executionStatus === 'succeeded' && args.sawOutput) { + return { kind: 'none' }; + } + if ( + args.completedRole !== 'reviewer' && + args.completedRole !== 'arbiter' + ) { + return { kind: 'none' }; + } + return args.nextTurnAction.kind === 'reviewer-turn' || + args.nextTurnAction.kind === 'arbiter-turn' || + args.nextTurnAction.kind === 'finalize-owner-turn' + ? { kind: 'enqueue', queueKind: 'paired-follow-up' } + : { kind: 'none' }; + } +} + /** Cursor key for a role. Owner uses chatJid, others use chatJid:role. */ export function resolveCursorKey( chatJid: string, diff --git a/src/message-runtime-shared.ts b/src/message-runtime-shared.ts new file mode 100644 index 0000000..8df6de6 --- /dev/null +++ b/src/message-runtime-shared.ts @@ -0,0 +1,40 @@ +import type { PairedRoomRole } from './types.js'; +import type { ServiceHandoff } from './db.js'; + +export function resolveHandoffRoleOverride( + handoff: Pick, +): PairedRoomRole | undefined { + if (handoff.target_role) { + return handoff.target_role; + } + if (handoff.intended_role) { + return handoff.intended_role; + } + if (handoff.reason?.startsWith('reviewer-')) { + return 'reviewer'; + } + if (handoff.reason?.startsWith('arbiter-')) { + return 'arbiter'; + } + return undefined; +} + +export function resolveHandoffCursorKey( + chatJid: string, + role?: PairedRoomRole, +): string { + if (!role || role === 'owner') { + return chatJid; + } + return `${chatJid}:${role}`; +} + +export function getFixedRoleChannelName(role: 'reviewer' | 'arbiter'): string { + return role === 'reviewer' ? 'discord-review' : 'discord-arbiter'; +} + +export function getMissingRoleChannelMessage( + role: 'reviewer' | 'arbiter', +): string { + return `Missing configured ${role} Discord bot channel (${getFixedRoleChannelName(role)}) for role-fixed delivery`; +} diff --git a/src/message-runtime.test.ts b/src/message-runtime.test.ts index 28d5ee8..abd0dc8 100644 --- a/src/message-runtime.test.ts +++ b/src/message-runtime.test.ts @@ -219,6 +219,7 @@ import { } from './message-runtime-flow.js'; import * as config from './config.js'; import { logger } from './logger.js'; +import { resetPairedFollowUpScheduleState } from './paired-follow-up-scheduler.js'; import * as serviceRouting from './service-routing.js'; import type { Channel, RegisteredGroup } from './types.js'; @@ -254,6 +255,7 @@ function makeChannel( describe('createMessageRuntime', () => { beforeEach(() => { vi.resetAllMocks(); + resetPairedFollowUpScheduleState(); vi.mocked(db.getLastBotFinalMessage).mockReturnValue([]); vi.mocked(serviceRouting.hasReviewerLease).mockReturnValue(false); vi.mocked(db.getRecentChatMessages).mockReturnValue([]); @@ -1697,6 +1699,7 @@ If your first line is DONE_WITH_CONCERNS, the system will reopen review instead task, cursor: 42, cursorKey: `${chatJid}:reviewer`, + intentKind: 'reviewer-turn', nextRole: 'reviewer', }); }); @@ -1806,6 +1809,115 @@ If your first line is DONE_WITH_CONCERNS, the system will reopen review instead ); }); + it('does not enqueue the same reviewer follow-up twice within the same run', 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-dedup', + 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; + + vi.mocked(serviceRouting.hasReviewerLease).mockReturnValue(true); + 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(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; + await onOutput?.({ + status: 'success', + phase: 'final', + result: 'DONE_WITH_CONCERNS\nowner complete', + newSessionId: 'session-owner-delivery-dedup', + }); + return { + status: 'success', + result: 'DONE_WITH_CONCERNS\nowner complete', + newSessionId: 'session-owner-delivery-dedup', + }; + }, + ); + + 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(), + }); + + await runtime.processGroupMessages(chatJid, { + runId: 'run-owner-delivery-dedup', + reason: 'messages', + }); + await runtime.processGroupMessages(chatJid, { + runId: 'run-owner-delivery-dedup', + reason: 'messages', + }); + + 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 in the same run', + ); + }); + it('routes fresh human input to owner even when the latest task is review_ready', async () => { const chatJid = 'group@test'; const group = makeGroup('codex'); diff --git a/src/message-runtime.ts b/src/message-runtime.ts index 495074d..d560cee 100644 --- a/src/message-runtime.ts +++ b/src/message-runtime.ts @@ -1,25 +1,11 @@ import { AgentOutput } from './agent-runner.js'; import { getAgentOutputText } from './agent-output.js'; -import { getErrorMessage } from './utils.js'; import { - claimServiceHandoff, - completeServiceHandoffAndAdvanceTargetCursor, createProducedWorkItem, - failServiceHandoff, - getAllPendingServiceHandoffs, getOpenWorkItemForChat, getMessagesSinceSeq, - getNewMessagesBySeq, - markWorkItemDelivered, - markWorkItemDeliveryRetry, getLastBotFinalMessage, - getLastHumanMessageContent, - getRecentChatMessages, getLatestOpenPairedTaskForChat, - getPairedTurnOutputs, - updatePairedTask, - type ServiceHandoff, - type WorkItem, } from './db.js'; import { isSessionCommandSenderAllowed } from './config.js'; import { GroupQueue, GroupRunContext } from './group-queue.js'; @@ -30,43 +16,43 @@ import { normalizeMessageForDedupe, } from './router.js'; import { isTriggerAllowed, loadSenderAllowlist } from './sender-allowlist.js'; +import { + enqueueGenericFollowUpAfterDeliveryRetry as enqueueDeliveryRetryFollowUp, +} from './message-runtime-dispatch.js'; +import { + deliverOpenWorkItem, + processOpenWorkItemDelivery, +} from './message-runtime-delivery.js'; +import { handleQueuedRunGates } from './message-runtime-gating.js'; +import { + enqueuePendingHandoffs as enqueueClaimedServiceHandoffs, + processClaimedHandoff as processClaimedServiceHandoff, +} from './message-runtime-handoffs.js'; +import { + buildScopedMessageCheckEnqueuer, + processMessageLoopTick, + recoverPendingMessages as recoverRuntimePendingMessages, +} from './message-runtime-loop.js'; +import { + runPendingPairedTurnIfNeeded, + runQueuedGroupTurn, +} from './message-runtime-queue.js'; import { advanceLastAgentCursor, createImplicitContinuationTracker, resolveNextTurnAction, - resolveActiveRole, - resolveCursorKeyForRole, - resolveCursorKey, - resolveQueuedTurnRole, + resolveFollowUpDispatch, filterLoopingPairedBotMessages, getProcessableMessages, - hasAllowedTrigger, shouldSkipBotOnlyCollaboration, } from './message-runtime-rules.js'; import { runAgentForGroup } from './message-agent-executor.js'; -import { - buildQueuedTurnDispatch, - buildPendingPairedTurn, - executeBotOnlyPairedFollowUpAction, - executePendingPairedTurn, - isBotOnlyPairedRoomTurn, - shouldSkipGenericFollowUpAfterDeliveryRetry, -} from './message-runtime-flow.js'; import { MessageTurnController } from './message-turn-controller.js'; import { - buildArbiterPromptForTask, - buildFinalizePendingPrompt, - buildOwnerPendingPrompt, - buildPairedTurnPrompt, - buildReviewerPendingPrompt, -} from './message-runtime-prompts.js'; + schedulePairedFollowUpOnce, + type ScheduledPairedFollowUpIntentKind, +} from './paired-follow-up-scheduler.js'; import { transitionPairedTaskStatus } from './paired-execution-context-shared.js'; -import { - extractSessionCommand, - handleSessionCommand, - isSessionCommandAllowed, - isSessionCommandControlMessage, -} from './session-commands.js'; import { Channel, NewMessage, @@ -76,8 +62,15 @@ import { type PairedTask, } from './types.js'; import { createScopedLogger, logger } from './logger.js'; -import { resolveGroupIpcPath } from './group-folder.js'; import { hasReviewerLease } from './service-routing.js'; +export { + resolveHandoffCursorKey, + resolveHandoffRoleOverride, +} from './message-runtime-shared.js'; +import { + getFixedRoleChannelName, + getMissingRoleChannelMessage, +} from './message-runtime-shared.js'; /** * Check if a message is a duplicate of the last bot final message in a paired room. @@ -104,34 +97,6 @@ export function isDuplicateOfLastBotFinal( return normalizedLast === normalizedCurrent && normalizedLast.length > 0; } -export function resolveHandoffRoleOverride( - handoff: Pick, -): PairedRoomRole | undefined { - if (handoff.target_role) { - return handoff.target_role; - } - if (handoff.intended_role) { - return handoff.intended_role; - } - if (handoff.reason?.startsWith('reviewer-')) { - return 'reviewer'; - } - if (handoff.reason?.startsWith('arbiter-')) { - return 'arbiter'; - } - return undefined; -} - -export function resolveHandoffCursorKey( - chatJid: string, - role?: PairedRoomRole, -): string { - if (!role || role === 'owner') { - return chatJid; - } - return `${chatJid}:${role}`; -} - export interface MessageRuntimeDeps { assistantName: string; idleTimeout: number; @@ -150,21 +115,6 @@ export interface MessageRuntimeDeps { clearSession: (groupFolder: string, opts?: { allRoles?: boolean }) => void; } -function getFixedRoleChannelName(role: 'reviewer' | 'arbiter'): string { - return role === 'reviewer' ? 'discord-review' : 'discord-arbiter'; -} - -function getMissingRoleChannelMessage(role: 'reviewer' | 'arbiter'): string { - return `Missing configured ${role} Discord bot channel (${getFixedRoleChannelName(role)}) for role-fixed delivery`; -} - -function isTerminalStatusMessage(text: string): boolean { - const trimmed = text.trimStart(); - return /^(?:\*\*)?(DONE(?:_WITH_CONCERNS)?|BLOCKED|NEEDS_CONTEXT)\b/.test( - trimmed, - ); -} - export function createMessageRuntime(deps: MessageRuntimeDeps): { processGroupMessages: ( chatJid: string, @@ -210,11 +160,27 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { }); }; - const enqueueScopedGroupMessageCheck = ( - chatJid: string, - groupFolder: string, - ): void => { - deps.queue.enqueueMessageCheck(chatJid, resolveGroupIpcPath(groupFolder)); + const enqueueScopedGroupMessageCheck = buildScopedMessageCheckEnqueuer( + deps.queue, + ); + + const scheduleQueuedPairedFollowUp = (args: { + chatJid: string; + runId: string; + task: PairedTask | null | undefined; + intentKind: ScheduledPairedFollowUpIntentKind; + enqueue: () => void; + }): boolean => { + if (!args.task) { + return false; + } + return schedulePairedFollowUpOnce({ + chatJid: args.chatJid, + runId: args.runId, + task: args.task, + intentKind: args.intentKind, + enqueue: args.enqueue, + }); }; /** @@ -228,119 +194,6 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { return isDuplicateOfLastBotFinal(chatJid, text); }; - const buildDeliveryLogContext = ( - channel: Channel, - item: WorkItem, - extra: Record = {}, - ): Record => ({ - chatJid: item.chat_jid, - channelName: channel.name, - workItemId: item.id, - deliveryRole: item.delivery_role ?? null, - ...extra, - }); - - const deliverOpenWorkItem = async ( - channel: Channel, - item: WorkItem, - options?: { - replaceMessageId?: string | null; - }, - ): Promise => { - const replaceMessageId = options?.replaceMessageId ?? null; - - // Check for duplicate in paired rooms before attempting delivery - const isDuplicate = checkDuplicateOfLastBotFinal( - item.chat_jid, - item.result_payload, - ); - - if (isDuplicate) { - // Mark as delivered without sending, and don't open continuation - markWorkItemDelivered(item.id, null); - logger.info( - buildDeliveryLogContext(channel, item, { - preview: item.result_payload.slice(0, 100), - suppressionReason: 'paired-final-duplicate', - }), - 'Suppressed duplicate final message in paired room (marked as delivered)', - ); - return true; - } - - try { - if (replaceMessageId && channel.editMessage) { - logger.info( - buildDeliveryLogContext(channel, item, { - deliveryAttempts: item.delivery_attempts + 1, - deliveryMode: 'edit', - replacedMessageId: replaceMessageId, - }), - 'Attempting to deliver produced work item by replacing tracked progress message', - ); - await channel.editMessage( - item.chat_jid, - replaceMessageId, - item.result_payload, - ); - markWorkItemDelivered(item.id, replaceMessageId); - continuationTracker.open(item.chat_jid); - logger.info( - buildDeliveryLogContext(channel, item, { - deliveryAttempts: item.delivery_attempts + 1, - deliveryMode: 'edit', - replacedMessageId: replaceMessageId, - }), - 'Delivered produced work item by replacing tracked progress message', - ); - return true; - } - } catch (err) { - logger.warn( - buildDeliveryLogContext(channel, item, { - deliveryAttempts: item.delivery_attempts + 1, - deliveryMode: 'edit', - replacedMessageId: replaceMessageId, - err, - }), - 'Failed to replace tracked progress message; falling back to a new message', - ); - } - - try { - logger.info( - buildDeliveryLogContext(channel, item, { - deliveryAttempts: item.delivery_attempts + 1, - deliveryMode: 'send', - }), - 'Attempting to deliver produced work item as a new message', - ); - await channel.sendMessage(item.chat_jid, item.result_payload); - markWorkItemDelivered(item.id); - continuationTracker.open(item.chat_jid); - logger.info( - buildDeliveryLogContext(channel, item, { - deliveryAttempts: item.delivery_attempts + 1, - deliveryMode: 'send', - }), - 'Delivered produced work item', - ); - return true; - } catch (err) { - const errorMessage = getErrorMessage(err); - markWorkItemDeliveryRetry(item.id, errorMessage); - logger.warn( - buildDeliveryLogContext(channel, item, { - deliveryAttempts: item.delivery_attempts + 1, - deliveryMode: 'send', - err, - }), - 'Failed to deliver produced work item', - ); - return false; - } - }; - const runAgent = async ( group: RegisteredGroup, prompt: string, @@ -451,7 +304,14 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { end_seq: endSeq, result_payload: text, }); - return deliverOpenWorkItem(channel, workItem); + return deliverOpenWorkItem({ + channel, + item: workItem, + log: logger, + isDuplicateOfLastBotFinal: checkDuplicateOfLastBotFinal, + openContinuation: (targetChatJid) => + continuationTracker.open(targetChatJid), + }); } catch (err) { logger.warn( { group: group.name, chatJid, runId, err }, @@ -492,8 +352,22 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { taskStatus: pendingTaskAfterDelivery?.status, lastTurnOutputRole: 'owner', }); - if (nextTurnAction.kind === 'reviewer-turn') { - deps.queue.enqueueMessageCheck(chatJid); + const dispatch = resolveFollowUpDispatch({ + source: 'owner-delivery-success', + nextTurnAction, + }); + 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), + }); logger.info( { chatJid, @@ -501,8 +375,11 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { completedRole: resolvedDeliveryRole, taskId: pendingTaskAfterDelivery?.id ?? null, taskStatus: pendingTaskAfterDelivery?.status ?? null, + scheduled, }, - 'Queued paired follow-up after successful owner delivery', + scheduled + ? 'Queued paired follow-up after successful owner delivery' + : 'Skipped duplicate paired follow-up after successful owner delivery in the same run', ); } } @@ -530,135 +407,21 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { }; const enqueuePendingHandoffs = (): void => { - // Unified runtime claims pending handoffs once and resolves delivery from role context. - for (const handoff of getAllPendingServiceHandoffs()) { - if (!claimServiceHandoff(handoff.id)) { - continue; - } - - deps.queue.enqueueTask( - handoff.chat_jid, - `handoff:${handoff.id}`, - async () => { - await processClaimedHandoff(handoff); - }, - ); - } - }; - - const processClaimedHandoff = async ( - handoff: ServiceHandoff, - ): Promise => { - const group = deps.getRegisteredGroups()[handoff.chat_jid]; - if (!group) { - failServiceHandoff(handoff.id, 'Group not registered on target service'); - return; - } - - const channel = findChannel(deps.channels, handoff.chat_jid); - if (!channel) { - failServiceHandoff(handoff.id, 'No channel owns handoff jid'); - return; - } - - // Reviewer/arbiter failover handoffs should run via the appropriate - // channel so they execute in the correct role mode. - const handoffRole = resolveHandoffRoleOverride(handoff); - let handoffChannel = channel; - if (handoffRole === 'reviewer') { - // Role-fixed delivery intentionally does not follow fallback agent type. - const reviewerChannel = findChannelByName( - deps.channels, - getFixedRoleChannelName('reviewer'), - ); - if (!reviewerChannel) { - failServiceHandoff( - handoff.id, - getMissingRoleChannelMessage('reviewer'), - ); - return; - } - handoffChannel = reviewerChannel; - } else if (handoffRole === 'arbiter') { - const arbiterChannel = findChannelByName( - deps.channels, - getFixedRoleChannelName('arbiter'), - ); - if (!arbiterChannel) { - failServiceHandoff(handoff.id, getMissingRoleChannelMessage('arbiter')); - return; - } - handoffChannel = arbiterChannel; - } - - const runId = `handoff-${handoff.id}`; - try { - logger.info( - { - chatJid: handoff.chat_jid, - handoffId: handoff.id, - runId, - handoffRole, - targetRole: handoff.target_role ?? null, - targetServiceId: handoff.target_service_id, - targetAgentType: handoff.target_agent_type, - reason: handoff.reason, - intendedRole: handoff.intended_role ?? null, - channelName: handoffChannel.name, - }, - 'Dispatching claimed service handoff', - ); - const result = await executeTurn({ - group, - prompt: handoff.prompt, - chatJid: handoff.chat_jid, - runId, - channel: handoffChannel, - startSeq: handoff.start_seq, - endSeq: handoff.end_seq, - forcedRole: handoffRole, - forcedAgentType: handoff.target_agent_type, - }); - - if (!result.deliverySucceeded) { - failServiceHandoff(handoff.id, 'Handoff delivery failed'); - return; - } - - const cursorKey = resolveHandoffCursorKey(handoff.chat_jid, handoffRole); - const appliedCursor = completeServiceHandoffAndAdvanceTargetCursor({ - id: handoff.id, - chat_jid: handoff.chat_jid, - cursor_key: cursorKey, - end_seq: handoff.end_seq, - }); - if (appliedCursor) { - deps.getLastAgentTimestamps()[cursorKey] = appliedCursor; - deps.saveState(); - } - logger.info( - { - chatJid: handoff.chat_jid, - handoffId: handoff.id, - runId, - outputStatus: result.outputStatus, - visiblePhase: result.visiblePhase, - appliedCursor, - cursorKey: - appliedCursor != null - ? resolveHandoffCursorKey(handoff.chat_jid, handoffRole) - : null, - }, - 'Completed claimed service handoff', - ); - } catch (err) { - const errorMessage = getErrorMessage(err); - failServiceHandoff(handoff.id, errorMessage); - logger.error( - { chatJid: handoff.chat_jid, handoffId: handoff.id, err }, - 'Claimed service handoff failed', - ); - } + enqueueClaimedServiceHandoffs({ + enqueueTask: (chatJid, taskId, task) => { + deps.queue.enqueueTask?.(chatJid, taskId, task); + }, + processClaimedHandoff: async (handoff) => { + await processClaimedServiceHandoff({ + handoff, + getRegisteredGroups: deps.getRegisteredGroups, + channels: deps.channels, + executeTurn, + lastAgentTimestamps: deps.getLastAgentTimestamps(), + saveState: deps.saveState, + }); + }, + }); }; const processGroupMessages = async ( @@ -695,49 +458,11 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { ); // Resolve the correct Discord channel for a given task status. - const roleToChannel: Record = { + const roleToChannel: Record<'owner' | 'reviewer' | 'arbiter', Channel | null> = { owner: channel, reviewer: foundReviewerChannel || null, arbiter: foundArbiterChannel || null, }; - const resolveChannelForRole = ( - role: 'owner' | 'reviewer' | 'arbiter', - ): Channel | null => { - return role === 'owner' ? channel : roleToChannel[role]; - }; - const resolveChannel = (taskStatus?: string | null): Channel | null => - resolveChannelForRole(resolveActiveRole(taskStatus)); - - const enqueueGroupMessageCheck = (): void => { - enqueueScopedGroupMessageCheck(chatJid, group.folder); - }; - - const enqueueGenericFollowUpAfterDeliveryRetry = (args: { - deliveryRole: PairedRoomRole; - pendingTask: PairedTask | null | undefined; - workItemId: string | number; - }): void => { - if ( - shouldSkipGenericFollowUpAfterDeliveryRetry({ - chatJid, - deliveryRole: args.deliveryRole, - pendingTask: args.pendingTask, - }) - ) { - log.info( - { - workItemId: args.workItemId, - chatJid, - deliveryRole: args.deliveryRole, - pendingTaskStatus: args.pendingTask?.status ?? null, - }, - 'Skipping queued follow-up after reviewer merge_ready delivery because inline finalize will handle the handoff', - ); - return; - } - deps.queue.enqueueMessageCheck(chatJid); - }; - if (hasReviewerLease(chatJid)) { log.info( { @@ -754,49 +479,48 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { // Delivery retries can come from forced fallback runs whose agent_type // differs from the room owner's registered agent type. const openWorkItem = getOpenWorkItemForChat(chatJid); - if (openWorkItem) { - const pendingTask = hasReviewerLease(chatJid) + const openWorkItemOutcome = await processOpenWorkItemDelivery({ + chatJid, + runId, + openWorkItem, + pendingTask: hasReviewerLease(chatJid) ? getLatestOpenPairedTaskForChat(chatJid) - : null; - if (hasReviewerLease(chatJid) && openWorkItem.delivery_role == null) { - log.warn( - { - workItemId: openWorkItem.id, - chatJid, - pendingTaskStatus: pendingTask?.status ?? null, - }, - 'Paired-room delivery retry is missing a persisted delivery role; falling back to inferred routing', - ); - } - const deliveryRole = - openWorkItem.delivery_role ?? - (pendingTask ? resolveActiveRole(pendingTask.status) : 'owner'); - const deliveryChannel = - deliveryRole === 'owner' ? channel : roleToChannel[deliveryRole]; - if (!deliveryChannel) { - const missingRole = deliveryRole === 'arbiter' ? 'arbiter' : 'reviewer'; - const errorMessage = getMissingRoleChannelMessage(missingRole); - markWorkItemDeliveryRetry(openWorkItem.id, errorMessage); - log.error( - { - workItemId: openWorkItem.id, - role: deliveryRole, - requiredChannel: getFixedRoleChannelName(missingRole), - }, - 'Unable to deliver paired-room work item because the dedicated Discord role channel is not configured', - ); - return false; - } - const delivered = await deliverOpenWorkItem( - deliveryChannel, - openWorkItem, - ); - if (!delivered) return false; - enqueueGenericFollowUpAfterDeliveryRetry({ + : null, + channel, + roleToChannel, + log, + isPairedRoom: hasReviewerLease(chatJid), + getMissingRoleChannelMessage, + isDuplicateOfLastBotFinal: checkDuplicateOfLastBotFinal, + openContinuation: (targetChatJid) => + continuationTracker.open(targetChatJid), + enqueueFollowUpAfterDeliveryRetry: ({ deliveryRole, pendingTask, - workItemId: openWorkItem.id, - }); + workItemId, + }) => + enqueueDeliveryRetryFollowUp({ + chatJid, + runId, + deliveryRole, + pendingTask, + 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') { + return false; + } + if (openWorkItemOutcome === 'delivered') { return true; } @@ -815,38 +539,26 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { ); if (missedMessages.length === 0) { - // Check if a paired review is pending — run reviewer even without new messages - const pendingReviewTask = hasReviewerLease(chatJid) - ? getLatestOpenPairedTaskForChat(chatJid) - : null; - const pendingTurn = pendingReviewTask - ? buildPendingPairedTurn({ - chatJid, - timezone: deps.timezone, - task: pendingReviewTask, - rawMissedMessages, - recentHumanMessages: getRecentChatMessages(chatJid, 20).filter( - (message) => !message.is_bot_message, - ), - labeledRecentMessages: labelPairedSenders( - chatJid, - getRecentChatMessages(chatJid, 20), - ), - resolveChannel, - }) - : null; - if (pendingTurn) { - return executePendingPairedTurn({ - pendingTurn, - chatJid, - group, - runId, - log, - saveState: deps.saveState, - lastAgentTimestamps: deps.getLastAgentTimestamps(), - executeTurn, - getFixedRoleChannelName, - }); + const pendingTurnOutcome = await runPendingPairedTurnIfNeeded({ + chatJid, + group, + runId, + log, + timezone: deps.timezone, + task: hasReviewerLease(chatJid) + ? getLatestOpenPairedTaskForChat(chatJid) + : null, + rawMissedMessages, + saveState: deps.saveState, + lastAgentTimestamps: deps.getLastAgentTimestamps(), + executeTurn, + getFixedRoleChannelName, + roleToChannel, + labelPairedSenders, + mode: 'idle', + }); + if (pendingTurnOutcome !== null) { + return pendingTurnOutcome; } const lastIgnored = rawMissedMessages[rawMissedMessages.length - 1]; @@ -861,38 +573,27 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { return true; } - const botOnlyPendingTask = hasReviewerLease(chatJid) - ? getLatestOpenPairedTaskForChat(chatJid) - : null; - const botOnlyPendingTurn = - botOnlyPendingTask && isBotOnlyPairedRoomTurn(chatJid, missedMessages) - ? buildPendingPairedTurn({ - chatJid, - timezone: deps.timezone, - task: botOnlyPendingTask, - rawMissedMessages, - recentHumanMessages: getRecentChatMessages(chatJid, 20).filter( - (message) => !message.is_bot_message, - ), - labeledRecentMessages: labelPairedSenders( - chatJid, - getRecentChatMessages(chatJid, 20), - ), - resolveChannel, - }) - : null; - if (botOnlyPendingTurn) { - return executePendingPairedTurn({ - pendingTurn: botOnlyPendingTurn, - chatJid, - group, - runId, - log, - saveState: deps.saveState, - lastAgentTimestamps: deps.getLastAgentTimestamps(), - executeTurn, - getFixedRoleChannelName, - }); + const botOnlyPendingTurnOutcome = await runPendingPairedTurnIfNeeded({ + chatJid, + group, + runId, + log, + timezone: deps.timezone, + task: hasReviewerLease(chatJid) + ? getLatestOpenPairedTaskForChat(chatJid) + : null, + rawMissedMessages, + saveState: deps.saveState, + lastAgentTimestamps: deps.getLastAgentTimestamps(), + executeTurn, + getFixedRoleChannelName, + roleToChannel, + labelPairedSenders, + mode: 'bot-only', + missedMessages, + }); + if (botOnlyPendingTurnOutcome !== null) { + return botOnlyPendingTurnOutcome; } if (shouldSkipBotOnlyCollaboration(chatJid, missedMessages)) { @@ -911,14 +612,15 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { return true; } - const cmdResult = await handleSessionCommand({ - missedMessages, - isMainGroup, - groupName: group.name, + const gateResult = await handleQueuedRunGates({ + chatJid, + group, runId, + missedMessages, triggerPattern: deps.triggerPattern, timezone: deps.timezone, - deps: { + hasImplicitContinuationWindow: continuationTracker.has, + sessionCommandDeps: { sendMessage: (text) => channel.sendMessage(chatJid, text), setTyping: (typing) => channel.setTyping?.(chatJid, typing) ?? Promise.resolve(), @@ -971,320 +673,30 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { killProcess: () => deps.queue.killProcess(chatJid), }, }); - if (cmdResult.handled) return cmdResult.success; - - if ( - !hasAllowedTrigger({ - chatJid, - messages: missedMessages, - group, - triggerPattern: deps.triggerPattern, - hasImplicitContinuationWindow: continuationTracker.has, - }) - ) { - log.info('Skipping queued run because no allowed trigger was found'); - return true; + if (gateResult.handled) { + return gateResult.success; } - // Determine role BEFORE advancing cursor — paired rooms use - // separate cursors for owner and reviewer so neither misses - // the other's messages. - const pendingTaskForChannel = hasReviewerLease(chatJid) - ? getLatestOpenPairedTaskForChat(chatJid) - : null; - const taskStatus = pendingTaskForChannel?.status; - const hasHumanMsg = !isBotOnlyPairedRoomTurn(chatJid, missedMessages); - const turnRole = pendingTaskForChannel - ? resolveQueuedTurnRole({ - taskStatus, - hasHumanMessage: hasHumanMsg, - }) - : 'owner'; - const turnChannel = resolveChannelForRole(turnRole); - const cursorKey = resolveCursorKeyForRole(chatJid, turnRole); - const forcedRole = - pendingTaskForChannel && turnRole !== resolveActiveRole(taskStatus) - ? turnRole - : undefined; - - // Arbiter turns use a dedicated context prompt; regular turns use formatted messages. - let prompt: string; - if (turnRole === 'arbiter' && pendingTaskForChannel) { - const recentMessages = getRecentChatMessages(chatJid, 20); - prompt = buildArbiterPromptForTask({ - task: pendingTaskForChannel, - chatJid, - timezone: deps.timezone, - turnOutputs: getPairedTurnOutputs(pendingTaskForChannel.id), - recentMessages, - labeledRecentMessages: labelPairedSenders(chatJid, recentMessages), - }); - } else if (pendingTaskForChannel) { - prompt = buildPairedTurnPrompt({ - taskId: pendingTaskForChannel.id, - chatJid, - timezone: deps.timezone, - missedMessages, - labeledFallbackMessages: labelPairedSenders(chatJid, missedMessages), - turnOutputs: getPairedTurnOutputs(pendingTaskForChannel.id), - }); - } else { - prompt = formatMessages( - labelPairedSenders(chatJid, missedMessages), - deps.timezone, - ); - } - const startSeq = missedMessages[0].seq ?? null; - const endSeq = missedMessages[missedMessages.length - 1].seq ?? null; - log.info( - { - messageCount: missedMessages.length, - messageSeqStart: startSeq, - messageSeqEnd: endSeq, - }, - 'Dispatching queued messages to agent', - ); - - if (!turnChannel) { - const missingRole = turnRole === 'arbiter' ? 'arbiter' : 'reviewer'; - log.error( - { - taskStatus, - role: turnRole, - requiredChannel: getFixedRoleChannelName(missingRole), - }, - 'Skipping paired-room run because the dedicated Discord role channel is not configured', - ); - return false; - } - - if (endSeq !== null) { - advanceLastAgentCursor( - deps.getLastAgentTimestamps(), - deps.saveState, - chatJid, - endSeq, - cursorKey, - ); - } - - const { deliverySucceeded, visiblePhase } = await executeTurn({ - group, - prompt, + return runQueuedGroupTurn({ chatJid, + group, runId, - channel: turnChannel, - deliveryRole: pendingTaskForChannel ? turnRole : undefined, - startSeq, - endSeq, - hasHumanMessage: hasHumanMsg, - forcedRole, - }); - - if (!deliverySucceeded) { - log.warn( - { - messageSeqStart: startSeq, - messageSeqEnd: endSeq, - }, - 'Persisted produced output for delivery retry without rerunning agent', - ); - return false; - } - - log.info( - { - visiblePhase, - messageSeqStart: startSeq, - messageSeqEnd: endSeq, - }, - 'Queued run completed successfully', - ); - - return true; - } - }; - - const processQueuedGroupDispatch = async (args: { - chatJid: string; - group: RegisteredGroup; - channel: Channel; - processableGroupMessages: NewMessage[]; - }): Promise => { - const { chatJid, group, channel, processableGroupMessages } = args; - const loopPendingTask = hasReviewerLease(chatJid) - ? getLatestOpenPairedTaskForChat(chatJid) - : null; - const loopCursorKey = resolveCursorKey(chatJid, loopPendingTask?.status); - const rawPendingMessages = getMessagesSinceSeq( - chatJid, - deps.getLastAgentTimestamps()[loopCursorKey] || '0', - deps.assistantName, - ); - const pendingMessages = filterLoopingPairedBotMessages( - chatJid, - getProcessableMessages(chatJid, rawPendingMessages, channel), - FAILURE_FINAL_TEXT, - ); - const messagesToSend = - pendingMessages.length > 0 ? pendingMessages : processableGroupMessages; - const labeledMessagesToSend = labelPairedSenders(chatJid, messagesToSend); - const { - formatted, - botOnlyFollowUpAction, - isBotOnlyPairedFollowUp, - loopCursorKey: dispatchCursorKey, - endSeq, - } = buildQueuedTurnDispatch({ - chatJid, - timezone: deps.timezone, - loopPendingTask, - rawPendingMessages, - messagesToSend, - labeledMessagesToSend, - formatMessages, - }); - - if ( - await executeBotOnlyPairedFollowUpAction({ - action: botOnlyFollowUpAction, - chatJid, - group, - runId: `loop-merge-ready-${Date.now().toString(36)}`, - channel, - log: logger, - saveState: deps.saveState, + log, + timezone: deps.timezone, + missedMessages, + task: hasReviewerLease(chatJid) + ? getLatestOpenPairedTaskForChat(chatJid) + : null, + roleToChannel, + ownerChannel: channel, lastAgentTimestamps: deps.getLastAgentTimestamps(), + saveState: deps.saveState, executeTurn, - enqueueGroupMessageCheck: () => - enqueueScopedGroupMessageCheck(chatJid, group.folder), - closeStdin: () => - deps.queue.closeStdin(chatJid, { - reason: 'paired-pending-turn-follow-up', - }), - }) - ) { - return; + getFixedRoleChannelName, + labelPairedSenders, + formatMessages, + }); } - - if (deps.queue.sendMessage(chatJid, formatted)) { - if (endSeq != null) { - advanceLastAgentCursor( - deps.getLastAgentTimestamps(), - deps.saveState, - chatJid, - endSeq, - dispatchCursorKey, - ); - } - logger.debug( - { - transition: 'typing:on', - source: 'follow-up-queued', - chatJid, - group: group.name, - groupFolder: group.folder, - endSeq: endSeq ?? null, - suppressed: isBotOnlyPairedFollowUp, - }, - 'Typing indicator transition', - ); - if (!isBotOnlyPairedFollowUp) { - await channel - .setTyping?.(chatJid, true) - ?.catch((err) => - logger.warn({ chatJid, err }, 'Failed to set typing indicator'), - ); - } - return; - } - - enqueueScopedGroupMessageCheck(chatJid, group.folder); - }; - - const processLoopGroupMessages = async (args: { - chatJid: string; - group: RegisteredGroup; - groupMessages: NewMessage[]; - channel: Channel; - }): Promise => { - const { chatJid, group, groupMessages, channel } = args; - const isMainGroup = group.isMain === true; - const processableGroupMessages = getProcessableMessages( - chatJid, - groupMessages, - channel, - ); - - if (processableGroupMessages.length === 0) { - const lastIgnored = groupMessages[groupMessages.length - 1]; - if (lastIgnored?.seq != null) { - advanceLastAgentCursor( - deps.getLastAgentTimestamps(), - deps.saveState, - chatJid, - lastIgnored.seq, - ); - } - return; - } - - if (shouldSkipBotOnlyCollaboration(chatJid, processableGroupMessages)) { - const lastIgnored = - processableGroupMessages[processableGroupMessages.length - 1]; - if (lastIgnored?.seq != null) { - advanceLastAgentCursor( - deps.getLastAgentTimestamps(), - deps.saveState, - chatJid, - lastIgnored.seq, - ); - } - logger.info( - { chatJid, group: group.name, groupFolder: group.folder }, - 'Bot-collaboration timeout: no recent human message, skipping', - ); - return; - } - - const loopCmdMsg = groupMessages.find( - (msg) => extractSessionCommand(msg.content, deps.triggerPattern) !== null, - ); - - if (loopCmdMsg) { - if ( - isSessionCommandAllowed( - isMainGroup, - loopCmdMsg.is_from_me === true, - isSessionCommandSenderAllowed(loopCmdMsg.sender), - ) - ) { - deps.queue.closeStdin(chatJid, { - reason: 'session-command-detected', - }); - } - deps.queue.enqueueMessageCheck(chatJid); - return; - } - - if ( - !hasAllowedTrigger({ - chatJid, - messages: processableGroupMessages, - group, - triggerPattern: deps.triggerPattern, - hasImplicitContinuationWindow: continuationTracker.has, - }) - ) { - return; - } - - await processQueuedGroupDispatch({ - chatJid, - group, - channel, - processableGroupMessages, - }); }; const startMessageLoop = async (): Promise => { @@ -1298,51 +710,29 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { while (true) { try { - enqueuePendingHandoffs(); - const registeredGroups = deps.getRegisteredGroups(); - const jids = Object.keys(registeredGroups); - const { messages, newSeqCursor } = getNewMessagesBySeq( - jids, - deps.getLastTimestamp(), - deps.assistantName, - ); - - if (messages.length > 0) { - logger.info({ count: messages.length }, 'New messages'); - - deps.setLastTimestamp(newSeqCursor); - deps.saveState(); - - const messagesByGroup = new Map(); - for (const msg of messages) { - const existing = messagesByGroup.get(msg.chat_jid); - if (existing) { - existing.push(msg); - } else { - messagesByGroup.set(msg.chat_jid, [msg]); - } - } - - for (const [chatJid, groupMessages] of messagesByGroup) { - const group = registeredGroups[chatJid]; - if (!group) continue; - - const channel = findChannel(deps.channels, chatJid); - if (!channel) { - logger.warn( - { chatJid }, - 'No channel owns JID, skipping messages', - ); - continue; - } - await processLoopGroupMessages({ - chatJid, - group, - groupMessages, - channel, - }); - } - } + await processMessageLoopTick({ + assistantName: deps.assistantName, + failureFinalText: FAILURE_FINAL_TEXT, + triggerPattern: deps.triggerPattern, + timezone: deps.timezone, + channels: deps.channels, + getRegisteredGroups: deps.getRegisteredGroups, + getLastTimestamp: deps.getLastTimestamp, + setLastTimestamp: deps.setLastTimestamp, + lastAgentTimestamps: deps.getLastAgentTimestamps(), + saveState: deps.saveState, + hasImplicitContinuationWindow: continuationTracker.has, + executeTurn, + enqueuePendingHandoffs, + scheduleQueuedPairedFollowUp, + enqueueScopedGroupMessageCheck, + sendQueuedMessage: deps.queue.sendMessage.bind(deps.queue), + closeStdin: (chatJid, reason) => + deps.queue.closeStdin(chatJid, { + reason, + }), + labelPairedSenders, + }); } catch (err) { logger.error({ err }, 'Error in message loop'); } @@ -1351,49 +741,14 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { }; const recoverPendingMessages = (): void => { - const registeredGroups = deps.getRegisteredGroups(); - for (const [chatJid, group] of Object.entries(registeredGroups)) { - const openWorkItem = getOpenWorkItemForChat(chatJid); - if (openWorkItem) { - logger.info( - { chatJid, group: group.name, workItemId: openWorkItem.id }, - 'Recovery: found open work item awaiting delivery', - ); - enqueueScopedGroupMessageCheck(chatJid, group.folder); - continue; - } - - const sinceSeqCursor = deps.getLastAgentTimestamps()[chatJid] || ''; - const rawPending = getMessagesSinceSeq( - chatJid, - sinceSeqCursor, - deps.assistantName, - ); - const recoveryChannel = findChannel(deps.channels, chatJid); - const pending = getProcessableMessages( - chatJid, - rawPending, - recoveryChannel ?? undefined, - ); - if (pending.length > 0) { - logger.info( - { group: group.name, pendingCount: pending.length }, - 'Recovery: found unprocessed messages', - ); - enqueueScopedGroupMessageCheck(chatJid, group.folder); - continue; - } else if (rawPending.length > 0) { - const endSeq = rawPending[rawPending.length - 1].seq; - if (endSeq != null) { - advanceLastAgentCursor( - deps.getLastAgentTimestamps(), - deps.saveState, - chatJid, - endSeq, - ); - } - } - } + recoverRuntimePendingMessages({ + assistantName: deps.assistantName, + channels: deps.channels, + getRegisteredGroups: deps.getRegisteredGroups, + lastAgentTimestamps: deps.getLastAgentTimestamps(), + saveState: deps.saveState, + enqueueScopedGroupMessageCheck, + }); }; return { diff --git a/src/paired-execution-context-owner.ts b/src/paired-execution-context-owner.ts index 2694a1f..f26eedd 100644 --- a/src/paired-execution-context-owner.ts +++ b/src/paired-execution-context-owner.ts @@ -7,9 +7,10 @@ import { logger } from './logger.js'; import { markPairedTaskReviewReady } from './paired-workspace-manager.js'; import { applyPairedTaskPatch, - classifyVerdict, hasCodeChangesSinceRef, + parseVisibleVerdict, requestArbiterOrEscalate, + resolveOwnerCompletionSignal, resolveCanonicalSourceRef, transitionPairedTaskStatus, } from './paired-execution-context-shared.js'; @@ -44,93 +45,68 @@ function handleOwnerFinalizeCompletion(args: { now: string; }): OwnerFinalizeOutcome { const { task, taskId, summary, now } = args; - const ownerVerdict = classifyVerdict(summary); + const ownerVerdict = parseVisibleVerdict(summary); + const workspace = getPairedWorkspace(task.id, 'owner'); + const hasNewChanges = workspace?.workspace_dir + ? hasCodeChangesSinceRef(workspace.workspace_dir, task.source_ref) + : null; + const signal = resolveOwnerCompletionSignal({ + phase: 'finalize', + visibleVerdict: ownerVerdict, + hasChangesSinceApproval: hasNewChanges, + roundTripCount: task.round_trip_count, + deadlockThreshold: ARBITER_DEADLOCK_THRESHOLD, + }); - if (ownerVerdict === 'blocked' || ownerVerdict === 'needs_context') { + if (signal.kind === 'request_arbiter') { + const arbiterLogMessage = + ownerVerdict === 'blocked' || ownerVerdict === 'needs_context' + ? 'Owner blocked during finalize — requesting arbiter' + : ownerVerdict === 'done_with_concerns' + ? 'Owner finalize loop detected — requesting arbiter' + : 'Owner finalize DONE loop detected — requesting arbiter'; + const escalateLogMessage = + ownerVerdict === 'blocked' || ownerVerdict === 'needs_context' + ? 'Owner blocked during finalize — escalating to user' + : ownerVerdict === 'done_with_concerns' + ? 'Owner finalize loop detected — escalating to user' + : 'Owner finalize DONE loop detected — escalating to user'; requestArbiterOrEscalate({ taskId, currentStatus: task.status, now, - arbiterLogMessage: 'Owner blocked during finalize — requesting arbiter', - escalateLogMessage: 'Owner blocked during finalize — escalating to user', + arbiterLogMessage, + escalateLogMessage, logContext: { taskId, ownerVerdict, + roundTrips: task.round_trip_count, + hasNewChanges, summary: summary?.slice(0, 100), }, }); return 'stop'; } - if (ownerVerdict === 'done_with_concerns') { - if (task.round_trip_count >= ARBITER_DEADLOCK_THRESHOLD) { - requestArbiterOrEscalate({ + if (signal.kind === 'request_reviewer') { + if (signal.resetStatusToActive) { + transitionPairedTaskStatus({ taskId, currentStatus: task.status, - now, - arbiterLogMessage: 'Owner finalize loop detected — requesting arbiter', - escalateLogMessage: 'Owner finalize loop detected — escalating to user', - logContext: { - taskId, - ownerVerdict, - roundTrips: task.round_trip_count, - }, + nextStatus: 'active', + updatedAt: now, }); - return 'stop'; } - transitionPairedTaskStatus({ - taskId, - currentStatus: task.status, - nextStatus: 'active', - updatedAt: now, - }); logger.info( { taskId, ownerVerdict, + hasNewChanges, summary: summary?.slice(0, 100), }, - 'Owner raised concerns during finalize — task set back to active', - ); - return 're_review'; - } - - const workspace = getPairedWorkspace(task.id, 'owner'); - const hasNewChanges = workspace?.workspace_dir - ? hasCodeChangesSinceRef(workspace.workspace_dir, task.source_ref) - : null; - - if (hasNewChanges === true) { - if (task.round_trip_count >= ARBITER_DEADLOCK_THRESHOLD) { - requestArbiterOrEscalate({ - taskId, - currentStatus: task.status, - now, - arbiterLogMessage: - 'Owner finalize DONE loop detected — requesting arbiter', - escalateLogMessage: - 'Owner finalize DONE loop detected — escalating to user', - logContext: { - taskId, - roundTrips: task.round_trip_count, - hasNewChanges, - }, - }); - return 'stop'; - } - transitionPairedTaskStatus({ - taskId, - currentStatus: task.status, - nextStatus: 'active', - updatedAt: now, - }); - logger.info( - { - taskId, - sourceRef: task.source_ref, - hasNewChanges, - }, - 'Owner made changes after reviewer approval — task set back to active before re-review', + ownerVerdict === 'done_with_concerns' + ? 'Owner raised concerns during finalize — task set back to active' + : 'Owner made changes after reviewer approval — task set back to active before re-review', ); return 're_review'; } @@ -219,8 +195,13 @@ export function handleOwnerCompletion(args: { return; } - const ownerVerdict = classifyVerdict(summary); - if (ownerVerdict === 'blocked' || ownerVerdict === 'needs_context') { + const ownerVerdict = parseVisibleVerdict(summary); + const signal = resolveOwnerCompletionSignal({ + phase: 'normal', + visibleVerdict: ownerVerdict, + }); + + if (signal.kind === 'request_arbiter') { requestArbiterOrEscalate({ taskId, currentStatus: task.status, diff --git a/src/paired-execution-context-reviewer.ts b/src/paired-execution-context-reviewer.ts index b3dc496..39b5602 100644 --- a/src/paired-execution-context-reviewer.ts +++ b/src/paired-execution-context-reviewer.ts @@ -2,8 +2,10 @@ import { ARBITER_DEADLOCK_THRESHOLD } from './config.js'; import { getPairedWorkspace, updatePairedTask } from './db.js'; import { logger } from './logger.js'; import { - classifyVerdict, + parseVisibleVerdict, requestArbiterOrEscalate, + resolveReviewerCompletionSignal, + resolveReviewerFailureSignal, resolveCanonicalSourceRef, transitionPairedTaskStatus, } from './paired-execution-context-shared.js'; @@ -18,26 +20,37 @@ export function handleFailedReviewerExecution(args: { const now = new Date().toISOString(); if (summary) { - const verdict = classifyVerdict(summary); + const verdict = parseVisibleVerdict(summary); + const signal = resolveReviewerFailureSignal({ + visibleVerdict: verdict, + }); if ( - verdict === 'done' || - verdict === 'blocked' || - verdict === 'needs_context' + signal.kind === 'request_owner_finalize' || + signal.kind === 'complete' ) { const ownerWs = - verdict === 'done' ? getPairedWorkspace(taskId, 'owner') : null; + signal.kind === 'request_owner_finalize' + ? getPairedWorkspace(taskId, 'owner') + : null; const approvedSourceRef = - verdict === 'done' && ownerWs?.workspace_dir + signal.kind === 'request_owner_finalize' && ownerWs?.workspace_dir ? resolveCanonicalSourceRef(ownerWs.workspace_dir) : task.source_ref; transitionPairedTaskStatus({ taskId, currentStatus: task.status, - nextStatus: verdict === 'done' ? 'merge_ready' : 'completed', + nextStatus: + signal.kind === 'request_owner_finalize' + ? 'merge_ready' + : 'completed', updatedAt: now, patch: { - ...(verdict === 'done' ? { source_ref: approvedSourceRef } : {}), - ...(verdict !== 'done' ? { completion_reason: 'escalated' } : {}), + ...(signal.kind === 'request_owner_finalize' + ? { source_ref: approvedSourceRef } + : {}), + ...(signal.kind === 'complete' + ? { completion_reason: signal.completionReason } + : {}), }, }); logger.info( @@ -83,10 +96,15 @@ export function handleReviewerCompletion(args: { }): void { const { task, taskId, summary } = args; const now = new Date().toISOString(); - const verdict = classifyVerdict(summary); + const verdict = parseVisibleVerdict(summary); + const signal = resolveReviewerCompletionSignal({ + visibleVerdict: verdict, + roundTripCount: task.round_trip_count, + deadlockThreshold: ARBITER_DEADLOCK_THRESHOLD, + }); - switch (verdict) { - case 'done': { + switch (signal.kind) { + case 'request_owner_finalize': { const ownerWs = getPairedWorkspace(taskId, 'owner'); const approvedSourceRef = ownerWs?.workspace_dir ? resolveCanonicalSourceRef(ownerWs.workspace_dir) @@ -112,39 +130,26 @@ export function handleReviewerCompletion(args: { return; } - case 'blocked': - case 'needs_context': + case 'request_arbiter': + const arbiterLogMessage = + verdict === 'blocked' || verdict === 'needs_context' + ? 'Reviewer blocked/needs_context — requesting arbiter before escalating' + : 'Deadlock detected — requesting arbiter intervention'; + const escalateLogMessage = + verdict === 'blocked' || verdict === 'needs_context' + ? 'Reviewer escalated to user — ping-pong stopped' + : 'Stopped ping-pong — escalating to user (arbiter not configured)'; requestArbiterOrEscalate({ taskId, currentStatus: task.status, now, - arbiterLogMessage: - 'Reviewer blocked/needs_context — requesting arbiter before escalating', - escalateLogMessage: 'Reviewer escalated to user — ping-pong stopped', + arbiterLogMessage, + escalateLogMessage, logContext: { taskId, verdict, summary: summary?.slice(0, 100) }, }); return; - case 'done_with_concerns': - case 'continue': - default: - if (task.round_trip_count >= ARBITER_DEADLOCK_THRESHOLD) { - requestArbiterOrEscalate({ - taskId, - currentStatus: task.status, - now, - arbiterLogMessage: - 'Deadlock detected — requesting arbiter intervention', - escalateLogMessage: - 'Stopped ping-pong — escalating to user (arbiter not configured)', - logContext: { - taskId, - verdict, - roundTrips: task.round_trip_count, - }, - }); - return; - } + case 'request_owner_changes': transitionPairedTaskStatus({ taskId, currentStatus: task.status, @@ -156,5 +161,7 @@ export function handleReviewerCompletion(args: { 'Reviewer has feedback, task set back to active for owner', ); return; + default: + return; } } diff --git a/src/paired-execution-context-shared.test.ts b/src/paired-execution-context-shared.test.ts new file mode 100644 index 0000000..b19568b --- /dev/null +++ b/src/paired-execution-context-shared.test.ts @@ -0,0 +1,113 @@ +import { describe, expect, it } from 'vitest'; + +import { + parseVisibleVerdict, + resolveOwnerCompletionSignal, + resolveReviewerCompletionSignal, + resolveReviewerFailureSignal, +} from './paired-execution-context-shared.js'; + +describe('paired execution context shared verdict helpers', () => { + it('parses visible verdicts from the first summary line only', () => { + expect( + parseVisibleVerdict( + 'DONE_WITH_CONCERNS\n\nfollow-up detail that should not affect parsing', + ), + ).toBe('done_with_concerns'); + expect(parseVisibleVerdict('BLOCKED\nextra detail')).toBe('blocked'); + expect(parseVisibleVerdict('random prose')).toBe('continue'); + }); + + it('maps normal owner completion verdicts to reviewer or arbiter signals', () => { + expect( + resolveOwnerCompletionSignal({ + phase: 'normal', + visibleVerdict: 'done', + }), + ).toEqual({ + kind: 'request_reviewer', + resetStatusToActive: false, + }); + expect( + resolveOwnerCompletionSignal({ + phase: 'normal', + visibleVerdict: 'blocked', + }), + ).toEqual({ kind: 'request_arbiter' }); + }); + + it('maps finalize owner outcomes to complete, re-review, or arbiter', () => { + expect( + resolveOwnerCompletionSignal({ + phase: 'finalize', + visibleVerdict: 'done', + hasChangesSinceApproval: false, + roundTripCount: 0, + deadlockThreshold: 2, + }), + ).toEqual({ kind: 'complete', completionReason: 'done' }); + expect( + resolveOwnerCompletionSignal({ + phase: 'finalize', + visibleVerdict: 'done_with_concerns', + hasChangesSinceApproval: false, + roundTripCount: 1, + deadlockThreshold: 3, + }), + ).toEqual({ + kind: 'request_reviewer', + resetStatusToActive: true, + }); + expect( + resolveOwnerCompletionSignal({ + phase: 'finalize', + visibleVerdict: 'done', + hasChangesSinceApproval: true, + roundTripCount: 3, + deadlockThreshold: 3, + }), + ).toEqual({ kind: 'request_arbiter' }); + }); + + it('maps reviewer completion verdicts to finalize, owner changes, or arbiter', () => { + expect( + resolveReviewerCompletionSignal({ + visibleVerdict: 'done', + roundTripCount: 0, + deadlockThreshold: 3, + }), + ).toEqual({ kind: 'request_owner_finalize' }); + expect( + resolveReviewerCompletionSignal({ + visibleVerdict: 'continue', + roundTripCount: 1, + deadlockThreshold: 3, + }), + ).toEqual({ kind: 'request_owner_changes' }); + expect( + resolveReviewerCompletionSignal({ + visibleVerdict: 'done_with_concerns', + roundTripCount: 3, + deadlockThreshold: 3, + }), + ).toEqual({ kind: 'request_arbiter' }); + }); + + it('maps reviewer failure verdicts to explicit failure signals', () => { + expect( + resolveReviewerFailureSignal({ + visibleVerdict: 'done', + }), + ).toEqual({ kind: 'request_owner_finalize' }); + expect( + resolveReviewerFailureSignal({ + visibleVerdict: 'needs_context', + }), + ).toEqual({ kind: 'complete', completionReason: 'escalated' }); + expect( + resolveReviewerFailureSignal({ + visibleVerdict: 'continue', + }), + ).toEqual({ kind: 'preserve_review_ready' }); + }); +}); diff --git a/src/paired-execution-context-shared.ts b/src/paired-execution-context-shared.ts index fa8328c..aa0e87b 100644 --- a/src/paired-execution-context-shared.ts +++ b/src/paired-execution-context-shared.ts @@ -5,14 +5,24 @@ import { updatePairedTask } from './db.js'; import { logger } from './logger.js'; import type { PairedTaskStatus } from './types.js'; -export type Verdict = +export type VisibleVerdict = | 'done' | 'done_with_concerns' | 'blocked' | 'needs_context' | 'continue'; -export function classifyVerdict(summary: string | null | undefined): Verdict { +export type CompletionSignal = + | { kind: 'request_reviewer'; resetStatusToActive: boolean } + | { kind: 'request_owner_finalize' } + | { kind: 'request_owner_changes' } + | { kind: 'request_arbiter' } + | { kind: 'complete'; completionReason: 'done' | 'escalated' } + | { kind: 'preserve_review_ready' }; + +export function parseVisibleVerdict( + summary: string | null | undefined, +): VisibleVerdict { if (!summary) return 'continue'; const cleaned = summary.replace(/[\s\S]*?<\/internal>/g, '').trim(); if (!cleaned) return 'continue'; @@ -27,6 +37,95 @@ export function classifyVerdict(summary: string | null | undefined): Verdict { return 'continue'; } +export function resolveOwnerCompletionSignal(args: { + phase: 'normal' | 'finalize'; + visibleVerdict: VisibleVerdict; + hasChangesSinceApproval?: boolean | null; + roundTripCount?: number; + deadlockThreshold?: number; +}): CompletionSignal { + const { + phase, + visibleVerdict, + hasChangesSinceApproval = false, + roundTripCount = 0, + deadlockThreshold = Number.POSITIVE_INFINITY, + } = args; + + if (visibleVerdict === 'blocked' || visibleVerdict === 'needs_context') { + return { kind: 'request_arbiter' }; + } + + if (phase === 'normal') { + return { + kind: 'request_reviewer', + resetStatusToActive: false, + }; + } + + const needsReReview = + visibleVerdict === 'done_with_concerns' || hasChangesSinceApproval === true; + + if (needsReReview) { + if (roundTripCount >= deadlockThreshold) { + return { kind: 'request_arbiter' }; + } + return { + kind: 'request_reviewer', + resetStatusToActive: true, + }; + } + + return { + kind: 'complete', + completionReason: 'done', + }; +} + +export function resolveReviewerCompletionSignal(args: { + visibleVerdict: VisibleVerdict; + roundTripCount: number; + deadlockThreshold: number; +}): CompletionSignal { + const { visibleVerdict, roundTripCount, deadlockThreshold } = args; + + switch (visibleVerdict) { + case 'done': + return { kind: 'request_owner_finalize' }; + case 'blocked': + case 'needs_context': + return { kind: 'request_arbiter' }; + case 'done_with_concerns': + case 'continue': + default: + if (roundTripCount >= deadlockThreshold) { + return { kind: 'request_arbiter' }; + } + return { kind: 'request_owner_changes' }; + } +} + +export function resolveReviewerFailureSignal(args: { + visibleVerdict: VisibleVerdict; +}): CompletionSignal { + const { visibleVerdict } = args; + + switch (visibleVerdict) { + case 'done': + return { kind: 'request_owner_finalize' }; + case 'blocked': + case 'needs_context': + return { + kind: 'complete', + completionReason: 'escalated', + }; + case 'done_with_concerns': + case 'continue': + default: + return { kind: 'preserve_review_ready' }; + } +} + export type ArbiterVerdictResult = | 'proceed' | 'revise' diff --git a/src/paired-follow-up-scheduler.test.ts b/src/paired-follow-up-scheduler.test.ts new file mode 100644 index 0000000..65fb0f1 --- /dev/null +++ b/src/paired-follow-up-scheduler.test.ts @@ -0,0 +1,119 @@ +import { beforeEach, describe, expect, it, vi } from 'vitest'; + +import { + SCHEDULED_PAIRED_FOLLOW_UP_TTL_MS, + buildPairedFollowUpKey, + resetPairedFollowUpScheduleState, + schedulePairedFollowUpOnce, +} from './paired-follow-up-scheduler.js'; + +describe('paired follow-up scheduler', () => { + beforeEach(() => { + resetPairedFollowUpScheduleState(); + vi.useRealTimers(); + }); + + it('deduplicates the same follow-up intent within one run', () => { + const enqueue = vi.fn(); + const task = { + id: 'task-1', + status: 'review_ready', + round_trip_count: 1, + } as const; + + const first = schedulePairedFollowUpOnce({ + chatJid: 'group@test', + runId: 'run-1', + task, + intentKind: 'reviewer-turn', + enqueue, + }); + const second = schedulePairedFollowUpOnce({ + chatJid: 'group@test', + runId: 'run-1', + task, + intentKind: 'reviewer-turn', + enqueue, + }); + + expect(first).toBe(true); + expect(second).toBe(false); + expect(enqueue).toHaveBeenCalledTimes(1); + }); + + it('keeps different round trips schedulable', () => { + const enqueue = vi.fn(); + + const first = schedulePairedFollowUpOnce({ + chatJid: 'group@test', + runId: 'run-1', + task: { + id: 'task-1', + status: 'review_ready', + round_trip_count: 1, + } as const, + intentKind: 'reviewer-turn', + enqueue, + }); + const second = schedulePairedFollowUpOnce({ + chatJid: 'group@test', + runId: 'run-1', + task: { + id: 'task-1', + status: 'review_ready', + round_trip_count: 2, + } as const, + intentKind: 'reviewer-turn', + enqueue, + }); + + expect(first).toBe(true); + expect(second).toBe(true); + expect(enqueue).toHaveBeenCalledTimes(2); + }); + + it('builds a key that includes round trip count and intent', () => { + expect( + buildPairedFollowUpKey({ + taskId: 'task-1', + taskStatus: 'review_ready', + roundTripCount: 3, + intentKind: 'reviewer-turn', + }), + ).toBe('task-1:review_ready:3:reviewer-turn'); + }); + + 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(); + const task = { + id: 'task-1', + status: 'review_ready', + round_trip_count: 1, + } as const; + + const first = schedulePairedFollowUpOnce({ + chatJid: 'group@test', + runId: 'run-1', + task, + intentKind: 'reviewer-turn', + enqueue, + }); + + vi.advanceTimersByTime(SCHEDULED_PAIRED_FOLLOW_UP_TTL_MS + 1); + + const second = schedulePairedFollowUpOnce({ + chatJid: 'group@test', + runId: 'run-1', + task, + intentKind: 'reviewer-turn', + enqueue, + }); + + expect(first).toBe(true); + expect(second).toBe(true); + expect(enqueue).toHaveBeenCalledTimes(2); + }); +}); diff --git a/src/paired-follow-up-scheduler.ts b/src/paired-follow-up-scheduler.ts new file mode 100644 index 0000000..11b4574 --- /dev/null +++ b/src/paired-follow-up-scheduler.ts @@ -0,0 +1,71 @@ +import type { PairedTask, PairedTaskStatus } from './types.js'; + +export type ScheduledPairedFollowUpIntentKind = + | 'reviewer-turn' + | 'arbiter-turn' + | 'owner-follow-up' + | 'finalize-owner-turn'; + +type ScheduledPairedFollowUpTask = Pick< + PairedTask, + 'id' | 'status' | 'round_trip_count' +>; + +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; + roundTripCount: number; + intentKind: ScheduledPairedFollowUpIntentKind; +}): string { + return [ + args.taskId, + args.taskStatus ?? 'unknown', + String(args.roundTripCount), + args.intentKind, + ].join(':'); +} + +export function schedulePairedFollowUpOnce(args: { + chatJid: string; + runId: string; + task: ScheduledPairedFollowUpTask; + intentKind: ScheduledPairedFollowUpIntentKind; + enqueue: () => void; +}): boolean { + const now = Date.now(); + pruneExpiredScheduledPairedFollowUps(now); + + const key = [ + args.chatJid, + args.runId, + buildPairedFollowUpKey({ + taskId: args.task.id, + taskStatus: args.task.status, + roundTripCount: args.task.round_trip_count, + intentKind: args.intentKind, + }), + ].join(':'); + + if (scheduledPairedFollowUps.has(key)) { + return false; + } + + scheduledPairedFollowUps.set(key, now); + args.enqueue(); + return true; +} + +export function resetPairedFollowUpScheduleState(): void { + scheduledPairedFollowUps.clear(); +}