From a858cc0cab5bfc61c98e5c02799fe44d202701fc Mon Sep 17 00:00:00 2001 From: ejclaw Date: Sat, 11 Apr 2026 09:18:47 +0900 Subject: [PATCH] paired: suppress stale finalize retries --- src/message-agent-executor-attempt-runner.ts | 11 ++- src/message-agent-executor-paired.ts | 49 +++++++++- src/message-agent-executor.test.ts | 98 ++++++++++++++++++++ src/message-runtime-turns.ts | 11 +++ src/message-runtime.test.ts | 1 + src/message-turn-controller.test.ts | 55 +++++++++++ src/message-turn-controller.ts | 17 ++++ src/paired-turn-run-ownership.ts | 34 +++++++ 8 files changed, 273 insertions(+), 3 deletions(-) create mode 100644 src/paired-turn-run-ownership.ts diff --git a/src/message-agent-executor-attempt-runner.ts b/src/message-agent-executor-attempt-runner.ts index e3adbea..7dd62a4 100644 --- a/src/message-agent-executor-attempt-runner.ts +++ b/src/message-agent-executor-attempt-runner.ts @@ -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 { @@ -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); diff --git a/src/message-agent-executor-paired.ts b/src/message-agent-executor-paired.ts index 40cdcbd..349bea2 100644 --- a/src/message-agent-executor-paired.ts +++ b/src/message-agent-executor-paired.ts @@ -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; @@ -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({ diff --git a/src/message-agent-executor.test.ts b/src/message-agent-executor.test.ts index 318febf..824cef9 100644 --- a/src/message-agent-executor.test.ts +++ b/src/message-agent-executor.test.ts @@ -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({ diff --git a/src/message-runtime-turns.ts b/src/message-runtime-turns.ts index 5553a19..c01964d 100644 --- a/src/message-runtime-turns.ts +++ b/src/message-runtime-turns.ts @@ -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({ diff --git a/src/message-runtime.test.ts b/src/message-runtime.test.ts index e55c8d9..9a7fd4a 100644 --- a/src/message-runtime.test.ts +++ b/src/message-runtime.test.ts @@ -132,6 +132,7 @@ vi.mock('./db.js', () => { }; }, ), + getPairedTurnAttempts: vi.fn(() => []), getOpenWorkItem, getOpenWorkItemForChat: vi.fn((chatJid: string) => getOpenWorkItem(chatJid), diff --git a/src/message-turn-controller.test.ts b/src/message-turn-controller.test.ts index 02f2948..64c0ba9 100644 --- a/src/message-turn-controller.test.ts +++ b/src/message-turn-controller.test.ts @@ -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', + }), + ]), + ); + }); }); diff --git a/src/message-turn-controller.ts b/src/message-turn-controller.ts index 09797b4..a149b00 100644 --- a/src/message-turn-controller.ts +++ b/src/message-turn-controller.ts @@ -34,6 +34,7 @@ interface MessageTurnControllerOptions { clearSession: () => void; requestClose: (reason: string) => void; deliverFinalText: (text: string) => Promise; + canDeliverFinalText?: () => boolean; allowProgressReplayWithoutFinal?: boolean; deliveryRole?: PairedRoomRole | null; deliveryServiceId?: string | null; @@ -582,6 +583,22 @@ export class MessageTurnController { private async deliverFinalText(text: string): Promise { 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, diff --git a/src/paired-turn-run-ownership.ts b/src/paired-turn-run-ownership.ts new file mode 100644 index 0000000..c0dec35 --- /dev/null +++ b/src/paired-turn-run-ownership.ts @@ -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, + }; +}