Merge pull request #167 from phj1081/codex/owner/ejclaw

fix: persist queued cursors after turn completion

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<string, string> = {};
+    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<string, string> = {};
+    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();
+  });
 });

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(
       {

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<string, string>;
-  saveState: () => void;
-  cursorKey: string;
   log: {
     warn: (bindings: Record<string, unknown>, 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,