claude-bot/EJClaw
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: persist direct terminal verdicts for paired reviewers

Browse Source
This commit is contained in:
ejclaw
2026-04-10 23:41:41 +09:00
parent 446223e3c8
commit bc42511771
6 changed files with 431 additions and 62 deletions
Download Patch File Download Diff File Expand all files Collapse all files

43
src/group-queue.test.ts
View File

@@ -124,6 +124,49 @@ describe('GroupQueue', () => {
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 () => {
class StubbornProcess extends EventEmitter {
exitCode: number | null = null;

28
src/group-queue.ts
View File

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

2
src/index.ts
View File

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

132
src/message-agent-executor-paired.ts
View File

@@ -14,7 +14,6 @@ import {
completePairedExecutionContext,
type PreparedPairedExecutionContext,
} from './paired-execution-context.js';
import { hasRecognizedPairedTerminalSummary } from './paired-execution-context-shared.js';
import { resolvePairedFollowUpQueueAction } from './message-agent-executor-rules.js';
import { enqueuePairedFollowUpAfterEvent } from './message-runtime-follow-up.js';
import type { PairedTurnIdentity } from './paired-turn-identity.js';
@@ -45,6 +44,7 @@ export function createPairedExecutionLifecycle(args: {
chatJid: string;
runId: string;
enqueueMessageCheck: () => void;
getDirectTerminalDeliveryText?: () => string | null;
onOutput?: (output: AgentOutput) => Promise<void>;
log: ExecutorLog;
}): PairedExecutionLifecycle {
@@ -55,20 +55,25 @@ export function createPairedExecutionLifecycle(args: {
chatJid,
runId,
enqueueMessageCheck,
getDirectTerminalDeliveryText,
onOutput,
log,
} = args;
let pairedExecutionStatus: 'succeeded' | 'failed' = 'failed';
let pairedExecutionSummary: string | null = null;
let pairedTerminalSummary: string | null = null;
let pairedTerminalOutput: string | null = null;
let pairedFinalOutput: string | null = null;
let pairedSummaryLocked = false;
let pairedExecutionCompleted = false;
let pairedExecutionDelegated = false;
let pairedSawOutput = false;
let pairedTurnOutputPersisted = false;
let pairedTurnStateFinalized = false;
let leaseHeartbeatTimer: ReturnType<typeof setInterval> | null = null;
const requiresVisibleVerdict =
pairedExecutionContext?.requiresVisibleVerdict === true;
const missingVisibleVerdictSummary =
'Execution completed without a visible terminal verdict.';
const finalizePairedTurnState = (
status: 'succeeded' | 'failed',
@@ -133,20 +138,12 @@ export function createPairedExecutionLifecycle(args: {
leaseHeartbeatTimer.unref?.();
}
const rememberTerminalOutputIfRecognized = (outputText: string) => {
if (!hasRecognizedPairedTerminalSummary(completedRole, outputText)) {
return;
}
pairedTerminalSummary = outputText.slice(0, 500);
pairedTerminalOutput = outputText;
};
const persistPairedTurnOutputIfNeeded = () => {
if (
!pairedExecutionContext ||
pairedTurnOutputPersisted ||
!pairedTerminalOutput ||
pairedTerminalOutput.length === 0
!pairedFinalOutput ||
pairedFinalOutput.length === 0
) {
return;
}
@@ -156,7 +153,7 @@ export function createPairedExecutionLifecycle(args: {
pairedExecutionContext.task.id,
turnNumber,
completedRole,
pairedTerminalOutput,
pairedFinalOutput,
);
pairedTurnOutputPersisted = true;
};
@@ -166,8 +163,8 @@ export function createPairedExecutionLifecycle(args: {
completedRole !== 'owner' ||
!pairedExecutionContext ||
pairedExecutionCompleted ||
!pairedTerminalOutput ||
pairedTerminalOutput.length === 0
!pairedFinalOutput ||
pairedFinalOutput.length === 0
) {
return;
}
@@ -181,16 +178,55 @@ export function createPairedExecutionLifecycle(args: {
role: completedRole,
status: 'succeeded',
runId,
summary: pairedTerminalSummary ?? pairedExecutionSummary,
summary: pairedExecutionSummary,
});
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 {
updateSummary({ outputText, errorText }) {
if (pairedSummaryLocked) {
return;
}
if (outputText && outputText.length > 0) {
pairedExecutionSummary = outputText.slice(0, 500);
rememberTerminalOutputIfRecognized(outputText);
return;
}
@@ -200,7 +236,7 @@ export function createPairedExecutionLifecycle(args: {
},
recordFinalOutputBeforeDelivery(outputText) {
rememberTerminalOutputIfRecognized(outputText);
lockVisibleVerdict(outputText);
completeSuccessfulOwnerTurnBeforeDeliveryIfNeeded();
persistPairedTurnOutputIfNeeded();
},
@@ -265,29 +301,33 @@ export function createPairedExecutionLifecycle(args: {
return;
}
const missingReviewerOrArbiterVerdict =
(completedRole === 'reviewer' || completedRole === 'arbiter') &&
pairedExecutionStatus === 'succeeded' &&
!pairedTerminalSummary;
const directTerminalOutput = adoptDirectTerminalDeliveryIfNeeded();
const missingVisibleVerdict =
requiresVisibleVerdict &&
(!pairedFinalOutput || pairedFinalOutput.length === 0);
if (missingVisibleVerdict) {
pairedExecutionSummary = missingVisibleVerdictSummary;
log.warn(
{
pairedTaskId: pairedExecutionContext?.task.id ?? null,
role: completedRole,
runId,
},
'Treating paired execution as failed because it ended without a visible terminal verdict',
);
}
const effectiveStatus =
completedRole === 'owner' &&
pairedExecutionStatus === 'succeeded' &&
!pairedSawOutput
? 'failed'
: missingReviewerOrArbiterVerdict
: missingVisibleVerdict && pairedExecutionStatus === 'succeeded'
? 'failed'
: pairedExecutionStatus;
if (missingReviewerOrArbiterVerdict) {
log.warn(
{
pairedTaskId: pairedExecutionContext?.task.id ?? null,
role: completedRole,
summary: pairedExecutionSummary,
},
'Downgraded paired reviewer/arbiter completion to failed because no explicit terminal verdict was emitted',
);
}
const sawOutputForFollowUp = missingVisibleVerdict
? false
: pairedSawOutput;
if (pairedExecutionContext && !pairedExecutionCompleted) {
if (effectiveStatus === 'succeeded') {
@@ -306,15 +346,15 @@ export function createPairedExecutionLifecycle(args: {
role: completedRole,
status: effectiveStatus,
runId,
summary:
effectiveStatus === 'succeeded'
? (pairedTerminalSummary ?? pairedExecutionSummary)
: (pairedExecutionSummary ?? pairedTerminalSummary),
summary: pairedExecutionSummary,
});
pairedExecutionCompleted = true;
}
finalizePairedTurnState(effectiveStatus);
finalizePairedTurnState(
effectiveStatus,
effectiveStatus === 'failed' ? pairedExecutionSummary : null,
);
if (!pairedExecutionContext) {
return;
@@ -342,10 +382,14 @@ export function createPairedExecutionLifecycle(args: {
}
}
const queueAction = resolvePairedFollowUpQueueAction({
const queueAction =
directTerminalOutput &&
(completedRole === 'reviewer' || completedRole === 'arbiter')
? 'none'
: resolvePairedFollowUpQueueAction({
completedRole,
executionStatus: effectiveStatus,
sawOutput: pairedSawOutput,
sawOutput: sawOutputForFollowUp,
taskStatus: finishedTask?.status ?? null,
});
if (queueAction !== 'pending' || !finishedTask) {
@@ -359,8 +403,8 @@ export function createPairedExecutionLifecycle(args: {
source: 'executor-recovery',
completedRole,
executionStatus: effectiveStatus,
sawOutput: pairedSawOutput,
fallbackLastTurnOutputRole: pairedSawOutput ? completedRole : null,
sawOutput: sawOutputForFollowUp,
fallbackLastTurnOutputRole: sawOutputForFollowUp ? completedRole : null,
enqueueMessageCheck,
});
if (followUpResult.kind !== 'paired-follow-up') {

249
src/message-agent-executor.test.ts
View File

@@ -981,6 +981,250 @@ describe('runAgentForGroup room memory', () => {
]);
});
it('fails paired reviewer turns that finish without a visible terminal verdict', async () => {
const group = {
...makeGroup(),
folder: 'test-group',
workDir: '/repo',
agentType: 'codex' as const,
};
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-no-verdict',
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: 0,
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(agentRunner.runAgentProcess).mockImplementation(
async (_group, _input, _onProcess, onOutput) => {
await onOutput?.({
status: 'success',
phase: 'progress',
result: '검토 중입니다.',
output: { visibility: 'public', text: '검토 중입니다.' },
} as any);
return {
status: 'success',
result: null,
};
},
);
const result = await runAgentForGroup(makeDeps(), {
group,
prompt: 'hello',
chatJid: 'group@test',
runId: 'run-review-no-verdict',
forcedRole: 'reviewer',
pairedTurnIdentity: {
turnId:
'paired-task-review-no-verdict:2026-04-10T00:00:00.000Z:reviewer-turn',
taskId: 'paired-task-review-no-verdict',
taskUpdatedAt: '2026-04-10T00:00:00.000Z',
intentKind: 'reviewer-turn',
role: 'reviewer',
},
});
expect(result).toBe('error');
expect(
pairedExecutionContext.completePairedExecutionContext,
).toHaveBeenCalledWith(
expect.objectContaining({
taskId: 'paired-task-review-no-verdict',
role: 'reviewer',
status: 'failed',
summary: 'Execution completed without a visible terminal verdict.',
}),
);
expect(db.failPairedTurn).toHaveBeenCalledWith({
turnIdentity: {
turnId:
'paired-task-review-no-verdict:2026-04-10T00:00:00.000Z:reviewer-turn',
taskId: 'paired-task-review-no-verdict',
taskUpdatedAt: '2026-04-10T00:00:00.000Z',
intentKind: 'reviewer-turn',
role: 'reviewer',
},
error: 'Execution completed without a visible terminal verdict.',
});
expect(db.completePairedTurn).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 () => {
const group = {
...makeGroup(),
@@ -1266,6 +1510,7 @@ describe('runAgentForGroup room memory', () => {
},
workspace: null,
envOverrides: {},
requiresVisibleVerdict: true,
});
vi.mocked(agentRunner.runAgentProcess).mockImplementation(
async (_group, _input, _onProcess, onOutput) => {
@@ -1322,9 +1567,7 @@ describe('runAgentForGroup room memory', () => {
taskId: 'paired-task-reviewer-progress-only',
role: 'reviewer',
status: 'failed',
summary: expect.stringContaining(
'우선 변경 파일과 현재 계약 노출 지점을',
),
summary: 'Execution completed without a visible terminal verdict.',
}),
);
});

31
src/message-agent-executor.ts
View File

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