claude-bot/EJClaw
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Unify paired status transitions and retry handling

Browse Source
This commit is contained in:
ejclaw
2026-04-07 02:49:02 +09:00
parent 4511ca5692
commit b8e8c9dabf
13 changed files with 574 additions and 280 deletions
Download Patch File Download Diff File Expand all files Collapse all files

164
src/agent-attempt-retry.ts Normal file
View File

@@ -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' };
}

42
src/message-agent-executor-rules.test.ts
View File

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

120
src/message-agent-executor-rules.ts
View File

@@ -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'

32
src/message-agent-executor.ts
View File

@@ -5,6 +5,10 @@ import { getErrorMessage } from './utils.js';
import { getAgentOutputText } from './agent-output.js';
import { createEvaluatedOutputHandler } from './agent-attempt.js';
import {
isRetryableClaudeSessionFailureAttempt,
resolveAttemptRetryAction,
} from './agent-attempt-retry.js';
import {
AgentOutput,
runAgentProcess,
@@ -31,9 +35,6 @@ import {
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 { buildRoomRoleContext } from './room-role-context.js';
@@ -683,22 +684,23 @@ export async function runAgentForGroup(
attempt: AgentAttempt,
rotationMessage?: string | null,
): Promise<'success' | 'error' | null> => {
const trigger = resolveClaudeRetryTrigger({
canRetryClaudeCredentials,
const retryAction = resolveAttemptRetryAction({
provider,
canRetryClaudeCredentials,
canRetryCodex: false,
attempt,
fallbackMessage: rotationMessage,
rotationMessage,
});
if (!trigger) {
if (retryAction.kind !== 'claude') {
return null;
}
const result = await retryClaudeWithRotation(
trigger,
rotationMessage ?? undefined,
retryAction.trigger,
retryAction.rotationMessage,
);
if (result === 'error') {
return maybeHandoffAfterError(trigger.reason, attempt);
return maybeHandoffAfterError(retryAction.trigger.reason, attempt);
}
pairedExecutionStatus = 'succeeded';
@@ -709,18 +711,20 @@ export async function runAgentForGroup(
attempt: AgentAttempt,
rotationMessage?: string | null,
): Promise<'success' | 'error' | null> => {
const trigger = resolveCodexRetryTrigger({
const retryAction = resolveAttemptRetryAction({
provider,
canRetryClaudeCredentials: false,
canRetryCodex: !isClaudeCodeAgent && getCodexAccountCount() > 1,
attempt,
rotationMessage,
});
if (!trigger) {
if (retryAction.kind !== 'codex') {
return null;
}
const result = await retryCodexWithRotation(
{ reason: trigger.reason },
rotationMessage ?? undefined,
retryAction.trigger,
retryAction.rotationMessage,
);
if (result === 'success') {
pairedExecutionStatus = 'succeeded';

69
src/message-runtime-flow.ts
View File

@@ -8,6 +8,7 @@ import {
buildArbiterPromptForTask,
buildFinalizePendingPrompt,
buildOwnerPendingPrompt,
buildPairedTurnPrompt,
buildReviewerPendingPrompt,
} from './message-runtime-prompts.js';
import {
@@ -47,6 +48,26 @@ export type BotOnlyPairedFollowUpAction =
nextRole: 'owner' | 'reviewer' | 'arbiter';
};
export type QueuedTurnDispatch = {
formatted: string;
botOnlyFollowUpAction: BotOnlyPairedFollowUpAction;
isBotOnlyPairedFollowUp: boolean;
loopCursorKey: string;
endSeq: number | null;
};
export function isBotOnlyPairedRoomTurn(
chatJid: string,
messages: NewMessage[],
): boolean {
return (
hasReviewerLease(chatJid) &&
messages.every(
(message) => message.is_from_me === true || !!message.is_bot_message,
)
);
}
export function resolveLastDeliveredBotRole(
messages: NewMessage[],
): PairedRoomRole | null {
@@ -384,6 +405,54 @@ 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,
lastDeliveredMessages: args.labeledMessagesToSend,
pendingCursorSource,
});
return {
formatted,
botOnlyFollowUpAction,
isBotOnlyPairedFollowUp,
loopCursorKey,
endSeq: args.messagesToSend[args.messagesToSend.length - 1]?.seq ?? null,
};
}
export function shouldSkipGenericFollowUpAfterDeliveryRetry(args: {
chatJid: string;
deliveryRole: PairedRoomRole;

66
src/message-runtime.ts
View File

@@ -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,
@@ -937,10 +929,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',
},
});
}
}
@@ -1223,30 +1220,20 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): {
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,
const {
formatted,
botOnlyFollowUpAction,
isBotOnlyPairedFollowUp,
lastDeliveredMessages: labeledMessagesToSend,
pendingCursorSource,
loopCursorKey: dispatchCursorKey,
endSeq,
} = buildQueuedTurnDispatch({
chatJid,
timezone: deps.timezone,
loopPendingTask,
rawPendingMessages,
messagesToSend,
labeledMessagesToSend,
formatMessages,
});
if (
await executeBotOnlyPairedFollowUpAction({
@@ -1271,14 +1258,13 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): {
}
if (deps.queue.sendMessage(chatJid, formatted)) {
const endSeq = messagesToSend[messagesToSend.length - 1]?.seq;
if (endSeq != null) {
advanceLastAgentCursor(
deps.getLastAgentTimestamps(),
deps.saveState,
chatJid,
endSeq,
loopCursorKey,
dispatchCursorKey,
);
}
logger.debug(

55
src/paired-execution-context-arbiter.ts
View File

@@ -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) },

31
src/paired-execution-context-owner.ts
View File

@@ -14,6 +14,7 @@ import {
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',

43
src/paired-execution-context-reviewer.ts
View File

@@ -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',

100
src/paired-execution-context-shared.ts
View File

@@ -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,108 @@ 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 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);
}

38
src/paired-execution-context.ts
View File

@@ -38,6 +38,7 @@ import {
import {
resolveCanonicalSourceRef,
requestArbiterOrEscalate,
transitionPairedTaskStatus,
} from './paired-execution-context-shared.js';
import {
markPairedTaskReviewReady,
@@ -226,11 +227,22 @@ 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 {
updatePairedTask(latestTask.id, {
...(hasHuman ? { round_trip_count: 0 } : {}),
updated_at: now,
});
}
}
// Use a stable per-channel worktree (not per-task) so the Claude SDK
// session persists across tasks. Different channels still get isolation.
@@ -253,9 +265,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 +279,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,
});
}
}

13
src/paired-workspace-manager.ts
View File

@@ -13,6 +13,7 @@ import {
} 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';
@@ -798,9 +799,15 @@ export function markPairedTaskReviewReady(taskId: string): {
);
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: now,
});
return { ownerWorkspace, reviewerWorkspace };

81
src/task-scheduler.ts
View File

@@ -3,6 +3,7 @@ import { CronExpressionParser } from 'cron-parser';
import fs from 'fs';
import { getAgentOutputText } from './agent-output.js';
import { createEvaluatedOutputHandler } from './agent-attempt.js';
import { resolveAttemptRetryAction } from './agent-attempt-retry.js';
import { getErrorMessage } from './utils.js';
import { ASSISTANT_NAME, SCHEDULER_POLL_INTERVAL, TIMEZONE } from './config.js';
@@ -42,7 +43,6 @@ import {
classifyRotationTrigger,
type AgentTriggerReason,
type CodexRotationReason,
isCodexRotationReason,
} from './agent-error-detection.js';
import {
getTokenCount,
@@ -512,65 +512,34 @@ async function runTask(
}
};
const provider = context.taskAgentType === 'codex' ? 'codex' : 'claude';
const provider = context.taskAgentType === 'codex' ? 'codex' : 'claude';
{
const attempt = await runTaskAttempt(provider);
result = attempt.attemptResult;
error = attempt.attemptError;
{
const attempt = await runTaskAttempt(provider);
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,
const retryAction = resolveAttemptRetryAction({
provider,
canRetryClaudeCredentials: provider === 'claude' && getTokenCount() > 0,
canRetryCodex: provider === 'codex' && getCodexAccountCount() > 1,
attempt,
rotationMessage: error,
});
if (retryAction.kind === 'claude') {
await retryClaudeTaskWithRotation(
retryAction.trigger,
retryAction.rotationMessage,
);
} else if (retryAction.kind === 'codex') {
await retryCodexTaskWithRotation(
retryAction.trigger,
retryAction.rotationMessage,
);
} else if (attempt.output.status === 'error') {
error = attempt.attemptError || 'Unknown error';
}
} else if (attempt.output.status === 'error') {
error = attempt.attemptError || 'Unknown error';
}
} // end else (non-exhausted path)
log.info(