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

Preserve owner Codex human messages on silent queued failure

src/message-runtime-queue-cursor.test.ts

@@ -0,0 +1,186 @@
+import { beforeEach, describe, expect, it, vi } from 'vitest';
+
+import { _initTestDatabase, createPairedTask } from './db.js';
+import { runQueuedGroupTurn } from './message-runtime-queue.js';
+import { resetPairedFollowUpScheduleState } from './paired-follow-up-scheduler.js';
+import type {
+  Channel,
+  NewMessage,
+  PairedTask,
+  RegisteredGroup,
+} from './types.js';
+
+type RunQueuedGroupTurnArgs = Parameters<typeof runQueuedGroupTurn>[0];
+
+function makeGroup(): RegisteredGroup {
+  return {
+    name: 'Test Group',
+    folder: 'test-group',
+    trigger: '@Andy',
+    added_at: '2026-03-30T00:00:00.000Z',
+    requiresTrigger: false,
+    agentType: 'codex',
+    workDir: '/repo',
+  };
+}
+
+function makeTask(overrides: Partial<PairedTask> = {}): PairedTask {
+  return {
+    id: 'task-queued-cursor',
+    chat_jid: 'group@test',
+    group_folder: 'test-group',
+    owner_service_id: 'claude',
+    reviewer_service_id: 'codex-main',
+    owner_agent_type: 'claude-code',
+    reviewer_agent_type: 'codex',
+    arbiter_agent_type: null,
+    title: null,
+    source_ref: 'HEAD',
+    plan_notes: null,
+    review_requested_at: null,
+    round_trip_count: 1,
+    status: 'active',
+    arbiter_verdict: null,
+    arbiter_requested_at: null,
+    completion_reason: null,
+    created_at: '2026-03-30T00:00:00.000Z',
+    updated_at: '2026-03-30T00:00:00.000Z',
+    ...overrides,
+  };
+}
+
+function makeChannel(): Channel {
+  return {
+    name: 'discord-owner',
+    connect: vi.fn(),
+    sendMessage: vi.fn(),
+    isConnected: vi.fn(() => true),
+    ownsJid: vi.fn(() => true),
+    disconnect: vi.fn(),
+  } as unknown as Channel;
+}
+
+function makeQueuedTurnArgs(overrides: {
+  task: PairedTask;
+  executeTurn: RunQueuedGroupTurnArgs['executeTurn'];
+  lastAgentTimestamps: Record<string, string>;
+  saveState: () => void;
+  log?: RunQueuedGroupTurnArgs['log'];
+}): RunQueuedGroupTurnArgs {
+  return {
+    chatJid: overrides.task.chat_jid,
+    group: makeGroup(),
+    runId: 'run-owner-queued-cursor',
+    log:
+      overrides.log ??
+      ({ info: vi.fn(), warn: vi.fn(), error: vi.fn() } as any),
+    timezone: 'UTC',
+    missedMessages: [
+      {
+        id: 'human-owner-cursor',
+        chat_jid: overrides.task.chat_jid,
+        sender: 'user@test',
+        sender_name: 'User',
+        content: '중간에 이 내용도 봐줘',
+        timestamp: '2026-03-30T00:00:04.000Z',
+        seq: 48,
+        is_bot_message: false,
+      },
+    ],
+    task: overrides.task,
+    roleToChannel: {
+      owner: null,
+      reviewer: makeChannel(),
+      arbiter: null,
+    },
+    ownerChannel: makeChannel(),
+    lastAgentTimestamps: overrides.lastAgentTimestamps,
+    saveState: overrides.saveState,
+    executeTurn: overrides.executeTurn,
+    getFixedRoleChannelName: () => 'discord-review',
+    labelPairedSenders: (_chatJid: string, messages: NewMessage[]) => messages,
+    formatMessages: () => 'formatted prompt',
+  };
+}
+
+describe('message-runtime queued cursor handling', () => {
+  beforeEach(() => {
+    _initTestDatabase();
+    resetPairedFollowUpScheduleState();
+  });
+
+  it('keeps the queued human cursor when the owner turn fails silently', async () => {
+    const task = makeTask({
+      owner_service_id: 'codex',
+      owner_agent_type: 'codex',
+    });
+    createPairedTask(task);
+    const executeTurn: RunQueuedGroupTurnArgs['executeTurn'] = vi.fn(
+      async () => ({
+        outputStatus: 'error' as const,
+        deliverySucceeded: true,
+        visiblePhase: 'silent',
+      }),
+    );
+    const lastAgentTimestamps: Record<string, string> = {};
+    const saveState = vi.fn(() => undefined);
+    const logMocks = {
+      info: vi.fn(),
+      warn: vi.fn(),
+      error: vi.fn(),
+    };
+    const log = logMocks as unknown as RunQueuedGroupTurnArgs['log'];
+
+    const outcome = await runQueuedGroupTurn(
+      makeQueuedTurnArgs({
+        task,
+        executeTurn,
+        lastAgentTimestamps,
+        saveState,
+        log,
+      }),
+    );
+
+    expect(outcome).toBe(false);
+    expect(executeTurn).toHaveBeenCalledTimes(1);
+    expect(lastAgentTimestamps).toEqual({});
+    expect(saveState).toHaveBeenCalledTimes(2);
+    expect(logMocks.warn).toHaveBeenCalledWith(
+      {
+        messageSeqStart: 48,
+        messageSeqEnd: 48,
+      },
+      'Queued run failed before producing visible output; keeping cursor for retry',
+    );
+  });
+
+  it('advances the queued human cursor when a silent owner turn succeeds', async () => {
+    const task = makeTask({
+      owner_service_id: 'codex',
+      owner_agent_type: 'codex',
+    });
+    createPairedTask(task);
+    const executeTurn: RunQueuedGroupTurnArgs['executeTurn'] = vi.fn(
+      async () => ({
+        outputStatus: 'success' as const,
+        deliverySucceeded: true,
+        visiblePhase: 'silent',
+      }),
+    );
+    const lastAgentTimestamps: Record<string, string> = {};
+    const saveState = vi.fn(() => undefined);
+
+    const outcome = await runQueuedGroupTurn(
+      makeQueuedTurnArgs({
+        task,
+        executeTurn,
+        lastAgentTimestamps,
+        saveState,
+      }),
+    );
+
+    expect(outcome).toBe(true);
+    expect(lastAgentTimestamps).toEqual({ 'group@test': '48' });
+    expect(saveState).toHaveBeenCalledTimes(1);
+  });
+});

