Merge branch 'codex/owner/ejclaw'

# Conflicts: # src/message-agent-executor-rules.test.ts # src/message-agent-executor-rules.ts # src/message-agent-executor.ts # src/message-runtime-flow.ts # src/message-runtime.test.ts # src/message-runtime.ts # src/paired-execution-context-arbiter.ts # src/paired-execution-context-owner.ts # src/paired-execution-context-reviewer.ts # src/paired-execution-context-shared.ts # src/paired-execution-context.ts # src/paired-workspace-manager.test.ts # src/task-scheduler.ts
2026-04-07 05:16:11 +09:00
parent 279cfc20c1 10e10b499c
commit 2b8c6c9cd0
28 changed files with 1200 additions and 708 deletions
--- a/CLAUDE.md
+++ b/CLAUDE.md
@@ -37,7 +37,8 @@ Run commands directly—don't tell the user to run them.
 ```bash
 bun run build                              # Build main project
 bun run build:runners                      # Install + build both runners
-bun run build:container                    # Rebuild reviewer Docker image
+bun run build:runtime                      # Build host runtime only
+bun run build:container                    # Optional: rebuild reviewer Docker image
 bun run dev                                # Dev mode with hot reload
 ```

@@ -53,6 +54,9 @@ Deploy:
 bun run deploy
 ```

+`deploy` rebuilds only the host runtime. Rebuild the reviewer Docker image
+separately with `bun run build:container` only when you need the container path.
+
 ## Service Stack Architecture

 Single unified service manages all three Discord bots in one process:
--- a/package.json
+++ b/package.json
@@ -8,9 +8,10 @@
  "scripts": {
    "build": "tsc",
    "build:runners": "bun install --cwd runners/agent-runner && bun run --cwd runners/agent-runner build && bun install --cwd runners/codex-runner && bun run --cwd runners/codex-runner build",
+    "build:runtime": "bun run build && bun run build:runners",
    "build:container": "docker build -f container/Dockerfile -t ejclaw-reviewer:latest .",
-    "build:all": "bun run build && bun run build:runners && bun run build:container",
-    "deploy": "git pull --ff-only && bun run build:all && systemctl --user restart ejclaw",
+    "build:all": "bun run build:runtime && bun run build:container",
+    "deploy": "git pull --ff-only && bun run build:runtime && systemctl --user restart ejclaw",
    "start": "bun dist/index.js",
    "dev": "bun --watch src/index.ts",
    "test": "vitest run",
--- a/prompts/arbiter-paired-room.md
+++ b/prompts/arbiter-paired-room.md
@@ -34,6 +34,8 @@ You may receive reference opinions from external models appended to your prompt.
 ## Rules

 - Base your verdict on evidence (code, test output, logs), not on who said what first
+- Distinguish reviewer snapshot limits from real product bugs. Reviewer workspaces may intentionally omit heavy artifacts like `node_modules`, `dist`, and `build`; inability to run direct local test/typecheck/build there is not, by itself, a blocker if dedicated verification evidence exists
+- When verification evidence exists from the dedicated verification path, judge that evidence on its merits instead of requiring the reviewer to reproduce the same result from the lightweight reviewer snapshot
 - Your verdict is final for this deadlock cycle — after it, work resumes normally
 - You do NOT implement or review code — you only judge the disagreement
 - Keep your verdict concise — state the decision, the evidence, and the required action
--- a/prompts/claude-paired-room.md
+++ b/prompts/claude-paired-room.md
@@ -30,6 +30,8 @@ Push back with evidence when the owner is wrong. Hold your ground when you are r
 ## Rules

 - Judge completion only by verification output. "It should work now" means run it. "I'm confident" means nothing — confidence is not evidence. "I tested earlier" means test again if code changed since. "It's a trivial change" means verify anyway
+- Reviewer workspaces may intentionally exclude heavy runtime artifacts such as `node_modules`, `dist`, and `build`. Do not treat the inability to run direct local test/typecheck/build from the reviewer snapshot as a product bug by itself
+- When test/typecheck/build evidence is needed, prefer the dedicated verification path (`run_verification`) over assuming the reviewer snapshot can execute the full project locally
 - Stagnation: **Spinning** (same error 3+), **Oscillation** (alternating approaches), **Diminishing returns** (shrinking improvement), **No progress** (discussion without change) — name the pattern and report: **Status**, **Attempted**, **Recommendation**
 - Implementation, commits, and pushes require agreement from both sides. Either can veto
 - Keep reviews concise — approve quickly when there is nothing to critique
--- a/runners/codex-runner/test/reviewer-runtime.test.ts
+++ b/runners/codex-runner/test/reviewer-runtime.test.ts
@@ -3,7 +3,7 @@ import fs from 'fs';
 import os from 'os';
 import path from 'path';

-import { describe, expect, it } from 'vitest';
+import { afterEach, beforeEach, describe, expect, it } from 'vitest';

 import {
  assertReadonlyWorkspaceRepoConnectivity,
@@ -11,6 +11,22 @@ import {
  isReviewerRuntime,
 } from '../src/reviewer-runtime.js';

+const ORIGINAL_UNSAFE_HOST_PAIRED_MODE =
+  process.env.EJCLAW_UNSAFE_HOST_PAIRED_MODE;
+
+afterEach(() => {
+  if (ORIGINAL_UNSAFE_HOST_PAIRED_MODE == null) {
+    delete process.env.EJCLAW_UNSAFE_HOST_PAIRED_MODE;
+  } else {
+    process.env.EJCLAW_UNSAFE_HOST_PAIRED_MODE =
+      ORIGINAL_UNSAFE_HOST_PAIRED_MODE;
+  }
+});
+
+beforeEach(() => {
+  delete process.env.EJCLAW_UNSAFE_HOST_PAIRED_MODE;
+});
+
 function createTempRepo(prefix: string): string {
  const cwd = fs.mkdtempSync(path.join(os.tmpdir(), prefix));
  execFileSync('git', ['init'], {
--- a/src/agent-attempt-orchestration.ts
+++ b/src/agent-attempt-orchestration.ts
@@ -0,0 +1,182 @@
+import type { AgentOutput } from './agent-runner.js';
+import {
+  resolveAttemptRetryAction,
+  type AttemptRetryAction,
+  type AttemptRetryState,
+} from './agent-attempt-retry.js';
+import type {
+  AgentTriggerReason,
+  CodexRotationReason,
+} from './agent-error-detection.js';
+import {
+  runClaudeRotationLoop,
+  runCodexRotationLoop,
+} from './provider-retry.js';
+
+type RotationLogContext = Record<string, unknown>;
+
+type ClaudeRetryAttemptState = Pick<
+  AttemptRetryState,
+  'sawOutput' | 'streamedTriggerReason'
+> & {
+  output?: Pick<AgentOutput, 'status' | 'result' | 'output' | 'error'>;
+  error?: unknown;
+  sawSuccessNullResultWithoutOutput?: boolean;
+};
+
+type CodexRetryAttemptState = Pick<
+  AttemptRetryState,
+  'sawOutput' | 'streamedTriggerReason'
+> & {
+  output?: Pick<AgentOutput, 'status' | 'result' | 'output' | 'error'>;
+  error?: unknown;
+};
+
+export type ExecutedAttemptRetryAction =
+  | { kind: 'none' }
+  | {
+      kind: 'claude';
+      trigger: { reason: AgentTriggerReason; retryAfterMs?: number };
+      rotationMessage?: string;
+      result: 'success' | 'error';
+    }
+  | {
+      kind: 'codex';
+      trigger: { reason: CodexRotationReason };
+      rotationMessage?: string;
+      result: 'success' | 'error';
+    };
+
+export async function runClaudeAttemptWithRotation<
+  TAttempt extends ClaudeRetryAttemptState,
+>(args: {
+  initialTrigger: {
+    reason: AgentTriggerReason;
+    retryAfterMs?: number;
+  };
+  runAttempt: () => Promise<TAttempt>;
+  logContext: RotationLogContext;
+  rotationMessage?: string;
+  afterAttempt?: (attempt: TAttempt) => Promise<void> | void;
+  onSuccess?: (outcome: { sawOutput: boolean }) => Promise<void> | void;
+}): Promise<'success' | 'error'> {
+  const outcome = await runClaudeRotationLoop(
+    args.initialTrigger,
+    async () => {
+      const attempt = await args.runAttempt();
+      await args.afterAttempt?.(attempt);
+      return {
+        output: attempt.output,
+        thrownError: attempt.error,
+        sawOutput: attempt.sawOutput,
+        sawSuccessNullResult: attempt.sawSuccessNullResultWithoutOutput,
+        streamedTriggerReason: attempt.streamedTriggerReason,
+      };
+    },
+    args.logContext,
+    args.rotationMessage,
+  );
+
+  if (outcome.type === 'success') {
+    await args.onSuccess?.({ sawOutput: outcome.sawOutput });
+    return 'success';
+  }
+
+  return 'error';
+}
+
+export async function runCodexAttemptWithRotation<
+  TAttempt extends CodexRetryAttemptState,
+>(args: {
+  initialTrigger: { reason: CodexRotationReason };
+  runAttempt: () => Promise<TAttempt>;
+  logContext: RotationLogContext;
+  rotationMessage?: string;
+  afterAttempt?: (attempt: TAttempt) => Promise<void> | void;
+}): Promise<'success' | 'error'> {
+  const outcome = await runCodexRotationLoop(
+    args.initialTrigger,
+    async () => {
+      const attempt = await args.runAttempt();
+      await args.afterAttempt?.(attempt);
+      return {
+        output: attempt.output,
+        thrownError: attempt.error,
+        sawOutput: attempt.sawOutput,
+        streamedTriggerReason: attempt.streamedTriggerReason,
+      };
+    },
+    args.logContext,
+    args.rotationMessage,
+  );
+
+  return outcome.type === 'success' ? 'success' : 'error';
+}
+
+export async function executeAttemptRetryAction(args: {
+  provider: 'claude' | 'codex';
+  canRetryClaudeCredentials: boolean;
+  canRetryCodex: boolean;
+  attempt: Pick<AttemptRetryState, 'sawOutput' | 'streamedTriggerReason'>;
+  rotationMessage?: string | null;
+  runClaude: (
+    trigger: { reason: AgentTriggerReason; retryAfterMs?: number },
+    rotationMessage?: string,
+  ) => Promise<'success' | 'error'>;
+  runCodex: (
+    trigger: { reason: CodexRotationReason },
+    rotationMessage?: string,
+  ) => Promise<'success' | 'error'>;
+}): Promise<ExecutedAttemptRetryAction> {
+  const retryAction = resolveAttemptRetryAction({
+    provider: args.provider,
+    canRetryClaudeCredentials: args.canRetryClaudeCredentials,
+    canRetryCodex: args.canRetryCodex,
+    attempt: args.attempt,
+    rotationMessage: args.rotationMessage,
+  });
+
+  return executeResolvedAttemptRetryAction({
+    retryAction,
+    runClaude: args.runClaude,
+    runCodex: args.runCodex,
+  });
+}
+
+async function executeResolvedAttemptRetryAction(args: {
+  retryAction: AttemptRetryAction;
+  runClaude: (
+    trigger: { reason: AgentTriggerReason; retryAfterMs?: number },
+    rotationMessage?: string,
+  ) => Promise<'success' | 'error'>;
+  runCodex: (
+    trigger: { reason: CodexRotationReason },
+    rotationMessage?: string,
+  ) => Promise<'success' | 'error'>;
+}): Promise<ExecutedAttemptRetryAction> {
+  if (args.retryAction.kind === 'claude') {
+    return {
+      kind: 'claude',
+      trigger: args.retryAction.trigger,
+      rotationMessage: args.retryAction.rotationMessage,
+      result: await args.runClaude(
+        args.retryAction.trigger,
+        args.retryAction.rotationMessage,
+      ),
+    };
+  }
+
+  if (args.retryAction.kind === 'codex') {
+    return {
+      kind: 'codex',
+      trigger: args.retryAction.trigger,
+      rotationMessage: args.retryAction.rotationMessage,
+      result: await args.runCodex(
+        args.retryAction.trigger,
+        args.retryAction.rotationMessage,
+      ),
+    };
+  }
+
+  return { kind: 'none' };
+}
--- a/src/agent-attempt-retry.ts
+++ b/src/agent-attempt-retry.ts
@@ -0,0 +1,164 @@
+import {
+  classifyRotationTrigger,
+  type AgentTriggerReason,
+  type CodexRotationReason,
+  isCodexRotationReason,
+} from './agent-error-detection.js';
+import { detectCodexRotationTrigger } from './codex-token-rotation.js';
+import { getErrorMessage } from './utils.js';
+
+export interface AttemptStreamedTrigger {
+  reason: AgentTriggerReason;
+  retryAfterMs?: number;
+}
+
+export interface AttemptRetryState {
+  sawOutput: boolean;
+  retryableSessionFailureDetected?: boolean;
+  streamedTriggerReason?: AttemptStreamedTrigger;
+  error?: unknown;
+  outputError?: string | null;
+}
+
+export function isRetryableClaudeSessionFailureAttempt(args: {
+  attempt: AttemptRetryState;
+  isClaudeCodeAgent: boolean;
+  provider: 'claude' | 'codex';
+  shouldRetryFreshSessionOnAgentFailure: (args: {
+    result: null;
+    error: string;
+  }) => boolean;
+}): boolean {
+  if (
+    !args.isClaudeCodeAgent ||
+    args.provider !== 'claude' ||
+    args.attempt.sawOutput
+  ) {
+    return false;
+  }
+
+  if (args.attempt.retryableSessionFailureDetected === true) {
+    return true;
+  }
+
+  if (args.attempt.error == null) {
+    return false;
+  }
+
+  return args.shouldRetryFreshSessionOnAgentFailure({
+    result: null,
+    error: getErrorMessage(args.attempt.error),
+  });
+}
+
+export function resolveClaudeRetryTrigger(args: {
+  canRetryClaudeCredentials: boolean;
+  provider: 'claude' | 'codex';
+  attempt: Pick<AttemptRetryState, 'sawOutput' | 'streamedTriggerReason'>;
+  fallbackMessage?: string | null;
+}): { reason: AgentTriggerReason; retryAfterMs?: number } | null {
+  if (
+    !(
+      args.canRetryClaudeCredentials &&
+      args.provider === 'claude' &&
+      !args.attempt.sawOutput
+    )
+  ) {
+    return null;
+  }
+
+  if (args.attempt.streamedTriggerReason) {
+    return {
+      reason: args.attempt.streamedTriggerReason.reason,
+      retryAfterMs: args.attempt.streamedTriggerReason.retryAfterMs,
+    };
+  }
+
+  const trigger = classifyRotationTrigger(args.fallbackMessage);
+  if (!trigger.shouldRetry) {
+    return null;
+  }
+
+  return {
+    reason: trigger.reason,
+    retryAfterMs: trigger.retryAfterMs,
+  };
+}
+
+export function resolveCodexRetryTrigger(args: {
+  canRetryCodex: boolean;
+  attempt: Pick<AttemptRetryState, 'streamedTriggerReason'>;
+  rotationMessage?: string | null;
+}): { reason: CodexRotationReason } | null {
+  if (!args.canRetryCodex) {
+    return null;
+  }
+
+  if (args.attempt.streamedTriggerReason) {
+    if (!isCodexRotationReason(args.attempt.streamedTriggerReason.reason)) {
+      return null;
+    }
+    return {
+      reason: args.attempt.streamedTriggerReason.reason,
+    };
+  }
+
+  const trigger = detectCodexRotationTrigger(args.rotationMessage);
+  if (!trigger.shouldRotate) {
+    return null;
+  }
+
+  return { reason: trigger.reason };
+}
+
+export type AttemptRetryAction =
+  | {
+      kind: 'claude';
+      trigger: { reason: AgentTriggerReason; retryAfterMs?: number };
+      rotationMessage?: string;
+    }
+  | {
+      kind: 'codex';
+      trigger: { reason: CodexRotationReason };
+      rotationMessage?: string;
+    }
+  | { kind: 'none' };
+
+export function resolveAttemptRetryAction(args: {
+  provider: 'claude' | 'codex';
+  canRetryClaudeCredentials: boolean;
+  canRetryCodex: boolean;
+  attempt: Pick<AttemptRetryState, 'sawOutput' | 'streamedTriggerReason'>;
+  rotationMessage?: string | null;
+}): AttemptRetryAction {
+  const normalizedRotationMessage = args.rotationMessage ?? undefined;
+
+  const claudeTrigger = resolveClaudeRetryTrigger({
+    canRetryClaudeCredentials: args.canRetryClaudeCredentials,
+    provider: args.provider,
+    attempt: args.attempt,
+    fallbackMessage: normalizedRotationMessage,
+  });
+  if (claudeTrigger) {
+    return {
+      kind: 'claude',
+      trigger: claudeTrigger,
+      rotationMessage: normalizedRotationMessage,
+    };
+  }
+
+  const codexTrigger = resolveCodexRetryTrigger({
+    canRetryCodex: args.canRetryCodex,
+    attempt: args.attempt,
+    rotationMessage: normalizedRotationMessage,
+  });
+  if (codexTrigger) {
+    return {
+      kind: 'codex',
+      trigger: codexTrigger,
+      rotationMessage: normalizedRotationMessage,
+    };
+  }
+
+  return { kind: 'none' };
+}
--- a/src/codex-token-rotation.test.ts
+++ b/src/codex-token-rotation.test.ts
@@ -121,4 +121,25 @@ describe('codex-token-rotation d7 ≥ 100% auto-skip', () => {
    expect(active).toBeDefined();
    expect(active!.index).toBe(1); // fallback picks next non-rate-limited
  });
+
+  it('warns when Codex rotation state cannot be persisted', async () => {
+    const mod = await import('./codex-token-rotation.js');
+    const utils = await import('./utils.js');
+    const { logger } = await import('./logger.js');
+
+    vi.mocked(utils.writeJsonFile).mockImplementation(() => {
+      throw new Error('disk full');
+    });
+
+    mod.initCodexTokenRotation();
+    expect(mod.rotateCodexToken('rate limit')).toBe(true);
+
+    expect(logger.warn).toHaveBeenCalledWith(
+      expect.objectContaining({
+        stateFile: '/tmp/ejclaw-codex-rot-data/codex-rotation-state.json',
+        err: expect.any(Error),
+      }),
+      'Failed to persist Codex rotation state',
+    );
+  });
 });
--- a/src/codex-token-rotation.ts
+++ b/src/codex-token-rotation.ts
@@ -138,8 +138,11 @@ function saveCodexState(): void {
      resetD7Ats: accounts.map((a) => a.resetD7At ?? null),
    };
    writeJsonFile(STATE_FILE, state);
-  } catch {
-    /* best effort */
+  } catch (err) {
+    logger.warn(
+      { stateFile: STATE_FILE, err },
+      'Failed to persist Codex rotation state',
+    );
  }
 }

--- a/src/logger.test.ts
+++ b/src/logger.test.ts
@@ -0,0 +1,33 @@
+import { describe, expect, it, vi } from 'vitest';
+
+describe('logger singleton', () => {
+  it('reuses the root logger and does not add duplicate process listeners across module reloads', async () => {
+    vi.resetModules();
+
+    const beforeUncaught = process.listeners('uncaughtException').length;
+    const beforeUnhandled = process.listeners('unhandledRejection').length;
+    const beforeExit = process.listeners('exit').length;
+
+    const first = await import('./logger.js');
+
+    const afterFirstUncaught = process.listeners('uncaughtException').length;
+    const afterFirstUnhandled = process.listeners('unhandledRejection').length;
+    const afterFirstExit = process.listeners('exit').length;
+
+    vi.resetModules();
+
+    const second = await import('./logger.js');
+
+    const afterSecondUncaught = process.listeners('uncaughtException').length;
+    const afterSecondUnhandled = process.listeners('unhandledRejection').length;
+    const afterSecondExit = process.listeners('exit').length;
+
+    expect(second.logger).toBe(first.logger);
+    expect(afterFirstUncaught).toBeGreaterThanOrEqual(beforeUncaught);
+    expect(afterFirstUnhandled).toBeGreaterThanOrEqual(beforeUnhandled);
+    expect(afterSecondUncaught).toBe(afterFirstUncaught);
+    expect(afterSecondUnhandled).toBe(afterFirstUnhandled);
+    expect(afterFirstExit).toBeGreaterThanOrEqual(beforeExit);
+    expect(afterSecondExit).toBe(afterFirstExit);
+  });
+});
--- a/src/logger.ts
+++ b/src/logger.ts
@@ -1,12 +1,34 @@
 import pino, { type Logger } from 'pino';

 const serviceName = (process.env.ASSISTANT_NAME || 'claude').toLowerCase();
+const isTestEnv =
+  process.env.VITEST === 'true' || process.env.NODE_ENV === 'test';

-export const logger = pino({
-  level: process.env.LOG_LEVEL || 'info',
-  name: serviceName,
-  transport: { target: 'pino-pretty', options: { colorize: true } },
-});
+type LoggerGlobalState = typeof globalThis & {
+  __ejclawLogger?: Logger;
+  __ejclawProcessHandlersInstalled?: boolean;
+};
+
+const globalState = globalThis as LoggerGlobalState;
+
+function createRootLogger(): Logger {
+  const baseOptions = {
+    level: process.env.LOG_LEVEL || 'info',
+    name: serviceName,
+  };
+
+  if (isTestEnv) {
+    return pino(baseOptions);
+  }
+
+  return pino({
+    ...baseOptions,
+    transport: { target: 'pino-pretty', options: { colorize: true } },
+  });
+}
+
+export const logger =
+  globalState.__ejclawLogger ?? (globalState.__ejclawLogger = createRootLogger());

 type LogBindings = Record<string, unknown>;

@@ -30,11 +52,17 @@ export function createScopedLogger(bindings: LogBindings): Logger {
 }

 // Route uncaught errors through pino so they get timestamps in stderr
-process.on('uncaughtException', (err) => {
+function handleUncaughtException(err: unknown): void {
  logger.fatal({ err }, 'Uncaught exception');
  process.exit(1);
-});
+}

-process.on('unhandledRejection', (reason) => {
+function handleUnhandledRejection(reason: unknown): void {
  logger.error({ err: reason }, 'Unhandled rejection');
-});
+}
+
+if (!globalState.__ejclawProcessHandlersInstalled) {
+  process.on('uncaughtException', handleUncaughtException);
+  process.on('unhandledRejection', handleUnhandledRejection);
+  globalState.__ejclawProcessHandlersInstalled = true;
+}
--- a/src/message-agent-executor-rules.test.ts
+++ b/src/message-agent-executor-rules.test.ts
@@ -23,6 +23,7 @@ import { classifyRotationTrigger } from './agent-error-detection.js';
 import { detectCodexRotationTrigger } from './codex-token-rotation.js';
 import {
  isRetryableClaudeSessionFailureAttempt,
+  resolveAttemptRetryAction,
  resolveClaudeRetryTrigger,
  resolveCodexRetryTrigger,
  resolvePairedFollowUpQueueAction,
@@ -88,6 +89,47 @@ describe('message-agent-executor-rules', () => {
    expect(detectCodexRotationTrigger).not.toHaveBeenCalled();
  });

+  it('resolves a shared Claude retry action from fallback output text', () => {
+    vi.mocked(classifyRotationTrigger).mockReturnValue({
+      shouldRetry: true,
+      reason: '429',
+      retryAfterMs: 30000,
+    });
+
+    expect(
+      resolveAttemptRetryAction({
+        provider: 'claude',
+        canRetryClaudeCredentials: true,
+        canRetryCodex: false,
+        attempt: { sawOutput: false },
+        rotationMessage: '429 rate limit',
+      }),
+    ).toEqual({
+      kind: 'claude',
+      trigger: { reason: '429', retryAfterMs: 30000 },
+      rotationMessage: '429 rate limit',
+    });
+  });
+
+  it('resolves a shared Codex retry action from streamed state', () => {
+    expect(
+      resolveAttemptRetryAction({
+        provider: 'codex',
+        canRetryClaudeCredentials: false,
+        canRetryCodex: true,
+        attempt: {
+          sawOutput: false,
+          streamedTriggerReason: { reason: 'auth-expired' },
+        },
+        rotationMessage: 'oauth token expired',
+      }),
+    ).toEqual({
+      kind: 'codex',
+      trigger: { reason: 'auth-expired' },
+      rotationMessage: 'oauth token expired',
+    });
+  });
+
  it('detects retryable Claude session failures from either flag or classifier', () => {
    expect(
      isRetryableClaudeSessionFailureAttempt({
--- a/src/message-agent-executor-rules.ts
+++ b/src/message-agent-executor-rules.ts
@@ -1,116 +1,14 @@
-import {
-  classifyRotationTrigger,
-  type AgentTriggerReason,
-  type CodexRotationReason,
-  isCodexRotationReason,
-} from './agent-error-detection.js';
-import { detectCodexRotationTrigger } from './codex-token-rotation.js';
 import { resolveNextTurnAction } from './message-runtime-rules.js';
 import type { PairedRoomRole, PairedTaskStatus } from './types.js';
-import { getErrorMessage } from './utils.js';
-
-export interface ExecutorStreamedTrigger {
-  reason: AgentTriggerReason;
-  retryAfterMs?: number;
-}
-
-export interface ExecutorAttemptState {
-  sawOutput: boolean;
-  retryableSessionFailureDetected?: boolean;
-  streamedTriggerReason?: ExecutorStreamedTrigger;
-  error?: unknown;
-}
-
-export function isRetryableClaudeSessionFailureAttempt(args: {
-  attempt: ExecutorAttemptState;
-  isClaudeCodeAgent: boolean;
-  provider: 'claude' | 'codex';
-  shouldRetryFreshSessionOnAgentFailure: (args: {
-    result: null;
-    error: string;
-  }) => boolean;
-}): boolean {
-  if (
-    !args.isClaudeCodeAgent ||
-    args.provider !== 'claude' ||
-    args.attempt.sawOutput
-  ) {
-    return false;
-  }
-
-  if (args.attempt.retryableSessionFailureDetected === true) {
-    return true;
-  }
-
-  if (args.attempt.error == null) {
-    return false;
-  }
-
-  return args.shouldRetryFreshSessionOnAgentFailure({
-    result: null,
-    error: getErrorMessage(args.attempt.error),
-  });
-}
-
-export function resolveClaudeRetryTrigger(args: {
-  canRetryClaudeCredentials: boolean;
-  provider: 'claude' | 'codex';
-  attempt: Pick<ExecutorAttemptState, 'sawOutput' | 'streamedTriggerReason'>;
-  fallbackMessage?: string | null;
-}): { reason: AgentTriggerReason; retryAfterMs?: number } | null {
-  if (
-    !(
-      args.canRetryClaudeCredentials &&
-      args.provider === 'claude' &&
-      !args.attempt.sawOutput
-    )
-  ) {
-    return null;
-  }
-
-  if (args.attempt.streamedTriggerReason) {
-    return {
-      reason: args.attempt.streamedTriggerReason.reason,
-      retryAfterMs: args.attempt.streamedTriggerReason.retryAfterMs,
-    };
-  }
-
-  const trigger = classifyRotationTrigger(args.fallbackMessage);
-  if (!trigger.shouldRetry) {
-    return null;
-  }
-
-  return {
-    reason: trigger.reason,
-    retryAfterMs: trigger.retryAfterMs,
-  };
-}
-
-export function resolveCodexRetryTrigger(args: {
-  canRetryCodex: boolean;
-  attempt: Pick<ExecutorAttemptState, 'streamedTriggerReason'>;
-  rotationMessage?: string | null;
-}): { reason: CodexRotationReason } | null {
-  if (!args.canRetryCodex) {
-    return null;
-  }
-
-  if (args.attempt.streamedTriggerReason) {
-    if (!isCodexRotationReason(args.attempt.streamedTriggerReason.reason)) {
-      return null;
-    }
-    return {
-      reason: args.attempt.streamedTriggerReason.reason,
-    };
-  }
-
-  const trigger = detectCodexRotationTrigger(args.rotationMessage);
-  if (!trigger.shouldRotate) {
-    return null;
-  }
-
-  return { reason: trigger.reason };
-}
+export {
+  isRetryableClaudeSessionFailureAttempt,
+  resolveAttemptRetryAction,
+  resolveClaudeRetryTrigger,
+  resolveCodexRetryTrigger,
+  type AttemptRetryAction,
+  type AttemptRetryState as ExecutorAttemptState,
+  type AttemptStreamedTrigger as ExecutorStreamedTrigger,
+} from './agent-attempt-retry.js';

 export type PairedFollowUpQueueAction =
  | 'generic'
--- a/src/message-agent-executor.ts
+++ b/src/message-agent-executor.ts
@@ -5,6 +5,12 @@ import { getErrorMessage } from './utils.js';

 import { getAgentOutputText } from './agent-output.js';
 import { createEvaluatedOutputHandler } from './agent-attempt.js';
+import {
+  executeAttemptRetryAction,
+  runClaudeAttemptWithRotation,
+  runCodexAttemptWithRotation,
+} from './agent-attempt-orchestration.js';
+import { isRetryableClaudeSessionFailureAttempt } from './agent-attempt-retry.js';
 import {
  AgentOutput,
  runAgentProcess,
@@ -30,18 +36,9 @@ import {
 } from './paired-execution-context.js';
 import { resolveCodexFallbackHandoff } from './paired-turn-fallback.js';
 import { resolveExecutionTarget } from './message-runtime-rules.js';
-import {
-  isRetryableClaudeSessionFailureAttempt,
-  resolveClaudeRetryTrigger,
-  resolveCodexRetryTrigger,
-  resolvePairedFollowUpQueueAction,
-} from './message-agent-executor-rules.js';
+import { resolvePairedFollowUpQueueAction } from './message-agent-executor-rules.js';
 import { buildRoomRoleContext } from './room-role-context.js';
 import { type AgentTriggerReason } from './agent-error-detection.js';
-import {
-  runClaudeRotationLoop,
-  runCodexRotationLoop,
-} from './provider-retry.js';
 import {
  shouldResetSessionOnAgentFailure,
  shouldRetryFreshSessionOnAgentFailure,
@@ -650,27 +647,17 @@ export async function runAgentForGroup(
    initialTrigger: { reason: CodexRotationReason },
    rotationMessage?: string,
  ): Promise<'success' | 'error'> => {
-    const outcome = await runCodexRotationLoop(
+    return runCodexAttemptWithRotation({
      initialTrigger,
-      async () => {
-        const attempt = await runAttempt('codex');
-        return {
-          output: attempt.output,
-          thrownError: attempt.error,
-          sawOutput: attempt.sawOutput,
-          streamedTriggerReason: attempt.streamedTriggerReason,
-        };
-      },
-      {
+      runAttempt: () => runAttempt('codex'),
+      logContext: {
        chatJid,
        group: group.name,
        groupFolder: group.folder,
        runId,
      },
      rotationMessage,
-    );
-
-    return outcome.type === 'success' ? 'success' : 'error';
+    });
  };

  type AgentAttempt = Awaited<ReturnType<typeof runAttempt>>;
@@ -689,78 +676,63 @@ export async function runAgentForGroup(
      runId,
    };

-    const outcome = await runClaudeRotationLoop(
+    return runClaudeAttemptWithRotation({
      initialTrigger,
-      async () => {
-        const attempt = await runAttempt('claude');
-        return {
-          output: attempt.output,
-          thrownError: attempt.error,
-          sawOutput: attempt.sawOutput,
-          sawSuccessNullResult: attempt.sawSuccessNullResultWithoutOutput,
-          streamedTriggerReason: attempt.streamedTriggerReason,
-        };
-      },
-      logCtx,
+      runAttempt: () => runAttempt('claude'),
+      logContext: logCtx,
      rotationMessage,
-    );
-
-    switch (outcome.type) {
-      case 'success':
-        pairedSawOutput = outcome.sawOutput;
-        return 'success';
-      case 'error':
-        return 'error';
-    }
+      onSuccess: ({ sawOutput }) => {
+        pairedSawOutput = sawOutput;
+      },
+    });
  };

  const retryClaudeAttemptIfNeeded = async (
    attempt: AgentAttempt,
    rotationMessage?: string | null,
  ): Promise<'success' | 'error' | null> => {
-    const trigger = resolveClaudeRetryTrigger({
-      canRetryClaudeCredentials,
+    const retryAction = await executeAttemptRetryAction({
      provider,
+      canRetryClaudeCredentials,
+      canRetryCodex: false,
      attempt,
-      fallbackMessage: rotationMessage,
+      rotationMessage,
+      runClaude: retryClaudeWithRotation,
+      runCodex: retryCodexWithRotation,
    });
-    if (!trigger) {
+    if (retryAction.kind !== 'claude') {
      return null;
    }

-    const result = await retryClaudeWithRotation(
-      trigger,
-      rotationMessage ?? undefined,
-    );
-    if (result === 'error') {
-      return maybeHandoffAfterError(trigger.reason, attempt);
+    if (retryAction.result === 'error') {
+      return maybeHandoffAfterError(retryAction.trigger.reason, attempt);
    }

    pairedExecutionStatus = 'succeeded';
-    return result;
+    return retryAction.result;
  };

  const retryCodexAttemptIfNeeded = async (
    attempt: AgentAttempt,
    rotationMessage?: string | null,
  ): Promise<'success' | 'error' | null> => {
-    const trigger = resolveCodexRetryTrigger({
+    const retryAction = await executeAttemptRetryAction({
+      provider,
+      canRetryClaudeCredentials: false,
      canRetryCodex: !isClaudeCodeAgent && getCodexAccountCount() > 1,
      attempt,
      rotationMessage,
+      runClaude: retryClaudeWithRotation,
+      runCodex: retryCodexWithRotation,
    });
-    if (!trigger) {
+    if (retryAction.kind !== 'codex') {
      return null;
    }

-    const result = await retryCodexWithRotation(
-      { reason: trigger.reason },
-      rotationMessage ?? undefined,
-    );
-    if (result === 'success') {
+    if (retryAction.result === 'success') {
      pairedExecutionStatus = 'succeeded';
    }
-    return result;
+    return retryAction.result;
  };

  const maybeHandoffAfterError = (
--- a/src/message-runtime-flow.ts
+++ b/src/message-runtime-flow.ts
@@ -8,6 +8,7 @@ import {
  buildArbiterPromptForTask,
  buildFinalizePendingPrompt,
  buildOwnerPendingPrompt,
+  buildPairedTurnPrompt,
  buildReviewerPendingPrompt,
 } from './message-runtime-prompts.js';
 import {
@@ -47,23 +48,24 @@ export type BotOnlyPairedFollowUpAction =
      nextRole: 'owner' | 'reviewer' | 'arbiter';
    };

-export function resolveLastDeliveredBotRole(
+export type QueuedTurnDispatch = {
+  formatted: string;
+  botOnlyFollowUpAction: BotOnlyPairedFollowUpAction;
+  isBotOnlyPairedFollowUp: boolean;
+  loopCursorKey: string;
+  endSeq: number | null;
+};
+
+export function isBotOnlyPairedRoomTurn(
+  chatJid: string,
  messages: NewMessage[],
-): PairedRoomRole | null {
-  for (let index = messages.length - 1; index >= 0; index -= 1) {
-    const message = messages[index];
-    if (!message?.is_bot_message) {
-      continue;
-    }
-    if (
-      message.sender_name === 'owner' ||
-      message.sender_name === 'reviewer' ||
-      message.sender_name === 'arbiter'
-    ) {
-      return message.sender_name;
-    }
-  }
-  return null;
+): boolean {
+  return (
+    hasReviewerLease(chatJid) &&
+    messages.every(
+      (message) => message.is_from_me === true || !!message.is_bot_message,
+    )
+  );
 }

 export function buildPendingPairedTurn(args: {
@@ -77,7 +79,6 @@ export function buildPendingPairedTurn(args: {
  labeledRecentMessages: Parameters<
    typeof buildArbiterPromptForTask
  >[0]['labeledRecentMessages'];
-  lastDeliveredMessages?: NewMessage[];
  resolveChannel: (taskStatus?: string | null) => Channel | null;
 }): PendingPairedTurn {
  const {
@@ -96,10 +97,7 @@ export function buildPendingPairedTurn(args: {
  const lastTurnOutput = turnOutputs[turnOutputs.length - 1];
  const nextTurnAction = resolveNextTurnAction({
    taskStatus,
-    lastTurnOutputRole:
-      resolveLastDeliveredBotRole(args.lastDeliveredMessages ?? []) ??
-      lastTurnOutput?.role ??
-      null,
+    lastTurnOutputRole: lastTurnOutput?.role ?? null,
  });
  const recentMessages = getRecentChatMessages(chatJid, 20);
  const lastHumanMessage = getLastHumanMessageContent(chatJid);
@@ -234,7 +232,6 @@ export function resolveBotOnlyPairedFollowUpAction(args: {
  chatJid: string;
  task: PairedTask | null | undefined;
  isBotOnlyPairedFollowUp: boolean;
-  lastDeliveredMessages?: NewMessage[];
  pendingCursorSource:
    | { seq?: number | null; timestamp?: string | null }
    | undefined;
@@ -249,10 +246,7 @@ export function resolveBotOnlyPairedFollowUpAction(args: {
  const lastTurnOutput = getPairedTurnOutputs(task.id).at(-1);
  const nextTurnAction = resolveNextTurnAction({
    taskStatus: task.status,
-    lastTurnOutputRole:
-      resolveLastDeliveredBotRole(args.lastDeliveredMessages ?? []) ??
-      lastTurnOutput?.role ??
-      null,
+    lastTurnOutputRole: lastTurnOutput?.role ?? null,
  });

  if (nextTurnAction.kind === 'finalize-owner-turn') {
@@ -384,6 +378,53 @@ export async function executeBotOnlyPairedFollowUpAction(args: {
  return true;
 }

+export function buildQueuedTurnDispatch(args: {
+  chatJid: string;
+  timezone: string;
+  loopPendingTask: PairedTask | null | undefined;
+  rawPendingMessages: NewMessage[];
+  messagesToSend: NewMessage[];
+  labeledMessagesToSend: NewMessage[];
+  formatMessages: (messages: NewMessage[], timezone: string) => string;
+}): QueuedTurnDispatch {
+  const loopCursorKey = resolveCursorKey(
+    args.chatJid,
+    args.loopPendingTask?.status,
+  );
+  const formatted = args.loopPendingTask
+    ? buildPairedTurnPrompt({
+        taskId: args.loopPendingTask.id,
+        chatJid: args.chatJid,
+        timezone: args.timezone,
+        missedMessages: args.messagesToSend,
+        labeledFallbackMessages: args.labeledMessagesToSend,
+        turnOutputs: getPairedTurnOutputs(args.loopPendingTask.id),
+      })
+    : args.formatMessages(args.labeledMessagesToSend, args.timezone);
+  const isBotOnlyPairedFollowUp = isBotOnlyPairedRoomTurn(
+    args.chatJid,
+    args.messagesToSend,
+  );
+  const pendingCursorSource =
+    args.rawPendingMessages.length > 0
+      ? args.rawPendingMessages[args.rawPendingMessages.length - 1]
+      : args.messagesToSend[args.messagesToSend.length - 1];
+  const botOnlyFollowUpAction = resolveBotOnlyPairedFollowUpAction({
+    chatJid: args.chatJid,
+    task: args.loopPendingTask,
+    isBotOnlyPairedFollowUp,
+    pendingCursorSource,
+  });
+
+  return {
+    formatted,
+    botOnlyFollowUpAction,
+    isBotOnlyPairedFollowUp,
+    loopCursorKey,
+    endSeq: args.messagesToSend[args.messagesToSend.length - 1]?.seq ?? null,
+  };
+}
+
 export function shouldSkipGenericFollowUpAfterDeliveryRetry(args: {
  chatJid: string;
  deliveryRole: PairedRoomRole;
--- a/src/message-runtime.test.ts
+++ b/src/message-runtime.test.ts
@@ -1827,124 +1827,6 @@ describe('createMessageRuntime', () => {
    expect(sendMessage).not.toHaveBeenCalled();
  });

-  it('requeues owner follow-ups from reviewer bot-only messages even when paired turn output storage is stale', async () => {
-    const chatJid = 'group@test';
-    const group = makeGroup('codex');
-    const ownerChannel = makeChannel(chatJid);
-    const reviewerChannel = {
-      ...makeChannel(chatJid, 'discord-review', false),
-      isOwnMessage: vi.fn((message) => message.sender === 'reviewer-bot@test'),
-    };
-    const enqueueMessageCheck = vi.fn();
-    const closeStdin = vi.fn();
-    const sendMessage = vi.fn(() => false);
-    const setLastTimestamp = vi.fn();
-    const saveState = vi.fn();
-    const lastAgentTimestamps: Record<string, string> = {};
-    const stopLoop = new Error('stop-message-loop');
-
-    vi.mocked(serviceRouting.hasReviewerLease).mockReturnValue(true);
-    vi.mocked(db.getLatestOpenPairedTaskForChat).mockReturnValue({
-      id: 'task-active-owner-follow-up-stale-output',
-      chat_jid: chatJid,
-      group_folder: group.folder,
-      owner_service_id: 'codex-main',
-      reviewer_service_id: 'claude',
-      title: null,
-      source_ref: 'HEAD',
-      plan_notes: null,
-      review_requested_at: '2026-03-30T00:00:00.000Z',
-      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',
-    });
-    vi.mocked(db.getPairedTurnOutputs).mockReturnValue([
-      {
-        id: 1,
-        task_id: 'task-active-owner-follow-up-stale-output',
-        turn_number: 1,
-        role: 'owner',
-        output_text: 'owner 초안',
-        created_at: '2026-03-30T00:00:01.000Z',
-      },
-    ]);
-    vi.mocked(db.getNewMessages).mockReturnValue({
-      messages: [
-        {
-          id: 'reviewer-bot-message-owner-follow-up-stale-output',
-          chat_jid: chatJid,
-          sender: 'reviewer-bot@test',
-          sender_name: '리뷰어',
-          content: 'DONE_WITH_CONCERNS\n\nreviewer direct message',
-          timestamp: '2026-03-30T00:00:04.000Z',
-          seq: 42,
-          is_bot_message: true,
-        } as any,
-      ],
-      newTimestamp: '42',
-    });
-
-    const runtime = createMessageRuntime({
-      assistantName: 'Andy',
-      idleTimeout: 60_000,
-      pollInterval: 123,
-      timezone: 'UTC',
-      triggerPattern: /^@Andy\b/i,
-      channels: [ownerChannel, reviewerChannel],
-      queue: {
-        registerProcess: vi.fn(),
-        closeStdin,
-        enqueueMessageCheck,
-        notifyIdle: vi.fn(),
-        sendMessage,
-      } as any,
-      getRegisteredGroups: () => ({ [chatJid]: group }),
-      getSessions: () => ({}),
-      getLastTimestamp: () => '',
-      setLastTimestamp,
-      getLastAgentTimestamps: () => lastAgentTimestamps,
-      saveState,
-      persistSession: vi.fn(),
-      clearSession: vi.fn(),
-    });
-
-    const originalSetTimeout = global.setTimeout;
-    const setTimeoutSpy = vi.spyOn(global, 'setTimeout').mockImplementation(((
-      handler: any,
-      timeout?: number,
-      ...args: any[]
-    ) => {
-      if (timeout === 123) {
-        throw stopLoop;
-      }
-      return (originalSetTimeout as any)(handler, timeout, ...args);
-    }) as typeof setTimeout);
-
-    try {
-      await expect(runtime.startMessageLoop()).rejects.toThrow(
-        stopLoop.message,
-      );
-    } finally {
-      setTimeoutSpy.mockRestore();
-    }
-
-    expect(setLastTimestamp).toHaveBeenCalledWith('42');
-    expect(agentRunner.runAgentProcess).not.toHaveBeenCalled();
-    expect(closeStdin).toHaveBeenCalledWith(
-      chatJid,
-      expect.objectContaining({ reason: 'paired-pending-turn-follow-up' }),
-    );
-    expect(enqueueMessageCheck).toHaveBeenCalledWith(
-      chatJid,
-      resolveGroupIpcPath(group.folder),
-    );
-    expect(sendMessage).not.toHaveBeenCalled();
-  });
-
  it('auto-runs an owner follow-up when a task returns to active after reviewer feedback', async () => {
    const chatJid = 'group@test';
    const group = makeGroup('codex');
--- a/src/message-runtime.ts
+++ b/src/message-runtime.ts
@@ -43,11 +43,11 @@ import {
 } from './message-runtime-rules.js';
 import { runAgentForGroup } from './message-agent-executor.js';
 import {
+  buildQueuedTurnDispatch,
  buildPendingPairedTurn,
  executeBotOnlyPairedFollowUpAction,
  executePendingPairedTurn,
-  resolveBotOnlyPairedFollowUpAction,
-  resolveLastDeliveredBotRole,
+  isBotOnlyPairedRoomTurn,
  shouldSkipGenericFollowUpAfterDeliveryRetry,
 } from './message-runtime-flow.js';
 import { MessageTurnController } from './message-turn-controller.js';
@@ -58,6 +58,7 @@ import {
  buildPairedTurnPrompt,
  buildReviewerPendingPrompt,
 } from './message-runtime-prompts.js';
+import { transitionPairedTaskStatus } from './paired-execution-context-shared.js';
 import {
  extractSessionCommand,
  handleSessionCommand,
@@ -207,15 +208,6 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): {
    });
  };

-  const isBotOnlyPairedRoomTurn = (
-    chatJid: string,
-    messages: NewMessage[],
-  ): boolean =>
-    hasReviewerLease(chatJid) &&
-    messages.every(
-      (message) => message.is_from_me === true || !!message.is_bot_message,
-    );
-
  const enqueueScopedGroupMessageCheck = (
    chatJid: string,
    groupFolder: string,
@@ -859,7 +851,6 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): {
                chatJid,
                getRecentChatMessages(chatJid, 20),
              ),
-              lastDeliveredMessages: labelPairedSenders(chatJid, missedMessages),
              resolveChannel,
            })
          : null;
@@ -937,10 +928,15 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): {
            if (hasReviewerLease(chatJid)) {
              const task = getLatestOpenPairedTaskForChat(chatJid);
              if (task) {
-                updatePairedTask(task.id, {
-                  status: 'completed',
-                  completion_reason: 'stopped',
-                  updated_at: new Date().toISOString(),
+                const now = new Date().toISOString();
+                transitionPairedTaskStatus({
+                  taskId: task.id,
+                  currentStatus: task.status,
+                  nextStatus: 'completed',
+                  updatedAt: now,
+                  patch: {
+                    completion_reason: 'stopped',
+                  },
                });
              }
            }
@@ -1072,6 +1068,191 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): {
    }
  };

+  const processQueuedGroupDispatch = async (args: {
+    chatJid: string;
+    group: RegisteredGroup;
+    channel: Channel;
+    processableGroupMessages: NewMessage[];
+  }): Promise<void> => {
+    const { chatJid, group, channel, processableGroupMessages } = args;
+    const loopPendingTask = hasReviewerLease(chatJid)
+      ? getLatestOpenPairedTaskForChat(chatJid)
+      : null;
+    const loopCursorKey = resolveCursorKey(chatJid, loopPendingTask?.status);
+    const rawPendingMessages = getMessagesSinceSeq(
+      chatJid,
+      deps.getLastAgentTimestamps()[loopCursorKey] || '0',
+      deps.assistantName,
+    );
+    const pendingMessages = filterLoopingPairedBotMessages(
+      chatJid,
+      getProcessableMessages(chatJid, rawPendingMessages, channel),
+      FAILURE_FINAL_TEXT,
+    );
+    const messagesToSend =
+      pendingMessages.length > 0 ? pendingMessages : processableGroupMessages;
+    const labeledMessagesToSend = labelPairedSenders(chatJid, messagesToSend);
+    const {
+      formatted,
+      botOnlyFollowUpAction,
+      isBotOnlyPairedFollowUp,
+      loopCursorKey: dispatchCursorKey,
+      endSeq,
+    } = buildQueuedTurnDispatch({
+      chatJid,
+      timezone: deps.timezone,
+      loopPendingTask,
+      rawPendingMessages,
+      messagesToSend,
+      labeledMessagesToSend,
+      formatMessages,
+    });
+
+    if (
+      await executeBotOnlyPairedFollowUpAction({
+        action: botOnlyFollowUpAction,
+        chatJid,
+        group,
+        runId: `loop-merge-ready-${Date.now().toString(36)}`,
+        channel,
+        log: logger,
+        saveState: deps.saveState,
+        lastAgentTimestamps: deps.getLastAgentTimestamps(),
+        executeTurn,
+        enqueueGroupMessageCheck: () =>
+          enqueueScopedGroupMessageCheck(chatJid, group.folder),
+        closeStdin: () =>
+          deps.queue.closeStdin(chatJid, {
+            reason: 'paired-pending-turn-follow-up',
+          }),
+      })
+    ) {
+      return;
+    }
+
+    if (deps.queue.sendMessage(chatJid, formatted)) {
+      if (endSeq != null) {
+        advanceLastAgentCursor(
+          deps.getLastAgentTimestamps(),
+          deps.saveState,
+          chatJid,
+          endSeq,
+          dispatchCursorKey,
+        );
+      }
+      logger.debug(
+        {
+          transition: 'typing:on',
+          source: 'follow-up-queued',
+          chatJid,
+          group: group.name,
+          groupFolder: group.folder,
+          endSeq: endSeq ?? null,
+          suppressed: isBotOnlyPairedFollowUp,
+        },
+        'Typing indicator transition',
+      );
+      if (!isBotOnlyPairedFollowUp) {
+        await channel
+          .setTyping?.(chatJid, true)
+          ?.catch((err) =>
+            logger.warn(
+              { chatJid, err },
+              'Failed to set typing indicator',
+            ),
+          );
+      }
+      return;
+    }
+
+    enqueueScopedGroupMessageCheck(chatJid, group.folder);
+  };
+
+  const processLoopGroupMessages = async (args: {
+    chatJid: string;
+    group: RegisteredGroup;
+    groupMessages: NewMessage[];
+    channel: Channel;
+  }): Promise<void> => {
+    const { chatJid, group, groupMessages, channel } = args;
+    const isMainGroup = group.isMain === true;
+    const processableGroupMessages = getProcessableMessages(
+      chatJid,
+      groupMessages,
+      channel,
+    );
+
+    if (processableGroupMessages.length === 0) {
+      const lastIgnored = groupMessages[groupMessages.length - 1];
+      if (lastIgnored?.seq != null) {
+        advanceLastAgentCursor(
+          deps.getLastAgentTimestamps(),
+          deps.saveState,
+          chatJid,
+          lastIgnored.seq,
+        );
+      }
+      return;
+    }
+
+    if (shouldSkipBotOnlyCollaboration(chatJid, processableGroupMessages)) {
+      const lastIgnored =
+        processableGroupMessages[processableGroupMessages.length - 1];
+      if (lastIgnored?.seq != null) {
+        advanceLastAgentCursor(
+          deps.getLastAgentTimestamps(),
+          deps.saveState,
+          chatJid,
+          lastIgnored.seq,
+        );
+      }
+      logger.info(
+        { chatJid, group: group.name, groupFolder: group.folder },
+        'Bot-collaboration timeout: no recent human message, skipping',
+      );
+      return;
+    }
+
+    const loopCmdMsg = groupMessages.find(
+      (msg) => extractSessionCommand(msg.content, deps.triggerPattern) !== null,
+    );
+
+    if (loopCmdMsg) {
+      if (
+        isSessionCommandAllowed(
+          isMainGroup,
+          loopCmdMsg.is_from_me === true,
+          isSessionCommandSenderAllowed(loopCmdMsg.sender),
+        )
+      ) {
+        deps.queue.closeStdin(chatJid, {
+          reason: 'session-command-detected',
+        });
+      }
+      deps.queue.enqueueMessageCheck(chatJid);
+      return;
+    }
+
+    if (
+      !hasAllowedTrigger({
+        chatJid,
+        messages: processableGroupMessages,
+        group,
+        triggerPattern: deps.triggerPattern,
+        hasImplicitContinuationWindow: continuationTracker.has,
+      })
+    ) {
+      return;
+    }
+
+    await processQueuedGroupDispatch({
+      chatJid,
+      group,
+      channel,
+      processableGroupMessages,
+    });
+  };
+
  const startMessageLoop = async (): Promise<void> => {
    if (messageLoopRunning) {
      logger.debug('Message loop already running, skipping duplicate start');
@@ -1120,193 +1301,12 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): {
              );
              continue;
            }
-
-            const isMainGroup = group.isMain === true;
-            const processableGroupMessages = getProcessableMessages(
+            await processLoopGroupMessages({
              chatJid,
+              group,
              groupMessages,
              channel,
-            );
-
-            if (processableGroupMessages.length === 0) {
-              const lastIgnored = groupMessages[groupMessages.length - 1];
-              if (lastIgnored?.seq != null) {
-                advanceLastAgentCursor(
-                  deps.getLastAgentTimestamps(),
-                  deps.saveState,
-                  chatJid,
-                  lastIgnored.seq,
-                );
-              }
-              continue;
-            }
-
-            if (
-              shouldSkipBotOnlyCollaboration(chatJid, processableGroupMessages)
-            ) {
-              const lastIgnored =
-                processableGroupMessages[processableGroupMessages.length - 1];
-              if (lastIgnored?.seq != null) {
-                advanceLastAgentCursor(
-                  deps.getLastAgentTimestamps(),
-                  deps.saveState,
-                  chatJid,
-                  lastIgnored.seq,
-                );
-              }
-              logger.info(
-                { chatJid, group: group.name, groupFolder: group.folder },
-                'Bot-collaboration timeout: no recent human message, skipping',
-              );
-              continue;
-            }
-
-            const loopCmdMsg = groupMessages.find(
-              (msg) =>
-                extractSessionCommand(msg.content, deps.triggerPattern) !==
-                null,
-            );
-
-            if (loopCmdMsg) {
-              if (
-                isSessionCommandAllowed(
-                  isMainGroup,
-                  loopCmdMsg.is_from_me === true,
-                  isSessionCommandSenderAllowed(loopCmdMsg.sender),
-                )
-              ) {
-                deps.queue.closeStdin(chatJid, {
-                  reason: 'session-command-detected',
-                });
-              }
-              deps.queue.enqueueMessageCheck(chatJid);
-              continue;
-            }
-
-            if (
-              !hasAllowedTrigger({
-                chatJid,
-                messages: processableGroupMessages,
-                group,
-                triggerPattern: deps.triggerPattern,
-                hasImplicitContinuationWindow: continuationTracker.has,
-              })
-            ) {
-              continue;
-            }
-
-            // Use role-aware cursor for paired rooms so the reviewer
-            // always sees the owner's last messages and vice versa.
-            const loopPendingTask = hasReviewerLease(chatJid)
-              ? getLatestOpenPairedTaskForChat(chatJid)
-              : null;
-            const loopCursorKey = resolveCursorKey(
-              chatJid,
-              loopPendingTask?.status,
-            );
-
-            const rawPendingMessages = getMessagesSinceSeq(
-              chatJid,
-              deps.getLastAgentTimestamps()[loopCursorKey] || '0',
-              deps.assistantName,
-            );
-            const pendingMessages = filterLoopingPairedBotMessages(
-              chatJid,
-              getProcessableMessages(chatJid, rawPendingMessages, channel),
-              FAILURE_FINAL_TEXT,
-            );
-            const messagesToSend =
-              pendingMessages.length > 0
-                ? pendingMessages
-                : processableGroupMessages;
-            const labeledMessagesToSend = labelPairedSenders(
-              chatJid,
-              messagesToSend,
-            );
-            const formatted = loopPendingTask
-              ? buildPairedTurnPrompt({
-                  taskId: loopPendingTask.id,
-                  chatJid,
-                  timezone: deps.timezone,
-                  missedMessages: messagesToSend,
-                  labeledFallbackMessages: labeledMessagesToSend,
-                  turnOutputs: getPairedTurnOutputs(loopPendingTask.id),
-                })
-              : formatMessages(labeledMessagesToSend, deps.timezone);
-            const isBotOnlyPairedFollowUp = isBotOnlyPairedRoomTurn(
-              chatJid,
-              messagesToSend,
-            );
-            const pendingCursorSource =
-              rawPendingMessages.length > 0
-                ? rawPendingMessages[rawPendingMessages.length - 1]
-                : messagesToSend[messagesToSend.length - 1];
-            const botOnlyFollowUpAction = resolveBotOnlyPairedFollowUpAction({
-              chatJid,
-              task: loopPendingTask,
-              isBotOnlyPairedFollowUp,
-              lastDeliveredMessages: labeledMessagesToSend,
-              pendingCursorSource,
            });
-            if (
-              await executeBotOnlyPairedFollowUpAction({
-                action: botOnlyFollowUpAction,
-                chatJid,
-                group,
-                runId: `loop-merge-ready-${Date.now().toString(36)}`,
-                channel,
-                log: logger,
-                saveState: deps.saveState,
-                lastAgentTimestamps: deps.getLastAgentTimestamps(),
-                executeTurn,
-                enqueueGroupMessageCheck: () =>
-                  enqueueScopedGroupMessageCheck(chatJid, group.folder),
-                closeStdin: () =>
-                  deps.queue.closeStdin(chatJid, {
-                    reason: 'paired-pending-turn-follow-up',
-                  }),
-              })
-            ) {
-              continue;
-            }
-
-            if (deps.queue.sendMessage(chatJid, formatted)) {
-              const endSeq = messagesToSend[messagesToSend.length - 1]?.seq;
-              if (endSeq != null) {
-                advanceLastAgentCursor(
-                  deps.getLastAgentTimestamps(),
-                  deps.saveState,
-                  chatJid,
-                  endSeq,
-                  loopCursorKey,
-                );
-              }
-              logger.debug(
-                {
-                  transition: 'typing:on',
-                  source: 'follow-up-queued',
-                  chatJid,
-                  group: group.name,
-                  groupFolder: group.folder,
-                  endSeq: endSeq ?? null,
-                  suppressed: isBotOnlyPairedFollowUp,
-                },
-                'Typing indicator transition',
-              );
-              if (!isBotOnlyPairedFollowUp) {
-                await channel
-                  .setTyping?.(chatJid, true)
-                  ?.catch((err) =>
-                    logger.warn(
-                      { chatJid, err },
-                      'Failed to set typing indicator',
-                    ),
-                  );
-              }
-              continue;
-            }
-
-            enqueueScopedGroupMessageCheck(chatJid, group.folder);
          }
        }
      } catch (err) {
--- a/src/paired-execution-context-arbiter.ts
+++ b/src/paired-execution-context-arbiter.ts
@@ -1,7 +1,9 @@
 import { ARBITER_DEADLOCK_THRESHOLD } from './config.js';
-import { updatePairedTask } from './db.js';
 import { logger } from './logger.js';
-import { classifyArbiterVerdict } from './paired-execution-context-shared.js';
+import {
+  classifyArbiterVerdict,
+  transitionPairedTaskStatus,
+} from './paired-execution-context-shared.js';
 import type { PairedTask } from './types.js';

 export function handleFailedArbiterExecution(args: {
@@ -15,7 +17,12 @@ export function handleFailedArbiterExecution(args: {
      ? 'arbiter_requested'
      : task.status;
  if (fallbackStatus !== task.status) {
-    updatePairedTask(taskId, { status: fallbackStatus, updated_at: now });
+    transitionPairedTaskStatus({
+      taskId,
+      currentStatus: task.status,
+      nextStatus: fallbackStatus,
+      updatedAt: now,
+    });
    logger.warn(
      {
        taskId,
@@ -45,11 +52,15 @@ export function handleArbiterCompletion(args: {
    case 'proceed':
    case 'revise':
    case 'reset':
-      updatePairedTask(taskId, {
-        status: 'active',
-        round_trip_count: Math.max(0, ARBITER_DEADLOCK_THRESHOLD - 1),
-        arbiter_verdict: arbiterVerdict,
-        updated_at: now,
+      transitionPairedTaskStatus({
+        taskId,
+        currentStatus: 'in_arbitration',
+        nextStatus: 'active',
+        updatedAt: now,
+        patch: {
+          round_trip_count: Math.max(0, ARBITER_DEADLOCK_THRESHOLD - 1),
+          arbiter_verdict: arbiterVerdict,
+        },
      });
      logger.info(
        { taskId, arbiterVerdict },
@@ -57,20 +68,28 @@ export function handleArbiterCompletion(args: {
      );
      return;
    case 'escalate':
-      updatePairedTask(taskId, {
-        status: 'completed',
-        arbiter_verdict: 'escalate',
-        completion_reason: 'arbiter_escalated',
-        updated_at: now,
+      transitionPairedTaskStatus({
+        taskId,
+        currentStatus: 'in_arbitration',
+        nextStatus: 'completed',
+        updatedAt: now,
+        patch: {
+          arbiter_verdict: 'escalate',
+          completion_reason: 'arbiter_escalated',
+        },
      });
      logger.info({ taskId }, 'Arbiter escalated to user — task completed');
      return;
    default:
-      updatePairedTask(taskId, {
-        status: 'active',
-        round_trip_count: Math.max(0, ARBITER_DEADLOCK_THRESHOLD - 1),
-        arbiter_verdict: 'unknown',
-        updated_at: now,
+      transitionPairedTaskStatus({
+        taskId,
+        currentStatus: 'in_arbitration',
+        nextStatus: 'active',
+        updatedAt: now,
+        patch: {
+          round_trip_count: Math.max(0, ARBITER_DEADLOCK_THRESHOLD - 1),
+          arbiter_verdict: 'unknown',
+        },
      });
      logger.warn(
        { taskId, summary: summary?.slice(0, 200) },
--- a/src/paired-execution-context-owner.ts
+++ b/src/paired-execution-context-owner.ts
@@ -5,15 +5,16 @@ import {
 import {
  getPairedWorkspace,
  hasActiveCiWatcherForChat,
-  updatePairedTask,
 } from './db.js';
 import { logger } from './logger.js';
 import { markPairedTaskReviewReady } from './paired-workspace-manager.js';
 import {
+  applyPairedTaskPatch,
  classifyVerdict,
  hasCodeChangesSinceRef,
  requestArbiterOrEscalate,
  resolveCanonicalSourceRef,
+  transitionPairedTaskStatus,
 } from './paired-execution-context-shared.js';
 import type { PairedTask } from './types.js';

@@ -24,7 +25,12 @@ export function handleFailedOwnerExecution(args: {
  const { task, taskId } = args;
  if (task.status !== 'active') {
    const now = new Date().toISOString();
-    updatePairedTask(taskId, { status: 'active', updated_at: now });
+    transitionPairedTaskStatus({
+      taskId,
+      currentStatus: task.status,
+      nextStatus: 'active',
+      updatedAt: now,
+    });
    logger.info(
      { taskId, role: 'owner', previousStatus: task.status },
      'Reset task to active after failed execution',
@@ -44,6 +50,7 @@ function handleOwnerFinalizeCompletion(args: {
  if (ownerVerdict === 'blocked' || ownerVerdict === 'needs_context') {
    requestArbiterOrEscalate({
      taskId,
+      currentStatus: task.status,
      now,
      arbiterLogMessage: 'Owner blocked during finalize — requesting arbiter',
      escalateLogMessage: 'Owner blocked during finalize — escalating to user',
@@ -60,6 +67,7 @@ function handleOwnerFinalizeCompletion(args: {
    if (task.round_trip_count >= ARBITER_DEADLOCK_THRESHOLD) {
      requestArbiterOrEscalate({
        taskId,
+        currentStatus: task.status,
        now,
        arbiterLogMessage: 'Owner finalize loop detected — requesting arbiter',
        escalateLogMessage: 'Owner finalize loop detected — escalating to user',
@@ -71,7 +79,12 @@ function handleOwnerFinalizeCompletion(args: {
      });
      return true;
    }
-    updatePairedTask(taskId, { status: 'active', updated_at: now });
+    transitionPairedTaskStatus({
+      taskId,
+      currentStatus: task.status,
+      nextStatus: 'active',
+      updatedAt: now,
+    });
    logger.info(
      {
        taskId,
@@ -92,6 +105,7 @@ function handleOwnerFinalizeCompletion(args: {
    if (task.round_trip_count >= ARBITER_DEADLOCK_THRESHOLD) {
      requestArbiterOrEscalate({
        taskId,
+        currentStatus: task.status,
        now,
        arbiterLogMessage:
          'Owner finalize DONE loop detected — requesting arbiter',
@@ -116,10 +130,14 @@ function handleOwnerFinalizeCompletion(args: {
    return false;
  }

-  updatePairedTask(taskId, {
-    status: 'completed',
-    completion_reason: 'done',
-    updated_at: now,
+  transitionPairedTaskStatus({
+    taskId,
+    currentStatus: task.status,
+    nextStatus: 'completed',
+    updatedAt: now,
+    patch: {
+      completion_reason: 'done',
+    },
  });
  logger.info(
    { taskId, hasNewChanges, summary: summary?.slice(0, 100) },
@@ -161,6 +179,7 @@ export function handleOwnerCompletion(args: {
    if (ownerVerdict === 'blocked' || ownerVerdict === 'needs_context') {
      requestArbiterOrEscalate({
        taskId,
+        currentStatus: task.status,
        now,
        arbiterLogMessage: 'Owner blocked/needs_context — requesting arbiter',
        escalateLogMessage: 'Owner blocked/needs_context — escalating to user',
@@ -188,10 +207,12 @@ export function handleOwnerCompletion(args: {

  const result = markPairedTaskReviewReady(taskId);
  if (result) {
-    updatePairedTask(taskId, {
-      round_trip_count: task.round_trip_count + 1,
-      review_requested_at: now,
-      updated_at: now,
+    applyPairedTaskPatch({
+      taskId,
+      updatedAt: now,
+      patch: {
+        round_trip_count: task.round_trip_count + 1,
+      },
    });
    logger.info(
      { taskId, roundTrip: task.round_trip_count + 1 },
--- a/src/paired-execution-context-reviewer.ts
+++ b/src/paired-execution-context-reviewer.ts
@@ -5,6 +5,7 @@ import {
  classifyVerdict,
  requestArbiterOrEscalate,
  resolveCanonicalSourceRef,
+  transitionPairedTaskStatus,
 } from './paired-execution-context-shared.js';
 import type { PairedTask } from './types.js';

@@ -29,11 +30,15 @@ export function handleFailedReviewerExecution(args: {
        verdict === 'done' && ownerWs?.workspace_dir
          ? resolveCanonicalSourceRef(ownerWs.workspace_dir)
          : task.source_ref;
-      updatePairedTask(taskId, {
-        status: verdict === 'done' ? 'merge_ready' : 'completed',
-        ...(verdict === 'done' ? { source_ref: approvedSourceRef } : {}),
-        ...(verdict !== 'done' ? { completion_reason: 'escalated' } : {}),
-        updated_at: now,
+      transitionPairedTaskStatus({
+        taskId,
+        currentStatus: task.status,
+        nextStatus: verdict === 'done' ? 'merge_ready' : 'completed',
+        updatedAt: now,
+        patch: {
+          ...(verdict === 'done' ? { source_ref: approvedSourceRef } : {}),
+          ...(verdict !== 'done' ? { completion_reason: 'escalated' } : {}),
+        },
      });
      logger.info(
        {
@@ -53,7 +58,12 @@ export function handleFailedReviewerExecution(args: {
      ? 'review_ready'
      : task.status;
  if (fallbackStatus !== task.status) {
-    updatePairedTask(taskId, { status: fallbackStatus, updated_at: now });
+    transitionPairedTaskStatus({
+      taskId,
+      currentStatus: task.status,
+      nextStatus: fallbackStatus,
+      updatedAt: now,
+    });
    logger.warn(
      {
        taskId,
@@ -81,10 +91,14 @@ export function handleReviewerCompletion(args: {
      const approvedSourceRef = ownerWs?.workspace_dir
        ? resolveCanonicalSourceRef(ownerWs.workspace_dir)
        : task.source_ref;
-      updatePairedTask(taskId, {
-        status: 'merge_ready',
-        source_ref: approvedSourceRef,
-        updated_at: now,
+      transitionPairedTaskStatus({
+        taskId,
+        currentStatus: task.status,
+        nextStatus: 'merge_ready',
+        updatedAt: now,
+        patch: {
+          source_ref: approvedSourceRef,
+        },
      });
      logger.info(
        {
@@ -102,6 +116,7 @@ export function handleReviewerCompletion(args: {
    case 'needs_context':
      requestArbiterOrEscalate({
        taskId,
+        currentStatus: task.status,
        now,
        arbiterLogMessage:
          'Reviewer blocked/needs_context — requesting arbiter before escalating',
@@ -116,6 +131,7 @@ export function handleReviewerCompletion(args: {
      if (task.round_trip_count >= ARBITER_DEADLOCK_THRESHOLD) {
        requestArbiterOrEscalate({
          taskId,
+          currentStatus: task.status,
          now,
          arbiterLogMessage:
            'Deadlock detected — requesting arbiter intervention',
@@ -129,7 +145,12 @@ export function handleReviewerCompletion(args: {
        });
        return;
      }
-      updatePairedTask(taskId, { status: 'active', updated_at: now });
+      transitionPairedTaskStatus({
+        taskId,
+        currentStatus: task.status,
+        nextStatus: 'active',
+        updatedAt: now,
+      });
      logger.info(
        { taskId, verdict },
        'Reviewer has feedback, task set back to active for owner',
--- a/src/paired-execution-context-shared.ts
+++ b/src/paired-execution-context-shared.ts
@@ -3,6 +3,7 @@ import { execFileSync } from 'child_process';
 import { isArbiterEnabled } from './config.js';
 import { updatePairedTask } from './db.js';
 import { logger } from './logger.js';
+import type { PairedTaskStatus } from './types.js';

 export type Verdict =
  | 'done'
@@ -93,29 +94,116 @@ export function hasCodeChangesSinceRef(
  }
 }

+const ALLOWED_PAIRED_STATUS_TRANSITIONS: Record<
+  PairedTaskStatus,
+  ReadonlySet<PairedTaskStatus>
+> = {
+  active: new Set(['review_ready', 'arbiter_requested', 'completed']),
+  review_ready: new Set([
+    'active',
+    'in_review',
+    'arbiter_requested',
+    'completed',
+  ]),
+  in_review: new Set([
+    'active',
+    'review_ready',
+    'merge_ready',
+    'arbiter_requested',
+    'completed',
+  ]),
+  merge_ready: new Set(['active', 'arbiter_requested', 'completed']),
+  completed: new Set(),
+  arbiter_requested: new Set(['in_arbitration', 'completed']),
+  in_arbitration: new Set(['active', 'arbiter_requested', 'completed']),
+};
+
+export function assertPairedTaskStatusTransition(args: {
+  currentStatus: PairedTaskStatus;
+  nextStatus: PairedTaskStatus;
+}): void {
+  const { currentStatus, nextStatus } = args;
+  if (currentStatus === nextStatus) {
+    return;
+  }
+
+  if (ALLOWED_PAIRED_STATUS_TRANSITIONS[currentStatus].has(nextStatus)) {
+    return;
+  }
+
+  throw new Error(
+    `Invalid paired task status transition: ${currentStatus} -> ${nextStatus}`,
+  );
+}
+
+export function transitionPairedTaskStatus(args: {
+  taskId: string;
+  currentStatus: PairedTaskStatus;
+  nextStatus: PairedTaskStatus;
+  updatedAt: string;
+  patch?: Omit<Parameters<typeof updatePairedTask>[1], 'status' | 'updated_at'>;
+}): void {
+  assertPairedTaskStatusTransition({
+    currentStatus: args.currentStatus,
+    nextStatus: args.nextStatus,
+  });
+
+  updatePairedTask(args.taskId, {
+    ...args.patch,
+    status: args.nextStatus,
+    updated_at: args.updatedAt,
+  });
+}
+
+export function applyPairedTaskPatch(args: {
+  taskId: string;
+  updatedAt: string;
+  patch: Omit<Parameters<typeof updatePairedTask>[1], 'updated_at'>;
+}): void {
+  updatePairedTask(args.taskId, {
+    ...args.patch,
+    updated_at: args.updatedAt,
+  });
+}
+
 export function requestArbiterOrEscalate(args: {
  taskId: string;
+  currentStatus: PairedTaskStatus;
  now: string;
  arbiterLogMessage: string;
  escalateLogMessage: string;
  logContext?: Record<string, unknown>;
 }): void {
-  const { taskId, now, arbiterLogMessage, escalateLogMessage, logContext } =
-    args;
+  const {
+    taskId,
+    currentStatus,
+    now,
+    arbiterLogMessage,
+    escalateLogMessage,
+    logContext,
+  } = args;
  if (isArbiterEnabled()) {
-    updatePairedTask(taskId, {
-      status: 'arbiter_requested',
-      arbiter_requested_at: now,
-      updated_at: now,
+    transitionPairedTaskStatus({
+      taskId,
+      currentStatus,
+      nextStatus: 'arbiter_requested',
+      updatedAt: now,
+      patch: {
+        arbiter_requested_at: now,
+      },
    });
    logger.info(logContext ?? { taskId }, arbiterLogMessage);
    return;
  }

-  updatePairedTask(taskId, {
-    status: 'completed',
-    completion_reason: 'escalated',
-    updated_at: now,
+  transitionPairedTaskStatus({
+    taskId,
+    currentStatus,
+    nextStatus: 'completed',
+    updatedAt: now,
+    patch: {
+      completion_reason: 'escalated',
+    },
  });
  logger.info(logContext ?? { taskId }, escalateLogMessage);
 }
--- a/src/paired-execution-context.test.ts
+++ b/src/paired-execution-context.test.ts
@@ -565,7 +565,6 @@ describe('paired execution context', () => {
      'task-1',
      expect.objectContaining({
        round_trip_count: 2,
-        review_requested_at: expect.any(String),
      }),
    );
  });
--- a/src/paired-execution-context.ts
+++ b/src/paired-execution-context.ts
@@ -19,7 +19,6 @@ import {
  getPairedTaskById,
  getPairedWorkspace,
  hasActiveCiWatcherForChat,
-  updatePairedTask,
  upsertPairedProject,
 } from './db.js';
 import { logger } from './logger.js';
@@ -36,8 +35,10 @@ import {
  handleReviewerCompletion,
 } from './paired-execution-context-reviewer.js';
 import {
+  applyPairedTaskPatch,
  resolveCanonicalSourceRef,
  requestArbiterOrEscalate,
+  transitionPairedTaskStatus,
 } from './paired-execution-context-shared.js';
 import {
  markPairedTaskReviewReady,
@@ -226,11 +227,25 @@ export function preparePairedExecutionContext(args: {
      latestTask.status === 'review_ready' ||
      latestTask.status === 'in_review';
    if (hasHuman || needsStatusReset) {
-      updatePairedTask(latestTask.id, {
-        ...(hasHuman ? { round_trip_count: 0 } : {}),
-        ...(needsStatusReset ? { status: 'active' as const } : {}),
-        updated_at: now,
-      });
+      if (needsStatusReset) {
+        transitionPairedTaskStatus({
+          taskId: latestTask.id,
+          currentStatus: latestTask.status,
+          nextStatus: 'active',
+          updatedAt: now,
+          patch: {
+            ...(hasHuman ? { round_trip_count: 0 } : {}),
+          },
+        });
+      } else {
+        applyPairedTaskPatch({
+          taskId: latestTask.id,
+          updatedAt: now,
+          patch: {
+            ...(hasHuman ? { round_trip_count: 0 } : {}),
+          },
+        });
+      }
    }
    // Use a stable per-channel worktree (not per-task) so the Claude SDK
    // session persists across tasks. Different channels still get isolation.
@@ -241,9 +256,12 @@ export function preparePairedExecutionContext(args: {
    if (workspace?.workspace_dir && latestTask.status === 'active') {
      const wsRef = resolveCanonicalSourceRef(workspace.workspace_dir);
      if (wsRef !== latestTask.source_ref) {
-        updatePairedTask(latestTask.id, {
-          source_ref: wsRef,
-          updated_at: now,
+        applyPairedTaskPatch({
+          taskId: latestTask.id,
+          updatedAt: now,
+          patch: {
+            source_ref: wsRef,
+          },
        });
      }
    }
@@ -253,9 +271,11 @@ export function preparePairedExecutionContext(args: {
    blockMessage = reviewerWorkspace.blockMessage;
    const refreshedTask = getPairedTaskById(latestTask.id) ?? latestTask;
    if (workspace && refreshedTask.status === 'review_ready') {
-      updatePairedTask(latestTask.id, {
-        status: 'in_review',
-        updated_at: now,
+      transitionPairedTaskStatus({
+        taskId: latestTask.id,
+        currentStatus: refreshedTask.status,
+        nextStatus: 'in_review',
+        updatedAt: now,
      });
    }
  } else if (roomRoleContext.role === 'arbiter') {
@@ -265,9 +285,11 @@ export function preparePairedExecutionContext(args: {
    blockMessage = reviewerWorkspace.blockMessage;
    const refreshedTask = getPairedTaskById(latestTask.id) ?? latestTask;
    if (workspace && refreshedTask.status === 'arbiter_requested') {
-      updatePairedTask(latestTask.id, {
-        status: 'in_arbitration',
-        updated_at: now,
+      transitionPairedTaskStatus({
+        taskId: latestTask.id,
+        currentStatus: refreshedTask.status,
+        nextStatus: 'in_arbitration',
+        updatedAt: now,
      });
    }
  }
--- a/src/paired-workspace-manager.test.ts
+++ b/src/paired-workspace-manager.test.ts
@@ -258,6 +258,31 @@ describe('paired workspace manager', () => {
    );
  });

+  it('leaves review_requested_at untouched when review handoff aborts before an owner workspace exists', async () => {
+    const { db, manager } = await loadModules();
+    db._initTestDatabase();
+
+    const canonicalDir = path.join(tempRoot, 'canonical');
+    initCanonicalRepo(canonicalDir);
+    seedPairedTask(db, canonicalDir, {
+      taskId: 'paired-task-no-owner-workspace',
+      groupFolder: 'no-owner-workspace-room',
+    });
+
+    const result = manager.markPairedTaskReviewReady(
+      'paired-task-no-owner-workspace',
+    );
+
+    expect(result).toBeNull();
+    expect(
+      db.getPairedTaskById('paired-task-no-owner-workspace')
+        ?.review_requested_at,
+    ).toBeNull();
+    expect(
+      db.getPairedTaskById('paired-task-no-owner-workspace')?.status,
+    ).toBe('active');
+  });
+
  it('uses the shared DB owner workspace across service-local data dirs', async () => {
    const canonicalDir = path.join(tempRoot, 'canonical');
    fs.mkdirSync(canonicalDir, { recursive: true });
--- a/src/paired-workspace-manager.ts
+++ b/src/paired-workspace-manager.ts
@@ -8,11 +8,11 @@ import {
  getPairedProject,
  getPairedTaskById,
  getPairedWorkspace,
-  updatePairedTask,
  upsertPairedWorkspace,
 } from './db.js';
 import { resolvePairedTaskWorkspacePath } from './group-folder.js';
 import { logger } from './logger.js';
+import { transitionPairedTaskStatus } from './paired-execution-context-shared.js';
 import type { PairedTask, PairedWorkspace } from './types.js';
 import { ensureWorkspaceDependenciesInstalled } from './workspace-package-manager.js';

@@ -757,10 +757,6 @@ export function markPairedTaskReviewReady(taskId: string): {
  reviewerWorkspace: PairedWorkspace;
 } | null {
  const requestedAt = new Date().toISOString();
-  updatePairedTask(taskId, {
-    review_requested_at: requestedAt,
-    updated_at: requestedAt,
-  });

  const ownerWorkspace = getPairedWorkspace(taskId, 'owner');
  if (!ownerWorkspace) {
@@ -797,10 +793,18 @@ export function markPairedTaskReviewReady(taskId: string): {
    'Reviewer will mount owner workspace directly',
  );

-  const now = new Date().toISOString();
-  updatePairedTask(taskId, {
-    status: 'review_ready',
-    updated_at: now,
+  const task = getPairedTaskById(taskId);
+  if (!task) {
+    return null;
+  }
+  transitionPairedTaskStatus({
+    taskId,
+    currentStatus: task.status,
+    nextStatus: 'review_ready',
+    updatedAt: requestedAt,
+    patch: {
+      review_requested_at: requestedAt,
+    },
  });

  return { ownerWorkspace, reviewerWorkspace };
--- a/src/task-scheduler.ts
+++ b/src/task-scheduler.ts
@@ -3,6 +3,11 @@ import { CronExpressionParser } from 'cron-parser';
 import fs from 'fs';
 import { getAgentOutputText } from './agent-output.js';
 import { createEvaluatedOutputHandler } from './agent-attempt.js';
+import {
+  executeAttemptRetryAction,
+  runClaudeAttemptWithRotation,
+  runCodexAttemptWithRotation,
+} from './agent-attempt-orchestration.js';
 import { getErrorMessage } from './utils.js';

 import { ASSISTANT_NAME, SCHEDULER_POLL_INTERVAL, TIMEZONE } from './config.js';
@@ -28,10 +33,6 @@ import {
 } from './group-folder.js';
 import { createScopedLogger, logger } from './logger.js';
 import { createTaskStatusTracker } from './task-status-tracker.js';
-import {
-  runClaudeRotationLoop,
-  runCodexRotationLoop,
-} from './provider-retry.js';
 import {
  detectCodexRotationTrigger,
  rotateCodexToken,
@@ -42,7 +43,6 @@ import {
  classifyRotationTrigger,
  type AgentTriggerReason,
  type CodexRotationReason,
-  isCodexRotationReason,
 } from './agent-error-detection.js';
 import {
  getTokenCount,
@@ -447,69 +447,53 @@ async function runTask(
        retryAfterMs?: number;
      },
      rotationMessage?: string,
-    ): Promise<void> => {
-      const logCtx = {
+    ): Promise<'success' | 'error'> => {
+      const logContext = {
        taskId: task.id,
        group: context.group.name,
        groupFolder: task.group_folder,
      };

-      const outcome = await runClaudeRotationLoop(
+      const outcome = await runClaudeAttemptWithRotation({
        initialTrigger,
-        async () => {
-          const attempt = await runTaskAttempt('claude');
+        runAttempt: () => runTaskAttempt('claude'),
+        logContext,
+        rotationMessage,
+        afterAttempt: (attempt) => {
          result = attempt.attemptResult;
          error = attempt.attemptError;
-          return {
-            output: attempt.output,
-            sawOutput: attempt.sawOutput,
-            streamedTriggerReason: attempt.streamedTriggerReason,
-          };
        },
-        logCtx,
-        rotationMessage,
-      );
+      });

-      switch (outcome.type) {
-        case 'success':
-          error = null;
-          return;
-        case 'error':
-          if (outcome.trigger) {
-            error = `Claude ${outcome.trigger.reason}`;
-          }
-          return;
+      if (outcome === 'success') {
+        error = null;
      }
+      return outcome;
    };

    const retryCodexTaskWithRotation = async (
      initialTrigger: { reason: CodexRotationReason },
      rotationMessage?: string,
-    ): Promise<void> => {
-      const outcome = await runCodexRotationLoop(
+    ): Promise<'success' | 'error'> => {
+      const outcome = await runCodexAttemptWithRotation({
        initialTrigger,
-        async () => {
-          const retryAttempt = await runTaskAttempt('codex');
-          result = retryAttempt.attemptResult;
-          error = retryAttempt.attemptError;
-          return {
-            output: retryAttempt.output,
-            thrownError: null,
-            sawOutput: retryAttempt.sawOutput,
-            streamedTriggerReason: retryAttempt.streamedTriggerReason,
-          };
-        },
-        {
+        runAttempt: () => runTaskAttempt('codex'),
+        logContext: {
          taskId: task.id,
          group: context.group.name,
          groupFolder: task.group_folder,
        },
        rotationMessage,
-      );
+        afterAttempt: (attempt) => {
+          result = attempt.attemptResult;
+          error = attempt.attemptError;
+        },
+      });

-      if (outcome.type === 'success') {
+      if (outcome === 'success') {
        error = null;
      }
+      return outcome;
    };

    const provider = context.taskAgentType === 'codex' ? 'codex' : 'claude';
@@ -519,56 +503,17 @@ async function runTask(
      result = attempt.attemptResult;
      error = attempt.attemptError;

-      if (
-        provider === 'claude' &&
-        attempt.streamedTriggerReason &&
-        !attempt.sawOutput
-      ) {
-        await retryClaudeTaskWithRotation(attempt.streamedTriggerReason);
-      } else if (
-        provider === 'codex' &&
-        attempt.streamedTriggerReason &&
-        !attempt.sawOutput &&
-        isCodexRotationReason(attempt.streamedTriggerReason.reason)
-      ) {
-        await retryCodexTaskWithRotation(
-          {
-            reason: attempt.streamedTriggerReason.reason,
-          },
-          typeof attempt.output.error === 'string'
-            ? attempt.output.error
-            : undefined,
-        );
-      } else if (attempt.output.status === 'error' && provider === 'claude') {
-        const trigger = attempt.streamedTriggerReason
-          ? {
-              shouldRetry: true,
-              reason: attempt.streamedTriggerReason.reason,
-              retryAfterMs: attempt.streamedTriggerReason.retryAfterMs,
-            }
-          : classifyRotationTrigger(error);
-        if (trigger.shouldRetry) {
-          await retryClaudeTaskWithRotation({
-            reason: trigger.reason,
-            retryAfterMs: trigger.retryAfterMs,
-          });
-        }
-      } else if (attempt.output.status === 'error' && provider === 'codex') {
-        const trigger =
-          attempt.streamedTriggerReason &&
-          isCodexRotationReason(attempt.streamedTriggerReason.reason)
-            ? {
-                shouldRotate: true as const,
-                reason: attempt.streamedTriggerReason.reason,
-              }
-            : detectCodexRotationTrigger(error);
-        if (trigger.shouldRotate) {
-          await retryCodexTaskWithRotation(
-            { reason: trigger.reason },
-            error || undefined,
-          );
-        }
-      } else if (attempt.output.status === 'error') {
+      const retryAction = await executeAttemptRetryAction({
+        provider,
+        canRetryClaudeCredentials: provider === 'claude' && getTokenCount() > 0,
+        canRetryCodex: provider === 'codex' && getCodexAccountCount() > 1,
+        attempt,
+        rotationMessage: error,
+        runClaude: retryClaudeTaskWithRotation,
+        runCodex: retryCodexTaskWithRotation,
+      });
+
+      if (retryAction.kind === 'none' && attempt.output.status === 'error') {
        error = attempt.attemptError || 'Unknown error';
      }
    } // end else (non-exhausted path)
--- a/src/token-rotation.test.ts
+++ b/src/token-rotation.test.ts
@@ -65,4 +65,44 @@ describe('token-rotation runtime reselection', () => {
      rateLimited: 0,
    });
  });
+
+  it('reloads Claude rotation state from disk written by another service', async () => {
+    const mod = await import('./token-rotation.js');
+    const utils = await import('./utils.js');
+    const readJsonFile = vi.mocked(utils.readJsonFile);
+
+    readJsonFile.mockReturnValueOnce(null);
+    mod.initTokenRotation();
+    expect(mod.getCurrentToken()).toBe('token-1');
+
+    readJsonFile.mockReturnValueOnce({
+      currentIndex: 1,
+      rateLimits: [Date.now() + 60_000, null],
+    });
+    mod.reloadTokenRotationStateFromDisk();
+
+    expect(mod.getCurrentTokenIndex()).toBe(1);
+    expect(mod.getCurrentToken()).toBe('token-2');
+  });
+
+  it('warns when Claude rotation state cannot be persisted', async () => {
+    const mod = await import('./token-rotation.js');
+    const utils = await import('./utils.js');
+    const { logger } = await import('./logger.js');
+
+    vi.mocked(utils.writeJsonFile).mockImplementation(() => {
+      throw new Error('disk full');
+    });
+
+    mod.initTokenRotation();
+    expect(mod.rotateToken('rate limit')).toBe(true);
+
+    expect(logger.warn).toHaveBeenCalledWith(
+      expect.objectContaining({
+        stateFile: '/tmp/ejclaw-claude-rot-data/token-rotation-state.json',
+        err: expect.any(Error),
+      }),
+      'Failed to persist Claude token rotation state',
+    );
+  });
 });
--- a/src/token-rotation.ts
+++ b/src/token-rotation.ts
@@ -69,12 +69,15 @@ function saveState(): void {
      rateLimits: tokens.map((t) => t.rateLimitedUntil),
    };
    writeJsonFile(STATE_FILE, state);
-  } catch {
-    /* best effort */
+  } catch (err) {
+    logger.warn(
+      { stateFile: STATE_FILE, err },
+      'Failed to persist Claude token rotation state',
+    );
  }
 }

-function loadState(): void {
+function loadState(quiet = false): void {
  const state = readJsonFile<{
    currentIndex?: number;
    rateLimits?: (number | null)[];
@@ -93,13 +96,27 @@ function loadState(): void {
      const until = state.rateLimits[i];
      if (typeof until === 'number' && until > now) {
        tokens[i].rateLimitedUntil = until;
+      } else {
+        tokens[i].rateLimitedUntil = null;
      }
    }
  }
-  logger.info(
-    { currentIndex, tokenCount: tokens.length },
-    'Token rotation state restored',
-  );
+  if (!quiet) {
+    logger.info(
+      { currentIndex, tokenCount: tokens.length },
+      'Token rotation state restored',
+    );
+  }
+}
+
+/**
+ * Re-read the on-disk rotation state (written by any service).
+ * Call before dashboard renders so the renderer picks up rotations
+ * performed by another Claude service process.
+ */
+export function reloadTokenRotationStateFromDisk(): void {
+  if (tokens.length <= 1) return;
+  loadState(true);
 }

 function refreshRuntimeTokenSelection(): void {