fix: persist direct terminal verdicts for paired reviewers

This commit is contained in:
ejclaw
2026-04-10 23:41:41 +09:00
parent 921c241873
commit d8b34db92c
6 changed files with 292 additions and 18 deletions

View File

@@ -124,6 +124,49 @@ describe('GroupQueue', () => {
await vi.advanceTimersByTimeAsync(10); await vi.advanceTimersByTimeAsync(10);
}); });
it('stores direct terminal text for the active run and clears it when the run ends', async () => {
let releaseRun!: (value: boolean) => void;
let runId: string | undefined;
const blocker = new Promise<boolean>((resolve) => {
releaseRun = resolve;
});
queue.setProcessMessagesFn(
async (_groupJid: string, context: GroupRunContext) => {
runId = context.runId;
return await blocker;
},
);
queue.enqueueMessageCheck('group1@g.us');
await vi.advanceTimersByTimeAsync(10);
expect(runId).toEqual(expect.any(String));
queue.noteDirectTerminalDelivery(
'group1@g.us',
'reviewer',
'DONE_WITH_CONCERNS\n리뷰 완료',
);
expect(
queue.hasDirectTerminalDeliveryForRun('group1@g.us', runId!, 'reviewer'),
).toBe(true);
expect(
queue.getDirectTerminalDeliveryForRun('group1@g.us', runId!, 'reviewer'),
).toBe('DONE_WITH_CONCERNS\n리뷰 완료');
releaseRun(true);
await vi.advanceTimersByTimeAsync(10);
expect(
queue.hasDirectTerminalDeliveryForRun('group1@g.us', runId!, 'reviewer'),
).toBe(false);
expect(
queue.getDirectTerminalDeliveryForRun('group1@g.us', runId!, 'reviewer'),
).toBeNull();
});
it('force-terminates a lingering process after output was delivered', async () => { it('force-terminates a lingering process after output was delivered', async () => {
class StubbornProcess extends EventEmitter { class StubbornProcess extends EventEmitter {
exitCode: number | null = null; exitCode: number | null = null;

View File

@@ -48,7 +48,7 @@ interface GroupState {
runPhase: RunPhase; runPhase: RunPhase;
runningTaskId: string | null; runningTaskId: string | null;
currentRunId: string | null; currentRunId: string | null;
directTerminalDeliveryRoles: Set<string>; directTerminalDeliveries: Map<string, string>;
pendingMessages: boolean; pendingMessages: boolean;
pendingTasks: QueuedTask[]; pendingTasks: QueuedTask[];
process: ChildProcess | null; process: ChildProcess | null;
@@ -114,7 +114,7 @@ function resetRunState(state: GroupState, groupJid: string): void {
state.process = null; state.process = null;
state.processName = null; state.processName = null;
state.ipcDir = null; state.ipcDir = null;
state.directTerminalDeliveryRoles.clear(); state.directTerminalDeliveries.clear();
transitionRunPhase(state, groupJid, 'idle'); transitionRunPhase(state, groupJid, 'idle');
} }
@@ -199,7 +199,7 @@ export class GroupQueue {
runPhase: 'idle', runPhase: 'idle',
runningTaskId: null, runningTaskId: null,
currentRunId: null, currentRunId: null,
directTerminalDeliveryRoles: new Set(), directTerminalDeliveries: new Map(),
pendingMessages: false, pendingMessages: false,
pendingTasks: [], pendingTasks: [],
process: null, process: null,
@@ -427,21 +427,25 @@ export class GroupQueue {
noteDirectTerminalDelivery( noteDirectTerminalDelivery(
groupJid: string, groupJid: string,
senderRole?: string | null, senderRole?: string | null,
text?: string | null,
): void { ): void {
const state = this.getGroup(groupJid); const state = this.getGroup(groupJid);
if ( if (
state.runPhase !== 'running_messages' || state.runPhase !== 'running_messages' ||
!state.currentRunId || !state.currentRunId ||
!senderRole !senderRole ||
!text ||
text.length === 0
) { ) {
return; return;
} }
state.directTerminalDeliveryRoles.add(senderRole); state.directTerminalDeliveries.set(senderRole, text);
logger.info( logger.info(
{ {
groupJid, groupJid,
runId: state.currentRunId, runId: state.currentRunId,
senderRole, senderRole,
textLength: text.length,
}, },
'Recorded direct terminal delivery for active run', 'Recorded direct terminal delivery for active run',
); );
@@ -456,7 +460,19 @@ export class GroupQueue {
if (state.currentRunId !== runId || !senderRole) { if (state.currentRunId !== runId || !senderRole) {
return false; return false;
} }
return state.directTerminalDeliveryRoles.has(senderRole); return state.directTerminalDeliveries.has(senderRole);
}
getDirectTerminalDeliveryForRun(
groupJid: string,
runId: string,
senderRole?: string | null,
): string | null {
const state = this.getGroup(groupJid);
if (state.currentRunId !== runId || !senderRole) {
return null;
}
return state.directTerminalDeliveries.get(senderRole) ?? null;
} }
private clearPostCloseTimers(state: GroupState): void { private clearPostCloseTimers(state: GroupState): void {

View File

@@ -391,7 +391,7 @@ async function main(): Promise<void> {
(senderRole === 'reviewer' || senderRole === 'arbiter') && (senderRole === 'reviewer' || senderRole === 'arbiter') &&
isTerminalStatusMessage(text) isTerminalStatusMessage(text)
) { ) {
queue.noteDirectTerminalDelivery(jid, senderRole); queue.noteDirectTerminalDelivery(jid, senderRole, text);
} }
}, },
nudgeScheduler: nudgeSchedulerLoop, nudgeScheduler: nudgeSchedulerLoop,

View File

@@ -44,6 +44,7 @@ export function createPairedExecutionLifecycle(args: {
chatJid: string; chatJid: string;
runId: string; runId: string;
enqueueMessageCheck: () => void; enqueueMessageCheck: () => void;
getDirectTerminalDeliveryText?: () => string | null;
onOutput?: (output: AgentOutput) => Promise<void>; onOutput?: (output: AgentOutput) => Promise<void>;
log: ExecutorLog; log: ExecutorLog;
}): PairedExecutionLifecycle { }): PairedExecutionLifecycle {
@@ -54,6 +55,7 @@ export function createPairedExecutionLifecycle(args: {
chatJid, chatJid,
runId, runId,
enqueueMessageCheck, enqueueMessageCheck,
getDirectTerminalDeliveryText,
onOutput, onOutput,
log, log,
} = args; } = args;
@@ -61,6 +63,7 @@ export function createPairedExecutionLifecycle(args: {
let pairedExecutionStatus: 'succeeded' | 'failed' = 'failed'; let pairedExecutionStatus: 'succeeded' | 'failed' = 'failed';
let pairedExecutionSummary: string | null = null; let pairedExecutionSummary: string | null = null;
let pairedFinalOutput: string | null = null; let pairedFinalOutput: string | null = null;
let pairedSummaryLocked = false;
let pairedExecutionCompleted = false; let pairedExecutionCompleted = false;
let pairedExecutionDelegated = false; let pairedExecutionDelegated = false;
let pairedSawOutput = false; let pairedSawOutput = false;
@@ -180,8 +183,48 @@ export function createPairedExecutionLifecycle(args: {
pairedExecutionCompleted = true; pairedExecutionCompleted = true;
}; };
const lockVisibleVerdict = (outputText: string) => {
if (outputText.length === 0) {
return;
}
if (!pairedFinalOutput || pairedFinalOutput.length === 0) {
pairedFinalOutput = outputText;
}
if (!pairedSummaryLocked) {
pairedExecutionSummary = outputText.slice(0, 500);
pairedSummaryLocked = true;
}
pairedSawOutput = true;
};
const adoptDirectTerminalDeliveryIfNeeded = () => {
const outputText = getDirectTerminalDeliveryText?.();
if (!outputText || outputText.length === 0) {
return null;
}
if (!pairedFinalOutput || pairedFinalOutput.length === 0) {
lockVisibleVerdict(outputText);
log.info(
{
pairedTaskId: pairedExecutionContext?.task.id ?? null,
role: completedRole,
runId,
},
'Adopted direct terminal delivery as paired final output',
);
} else if (!pairedSummaryLocked) {
pairedExecutionSummary = pairedFinalOutput.slice(0, 500);
pairedSummaryLocked = true;
}
return outputText;
};
return { return {
updateSummary({ outputText, errorText }) { updateSummary({ outputText, errorText }) {
if (pairedSummaryLocked) {
return;
}
if (outputText && outputText.length > 0) { if (outputText && outputText.length > 0) {
pairedExecutionSummary = outputText.slice(0, 500); pairedExecutionSummary = outputText.slice(0, 500);
return; return;
@@ -193,7 +236,7 @@ export function createPairedExecutionLifecycle(args: {
}, },
recordFinalOutputBeforeDelivery(outputText) { recordFinalOutputBeforeDelivery(outputText) {
pairedFinalOutput = outputText; lockVisibleVerdict(outputText);
completeSuccessfulOwnerTurnBeforeDeliveryIfNeeded(); completeSuccessfulOwnerTurnBeforeDeliveryIfNeeded();
persistPairedTurnOutputIfNeeded(); persistPairedTurnOutputIfNeeded();
}, },
@@ -258,6 +301,8 @@ export function createPairedExecutionLifecycle(args: {
return; return;
} }
const directTerminalOutput = adoptDirectTerminalDeliveryIfNeeded();
const missingVisibleVerdict = const missingVisibleVerdict =
requiresVisibleVerdict && requiresVisibleVerdict &&
(!pairedFinalOutput || pairedFinalOutput.length === 0); (!pairedFinalOutput || pairedFinalOutput.length === 0);
@@ -337,12 +382,16 @@ export function createPairedExecutionLifecycle(args: {
} }
} }
const queueAction = resolvePairedFollowUpQueueAction({ const queueAction =
completedRole, directTerminalOutput &&
executionStatus: effectiveStatus, (completedRole === 'reviewer' || completedRole === 'arbiter')
sawOutput: sawOutputForFollowUp, ? 'none'
taskStatus: finishedTask?.status ?? null, : resolvePairedFollowUpQueueAction({
}); completedRole,
executionStatus: effectiveStatus,
sawOutput: sawOutputForFollowUp,
taskStatus: finishedTask?.status ?? null,
});
if (queueAction !== 'pending' || !finishedTask) { if (queueAction !== 'pending' || !finishedTask) {
return; return;
} }

View File

@@ -1083,6 +1083,149 @@ describe('runAgentForGroup room memory', () => {
expect(db.insertPairedTurnOutput).not.toHaveBeenCalled(); expect(db.insertPairedTurnOutput).not.toHaveBeenCalled();
}); });
it('adopts a direct terminal reviewer delivery as the paired final output and avoids requeue', async () => {
const group = {
...makeGroup(),
folder: 'test-group',
workDir: '/repo',
agentType: 'codex' as const,
};
const deps = {
...makeDeps(),
queue: {
registerProcess: vi.fn(),
enqueueMessageCheck: vi.fn(),
getDirectTerminalDeliveryForRun: vi.fn(
(groupJid: string, runId: string, senderRole?: string | null) =>
groupJid === 'group@test' &&
runId === 'run-review-direct-terminal' &&
senderRole === 'reviewer'
? 'DONE_WITH_CONCERNS\n핵심 concern이 있습니다.'
: null,
),
},
};
vi.mocked(serviceRouting.getEffectiveChannelLease).mockReturnValue({
chat_jid: 'group@test',
owner_agent_type: 'claude-code',
reviewer_agent_type: 'codex',
arbiter_agent_type: null,
owner_service_id: 'claude',
reviewer_service_id: 'codex-review',
arbiter_service_id: null,
activated_at: null,
reason: null,
explicit: false,
});
vi.mocked(
pairedExecutionContext.preparePairedExecutionContext,
).mockReturnValue({
task: {
id: 'paired-task-review-direct-terminal',
chat_jid: 'group@test',
group_folder: 'test-group',
owner_service_id: 'claude',
reviewer_service_id: 'codex-review',
title: null,
source_ref: 'HEAD',
review_requested_at: '2026-04-10T00:00:00.000Z',
plan_notes: null,
round_trip_count: 1,
status: 'in_review',
arbiter_verdict: null,
arbiter_requested_at: null,
completion_reason: null,
created_at: '2026-04-10T00:00:00.000Z',
updated_at: '2026-04-10T00:00:00.000Z',
},
claimedTaskUpdatedAt: '2026-04-10T00:00:00.000Z',
workspace: null,
envOverrides: {},
requiresVisibleVerdict: true,
});
vi.mocked(db.getPairedTaskById).mockReturnValue({
id: 'paired-task-review-direct-terminal',
chat_jid: 'group@test',
group_folder: 'test-group',
owner_service_id: 'claude',
reviewer_service_id: 'codex-review',
title: null,
source_ref: 'HEAD',
plan_notes: null,
round_trip_count: 1,
review_requested_at: '2026-04-10T00:00:00.000Z',
status: 'review_ready',
arbiter_verdict: null,
arbiter_requested_at: null,
completion_reason: null,
created_at: '2026-04-10T00:00:00.000Z',
updated_at: '2026-04-10T00:00:00.000Z',
});
vi.mocked(agentRunner.runAgentProcess).mockImplementation(
async (_group, _input, _onProcess, onOutput) => {
await onOutput?.({
status: 'success',
phase: 'progress',
result: '리뷰 완료했습니다. 핵심 concern을 정리 중입니다.',
output: {
visibility: 'public',
text: '리뷰 완료했습니다. 핵심 concern을 정리 중입니다.',
},
} as any);
return {
status: 'success',
result: null,
};
},
);
const result = await runAgentForGroup(deps, {
group,
prompt: 'please review',
chatJid: 'group@test',
runId: 'run-review-direct-terminal',
forcedRole: 'reviewer',
pairedTurnIdentity: {
turnId:
'paired-task-review-direct-terminal:2026-04-10T00:00:00.000Z:reviewer-turn',
taskId: 'paired-task-review-direct-terminal',
taskUpdatedAt: '2026-04-10T00:00:00.000Z',
intentKind: 'reviewer-turn',
role: 'reviewer',
},
onOutput: async () => {},
});
expect(result).toBe('success');
expect(
pairedExecutionContext.completePairedExecutionContext,
).toHaveBeenCalledWith(
expect.objectContaining({
taskId: 'paired-task-review-direct-terminal',
role: 'reviewer',
status: 'succeeded',
summary: 'DONE_WITH_CONCERNS\n핵심 concern이 있습니다.',
}),
);
expect(db.insertPairedTurnOutput).toHaveBeenCalledWith(
'paired-task-review-direct-terminal',
1,
'reviewer',
'DONE_WITH_CONCERNS\n핵심 concern이 있습니다.',
);
expect(db.completePairedTurn).toHaveBeenCalledWith({
turnId:
'paired-task-review-direct-terminal:2026-04-10T00:00:00.000Z:reviewer-turn',
taskId: 'paired-task-review-direct-terminal',
taskUpdatedAt: '2026-04-10T00:00:00.000Z',
intentKind: 'reviewer-turn',
role: 'reviewer',
});
expect(db.failPairedTurn).not.toHaveBeenCalled();
expect(deps.queue.enqueueMessageCheck).not.toHaveBeenCalled();
});
it('passes paired workspace env overrides into the runner when execution metadata exists', async () => { it('passes paired workspace env overrides into the runner when execution metadata exists', async () => {
const group = { const group = {
...makeGroup(), ...makeGroup(),

View File

@@ -68,7 +68,8 @@ import type { AgentType, PairedRoomRole, RegisteredGroup } from './types.js';
export interface MessageAgentExecutorDeps { export interface MessageAgentExecutorDeps {
assistantName: string; assistantName: string;
queue: Pick<GroupQueue, 'registerProcess' | 'enqueueMessageCheck'>; queue: Pick<GroupQueue, 'registerProcess' | 'enqueueMessageCheck'> &
Partial<Pick<GroupQueue, 'getDirectTerminalDeliveryForRun'>>;
getRoomBindings: () => Record<string, RegisteredGroup>; getRoomBindings: () => Record<string, RegisteredGroup>;
getSessions: () => Record<string, string>; getSessions: () => Record<string, string>;
persistSession: (groupFolder: string, sessionId: string) => void; persistSession: (groupFolder: string, sessionId: string) => void;
@@ -402,9 +403,21 @@ export async function runAgentForGroup(
chatJid, chatJid,
runId, runId,
enqueueMessageCheck: () => deps.queue.enqueueMessageCheck(chatJid), enqueueMessageCheck: () => deps.queue.enqueueMessageCheck(chatJid),
getDirectTerminalDeliveryText: () =>
deps.queue.getDirectTerminalDeliveryForRun?.(
chatJid,
runId,
runtimePairedTurnIdentity?.role ?? activeRole,
) ?? null,
onOutput, onOutput,
log, log,
}); });
const hasDirectTerminalDelivery = (): boolean =>
!!deps.queue.getDirectTerminalDeliveryForRun?.(
chatJid,
runId,
runtimePairedTurnIdentity?.role ?? activeRole,
);
const maybeHandoffToCodex = ( const maybeHandoffToCodex = (
reason: AgentTriggerReason, reason: AgentTriggerReason,
@@ -917,7 +930,11 @@ export async function runAgentForGroup(
}); });
} }
if (!attempt.sawOutput && output.status !== 'error') { if (
!attempt.sawOutput &&
!hasDirectTerminalDelivery() &&
output.status !== 'error'
) {
const claudeRetryResult = await retryClaudeAttemptIfNeeded(attempt); const claudeRetryResult = await retryClaudeAttemptIfNeeded(attempt);
if (claudeRetryResult) { if (claudeRetryResult) {
return claudeRetryResult; return claudeRetryResult;
@@ -973,7 +990,11 @@ export async function runAgentForGroup(
// success-null-result with no visible output — agent returned nothing useful. // success-null-result with no visible output — agent returned nothing useful.
// But if output was already delivered to Discord (sawOutput), treat as success. // But if output was already delivered to Discord (sawOutput), treat as success.
if (attempt.sawSuccessNullResultWithoutOutput && !attempt.sawOutput) { if (
attempt.sawSuccessNullResultWithoutOutput &&
!attempt.sawOutput &&
!hasDirectTerminalDelivery()
) {
log.error( log.error(
'Agent returned success with null result and no visible output', 'Agent returned success with null result and no visible output',
); );
@@ -981,7 +1002,9 @@ export async function runAgentForGroup(
} }
pairedExecutionLifecycle.markStatus('succeeded'); pairedExecutionLifecycle.markStatus('succeeded');
pairedExecutionLifecycle.markSawOutput(attempt.sawOutput); pairedExecutionLifecycle.markSawOutput(
attempt.sawOutput || hasDirectTerminalDelivery(),
);
return 'success'; return 'success';
}; };