src/message-runtime-queue.ts

@@ -14,6 +14,7 @@ import { buildPairedTurnIdentity } from './paired-turn-identity.js';
 import { resolveStoredVisibleVerdict } from './paired-verdict.js';
 import {
   advanceLastAgentCursor,
+  finalizeQueuedRunCursor,
   resolveActiveRole,
   resolveCursorKeyForRole,
   resolveQueuedPairedTurnRole,
@@ -28,6 +29,7 @@ import type {
   Channel,
   NewMessage,
   PairedTask,
+  PairedTaskStatus,
   PairedTurnReservationIntentKind,
   RegisteredGroup,
 } from './types.js';
@@ -52,6 +54,86 @@ function resolveQueuedTurnReservationIntent(args: {
   return 'owner-follow-up';
 }
 
+function buildQueuedGroupTurnPrompt(args: {
+  turnRole: 'owner' | 'reviewer' | 'arbiter';
+  currentTask: PairedTask | null | undefined;
+  chatJid: string;
+  timezone: string;
+  missedMessages: NewMessage[];
+  fallbackMessages: NewMessage[];
+  turnOutputs: ReturnType<typeof getPairedTurnOutputs>;
+  labelPairedSenders: (chatJid: string, messages: NewMessage[]) => NewMessage[];
+  formatMessages: (messages: NewMessage[], timezone: string) => string;
+}): string {
+  if (args.turnRole === 'arbiter' && args.currentTask) {
+    const recentMessages = getRecentChatMessages(args.chatJid, 20);
+    return buildArbiterPromptForTask({
+      task: args.currentTask,
+      chatJid: args.chatJid,
+      timezone: args.timezone,
+      turnOutputs: args.turnOutputs,
+      recentMessages,
+      labeledRecentMessages: args.labelPairedSenders(
+        args.chatJid,
+        recentMessages,
+      ),
+    });
+  }
+
+  if (args.currentTask) {
+    return buildPairedTurnPrompt({
+      taskId: args.currentTask.id,
+      chatJid: args.chatJid,
+      timezone: args.timezone,
+      missedMessages: args.missedMessages,
+      labeledFallbackMessages: args.labelPairedSenders(
+        args.chatJid,
+        args.fallbackMessages,
+      ),
+      turnOutputs: args.turnOutputs,
+    });
+  }
+
+  return args.formatMessages(
+    args.labelPairedSenders(args.chatJid, args.missedMessages),
+    args.timezone,
+  );
+}
+
+function claimQueuedPairedTurn(args: {
+  chatJid: string;
+  runId: string;
+  task: PairedTask | null | undefined;
+  taskStatus: PairedTaskStatus | null | undefined;
+  intentKind: PairedTurnReservationIntentKind | null;
+  turnRole: 'owner' | 'reviewer' | 'arbiter';
+  log: typeof logger;
+}): boolean {
+  if (!args.task) {
+    return true;
+  }
+  const claimed = claimPairedTurnExecution({
+    chatJid: args.chatJid,
+    runId: args.runId,
+    task: args.task,
+    intentKind: args.intentKind!,
+  });
+  if (claimed) {
+    return true;
+  }
+  args.log.info(
+    {
+      taskId: args.task.id,
+      taskStatus: args.taskStatus,
+      taskUpdatedAt: args.task.updated_at,
+      intentKind: args.intentKind,
+      turnRole: args.turnRole,
+    },
+    'Skipped queued paired turn because the task revision was already claimed elsewhere',
+  );
+  return false;
+}
+
 export async function runPendingPairedTurnIfNeeded(args: {
   chatJid: string;
   group: RegisteredGroup;
@@ -246,35 +328,17 @@ export async function runQueuedGroupTurn(args: {
         })
       : undefined;
 
-  let prompt: string;
+  const prompt = buildQueuedGroupTurnPrompt({
-  if (turnRole === 'arbiter' && currentTask) {
+    turnRole,
-    const recentMessages = getRecentChatMessages(chatJid, 20);
+    currentTask,
-    prompt = buildArbiterPromptForTask({
+    chatJid,
-      task: currentTask,
+    timezone: args.timezone,
-      chatJid,
+    missedMessages,
-      timezone: args.timezone,
+    fallbackMessages,
-      turnOutputs,
+    turnOutputs,
-      recentMessages,
+    labelPairedSenders: args.labelPairedSenders,
-      labeledRecentMessages: args.labelPairedSenders(chatJid, recentMessages),
+    formatMessages: args.formatMessages,
-    });
+  });
-  } else if (currentTask) {
-    prompt = buildPairedTurnPrompt({
-      taskId: currentTask.id,
-      chatJid,
-      timezone: args.timezone,
-      missedMessages,
-      labeledFallbackMessages: args.labelPairedSenders(
-        chatJid,
-        fallbackMessages,
-      ),
-      turnOutputs,
-    });
-  } else {
-    prompt = args.formatMessages(
-      args.labelPairedSenders(chatJid, missedMessages),
-      args.timezone,
-    );
-  }
 
   const startSeq = missedMessages[0].seq ?? null;
   const endSeq = missedMessages[missedMessages.length - 1].seq ?? null;
@@ -300,29 +364,23 @@ export async function runQueuedGroupTurn(args: {
     return false;
   }
 
-  if (currentTask) {
+  if (
-    const claimed = claimPairedTurnExecution({
+    !claimQueuedPairedTurn({
       chatJid,
       runId,
       task: currentTask,
+      taskStatus,
       intentKind: queuedIntentKind!,
-    });
+      turnRole,
-    if (!claimed) {
+      log,
-      log.info(
+    })
-        {
+  ) {
-          taskId: currentTask.id,
+    return true;
-          taskStatus,
-          taskUpdatedAt: currentTask.updated_at,
-          intentKind: queuedIntentKind,
-          turnRole,
-        },
-        'Skipped queued paired turn because the task revision was already claimed elsewhere',
-      );
-      return true;
-    }
   }
 
-  if (endSeq !== null) {
+  const previousCursor = args.lastAgentTimestamps[cursorKey];
+  const cursorAdvanced = endSeq !== null;
+  if (cursorAdvanced) {
     advanceLastAgentCursor(
       args.lastAgentTimestamps,
       args.saveState,
@@ -332,19 +390,41 @@ export async function runQueuedGroupTurn(args: {
     );
   }
 
-  const { deliverySucceeded, visiblePhase } = await args.executeTurn({
+  const { outputStatus, deliverySucceeded, visiblePhase } =
-    group,
+    await args.executeTurn({
-    prompt,
+      group,
-    chatJid,
+      prompt,
-    runId,
+      chatJid,
-    channel: turnChannel,
+      runId,
-    deliveryRole: currentTask ? turnRole : undefined,
+      channel: turnChannel,
-    startSeq,
+      deliveryRole: currentTask ? turnRole : undefined,
-    endSeq,
+      startSeq,
-    hasHumanMessage: hasHumanMsg,
+      endSeq,
-    forcedRole,
+      hasHumanMessage: hasHumanMsg,
-    pairedTurnIdentity,
+      forcedRole,
-  });
+      pairedTurnIdentity,
+    });
+
+  if (
+    !finalizeQueuedRunCursor({
+      outputStatus,
+      visiblePhase,
+      startSeq,
+      endSeq,
+      rollbackOnSilentError:
+        hasHumanMsg &&
+        turnRole === 'owner' &&
+        currentTask?.owner_agent_type === 'codex',
+      cursorAdvanced,
+      previousCursor,
+      lastAgentTimestamps: args.lastAgentTimestamps,
+      saveState: args.saveState,
+      cursorKey,
+      log,
+    })
+  ) {
+    return false;
+  }
 
   if (!deliverySucceeded) {
     log.warn(

src/message-runtime-rules.ts

@@ -46,6 +46,47 @@ export function advanceLastAgentCursor(
   saveState();
 }
 
+export function finalizeQueuedRunCursor(args: {
+  outputStatus: 'success' | 'error';
+  visiblePhase: unknown;
+  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;
+  };
+}): boolean {
+  if (
+    args.rollbackOnSilentError &&
+    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,
+        messageSeqEnd: args.endSeq,
+      },
+      'Queued run failed before producing visible output; keeping cursor for retry',
+    );
+    return false;
+  }
+
+  return true;
+}
+
 /** Map task status to the active role. */
 export function resolveActiveRole(
   taskStatus?: string | null,