Fix merge-ready task supersession handling
This commit is contained in:
@@ -74,6 +74,7 @@ export {
|
||||
} from './db/runtime-tasks-rooms.js';
|
||||
export {
|
||||
_clearPairedTurnReservationsForTests,
|
||||
cancelPairedTurn,
|
||||
claimPairedTurnReservation,
|
||||
claimServiceHandoff,
|
||||
clearChannelOwnerLease,
|
||||
|
||||
@@ -463,6 +463,20 @@ export function failPairedTurnInDatabase(
|
||||
});
|
||||
}
|
||||
|
||||
export function cancelPairedTurnInDatabase(
|
||||
database: Database,
|
||||
args: {
|
||||
turnIdentity: PairedTurnIdentity;
|
||||
error?: string | null;
|
||||
},
|
||||
): void {
|
||||
markPairedTurnTerminalStateInDatabase(database, {
|
||||
turnIdentity: args.turnIdentity,
|
||||
state: 'cancelled',
|
||||
error: args.error,
|
||||
});
|
||||
}
|
||||
|
||||
export function getPairedTurnByIdFromDatabase(
|
||||
database: Database,
|
||||
turnId: string,
|
||||
|
||||
@@ -50,6 +50,7 @@ import {
|
||||
insertPairedTurnOutputInDatabase,
|
||||
} from './paired-turn-outputs.js';
|
||||
import {
|
||||
cancelPairedTurnInDatabase,
|
||||
clearPairedTurnsInDatabase,
|
||||
completePairedTurnInDatabase,
|
||||
failPairedTurnInDatabase,
|
||||
@@ -176,6 +177,13 @@ export function failPairedTurn(args: {
|
||||
failPairedTurnInDatabase(requireDatabase(), args);
|
||||
}
|
||||
|
||||
export function cancelPairedTurn(args: {
|
||||
turnIdentity: PairedTurnIdentity;
|
||||
error?: string | null;
|
||||
}): void {
|
||||
cancelPairedTurnInDatabase(requireDatabase(), args);
|
||||
}
|
||||
|
||||
export function getPairedTurnById(
|
||||
turnId: string,
|
||||
): PairedTurnRecord | undefined {
|
||||
|
||||
@@ -3,8 +3,11 @@ import { beforeEach, describe, expect, it, vi } from 'vitest';
|
||||
import {
|
||||
_initTestDatabase,
|
||||
createPairedTask,
|
||||
getLatestOpenPairedTaskForChat,
|
||||
getPairedTaskById,
|
||||
getPairedTurnById,
|
||||
insertPairedTurnOutput,
|
||||
markPairedTurnRunning,
|
||||
updatePairedTaskIfUnchanged,
|
||||
} from './db.js';
|
||||
import { claimPairedTurnExecution } from './paired-follow-up-scheduler.js';
|
||||
@@ -13,6 +16,7 @@ import {
|
||||
runQueuedGroupTurn,
|
||||
} from './message-runtime-queue.js';
|
||||
import { resetPairedFollowUpScheduleState } from './paired-follow-up-scheduler.js';
|
||||
import { buildPairedTurnIdentity } from './paired-turn-identity.js';
|
||||
import type { Channel, PairedTask, RegisteredGroup } from './types.js';
|
||||
|
||||
function makeGroup(): RegisteredGroup {
|
||||
@@ -23,6 +27,7 @@ function makeGroup(): RegisteredGroup {
|
||||
added_at: '2026-03-30T00:00:00.000Z',
|
||||
requiresTrigger: false,
|
||||
agentType: 'codex',
|
||||
workDir: '/repo',
|
||||
};
|
||||
}
|
||||
|
||||
@@ -343,4 +348,88 @@ describe('message-runtime-queue', () => {
|
||||
}),
|
||||
);
|
||||
});
|
||||
|
||||
it('splits merge_ready owner turns into a fresh task and cancels the stale finalize turn', async () => {
|
||||
const task = makeTask({
|
||||
id: 'task-merge-ready-owner',
|
||||
status: 'merge_ready',
|
||||
updated_at: '2026-03-30T00:00:00.000Z',
|
||||
});
|
||||
createPairedTask(task);
|
||||
insertPairedTurnOutput(task.id, 1, 'reviewer', 'DONE\nreview finished');
|
||||
const staleFinalizeTurn = buildPairedTurnIdentity({
|
||||
taskId: task.id,
|
||||
taskUpdatedAt: task.updated_at,
|
||||
intentKind: 'finalize-owner-turn',
|
||||
role: 'owner',
|
||||
});
|
||||
markPairedTurnRunning({
|
||||
turnIdentity: staleFinalizeTurn,
|
||||
executorServiceId: 'claude',
|
||||
executorAgentType: 'claude-code',
|
||||
runId: 'run-stale-finalize',
|
||||
});
|
||||
const executeTurn = vi.fn(async () => ({
|
||||
outputStatus: 'success' as const,
|
||||
deliverySucceeded: true,
|
||||
visiblePhase: 'final',
|
||||
}));
|
||||
|
||||
const outcome = await runQueuedGroupTurn({
|
||||
chatJid: task.chat_jid,
|
||||
group: makeGroup(),
|
||||
runId: 'run-owner-after-merge-ready',
|
||||
log: { info: vi.fn(), warn: vi.fn(), error: vi.fn() } as any,
|
||||
timezone: 'UTC',
|
||||
missedMessages: [
|
||||
{
|
||||
id: 'human-merge-ready-1',
|
||||
chat_jid: task.chat_jid,
|
||||
sender: 'user@test',
|
||||
sender_name: 'User',
|
||||
content: '새 질문입니다',
|
||||
timestamp: '2026-03-30T00:00:02.000Z',
|
||||
seq: 47,
|
||||
is_bot_message: false,
|
||||
},
|
||||
],
|
||||
task,
|
||||
roleToChannel: {
|
||||
owner: null,
|
||||
reviewer: makeChannel(),
|
||||
arbiter: null,
|
||||
},
|
||||
ownerChannel: makeChannel(),
|
||||
lastAgentTimestamps: {},
|
||||
saveState: vi.fn(),
|
||||
executeTurn,
|
||||
getFixedRoleChannelName: () => 'discord-review',
|
||||
labelPairedSenders: (_chatJid, messages) => messages,
|
||||
formatMessages: () => 'formatted prompt',
|
||||
});
|
||||
|
||||
expect(outcome).toBe(true);
|
||||
const freshTask = getLatestOpenPairedTaskForChat(task.chat_jid);
|
||||
expect(freshTask).toBeDefined();
|
||||
expect(freshTask?.id).not.toBe(task.id);
|
||||
expect(getPairedTaskById(task.id)).toMatchObject({
|
||||
status: 'completed',
|
||||
completion_reason: 'superseded',
|
||||
});
|
||||
expect(getPairedTurnById(staleFinalizeTurn.turnId)).toMatchObject({
|
||||
state: 'cancelled',
|
||||
});
|
||||
expect(executeTurn).toHaveBeenCalledWith(
|
||||
expect.objectContaining({
|
||||
prompt: expect.not.stringContaining('DONE\nreview finished'),
|
||||
deliveryRole: 'owner',
|
||||
forcedRole: 'owner',
|
||||
pairedTurnIdentity: expect.objectContaining({
|
||||
taskId: freshTask?.id,
|
||||
intentKind: 'owner-turn',
|
||||
role: 'owner',
|
||||
}),
|
||||
}),
|
||||
);
|
||||
});
|
||||
});
|
||||
|
||||
@@ -4,6 +4,7 @@ import {
|
||||
buildArbiterPromptForTask,
|
||||
buildPairedTurnPrompt,
|
||||
} from './message-runtime-prompts.js';
|
||||
import { resolveOwnerTaskForHumanMessage } from './paired-execution-context.js';
|
||||
import {
|
||||
buildPendingPairedTurn,
|
||||
executePendingPairedTurn,
|
||||
@@ -157,16 +158,32 @@ export async function runQueuedGroupTurn(args: {
|
||||
}): Promise<boolean> {
|
||||
const { chatJid, group, runId, log, missedMessages, task, roleToChannel } =
|
||||
args;
|
||||
const taskStatus = task?.status;
|
||||
let currentTask = task;
|
||||
const hasHumanMsg = task
|
||||
? !missedMessages.every(
|
||||
(message) => message.is_from_me === true || !!message.is_bot_message,
|
||||
)
|
||||
: !isBotOnlyPairedRoomTurn(chatJid, missedMessages);
|
||||
const lastTurnOutputRole = task
|
||||
? (getPairedTurnOutputs(task.id).at(-1)?.role ?? null)
|
||||
let fallbackMessages = missedMessages;
|
||||
if (currentTask && hasHumanMsg) {
|
||||
const resolvedTask = resolveOwnerTaskForHumanMessage({
|
||||
group,
|
||||
chatJid,
|
||||
existingTask: currentTask,
|
||||
});
|
||||
currentTask = resolvedTask.task;
|
||||
if (resolvedTask.supersededTask) {
|
||||
fallbackMessages = getRecentChatMessages(chatJid, 20).filter(
|
||||
(message) => !message.is_bot_message,
|
||||
);
|
||||
}
|
||||
}
|
||||
const taskStatus = currentTask?.status;
|
||||
const turnOutputs = currentTask ? getPairedTurnOutputs(currentTask.id) : [];
|
||||
const lastTurnOutputRole = currentTask
|
||||
? (turnOutputs.at(-1)?.role ?? null)
|
||||
: null;
|
||||
const turnRole = task
|
||||
const turnRole = currentTask
|
||||
? hasHumanMsg
|
||||
? resolveQueuedTurnRole({
|
||||
taskStatus,
|
||||
@@ -203,43 +220,46 @@ export async function runQueuedGroupTurn(args: {
|
||||
const turnChannel =
|
||||
turnRole === 'owner' ? args.ownerChannel : roleToChannel[turnRole];
|
||||
const cursorKey = resolveCursorKeyForRole(chatJid, turnRole);
|
||||
const forcedRole = task ? turnRole : undefined;
|
||||
const queuedIntentKind = task
|
||||
const forcedRole = currentTask ? turnRole : undefined;
|
||||
const queuedIntentKind = currentTask
|
||||
? resolveQueuedTurnReservationIntent({
|
||||
task,
|
||||
task: currentTask,
|
||||
turnRole,
|
||||
hasHumanMessage: hasHumanMsg,
|
||||
})
|
||||
: null;
|
||||
const pairedTurnIdentity =
|
||||
task && queuedIntentKind
|
||||
currentTask && queuedIntentKind
|
||||
? buildPairedTurnIdentity({
|
||||
taskId: task.id,
|
||||
taskUpdatedAt: task.updated_at,
|
||||
taskId: currentTask.id,
|
||||
taskUpdatedAt: currentTask.updated_at,
|
||||
intentKind: queuedIntentKind,
|
||||
role: turnRole,
|
||||
})
|
||||
: undefined;
|
||||
|
||||
let prompt: string;
|
||||
if (turnRole === 'arbiter' && task) {
|
||||
if (turnRole === 'arbiter' && currentTask) {
|
||||
const recentMessages = getRecentChatMessages(chatJid, 20);
|
||||
prompt = buildArbiterPromptForTask({
|
||||
task,
|
||||
task: currentTask,
|
||||
chatJid,
|
||||
timezone: args.timezone,
|
||||
turnOutputs: getPairedTurnOutputs(task.id),
|
||||
turnOutputs,
|
||||
recentMessages,
|
||||
labeledRecentMessages: args.labelPairedSenders(chatJid, recentMessages),
|
||||
});
|
||||
} else if (task) {
|
||||
} else if (currentTask) {
|
||||
prompt = buildPairedTurnPrompt({
|
||||
taskId: task.id,
|
||||
taskId: currentTask.id,
|
||||
chatJid,
|
||||
timezone: args.timezone,
|
||||
missedMessages,
|
||||
labeledFallbackMessages: args.labelPairedSenders(chatJid, missedMessages),
|
||||
turnOutputs: getPairedTurnOutputs(task.id),
|
||||
labeledFallbackMessages: args.labelPairedSenders(
|
||||
chatJid,
|
||||
fallbackMessages,
|
||||
),
|
||||
turnOutputs,
|
||||
});
|
||||
} else {
|
||||
prompt = args.formatMessages(
|
||||
@@ -272,19 +292,19 @@ export async function runQueuedGroupTurn(args: {
|
||||
return false;
|
||||
}
|
||||
|
||||
if (task) {
|
||||
if (currentTask) {
|
||||
const claimed = claimPairedTurnExecution({
|
||||
chatJid,
|
||||
runId,
|
||||
task,
|
||||
task: currentTask,
|
||||
intentKind: queuedIntentKind!,
|
||||
});
|
||||
if (!claimed) {
|
||||
log.info(
|
||||
{
|
||||
taskId: task.id,
|
||||
taskId: currentTask.id,
|
||||
taskStatus,
|
||||
taskUpdatedAt: task.updated_at,
|
||||
taskUpdatedAt: currentTask.updated_at,
|
||||
intentKind: queuedIntentKind,
|
||||
turnRole,
|
||||
},
|
||||
@@ -310,7 +330,7 @@ export async function runQueuedGroupTurn(args: {
|
||||
chatJid,
|
||||
runId,
|
||||
channel: turnChannel,
|
||||
deliveryRole: task ? turnRole : undefined,
|
||||
deliveryRole: currentTask ? turnRole : undefined,
|
||||
startSeq,
|
||||
endSeq,
|
||||
hasHumanMessage: hasHumanMsg,
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
import { beforeEach, describe, expect, it, vi } from 'vitest';
|
||||
|
||||
import { TASK_STATUS_MESSAGE_PREFIX } from './task-watch-status.js';
|
||||
import * as pairedExecutionContext from './paired-execution-context.js';
|
||||
|
||||
/** Prefix helper for progress message assertions */
|
||||
const P = (text: string) => `${TASK_STATUS_MESSAGE_PREFIX}${text}`;
|
||||
@@ -34,6 +35,10 @@ vi.mock('./config.js', () => ({
|
||||
vi.mock('./paired-execution-context.js', () => ({
|
||||
preparePairedExecutionContext: vi.fn(() => undefined),
|
||||
completePairedExecutionContext: vi.fn(),
|
||||
resolveOwnerTaskForHumanMessage: vi.fn((args?: { existingTask?: unknown }) => ({
|
||||
task: args?.existingTask ?? null,
|
||||
supersededTask: null,
|
||||
})),
|
||||
}));
|
||||
|
||||
vi.mock('./db.js', () => {
|
||||
@@ -693,6 +698,131 @@ describe('createMessageRuntime', () => {
|
||||
expect(enqueueMessageCheck).toHaveBeenCalledWith(chatJid);
|
||||
});
|
||||
|
||||
it('suppresses a stale owner work item when a new human message supersedes a merge_ready task', async () => {
|
||||
const chatJid = 'group@test';
|
||||
const group = makeGroup('codex');
|
||||
const channel = makeChannel(chatJid);
|
||||
const enqueueMessageCheck = vi.fn();
|
||||
const oldTask = {
|
||||
id: 'task-merge-ready-old',
|
||||
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: '2026-03-30T00:00:00.000Z',
|
||||
round_trip_count: 1,
|
||||
status: 'merge_ready',
|
||||
arbiter_verdict: null,
|
||||
arbiter_requested_at: null,
|
||||
completion_reason: null,
|
||||
created_at: '2026-03-30T00:00:00.000Z',
|
||||
updated_at: '2026-03-30T00:00:00.000Z',
|
||||
} as any;
|
||||
const newTask = {
|
||||
...oldTask,
|
||||
id: 'task-fresh-owner-turn',
|
||||
status: 'active',
|
||||
completion_reason: null,
|
||||
created_at: '2026-03-30T00:00:05.000Z',
|
||||
updated_at: '2026-03-30T00:00:05.000Z',
|
||||
} as any;
|
||||
|
||||
vi.mocked(serviceRouting.hasReviewerLease).mockReturnValue(true);
|
||||
vi.mocked(db.getOpenWorkItem).mockReturnValue({
|
||||
id: 99,
|
||||
group_folder: group.folder,
|
||||
chat_jid: chatJid,
|
||||
agent_type: 'codex',
|
||||
service_id: 'claude',
|
||||
delivery_role: 'owner',
|
||||
status: 'delivery_retry',
|
||||
start_seq: 1,
|
||||
end_seq: 1,
|
||||
result_payload: '이전 final입니다.',
|
||||
delivery_attempts: 1,
|
||||
created_at: new Date().toISOString(),
|
||||
updated_at: new Date().toISOString(),
|
||||
delivered_at: null,
|
||||
delivery_message_id: null,
|
||||
last_error: 'discord send failed',
|
||||
});
|
||||
vi.mocked(db.getMessagesSince).mockReturnValue([
|
||||
{
|
||||
id: 'msg-1',
|
||||
chat_jid: chatJid,
|
||||
sender: 'user@test',
|
||||
sender_name: 'User',
|
||||
content: '새 작업입니다.',
|
||||
timestamp: '2026-03-30T00:00:10.000Z',
|
||||
seq: 10,
|
||||
},
|
||||
]);
|
||||
vi.mocked(db.getLatestOpenPairedTaskForChat)
|
||||
.mockReturnValueOnce(oldTask)
|
||||
.mockReturnValue(newTask);
|
||||
vi.mocked(
|
||||
pairedExecutionContext.resolveOwnerTaskForHumanMessage,
|
||||
).mockReturnValue({
|
||||
task: newTask,
|
||||
supersededTask: oldTask,
|
||||
});
|
||||
vi.mocked(agentRunner.runAgentProcess).mockImplementation(
|
||||
async (_group, input, _onProcess, onOutput) => {
|
||||
expect(input.prompt).toContain('새 작업입니다.');
|
||||
expect(input.prompt).not.toContain('이전 final입니다.');
|
||||
await onOutput?.({
|
||||
status: 'success',
|
||||
phase: 'final',
|
||||
result: 'DONE\n새 작업 처리',
|
||||
output: { visibility: 'public', text: 'DONE\n새 작업 처리' },
|
||||
} as any);
|
||||
return {
|
||||
status: 'success',
|
||||
result: 'DONE\n새 작업 처리',
|
||||
};
|
||||
},
|
||||
);
|
||||
|
||||
const runtime = createMessageRuntime({
|
||||
assistantName: 'Andy',
|
||||
idleTimeout: 1_000,
|
||||
pollInterval: 1_000,
|
||||
timezone: 'UTC',
|
||||
triggerPattern: /^@Andy\b/i,
|
||||
channels: [channel],
|
||||
queue: {
|
||||
registerProcess: vi.fn(),
|
||||
closeStdin: vi.fn(),
|
||||
notifyIdle: vi.fn(),
|
||||
enqueueMessageCheck,
|
||||
} as any,
|
||||
getRoomBindings: () => ({ [chatJid]: group }),
|
||||
getSessions: () => ({}),
|
||||
getLastTimestamp: () => '',
|
||||
setLastTimestamp: vi.fn(),
|
||||
getLastAgentTimestamps: () => ({}),
|
||||
saveState: vi.fn(),
|
||||
persistSession: vi.fn(),
|
||||
clearSession: vi.fn(),
|
||||
});
|
||||
|
||||
const result = await runtime.processGroupMessages(chatJid, {
|
||||
runId: 'run-supersede-owner-work-item',
|
||||
reason: 'messages',
|
||||
});
|
||||
|
||||
expect(result).toBe(true);
|
||||
expect(db.markWorkItemDelivered).toHaveBeenCalledWith(99);
|
||||
expect(agentRunner.runAgentProcess).toHaveBeenCalledTimes(1);
|
||||
expect(channel.sendMessage).not.toHaveBeenCalledWith(
|
||||
chatJid,
|
||||
'이전 final입니다.',
|
||||
);
|
||||
});
|
||||
|
||||
it('suppresses duplicate stale work item delivery and logs the suppression reason', async () => {
|
||||
const chatJid = 'group@test';
|
||||
const group = makeGroup('codex');
|
||||
|
||||
@@ -4,6 +4,7 @@ import {
|
||||
getOpenWorkItemForChat,
|
||||
getMessagesSinceSeq,
|
||||
getLatestOpenPairedTaskForChat,
|
||||
markWorkItemDelivered,
|
||||
getPairedTaskById,
|
||||
} from './db.js';
|
||||
import {
|
||||
@@ -39,6 +40,7 @@ import {
|
||||
getProcessableMessages,
|
||||
shouldSkipBotOnlyCollaboration,
|
||||
} from './message-runtime-rules.js';
|
||||
import { resolveOwnerTaskForHumanMessage } from './paired-execution-context.js';
|
||||
import {
|
||||
enqueuePairedFollowUpAfterEvent,
|
||||
schedulePairedFollowUpWithMessageCheck,
|
||||
@@ -345,14 +347,55 @@ 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, SERVICE_SESSION_SCOPE);
|
||||
let pendingTask = hasReviewerLease(chatJid)
|
||||
? getLatestOpenPairedTaskForChat(chatJid)
|
||||
: null;
|
||||
let openWorkItem = getOpenWorkItemForChat(chatJid, SERVICE_SESSION_SCOPE);
|
||||
if (
|
||||
pendingTask?.status === 'merge_ready' &&
|
||||
openWorkItem?.delivery_role === 'owner'
|
||||
) {
|
||||
const sinceSeqCursor = deps.getLastAgentTimestamps()[chatJid] || '0';
|
||||
const preflightRawMessages = getMessagesSinceSeq(
|
||||
chatJid,
|
||||
sinceSeqCursor,
|
||||
deps.assistantName,
|
||||
);
|
||||
const preflightMessages = filterLoopingPairedBotMessages(
|
||||
chatJid,
|
||||
getProcessableMessages(chatJid, preflightRawMessages, channel),
|
||||
FAILURE_FINAL_TEXT,
|
||||
);
|
||||
const hasFreshHumanMessage = preflightMessages.some(
|
||||
(message) => message.is_from_me !== true && !message.is_bot_message,
|
||||
);
|
||||
if (hasFreshHumanMessage) {
|
||||
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;
|
||||
}
|
||||
}
|
||||
}
|
||||
const openWorkItemOutcome = await processOpenWorkItemDelivery({
|
||||
chatJid,
|
||||
runId,
|
||||
openWorkItem,
|
||||
pendingTask: hasReviewerLease(chatJid)
|
||||
? getLatestOpenPairedTaskForChat(chatJid)
|
||||
: null,
|
||||
pendingTask,
|
||||
channel,
|
||||
roleToChannel,
|
||||
log,
|
||||
|
||||
@@ -12,10 +12,12 @@ import { afterAll, beforeEach, describe, expect, it, vi } from 'vitest';
|
||||
vi.mock('./db.js', () => {
|
||||
const updatePairedTask = vi.fn();
|
||||
return {
|
||||
cancelPairedTurn: vi.fn(),
|
||||
createPairedTask: vi.fn(),
|
||||
getLatestPairedTaskForChat: vi.fn(),
|
||||
getLatestOpenPairedTaskForChat: vi.fn(),
|
||||
getPairedTaskById: vi.fn(),
|
||||
getPairedTurnById: vi.fn(),
|
||||
getPairedWorkspace: vi.fn(),
|
||||
updatePairedTask,
|
||||
updatePairedTaskIfUnchanged: vi.fn((id, _expectedUpdatedAt, updates) => {
|
||||
|
||||
@@ -18,10 +18,12 @@ import {
|
||||
REVIEWER_AGENT_TYPE,
|
||||
} from './config.js';
|
||||
import {
|
||||
cancelPairedTurn,
|
||||
createPairedTask,
|
||||
getLatestPairedTaskForChat,
|
||||
getLatestOpenPairedTaskForChat,
|
||||
getPairedTaskById,
|
||||
getPairedTurnById,
|
||||
getPairedWorkspace,
|
||||
hasActiveCiWatcherForChat,
|
||||
releasePairedTaskExecutionLease,
|
||||
@@ -52,7 +54,10 @@ import {
|
||||
prepareReviewerWorkspaceForExecution,
|
||||
provisionOwnerWorkspaceForPairedTask,
|
||||
} from './paired-workspace-manager.js';
|
||||
import type { PairedTurnIdentity } from './paired-turn-identity.js';
|
||||
import {
|
||||
buildPairedTurnIdentity,
|
||||
type PairedTurnIdentity,
|
||||
} from './paired-turn-identity.js';
|
||||
import type {
|
||||
AgentType,
|
||||
PairedRoomRole,
|
||||
@@ -97,32 +102,12 @@ function resolvePairedTaskServiceShadow(
|
||||
return role === 'owner' ? CODEX_MAIN_SERVICE_ID : CODEX_REVIEW_SERVICE_ID;
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// ensureActiveTask
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
function ensureActiveTask(
|
||||
group: RegisteredGroup,
|
||||
chatJid: string,
|
||||
roomRoleContext: RoomRoleContext,
|
||||
hasHumanMessage?: boolean,
|
||||
): PairedTask | null {
|
||||
const canonicalWorkDir = ensurePairedProject(group, chatJid);
|
||||
if (!canonicalWorkDir) {
|
||||
return null;
|
||||
}
|
||||
|
||||
const existing = getLatestOpenPairedTaskForChat(chatJid);
|
||||
if (existing) {
|
||||
return existing;
|
||||
}
|
||||
|
||||
// Don't create a new task for bot-only messages — prevents
|
||||
// ESCALATE → completed → bot message triggers new task → loop.
|
||||
if (!hasHumanMessage) {
|
||||
return null;
|
||||
}
|
||||
|
||||
function createActiveTaskForRoom(args: {
|
||||
group: RegisteredGroup;
|
||||
chatJid: string;
|
||||
canonicalWorkDir: string;
|
||||
}): PairedTask {
|
||||
const { group, chatJid, canonicalWorkDir } = args;
|
||||
const now = new Date().toISOString();
|
||||
const rolePlan = resolveRoleAgentPlan({
|
||||
paired: true,
|
||||
@@ -172,6 +157,137 @@ function ensureActiveTask(
|
||||
return task;
|
||||
}
|
||||
|
||||
function cancelOutstandingFinalizeOwnerTurn(task: PairedTask): void {
|
||||
const turnIdentity = buildPairedTurnIdentity({
|
||||
taskId: task.id,
|
||||
taskUpdatedAt: task.updated_at,
|
||||
intentKind: 'finalize-owner-turn',
|
||||
role: 'owner',
|
||||
});
|
||||
const existingTurn = getPairedTurnById(turnIdentity.turnId);
|
||||
if (
|
||||
!existingTurn ||
|
||||
(existingTurn.state !== 'running' && existingTurn.state !== 'delegated')
|
||||
) {
|
||||
return;
|
||||
}
|
||||
cancelPairedTurn({
|
||||
turnIdentity,
|
||||
error: 'Superseded by a newer human message before owner finalize delivery.',
|
||||
});
|
||||
logger.info(
|
||||
{
|
||||
taskId: task.id,
|
||||
turnId: turnIdentity.turnId,
|
||||
turnState: existingTurn.state,
|
||||
},
|
||||
'Cancelled stale finalize-owner turn after a new human message superseded the merge_ready task',
|
||||
);
|
||||
}
|
||||
|
||||
export interface ResolvedOwnerHumanTask {
|
||||
task: PairedTask | null;
|
||||
supersededTask: PairedTask | null;
|
||||
}
|
||||
|
||||
export function resolveOwnerTaskForHumanMessage(args: {
|
||||
group: RegisteredGroup;
|
||||
chatJid: string;
|
||||
existingTask?: PairedTask | null;
|
||||
}): ResolvedOwnerHumanTask {
|
||||
const canonicalWorkDir = ensurePairedProject(args.group, args.chatJid);
|
||||
const existing =
|
||||
args.existingTask ?? getLatestOpenPairedTaskForChat(args.chatJid) ?? null;
|
||||
|
||||
if (!existing) {
|
||||
return {
|
||||
task: canonicalWorkDir
|
||||
? createActiveTaskForRoom({
|
||||
group: args.group,
|
||||
chatJid: args.chatJid,
|
||||
canonicalWorkDir,
|
||||
})
|
||||
: null,
|
||||
supersededTask: null,
|
||||
};
|
||||
}
|
||||
|
||||
if (existing.status !== 'merge_ready' || !canonicalWorkDir) {
|
||||
return {
|
||||
task: existing,
|
||||
supersededTask: null,
|
||||
};
|
||||
}
|
||||
|
||||
const now = new Date().toISOString();
|
||||
const superseded = transitionPairedTaskStatus({
|
||||
taskId: existing.id,
|
||||
currentStatus: existing.status,
|
||||
nextStatus: 'completed',
|
||||
expectedUpdatedAt: existing.updated_at,
|
||||
updatedAt: now,
|
||||
patch: {
|
||||
completion_reason: 'superseded',
|
||||
},
|
||||
});
|
||||
|
||||
if (!superseded) {
|
||||
return {
|
||||
task: getLatestOpenPairedTaskForChat(args.chatJid) ?? existing,
|
||||
supersededTask: null,
|
||||
};
|
||||
}
|
||||
|
||||
cancelOutstandingFinalizeOwnerTurn(existing);
|
||||
|
||||
return {
|
||||
task: createActiveTaskForRoom({
|
||||
group: args.group,
|
||||
chatJid: args.chatJid,
|
||||
canonicalWorkDir,
|
||||
}),
|
||||
supersededTask: existing,
|
||||
};
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// ensureActiveTask
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
function ensureActiveTask(
|
||||
group: RegisteredGroup,
|
||||
chatJid: string,
|
||||
roomRoleContext: RoomRoleContext,
|
||||
hasHumanMessage?: boolean,
|
||||
): PairedTask | null {
|
||||
const existing = getLatestOpenPairedTaskForChat(chatJid);
|
||||
if (roomRoleContext.role === 'owner' && hasHumanMessage) {
|
||||
return resolveOwnerTaskForHumanMessage({
|
||||
group,
|
||||
chatJid,
|
||||
existingTask: existing ?? null,
|
||||
}).task;
|
||||
}
|
||||
if (existing) {
|
||||
return existing;
|
||||
}
|
||||
|
||||
// Don't create a new task for bot-only messages — prevents
|
||||
// ESCALATE → completed → bot message triggers new task → loop.
|
||||
if (!hasHumanMessage) {
|
||||
return null;
|
||||
}
|
||||
const canonicalWorkDir = ensurePairedProject(group, chatJid);
|
||||
if (!canonicalWorkDir) {
|
||||
return null;
|
||||
}
|
||||
return createActiveTaskForRoom({
|
||||
group,
|
||||
chatJid,
|
||||
canonicalWorkDir,
|
||||
});
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// preparePairedExecutionContext
|
||||
// ---------------------------------------------------------------------------
|
||||
@@ -236,17 +352,12 @@ export function preparePairedExecutionContext(args: {
|
||||
const now = new Date().toISOString();
|
||||
|
||||
if (roomRoleContext.role === 'owner') {
|
||||
// New human message → new ping-pong cycle. Reset round trip counter
|
||||
// AND status so the owner turn is not treated as a finalize turn.
|
||||
// Reset status on new human message so the owner gets a fresh working
|
||||
// turn. merge_ready is only reset when a human message is present —
|
||||
// without it, this is a finalize turn after reviewer approval and
|
||||
// resetting would prevent task completion.
|
||||
// New human message keeps the same task only for active review loops.
|
||||
// merge_ready is split into a fresh task before this function runs.
|
||||
// Only reset round_trip_count when a human message is present —
|
||||
// bot-only ping-pong must accumulate the counter for loop detection.
|
||||
const hasHuman = args.hasHumanMessage === true;
|
||||
const needsStatusReset =
|
||||
(latestTask.status === 'merge_ready' && hasHuman) ||
|
||||
latestTask.status === 'review_ready' ||
|
||||
latestTask.status === 'in_review';
|
||||
if (hasHuman || needsStatusReset) {
|
||||
|
||||
Reference in New Issue
Block a user