diff --git a/src/message-runtime-queue-cursor.test.ts b/src/message-runtime-queue-cursor.test.ts index 3e543ad..7f67bd8 100644 --- a/src/message-runtime-queue-cursor.test.ts +++ b/src/message-runtime-queue-cursor.test.ts @@ -144,7 +144,7 @@ describe('message-runtime queued cursor handling', () => { expect(outcome).toBe(false); expect(executeTurn).toHaveBeenCalledTimes(1); expect(lastAgentTimestamps).toEqual({}); - expect(saveState).toHaveBeenCalledTimes(2); + expect(saveState).not.toHaveBeenCalled(); expect(logMocks.warn).toHaveBeenCalledWith( { messageSeqStart: 48, @@ -183,4 +183,69 @@ describe('message-runtime queued cursor handling', () => { expect(lastAgentTimestamps).toEqual({ 'group@test': '48' }); expect(saveState).toHaveBeenCalledTimes(1); }); + + it('does not advance the queued human cursor before the owner turn finishes', async () => { + const task = makeTask({ + owner_service_id: 'codex', + owner_agent_type: 'codex', + }); + createPairedTask(task); + const lastAgentTimestamps: Record = {}; + const saveState = vi.fn(() => undefined); + const executeTurn: RunQueuedGroupTurnArgs['executeTurn'] = vi.fn( + async () => { + expect(lastAgentTimestamps).toEqual({}); + expect(saveState).not.toHaveBeenCalled(); + return { + outputStatus: 'success' as const, + deliverySucceeded: true, + visiblePhase: 'final', + }; + }, + ); + + const outcome = await runQueuedGroupTurn( + makeQueuedTurnArgs({ + task, + executeTurn, + lastAgentTimestamps, + saveState, + }), + ); + + expect(outcome).toBe(true); + expect(executeTurn).toHaveBeenCalledTimes(1); + expect(lastAgentTimestamps).toEqual({ 'group@test': '48' }); + expect(saveState).toHaveBeenCalledTimes(1); + }); + + it('keeps the queued human cursor when the owner turn aborts before completion', async () => { + const task = makeTask({ + owner_service_id: 'codex', + owner_agent_type: 'codex', + }); + createPairedTask(task); + const executeTurn: RunQueuedGroupTurnArgs['executeTurn'] = vi.fn( + async () => { + throw new Error('process terminated during restart'); + }, + ); + const lastAgentTimestamps: Record = {}; + const saveState = vi.fn(() => undefined); + + await expect( + runQueuedGroupTurn( + makeQueuedTurnArgs({ + task, + executeTurn, + lastAgentTimestamps, + saveState, + }), + ), + ).rejects.toThrow('process terminated during restart'); + + expect(executeTurn).toHaveBeenCalledTimes(1); + expect(lastAgentTimestamps).toEqual({}); + expect(saveState).not.toHaveBeenCalled(); + }); }); diff --git a/src/message-runtime-queue.ts b/src/message-runtime-queue.ts index 2c30182..5fc2f23 100644 --- a/src/message-runtime-queue.ts +++ b/src/message-runtime-queue.ts @@ -379,17 +379,7 @@ export async function runQueuedGroupTurn(args: { return true; } - const previousCursor = args.lastAgentTimestamps[cursorKey]; const cursorAdvanced = endSeq !== null; - if (cursorAdvanced) { - advanceLastAgentCursor( - args.lastAgentTimestamps, - args.saveState, - chatJid, - endSeq, - cursorKey, - ); - } const { outputStatus, deliverySucceeded, visiblePhase } = await args.executeTurn({ @@ -416,17 +406,22 @@ export async function runQueuedGroupTurn(args: { hasHumanMsg && turnRole === 'owner' && currentTask?.owner_agent_type === 'codex', - cursorAdvanced, - previousCursor, - lastAgentTimestamps: args.lastAgentTimestamps, - saveState: args.saveState, - cursorKey, log, }) ) { return false; } + if (cursorAdvanced) { + advanceLastAgentCursor( + args.lastAgentTimestamps, + args.saveState, + chatJid, + endSeq, + cursorKey, + ); + } + if (!deliverySucceeded) { log.warn( { diff --git a/src/message-runtime-rules.ts b/src/message-runtime-rules.ts index d955b55..4943013 100644 --- a/src/message-runtime-rules.ts +++ b/src/message-runtime-rules.ts @@ -52,11 +52,6 @@ export function finalizeQueuedRunCursor(args: { startSeq: number | null; endSeq: number | null; rollbackOnSilentError: boolean; - cursorAdvanced: boolean; - previousCursor: string | undefined; - lastAgentTimestamps: Record; - saveState: () => void; - cursorKey: string; log: { warn: (bindings: Record, message: string) => void; }; @@ -66,14 +61,6 @@ export function finalizeQueuedRunCursor(args: { args.outputStatus === 'error' && args.visiblePhase === 'silent' ) { - if (args.cursorAdvanced) { - if (args.previousCursor === undefined) { - delete args.lastAgentTimestamps[args.cursorKey]; - } else { - args.lastAgentTimestamps[args.cursorKey] = args.previousCursor; - } - args.saveState(); - } args.log.warn( { messageSeqStart: args.startSeq,