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

paired: suppress stale finalize retries

Browse Source
This commit is contained in:
ejclaw
2026-04-11 09:18:47 +09:00
parent fe9d265c66
commit a858cc0cab
8 changed files with 273 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
src/message-agent-executor-attempt-runner.ts
View File

@@ -64,7 +64,7 @@ export async function runMessageAgentAttempt(args: {
outputText?: string | null;
errorText?: string | null;
}): void;
recordFinalOutputBeforeDelivery(outputText: string): void;
recordFinalOutputBeforeDelivery(outputText: string): boolean;
};
log: Logger;
}): Promise<MessageAgentAttempt> {
@@ -214,14 +214,21 @@ export async function runMessageAgentAttempt(args: {
outputText &&
outputText.length > 0
) {
let finalOutputAccepted = true;
try {
pairedExecutionLifecycle.recordFinalOutputBeforeDelivery(outputText);
finalOutputAccepted =
pairedExecutionLifecycle.recordFinalOutputBeforeDelivery(
outputText,
);
} catch (err) {
log.warn(
{ pairedTaskId: pairedExecutionContext?.task.id ?? null, err },
'Failed to persist paired turn output and status before delivery',
);
}
if (!finalOutputAccepted) {
return;
}
}
if (onOutput) {
await onOutput(output);

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

@@ -17,6 +17,7 @@ import {
import { resolvePairedFollowUpQueueAction } from './message-agent-executor-rules.js';
import { enqueuePairedFollowUpAfterEvent } from './message-runtime-follow-up.js';
import type { PairedTurnIdentity } from './paired-turn-identity.js';
import { resolvePairedTurnRunOwnership } from './paired-turn-run-ownership.js';
import type { PairedRoomRole } from './types.js';
type ExecutorLog = Pick<typeof logger, 'info' | 'warn'>;
@@ -28,7 +29,7 @@ export interface PairedExecutionLifecycle {
outputText?: string | null;
errorText?: string | null;
}): void;
recordFinalOutputBeforeDelivery(outputText: string): void;
recordFinalOutputBeforeDelivery(outputText: string): boolean;
completeImmediately(args: { status: 'succeeded' | 'failed' }): void;
markDelegated(): void;
markStatus(status: 'succeeded' | 'failed'): void;
@@ -75,6 +76,44 @@ export function createPairedExecutionLifecycle(args: {
const missingVisibleVerdictSummary =
'Execution completed without a visible terminal verdict.';
const currentRunOwnsActiveAttempt = (reason: string): boolean => {
if (!pairedTurnIdentity) {
return true;
}
const ownership = resolvePairedTurnRunOwnership({
turnId: pairedTurnIdentity.turnId,
runId,
});
if (ownership.state === 'active') {
return true;
}
if (ownership.state === 'missing') {
log.warn(
{
pairedTaskId: pairedExecutionContext?.task.id ?? null,
turnId: pairedTurnIdentity.turnId,
runId,
reason,
},
'Could not verify paired turn attempt ownership before final side effects; keeping legacy behavior',
);
return true;
}
log.warn(
{
pairedTaskId: pairedExecutionContext?.task.id ?? null,
turnId: pairedTurnIdentity.turnId,
runId,
reason,
currentAttemptNo: ownership.currentAttemptNo,
currentAttemptState: ownership.currentAttemptState,
currentAttemptRunId: ownership.currentAttemptRunId,
},
'Skipping paired final side effects because this run no longer owns the active attempt',
);
return false;
};
const finalizePairedTurnState = (
status: 'succeeded' | 'failed',
errorText?: string | null,
@@ -236,9 +275,13 @@ export function createPairedExecutionLifecycle(args: {
},
recordFinalOutputBeforeDelivery(outputText) {
if (!currentRunOwnsActiveAttempt('streamed-final-output')) {
return false;
}
lockVisibleVerdict(outputText);
completeSuccessfulOwnerTurnBeforeDeliveryIfNeeded();
persistPairedTurnOutputIfNeeded();
return true;
},
completeImmediately({ status }) {
@@ -281,6 +324,10 @@ export function createPairedExecutionLifecycle(args: {
async asyncFinalize() {
clearLeaseHeartbeat();
if (!currentRunOwnsActiveAttempt('async-finalize')) {
return;
}
if (pairedExecutionContext && pairedExecutionDelegated) {
try {
releasePairedTaskExecutionLease({

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

@@ -68,6 +68,7 @@ vi.mock('./db.js', () => {
getLatestOpenPairedTaskForChat: vi.fn(() => undefined),
getLatestTurnNumber: vi.fn(() => 0),
getPairedTaskById: vi.fn(() => undefined),
getPairedTurnAttempts: vi.fn(() => []),
getPairedTurnOutputs: vi.fn(() => []),
insertPairedTurnOutput: vi.fn(),
markPairedTurnRunning: vi.fn(),
@@ -909,6 +910,103 @@ describe('runAgentForGroup room memory', () => {
});
});
it('suppresses stale owner finalize output when another run already owns the active paired attempt', async () => {
const group = { ...makeGroup(), folder: 'test-group' };
const onOutput = vi.fn(async () => {});
vi.mocked(
pairedExecutionContext.preparePairedExecutionContext,
).mockReturnValue({
task: {
id: 'paired-task-stale-owner-final',
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: 0,
review_requested_at: null,
status: 'merge_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',
},
workspace: null,
envOverrides: {},
});
vi.mocked(db.getPairedTurnAttempts).mockReturnValue([
{
attempt_id:
'paired-task-stale-owner-final:2026-04-10T00:00:00.000Z:finalize-owner-turn:attempt:2',
parent_attempt_id:
'paired-task-stale-owner-final:2026-04-10T00:00:00.000Z:finalize-owner-turn:attempt:1',
parent_handoff_id: null,
continuation_handoff_id: null,
turn_id:
'paired-task-stale-owner-final:2026-04-10T00:00:00.000Z:finalize-owner-turn',
attempt_no: 2,
task_id: 'paired-task-stale-owner-final',
task_updated_at: '2026-04-10T00:00:00.000Z',
role: 'owner',
intent_kind: 'finalize-owner-turn',
state: 'running',
executor_service_id: 'claude',
executor_agent_type: 'claude-code',
active_run_id: 'run-new-owner-attempt',
created_at: '2026-04-10T00:00:01.000Z',
updated_at: '2026-04-10T00:00:01.000Z',
completed_at: null,
last_error: null,
},
]);
vi.mocked(agentRunner.runAgentProcess).mockImplementation(
async (_group, _input, _onProcess, emitOutput) => {
await emitOutput?.({
status: 'success',
result: 'DONE\nowner final from stale attempt',
output: {
visibility: 'public',
text: 'DONE\nowner final from stale attempt',
},
phase: 'final',
} as any);
return {
status: 'success',
result: 'DONE\nowner final from stale attempt',
};
},
);
const result = await runAgentForGroup(makeDeps(), {
group,
prompt: 'finalize please',
chatJid: 'group@test',
runId: 'run-stale-owner-attempt',
pairedTurnIdentity: {
turnId:
'paired-task-stale-owner-final:2026-04-10T00:00:00.000Z:finalize-owner-turn',
taskId: 'paired-task-stale-owner-final',
taskUpdatedAt: '2026-04-10T00:00:00.000Z',
intentKind: 'finalize-owner-turn',
role: 'owner',
},
onOutput,
});
expect(result).toBe('success');
expect(onOutput).not.toHaveBeenCalled();
expect(db.insertPairedTurnOutput).not.toHaveBeenCalled();
expect(
pairedExecutionContext.completePairedExecutionContext,
).not.toHaveBeenCalled();
expect(db.completePairedTurn).not.toHaveBeenCalled();
expect(db.failPairedTurn).not.toHaveBeenCalled();
});
it('allows silent reviewer outputs', async () => {
const group = { ...makeGroup(), folder: 'test-group', workDir: '/repo' };
vi.mocked(serviceRouting.getEffectiveChannelLease).mockReturnValue({

11
src/message-runtime-turns.ts
View File

@@ -7,6 +7,7 @@ import {
hasReviewerLease,
resolveLeaseServiceId,
} from './service-routing.js';
import { resolvePairedTurnRunOwnership } from './paired-turn-run-ownership.js';
import { normalizeMessageForDedupe } from './router.js';
import type { ExecuteTurnFn } from './message-runtime-types.js';
import type {
@@ -186,6 +187,16 @@ export function createExecuteTurn(deps: CreateExecuteTurnDeps): ExecuteTurnFn {
deliveryRole: resolvedDeliveryRole,
deliveryServiceId: resolvedDeliveryServiceId,
pairedTurnIdentity: args.pairedTurnIdentity ?? null,
canDeliverFinalText: () => {
if (!args.pairedTurnIdentity) {
return true;
}
const ownership = resolvePairedTurnRunOwnership({
turnId: args.pairedTurnIdentity.turnId,
runId,
});
return ownership.state !== 'inactive';
},
deliverFinalText: async (text) => {
try {
return await deps.deliverFinalText({

1
src/message-runtime.test.ts
View File

@@ -132,6 +132,7 @@ vi.mock('./db.js', () => {
};
},
),
getPairedTurnAttempts: vi.fn(() => []),
getOpenWorkItem,
getOpenWorkItemForChat: vi.fn((chatJid: string) =>
getOpenWorkItem(chatJid),

55
src/message-turn-controller.test.ts
View File

@@ -273,4 +273,59 @@ describe('MessageTurnController outbound audit logging', () => {
]),
);
});
it('suppresses replaying the last progress update as final when final delivery is disallowed', async () => {
const channel = makeChannel();
const deliverFinalText = vi.fn().mockResolvedValue(true);
const controller = new MessageTurnController({
chatJid: 'dc:test-room',
group: makeGroup(),
runId: 'run-stale-owner-attempt',
channel,
idleTimeout: 1_000,
failureFinalText: '실패',
isClaudeCodeAgent: true,
clearSession: vi.fn(),
requestClose: vi.fn(),
deliverFinalText,
canDeliverFinalText: () => false,
allowProgressReplayWithoutFinal: true,
deliveryRole: 'owner',
pairedTurnIdentity: {
turnId: 'paired-task:2026-04-10T00:00:00.000Z:finalize-owner-turn',
taskId: 'paired-task',
taskUpdatedAt: '2026-04-10T00:00:00.000Z',
intentKind: 'finalize-owner-turn',
role: 'owner',
},
});
await controller.start();
await controller.handleOutput({
status: 'success',
phase: 'progress',
result: '작업 중 1',
output: { visibility: 'public', text: '작업 중 1' },
} as any);
await controller.handleOutput({
status: 'success',
phase: 'progress',
result: '작업 중 2',
output: { visibility: 'public', text: '작업 중 2' },
} as any);
await flushAsync();
const finishResult = await controller.finish('success');
expect(finishResult.deliverySucceeded).toBe(true);
expect(channel.sendAndTrack).toHaveBeenCalledTimes(1);
expect(deliverFinalText).not.toHaveBeenCalled();
expect(getAuditEntries()).not.toEqual(
expect.arrayContaining([
expect.objectContaining({
auditEvent: 'final-delivery-attempt',
}),
]),
);
});
});

17
src/message-turn-controller.ts
View File

@@ -34,6 +34,7 @@ interface MessageTurnControllerOptions {
clearSession: () => void;
requestClose: (reason: string) => void;
deliverFinalText: (text: string) => Promise<boolean>;
canDeliverFinalText?: () => boolean;
allowProgressReplayWithoutFinal?: boolean;
deliveryRole?: PairedRoomRole | null;
deliveryServiceId?: string | null;
@@ -582,6 +583,22 @@ export class MessageTurnController {
private async deliverFinalText(text: string): Promise<void> {
await this.activateTyping('turn:deliver-final');
this.visiblePhase = toVisiblePhase('final');
if (
this.options.canDeliverFinalText &&
!this.options.canDeliverFinalText()
) {
this.log.info(
{
runId: this.options.runId,
deliveryRole: this.options.deliveryRole ?? null,
turnId: this.options.pairedTurnIdentity?.turnId ?? null,
textLength: text.length,
},
'Suppressed final delivery because this run no longer owns the active paired turn attempt',
);
this.latestProgressTextForFinal = null;
return;
}
this.logOutboundAudit('final-delivery-attempt', {
messageId: this.progressMessageId,
textLength: text.length,

34
src/paired-turn-run-ownership.ts Normal file
View File

@@ -0,0 +1,34 @@
import { getPairedTurnAttempts } from './db.js';
export interface PairedTurnRunOwnership {
state: 'active' | 'inactive' | 'missing';
currentAttemptNo: number | null;
currentAttemptState: string | null;
currentAttemptRunId: string | null;
}
export function resolvePairedTurnRunOwnership(args: {
turnId: string;
runId: string;
}): PairedTurnRunOwnership {
const currentAttempt = getPairedTurnAttempts(args.turnId).at(-1);
if (!currentAttempt) {
return {
state: 'missing',
currentAttemptNo: null,
currentAttemptState: null,
currentAttemptRunId: null,
};
}
return {
state:
currentAttempt.state === 'running' &&
currentAttempt.active_run_id === args.runId
? 'active'
: 'inactive',
currentAttemptNo: currentAttempt.attempt_no,
currentAttemptState: currentAttempt.state,
currentAttemptRunId: currentAttempt.active_run_id ?? null,
};
}