Merge pull request #177 from phj1081/codex/message-runtime-quality
refactor: split message runtime processing
This commit is contained in:
Eyejoker
@@ -155,10 +155,6 @@
|
||||
"maxNesting": 1,
|
||||
"owner": "message runtime test hotspot"
|
||||
},
|
||||
"src/message-runtime.ts": {
|
||||
"maxFunctionLines": 620,
|
||||
"owner": "message runtime hotspot"
|
||||
},
|
||||
"src/message-turn-controller.test.ts": {
|
||||
"maxFunctionLines": 730,
|
||||
"owner": "message turn controller test hotspot"
|
||||
|
||||
499
src/message-runtime-group-processing.ts
Normal file
499
src/message-runtime-group-processing.ts
Normal file
@@ -0,0 +1,499 @@
|
||||
import {
|
||||
getLatestOpenPairedTaskForChat,
|
||||
getMessagesSinceSeq,
|
||||
getOpenWorkItemForChat,
|
||||
markWorkItemDelivered,
|
||||
type WorkItem,
|
||||
} from './db.js';
|
||||
import {
|
||||
isSessionCommandSenderAllowed,
|
||||
SERVICE_SESSION_SCOPE,
|
||||
} from './config.js';
|
||||
import { resolveRuntimeAttachmentBaseDirs } from './attachment-base-dirs.js';
|
||||
import { enqueueGenericFollowUpAfterDeliveryRetry as enqueueDeliveryRetryFollowUp } from './message-runtime-dispatch.js';
|
||||
import { processOpenWorkItemDelivery } from './message-runtime-delivery.js';
|
||||
import { handleQueuedRunGates } from './message-runtime-gating.js';
|
||||
import {
|
||||
runPendingPairedTurnIfNeeded,
|
||||
runQueuedGroupTurn,
|
||||
} from './message-runtime-queue.js';
|
||||
import {
|
||||
advanceLastAgentCursor,
|
||||
filterLoopingPairedBotMessages,
|
||||
getProcessableMessages,
|
||||
shouldSkipBotOnlyCollaboration,
|
||||
} from './message-runtime-rules.js';
|
||||
import { resolveOwnerTaskForHumanMessage } from './paired-execution-context.js';
|
||||
import { transitionPairedTaskStatus } from './paired-task-status.js';
|
||||
import { hasReviewerLease } from './service-routing.js';
|
||||
import {
|
||||
getFixedRoleChannelName,
|
||||
getMissingRoleChannelMessage,
|
||||
} from './message-runtime-shared.js';
|
||||
import { resolvePairedRoleChannels } from './message-runtime-role-channels.js';
|
||||
import {
|
||||
getFreshHumanPreflightMessages,
|
||||
hasHumanMessageAfterWorkItem,
|
||||
} from './message-runtime-preflight-messages.js';
|
||||
import { deliverCanonicalOutboundMessage } from './ipc-outbound-delivery.js';
|
||||
import { findChannel, formatMessages } from './router.js';
|
||||
import { createScopedLogger, logger } from './logger.js';
|
||||
import type { AgentOutput } from './agent-runner.js';
|
||||
import type { GroupQueue, GroupRunContext } from './group-queue.js';
|
||||
import type {
|
||||
ExecuteTurnFn,
|
||||
RoleToChannelMap,
|
||||
} from './message-runtime-types.js';
|
||||
import type {
|
||||
Channel,
|
||||
NewMessage,
|
||||
PairedTask,
|
||||
RegisteredGroup,
|
||||
} from './types.js';
|
||||
|
||||
type RunAgentFn = (
|
||||
group: RegisteredGroup,
|
||||
prompt: string,
|
||||
chatJid: string,
|
||||
runId: string,
|
||||
onOutput?: (output: AgentOutput) => Promise<void>,
|
||||
) => Promise<'success' | 'error'>;
|
||||
|
||||
interface ProcessGroupMessagesDeps {
|
||||
assistantName: string;
|
||||
failureFinalText: string;
|
||||
timezone: string;
|
||||
triggerPattern: RegExp;
|
||||
channels: Channel[];
|
||||
queue: Pick<GroupQueue, 'enqueueMessageCheck' | 'closeStdin' | 'killProcess'>;
|
||||
getRoomBindings: () => Record<string, RegisteredGroup>;
|
||||
getLastAgentTimestamps: () => Record<string, string>;
|
||||
saveState: () => void;
|
||||
clearSession: (groupFolder: string, opts?: { allRoles?: boolean }) => void;
|
||||
runAgent: RunAgentFn;
|
||||
executeTurn: ExecuteTurnFn;
|
||||
hasImplicitContinuationWindow: (
|
||||
chatJid: string,
|
||||
messages: NewMessage[],
|
||||
) => boolean;
|
||||
openContinuation: (chatJid: string) => void;
|
||||
isDuplicateOfLastBotFinal: (chatJid: string, text: string) => boolean;
|
||||
labelPairedSenders: (chatJid: string, messages: NewMessage[]) => NewMessage[];
|
||||
}
|
||||
|
||||
interface RuntimeContext {
|
||||
chatJid: string;
|
||||
runId: string;
|
||||
group: RegisteredGroup;
|
||||
channel: Channel;
|
||||
log: ReturnType<typeof createScopedLogger>;
|
||||
roleToChannel: RoleToChannelMap;
|
||||
}
|
||||
|
||||
interface PendingDeliveryState {
|
||||
pendingTask: PairedTask | null;
|
||||
openWorkItem: WorkItem | undefined;
|
||||
}
|
||||
|
||||
export function createProcessGroupMessages(
|
||||
args: ProcessGroupMessagesDeps,
|
||||
): (chatJid: string, context: GroupRunContext) => Promise<boolean> {
|
||||
return async (chatJid, context): Promise<boolean> => {
|
||||
const runtime = resolveRuntimeContext(args, chatJid, context.runId);
|
||||
if (!runtime) return true;
|
||||
|
||||
const pendingDelivery = resolvePendingDeliveryState(args, runtime);
|
||||
const deliveryOutcome = await deliverPendingWorkItem(
|
||||
args,
|
||||
runtime,
|
||||
pendingDelivery,
|
||||
);
|
||||
if (deliveryOutcome !== null) {
|
||||
return deliveryOutcome;
|
||||
}
|
||||
|
||||
return processMissedMessages(args, runtime);
|
||||
};
|
||||
}
|
||||
|
||||
function resolveRuntimeContext(
|
||||
args: ProcessGroupMessagesDeps,
|
||||
chatJid: string,
|
||||
runId: string,
|
||||
): RuntimeContext | null {
|
||||
const group = args.getRoomBindings()[chatJid];
|
||||
if (!group) return null;
|
||||
const log = createScopedLogger({
|
||||
chatJid,
|
||||
groupName: group.name,
|
||||
groupFolder: group.folder,
|
||||
runId,
|
||||
});
|
||||
|
||||
const channel = findChannel(args.channels, chatJid);
|
||||
if (!channel) {
|
||||
log.warn('No channel owns JID, skipping messages');
|
||||
return null;
|
||||
}
|
||||
|
||||
const {
|
||||
roleToChannel,
|
||||
reviewerChannelName,
|
||||
foundReviewerChannel,
|
||||
arbiterChannelName,
|
||||
foundArbiterChannel,
|
||||
} = resolvePairedRoleChannels(args.channels, channel);
|
||||
if (hasReviewerLease(chatJid)) {
|
||||
log.info(
|
||||
{
|
||||
reviewerChannelName,
|
||||
foundChannel: foundReviewerChannel?.name ?? null,
|
||||
arbiterChannelName,
|
||||
foundArbiterChannel: foundArbiterChannel?.name ?? null,
|
||||
availableChannels: args.channels.map((c) => c.name),
|
||||
},
|
||||
'Paired room reviewer/arbiter channel resolution',
|
||||
);
|
||||
}
|
||||
|
||||
return { chatJid, runId, group, channel, log, roleToChannel };
|
||||
}
|
||||
|
||||
function resolvePendingDeliveryState(
|
||||
args: ProcessGroupMessagesDeps,
|
||||
runtime: RuntimeContext,
|
||||
): PendingDeliveryState {
|
||||
const { chatJid, channel, group, log } = runtime;
|
||||
let pendingTask: PairedTask | null = hasReviewerLease(chatJid)
|
||||
? (getLatestOpenPairedTaskForChat(chatJid) ?? null)
|
||||
: null;
|
||||
let openWorkItem = getOpenWorkItemForChat(chatJid, SERVICE_SESSION_SCOPE);
|
||||
if (openWorkItem?.delivery_role !== 'owner' || !pendingTask) {
|
||||
return { pendingTask, openWorkItem };
|
||||
}
|
||||
|
||||
const freshHumanMessages = getFreshHumanPreflightMessages({
|
||||
chatJid,
|
||||
channel,
|
||||
lastAgentTimestamps: args.getLastAgentTimestamps(),
|
||||
assistantName: args.assistantName,
|
||||
failureFinalText: args.failureFinalText,
|
||||
});
|
||||
if (pendingTask.status === 'merge_ready' && freshHumanMessages.length > 0) {
|
||||
const resolvedTask = resolveOwnerTaskForHumanMessage({
|
||||
group,
|
||||
chatJid,
|
||||
existingTask: pendingTask,
|
||||
});
|
||||
pendingTask = resolvedTask.task ?? null;
|
||||
if (resolvedTask.supersededTask) {
|
||||
markWorkItemDelivered(openWorkItem.id);
|
||||
log.info(
|
||||
{
|
||||
chatJid,
|
||||
workItemId: openWorkItem.id,
|
||||
supersededTaskId: resolvedTask.supersededTask.id,
|
||||
replacementTaskId: resolvedTask.task?.id ?? null,
|
||||
},
|
||||
'Suppressed stale owner delivery retry because a new human message superseded the merge_ready task',
|
||||
);
|
||||
openWorkItem = undefined;
|
||||
}
|
||||
} else if (
|
||||
pendingTask.status === 'active' &&
|
||||
hasHumanMessageAfterWorkItem(openWorkItem, freshHumanMessages)
|
||||
) {
|
||||
markWorkItemDelivered(openWorkItem.id);
|
||||
log.info(
|
||||
{
|
||||
chatJid,
|
||||
workItemId: openWorkItem.id,
|
||||
taskId: pendingTask.id,
|
||||
taskStatus: pendingTask.status,
|
||||
workItemEndSeq: openWorkItem.end_seq ?? null,
|
||||
freshHumanMessageCount: freshHumanMessages.length,
|
||||
},
|
||||
'Suppressed stale owner delivery retry because a new human message arrived while the paired task was still active',
|
||||
);
|
||||
openWorkItem = undefined;
|
||||
}
|
||||
|
||||
return { pendingTask, openWorkItem };
|
||||
}
|
||||
|
||||
async function deliverPendingWorkItem(
|
||||
args: ProcessGroupMessagesDeps,
|
||||
runtime: RuntimeContext,
|
||||
pendingDelivery: PendingDeliveryState,
|
||||
): Promise<boolean | null> {
|
||||
const { chatJid, runId, group, channel, roleToChannel, log } = runtime;
|
||||
const openWorkItemOutcome = await processOpenWorkItemDelivery({
|
||||
chatJid,
|
||||
runId,
|
||||
openWorkItem: pendingDelivery.openWorkItem,
|
||||
pendingTask: pendingDelivery.pendingTask,
|
||||
channel,
|
||||
roleToChannel,
|
||||
log,
|
||||
attachmentBaseDirs: resolveRuntimeAttachmentBaseDirs(group),
|
||||
isPairedRoom: hasReviewerLease(chatJid),
|
||||
getMissingRoleChannelMessage,
|
||||
isDuplicateOfLastBotFinal: args.isDuplicateOfLastBotFinal,
|
||||
openContinuation: args.openContinuation,
|
||||
enqueueFollowUpAfterDeliveryRetry: ({
|
||||
deliveryRole,
|
||||
pendingTask,
|
||||
workItemId,
|
||||
}) =>
|
||||
enqueueDeliveryRetryFollowUp({
|
||||
chatJid,
|
||||
runId,
|
||||
deliveryRole,
|
||||
pendingTask,
|
||||
workItemId,
|
||||
log,
|
||||
enqueueMessageCheck: () => args.queue.enqueueMessageCheck(chatJid),
|
||||
}),
|
||||
});
|
||||
if (openWorkItemOutcome === 'failed') {
|
||||
return false;
|
||||
}
|
||||
if (openWorkItemOutcome === 'delivered') {
|
||||
return true;
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
async function processMissedMessages(
|
||||
args: ProcessGroupMessagesDeps,
|
||||
runtime: RuntimeContext,
|
||||
): Promise<boolean> {
|
||||
while (true) {
|
||||
const rawMissedMessages = getMessagesSinceSeq(
|
||||
runtime.chatJid,
|
||||
args.getLastAgentTimestamps()[runtime.chatJid] || '0',
|
||||
args.assistantName,
|
||||
);
|
||||
const missedMessages = filterLoopingPairedBotMessages(
|
||||
runtime.chatJid,
|
||||
getProcessableMessages(
|
||||
runtime.chatJid,
|
||||
rawMissedMessages,
|
||||
runtime.channel,
|
||||
),
|
||||
args.failureFinalText,
|
||||
);
|
||||
|
||||
if (missedMessages.length === 0) {
|
||||
return handleNoMissedMessages(args, runtime, rawMissedMessages);
|
||||
}
|
||||
|
||||
const pendingTurnOutcome = await runBotOnlyPendingTurn(
|
||||
args,
|
||||
runtime,
|
||||
rawMissedMessages,
|
||||
missedMessages,
|
||||
);
|
||||
if (pendingTurnOutcome !== null) {
|
||||
return pendingTurnOutcome;
|
||||
}
|
||||
|
||||
if (shouldSkipBotOnlyCollaboration(runtime.chatJid, missedMessages)) {
|
||||
advancePastBotOnlyCollaboration(args, runtime, missedMessages);
|
||||
return true;
|
||||
}
|
||||
|
||||
const gateOutcome = await runQueuedRunGates(args, runtime, missedMessages);
|
||||
if (gateOutcome !== null) {
|
||||
return gateOutcome;
|
||||
}
|
||||
|
||||
return runQueuedGroupTurn({
|
||||
chatJid: runtime.chatJid,
|
||||
group: runtime.group,
|
||||
runId: runtime.runId,
|
||||
log: runtime.log,
|
||||
timezone: args.timezone,
|
||||
missedMessages,
|
||||
task: hasReviewerLease(runtime.chatJid)
|
||||
? (getLatestOpenPairedTaskForChat(runtime.chatJid) ?? null)
|
||||
: undefined,
|
||||
roleToChannel: runtime.roleToChannel,
|
||||
ownerChannel: runtime.channel,
|
||||
lastAgentTimestamps: args.getLastAgentTimestamps(),
|
||||
saveState: args.saveState,
|
||||
executeTurn: args.executeTurn,
|
||||
getFixedRoleChannelName,
|
||||
labelPairedSenders: args.labelPairedSenders,
|
||||
formatMessages,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
async function handleNoMissedMessages(
|
||||
args: ProcessGroupMessagesDeps,
|
||||
runtime: RuntimeContext,
|
||||
rawMissedMessages: NewMessage[],
|
||||
): Promise<boolean> {
|
||||
const pendingTurnOutcome = await runPendingPairedTurnIfNeeded({
|
||||
chatJid: runtime.chatJid,
|
||||
group: runtime.group,
|
||||
runId: runtime.runId,
|
||||
log: runtime.log,
|
||||
timezone: args.timezone,
|
||||
task: hasReviewerLease(runtime.chatJid)
|
||||
? getLatestOpenPairedTaskForChat(runtime.chatJid)
|
||||
: null,
|
||||
rawMissedMessages,
|
||||
saveState: args.saveState,
|
||||
lastAgentTimestamps: args.getLastAgentTimestamps(),
|
||||
executeTurn: args.executeTurn,
|
||||
getFixedRoleChannelName,
|
||||
roleToChannel: runtime.roleToChannel,
|
||||
labelPairedSenders: args.labelPairedSenders,
|
||||
mode: 'idle',
|
||||
});
|
||||
if (pendingTurnOutcome !== null) {
|
||||
return pendingTurnOutcome;
|
||||
}
|
||||
|
||||
const lastIgnored = rawMissedMessages[rawMissedMessages.length - 1];
|
||||
if (lastIgnored) {
|
||||
advanceLastAgentCursor(
|
||||
args.getLastAgentTimestamps(),
|
||||
args.saveState,
|
||||
runtime.chatJid,
|
||||
lastIgnored.timestamp,
|
||||
);
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
function runBotOnlyPendingTurn(
|
||||
args: ProcessGroupMessagesDeps,
|
||||
runtime: RuntimeContext,
|
||||
rawMissedMessages: NewMessage[],
|
||||
missedMessages: NewMessage[],
|
||||
): Promise<boolean | null> {
|
||||
return runPendingPairedTurnIfNeeded({
|
||||
chatJid: runtime.chatJid,
|
||||
group: runtime.group,
|
||||
runId: runtime.runId,
|
||||
log: runtime.log,
|
||||
timezone: args.timezone,
|
||||
task: hasReviewerLease(runtime.chatJid)
|
||||
? getLatestOpenPairedTaskForChat(runtime.chatJid)
|
||||
: null,
|
||||
rawMissedMessages,
|
||||
saveState: args.saveState,
|
||||
lastAgentTimestamps: args.getLastAgentTimestamps(),
|
||||
executeTurn: args.executeTurn,
|
||||
getFixedRoleChannelName,
|
||||
roleToChannel: runtime.roleToChannel,
|
||||
labelPairedSenders: args.labelPairedSenders,
|
||||
mode: 'bot-only',
|
||||
missedMessages,
|
||||
});
|
||||
}
|
||||
|
||||
function advancePastBotOnlyCollaboration(
|
||||
args: ProcessGroupMessagesDeps,
|
||||
runtime: RuntimeContext,
|
||||
missedMessages: NewMessage[],
|
||||
): void {
|
||||
const lastMessage = missedMessages[missedMessages.length - 1];
|
||||
if (lastMessage?.seq != null) {
|
||||
advanceLastAgentCursor(
|
||||
args.getLastAgentTimestamps(),
|
||||
args.saveState,
|
||||
runtime.chatJid,
|
||||
lastMessage.seq,
|
||||
);
|
||||
}
|
||||
runtime.log.info(
|
||||
'Skipping bot-only collaboration because no recent human message exists',
|
||||
);
|
||||
}
|
||||
|
||||
async function runQueuedRunGates(
|
||||
args: ProcessGroupMessagesDeps,
|
||||
runtime: RuntimeContext,
|
||||
missedMessages: NewMessage[],
|
||||
): Promise<boolean | null> {
|
||||
const gateResult = await handleQueuedRunGates({
|
||||
chatJid: runtime.chatJid,
|
||||
group: runtime.group,
|
||||
runId: runtime.runId,
|
||||
missedMessages,
|
||||
triggerPattern: args.triggerPattern,
|
||||
timezone: args.timezone,
|
||||
hasImplicitContinuationWindow: args.hasImplicitContinuationWindow,
|
||||
sessionCommandDeps: {
|
||||
sendMessage: (text) =>
|
||||
deliverSessionCommandMessage(args, runtime.chatJid, text),
|
||||
setTyping: (typing) =>
|
||||
runtime.channel.setTyping?.(runtime.chatJid, typing) ??
|
||||
Promise.resolve(),
|
||||
runAgent: (prompt, onOutput) =>
|
||||
args.runAgent(
|
||||
runtime.group,
|
||||
prompt,
|
||||
runtime.chatJid,
|
||||
runtime.runId,
|
||||
onOutput,
|
||||
),
|
||||
closeStdin: () =>
|
||||
args.queue.closeStdin(runtime.chatJid, {
|
||||
reason: 'session-command',
|
||||
}),
|
||||
clearSession: (opts) => args.clearSession(runtime.group.folder, opts),
|
||||
advanceCursor: (cursorOrTimestamp) => {
|
||||
advanceLastAgentCursor(
|
||||
args.getLastAgentTimestamps(),
|
||||
args.saveState,
|
||||
runtime.chatJid,
|
||||
cursorOrTimestamp,
|
||||
);
|
||||
},
|
||||
formatMessages,
|
||||
isAdminSender: (msg) => isSessionCommandSenderAllowed(msg.sender),
|
||||
canSenderInteract: () => true,
|
||||
resetPairedTask: () => completePairedTaskAsStopped(runtime.chatJid),
|
||||
killProcess: () => args.queue.killProcess(runtime.chatJid),
|
||||
},
|
||||
});
|
||||
return gateResult.handled ? gateResult.success : null;
|
||||
}
|
||||
|
||||
function completePairedTaskAsStopped(chatJid: string): void {
|
||||
if (!hasReviewerLease(chatJid)) return;
|
||||
const task = getLatestOpenPairedTaskForChat(chatJid);
|
||||
if (!task) return;
|
||||
|
||||
const now = new Date().toISOString();
|
||||
transitionPairedTaskStatus({
|
||||
taskId: task.id,
|
||||
currentStatus: task.status,
|
||||
nextStatus: 'completed',
|
||||
expectedUpdatedAt: task.updated_at,
|
||||
updatedAt: now,
|
||||
patch: {
|
||||
completion_reason: 'stopped',
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
async function deliverSessionCommandMessage(
|
||||
deps: Pick<ProcessGroupMessagesDeps, 'channels' | 'getRoomBindings'>,
|
||||
chatJid: string,
|
||||
text: string,
|
||||
): Promise<void> {
|
||||
await deliverCanonicalOutboundMessage(
|
||||
{ jid: chatJid, text },
|
||||
{
|
||||
channels: deps.channels,
|
||||
roomBindings: deps.getRoomBindings,
|
||||
log: logger,
|
||||
},
|
||||
);
|
||||
}
|
||||
@@ -1,61 +1,24 @@
|
||||
import {
|
||||
getOpenWorkItemForChat,
|
||||
getMessagesSinceSeq,
|
||||
getLatestOpenPairedTaskForChat,
|
||||
markWorkItemDelivered,
|
||||
updatePairedTurnProgressText,
|
||||
} from './db.js';
|
||||
import {
|
||||
isSessionCommandSenderAllowed,
|
||||
SERVICE_SESSION_SCOPE,
|
||||
} from './config.js';
|
||||
import { resolveRuntimeAttachmentBaseDirs } from './attachment-base-dirs.js';
|
||||
import { updatePairedTurnProgressText } from './db.js';
|
||||
import { GroupQueue, GroupRunContext } from './group-queue.js';
|
||||
import { findChannel, formatMessages } from './router.js';
|
||||
import { enqueueGenericFollowUpAfterDeliveryRetry as enqueueDeliveryRetryFollowUp } from './message-runtime-dispatch.js';
|
||||
import { processOpenWorkItemDelivery } from './message-runtime-delivery.js';
|
||||
import { deliverCanonicalOutboundMessage } from './ipc-outbound-delivery.js';
|
||||
import { handleQueuedRunGates } from './message-runtime-gating.js';
|
||||
import { enqueueMessageRuntimePendingHandoffs } 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,
|
||||
filterLoopingPairedBotMessages,
|
||||
getProcessableMessages,
|
||||
shouldSkipBotOnlyCollaboration,
|
||||
} from './message-runtime-rules.js';
|
||||
import { resolveOwnerTaskForHumanMessage } from './paired-execution-context.js';
|
||||
import { createImplicitContinuationTracker } from './message-runtime-rules.js';
|
||||
import { schedulePairedFollowUpWithMessageCheck } from './message-runtime-follow-up.js';
|
||||
import { transitionPairedTaskStatus } from './paired-task-status.js';
|
||||
import { Channel, NewMessage, RegisteredGroup } from './types.js';
|
||||
import { createScopedLogger, logger } from './logger.js';
|
||||
import { hasReviewerLease } from './service-routing.js';
|
||||
import {
|
||||
getFixedRoleChannelName,
|
||||
getMissingRoleChannelMessage,
|
||||
} from './message-runtime-shared.js';
|
||||
import { logger } from './logger.js';
|
||||
import {
|
||||
createRunAgent,
|
||||
createExecuteTurn,
|
||||
isDuplicateOfLastBotFinal,
|
||||
labelPairedSenders,
|
||||
} from './message-runtime-turns.js';
|
||||
import { resolvePairedRoleChannels } from './message-runtime-role-channels.js';
|
||||
import {
|
||||
getFreshHumanPreflightMessages,
|
||||
hasHumanMessageAfterWorkItem,
|
||||
} from './message-runtime-preflight-messages.js';
|
||||
import { handleMessageRuntimeAfterDeliverySuccess } from './message-runtime-after-delivery.js';
|
||||
import { deliverMessageRuntimeFinalText } from './message-runtime-final-delivery.js';
|
||||
import { createProcessGroupMessages } from './message-runtime-group-processing.js';
|
||||
|
||||
export { isDuplicateOfLastBotFinal };
|
||||
|
||||
@@ -77,19 +40,30 @@ export interface MessageRuntimeDeps {
|
||||
clearSession: (groupFolder: string, opts?: { allRoles?: boolean }) => void;
|
||||
}
|
||||
|
||||
async function deliverSessionCommandMessage(
|
||||
deps: Pick<MessageRuntimeDeps, 'channels' | 'getRoomBindings'>,
|
||||
chatJid: string,
|
||||
text: string,
|
||||
): Promise<void> {
|
||||
await deliverCanonicalOutboundMessage(
|
||||
{ jid: chatJid, text },
|
||||
{
|
||||
channels: deps.channels,
|
||||
roomBindings: deps.getRoomBindings,
|
||||
log: logger,
|
||||
},
|
||||
);
|
||||
function createStartMessageLoop(args: {
|
||||
pollInterval: number;
|
||||
processTick: () => Promise<void>;
|
||||
}): () => Promise<void> {
|
||||
let messageLoopRunning = false;
|
||||
|
||||
return async (): Promise<void> => {
|
||||
if (messageLoopRunning) {
|
||||
logger.debug('Message loop already running, skipping duplicate start');
|
||||
return;
|
||||
}
|
||||
messageLoopRunning = true;
|
||||
|
||||
logger.info('EJClaw running');
|
||||
|
||||
while (true) {
|
||||
try {
|
||||
await args.processTick();
|
||||
} catch (err) {
|
||||
logger.error({ err }, 'Error in message loop');
|
||||
}
|
||||
await new Promise((resolve) => setTimeout(resolve, args.pollInterval));
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
export function createMessageRuntime(deps: MessageRuntimeDeps): {
|
||||
@@ -100,7 +74,6 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): {
|
||||
recoverPendingMessages: () => void;
|
||||
startMessageLoop: () => Promise<void>;
|
||||
} {
|
||||
let messageLoopRunning = false;
|
||||
const FAILURE_FINAL_TEXT = '요청을 완료하지 못했습니다. 다시 시도해 주세요.';
|
||||
const continuationTracker = createImplicitContinuationTracker(
|
||||
deps.idleTimeout,
|
||||
@@ -113,10 +86,6 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): {
|
||||
deps.queue,
|
||||
);
|
||||
|
||||
/**
|
||||
* Check if a message is a duplicate of the last bot final message in a paired room.
|
||||
* Returns true if duplicate (should be suppressed).
|
||||
*/
|
||||
const checkDuplicateOfLastBotFinal = (
|
||||
chatJid: string,
|
||||
text: string,
|
||||
@@ -208,314 +177,28 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): {
|
||||
});
|
||||
};
|
||||
|
||||
const processGroupMessages = async (
|
||||
chatJid: string,
|
||||
context: GroupRunContext,
|
||||
): Promise<boolean> => {
|
||||
const { runId } = context;
|
||||
const group = deps.getRoomBindings()[chatJid];
|
||||
if (!group) return true;
|
||||
const log = createScopedLogger({
|
||||
chatJid,
|
||||
groupName: group.name,
|
||||
groupFolder: group.folder,
|
||||
runId,
|
||||
});
|
||||
|
||||
const channel = findChannel(deps.channels, chatJid);
|
||||
if (!channel) {
|
||||
log.warn('No channel owns JID, skipping messages');
|
||||
return true;
|
||||
}
|
||||
|
||||
const {
|
||||
roleToChannel,
|
||||
reviewerChannelName,
|
||||
foundReviewerChannel,
|
||||
arbiterChannelName,
|
||||
foundArbiterChannel,
|
||||
} = resolvePairedRoleChannels(deps.channels, channel);
|
||||
if (hasReviewerLease(chatJid)) {
|
||||
log.info(
|
||||
{
|
||||
reviewerChannelName,
|
||||
foundChannel: foundReviewerChannel?.name ?? null,
|
||||
arbiterChannelName,
|
||||
foundArbiterChannel: foundArbiterChannel?.name ?? null,
|
||||
availableChannels: deps.channels.map((c) => c.name),
|
||||
},
|
||||
'Paired room reviewer/arbiter channel resolution',
|
||||
);
|
||||
}
|
||||
|
||||
// Delivery retries can come from forced fallback runs whose agent_type
|
||||
// differs from the room owner's registered agent type.
|
||||
let pendingTask = hasReviewerLease(chatJid)
|
||||
? getLatestOpenPairedTaskForChat(chatJid)
|
||||
: null;
|
||||
let openWorkItem = getOpenWorkItemForChat(chatJid, SERVICE_SESSION_SCOPE);
|
||||
if (openWorkItem?.delivery_role === 'owner' && pendingTask) {
|
||||
const freshHumanMessages = getFreshHumanPreflightMessages({
|
||||
chatJid,
|
||||
channel,
|
||||
lastAgentTimestamps: deps.getLastAgentTimestamps(),
|
||||
const processGroupMessages = createProcessGroupMessages({
|
||||
assistantName: deps.assistantName,
|
||||
failureFinalText: FAILURE_FINAL_TEXT,
|
||||
});
|
||||
if (
|
||||
pendingTask.status === 'merge_ready' &&
|
||||
freshHumanMessages.length > 0
|
||||
) {
|
||||
const resolvedTask = resolveOwnerTaskForHumanMessage({
|
||||
group,
|
||||
chatJid,
|
||||
existingTask: pendingTask,
|
||||
});
|
||||
pendingTask = resolvedTask.task;
|
||||
if (resolvedTask.supersededTask) {
|
||||
markWorkItemDelivered(openWorkItem.id);
|
||||
log.info(
|
||||
{
|
||||
chatJid,
|
||||
workItemId: openWorkItem.id,
|
||||
supersededTaskId: resolvedTask.supersededTask.id,
|
||||
replacementTaskId: resolvedTask.task?.id ?? null,
|
||||
},
|
||||
'Suppressed stale owner delivery retry because a new human message superseded the merge_ready task',
|
||||
);
|
||||
openWorkItem = undefined;
|
||||
}
|
||||
} else if (
|
||||
pendingTask.status === 'active' &&
|
||||
hasHumanMessageAfterWorkItem(openWorkItem, freshHumanMessages)
|
||||
) {
|
||||
markWorkItemDelivered(openWorkItem.id);
|
||||
log.info(
|
||||
{
|
||||
chatJid,
|
||||
workItemId: openWorkItem.id,
|
||||
taskId: pendingTask.id,
|
||||
taskStatus: pendingTask.status,
|
||||
workItemEndSeq: openWorkItem.end_seq ?? null,
|
||||
freshHumanMessageCount: freshHumanMessages.length,
|
||||
},
|
||||
'Suppressed stale owner delivery retry because a new human message arrived while the paired task was still active',
|
||||
);
|
||||
openWorkItem = undefined;
|
||||
}
|
||||
}
|
||||
const openWorkItemOutcome = await processOpenWorkItemDelivery({
|
||||
chatJid,
|
||||
runId,
|
||||
openWorkItem,
|
||||
pendingTask,
|
||||
channel,
|
||||
roleToChannel,
|
||||
log,
|
||||
attachmentBaseDirs: resolveRuntimeAttachmentBaseDirs(group),
|
||||
isPairedRoom: hasReviewerLease(chatJid),
|
||||
getMissingRoleChannelMessage,
|
||||
isDuplicateOfLastBotFinal: checkDuplicateOfLastBotFinal,
|
||||
openContinuation: (targetChatJid) =>
|
||||
continuationTracker.open(targetChatJid),
|
||||
enqueueFollowUpAfterDeliveryRetry: ({
|
||||
deliveryRole,
|
||||
pendingTask,
|
||||
workItemId,
|
||||
}) =>
|
||||
enqueueDeliveryRetryFollowUp({
|
||||
chatJid,
|
||||
runId,
|
||||
deliveryRole,
|
||||
pendingTask,
|
||||
workItemId,
|
||||
log,
|
||||
enqueueMessageCheck: () => deps.queue.enqueueMessageCheck(chatJid),
|
||||
}),
|
||||
});
|
||||
if (openWorkItemOutcome === 'failed') {
|
||||
return false;
|
||||
}
|
||||
if (openWorkItemOutcome === 'delivered') {
|
||||
return true;
|
||||
}
|
||||
|
||||
while (true) {
|
||||
const sinceSeqCursor = deps.getLastAgentTimestamps()[chatJid] || '0';
|
||||
const rawMissedMessages = getMessagesSinceSeq(
|
||||
chatJid,
|
||||
sinceSeqCursor,
|
||||
deps.assistantName,
|
||||
);
|
||||
const missedMessages = filterLoopingPairedBotMessages(
|
||||
chatJid,
|
||||
getProcessableMessages(chatJid, rawMissedMessages, channel),
|
||||
FAILURE_FINAL_TEXT,
|
||||
);
|
||||
|
||||
if (missedMessages.length === 0) {
|
||||
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: labelPairedRuntimeSenders,
|
||||
mode: 'idle',
|
||||
});
|
||||
if (pendingTurnOutcome !== null) {
|
||||
return pendingTurnOutcome;
|
||||
}
|
||||
|
||||
const lastIgnored = rawMissedMessages[rawMissedMessages.length - 1];
|
||||
if (lastIgnored) {
|
||||
advanceLastAgentCursor(
|
||||
deps.getLastAgentTimestamps(),
|
||||
deps.saveState,
|
||||
chatJid,
|
||||
lastIgnored.timestamp,
|
||||
);
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
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: labelPairedRuntimeSenders,
|
||||
mode: 'bot-only',
|
||||
missedMessages,
|
||||
});
|
||||
if (botOnlyPendingTurnOutcome !== null) {
|
||||
return botOnlyPendingTurnOutcome;
|
||||
}
|
||||
|
||||
if (shouldSkipBotOnlyCollaboration(chatJid, missedMessages)) {
|
||||
const lastMessage = missedMessages[missedMessages.length - 1];
|
||||
if (lastMessage?.seq != null) {
|
||||
advanceLastAgentCursor(
|
||||
deps.getLastAgentTimestamps(),
|
||||
deps.saveState,
|
||||
chatJid,
|
||||
lastMessage.seq,
|
||||
);
|
||||
}
|
||||
log.info(
|
||||
'Skipping bot-only collaboration because no recent human message exists',
|
||||
);
|
||||
return true;
|
||||
}
|
||||
|
||||
const gateResult = await handleQueuedRunGates({
|
||||
chatJid,
|
||||
group,
|
||||
runId,
|
||||
missedMessages,
|
||||
triggerPattern: deps.triggerPattern,
|
||||
timezone: deps.timezone,
|
||||
hasImplicitContinuationWindow: continuationTracker.has,
|
||||
sessionCommandDeps: {
|
||||
sendMessage: (text) =>
|
||||
deliverSessionCommandMessage(deps, chatJid, text),
|
||||
setTyping: (typing) =>
|
||||
channel.setTyping?.(chatJid, typing) ?? Promise.resolve(),
|
||||
runAgent: (prompt, onOutput) =>
|
||||
runAgent(group, prompt, chatJid, runId, onOutput),
|
||||
closeStdin: () =>
|
||||
deps.queue.closeStdin(chatJid, {
|
||||
reason: 'session-command',
|
||||
}),
|
||||
clearSession: (opts) => deps.clearSession(group.folder, opts),
|
||||
advanceCursor: (cursorOrTimestamp) => {
|
||||
advanceLastAgentCursor(
|
||||
deps.getLastAgentTimestamps(),
|
||||
deps.saveState,
|
||||
chatJid,
|
||||
cursorOrTimestamp,
|
||||
);
|
||||
},
|
||||
formatMessages,
|
||||
isAdminSender: (msg) => isSessionCommandSenderAllowed(msg.sender),
|
||||
canSenderInteract: () => true,
|
||||
resetPairedTask: () => {
|
||||
if (hasReviewerLease(chatJid)) {
|
||||
const task = getLatestOpenPairedTaskForChat(chatJid);
|
||||
if (task) {
|
||||
const now = new Date().toISOString();
|
||||
transitionPairedTaskStatus({
|
||||
taskId: task.id,
|
||||
currentStatus: task.status,
|
||||
nextStatus: 'completed',
|
||||
expectedUpdatedAt: task.updated_at,
|
||||
updatedAt: now,
|
||||
patch: {
|
||||
completion_reason: 'stopped',
|
||||
},
|
||||
});
|
||||
}
|
||||
}
|
||||
},
|
||||
killProcess: () => deps.queue.killProcess(chatJid),
|
||||
},
|
||||
});
|
||||
if (gateResult.handled) {
|
||||
return gateResult.success;
|
||||
}
|
||||
|
||||
return runQueuedGroupTurn({
|
||||
chatJid,
|
||||
group,
|
||||
runId,
|
||||
log,
|
||||
timezone: deps.timezone,
|
||||
missedMessages,
|
||||
task: hasReviewerLease(chatJid)
|
||||
? (getLatestOpenPairedTaskForChat(chatJid) ?? null)
|
||||
: undefined,
|
||||
roleToChannel,
|
||||
ownerChannel: channel,
|
||||
lastAgentTimestamps: deps.getLastAgentTimestamps(),
|
||||
channels: deps.channels,
|
||||
queue: deps.queue,
|
||||
getRoomBindings: deps.getRoomBindings,
|
||||
getLastAgentTimestamps: deps.getLastAgentTimestamps,
|
||||
saveState: deps.saveState,
|
||||
clearSession: deps.clearSession,
|
||||
runAgent,
|
||||
executeTurn,
|
||||
getFixedRoleChannelName,
|
||||
hasImplicitContinuationWindow: continuationTracker.has,
|
||||
openContinuation: continuationTracker.open,
|
||||
isDuplicateOfLastBotFinal: checkDuplicateOfLastBotFinal,
|
||||
labelPairedSenders: labelPairedRuntimeSenders,
|
||||
formatMessages,
|
||||
});
|
||||
}
|
||||
};
|
||||
|
||||
const startMessageLoop = async (): Promise<void> => {
|
||||
if (messageLoopRunning) {
|
||||
logger.debug('Message loop already running, skipping duplicate start');
|
||||
return;
|
||||
}
|
||||
messageLoopRunning = true;
|
||||
|
||||
logger.info('EJClaw running');
|
||||
|
||||
while (true) {
|
||||
try {
|
||||
const startMessageLoop = createStartMessageLoop({
|
||||
pollInterval: deps.pollInterval,
|
||||
processTick: async () => {
|
||||
await processMessageLoopTick({
|
||||
assistantName: deps.assistantName,
|
||||
failureFinalText: FAILURE_FINAL_TEXT,
|
||||
@@ -538,16 +221,11 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): {
|
||||
reason,
|
||||
}),
|
||||
isRunningMessageTurn: (chatJid) =>
|
||||
deps.queue.getStatuses([chatJid])[0]?.runPhase ===
|
||||
'running_messages',
|
||||
deps.queue.getStatuses([chatJid])[0]?.runPhase === 'running_messages',
|
||||
labelPairedSenders: labelPairedRuntimeSenders,
|
||||
});
|
||||
} catch (err) {
|
||||
logger.error({ err }, 'Error in message loop');
|
||||
}
|
||||
await new Promise((resolve) => setTimeout(resolve, deps.pollInterval));
|
||||
}
|
||||
};
|
||||
},
|
||||
});
|
||||
|
||||
const recoverPendingMessages = (): void => {
|
||||
recoverRuntimePendingMessages({
|
||||
|
||||
Reference in New Issue
Block a user