fix: recover local paired attempts after restart
This commit is contained in:
@@ -7,7 +7,8 @@ import {
|
||||
markPairedTurnRunning,
|
||||
recoverInterruptedPairedTurnAttemptsForService,
|
||||
} from './db.js';
|
||||
import { CODEX_MAIN_SERVICE_ID } from './config.js';
|
||||
import { CLAUDE_SERVICE_ID, CODEX_MAIN_SERVICE_ID } from './config.js';
|
||||
import { requireDatabase } from './db/runtime-database.js';
|
||||
import { buildPairedTurnIdentity } from './paired-turn-identity.js';
|
||||
import type { PairedTask } from './types.js';
|
||||
|
||||
@@ -46,7 +47,7 @@ describe('paired turn attempt restart recovery', () => {
|
||||
_initTestDatabase();
|
||||
});
|
||||
|
||||
it('fails same-service running attempts so restart recovery can retry the turn', () => {
|
||||
it('fails local-service running attempts so restart recovery can retry the turn', () => {
|
||||
const task = makeTask();
|
||||
createPairedTask(task);
|
||||
const turnIdentity = buildPairedTurnIdentity({
|
||||
@@ -64,7 +65,7 @@ describe('paired turn attempt restart recovery', () => {
|
||||
});
|
||||
|
||||
const recovered = recoverInterruptedPairedTurnAttemptsForService({
|
||||
serviceId: CODEX_MAIN_SERVICE_ID,
|
||||
serviceIds: [CLAUDE_SERVICE_ID, CODEX_MAIN_SERVICE_ID],
|
||||
now: '2026-05-27T00:11:00.000Z',
|
||||
});
|
||||
|
||||
@@ -130,7 +131,7 @@ describe('paired turn attempt restart recovery', () => {
|
||||
|
||||
expect(
|
||||
recoverInterruptedPairedTurnAttemptsForService({
|
||||
serviceId: CODEX_MAIN_SERVICE_ID,
|
||||
serviceIds: [CODEX_MAIN_SERVICE_ID],
|
||||
now: '2026-05-27T00:11:00.000Z',
|
||||
}),
|
||||
).toEqual([]);
|
||||
@@ -141,4 +142,80 @@ describe('paired turn attempt restart recovery', () => {
|
||||
}),
|
||||
]);
|
||||
});
|
||||
|
||||
it('recovers codex executor attempts even when the orchestration lease used the claude service id', () => {
|
||||
const task = makeTask({ id: 'cross-service-lease-task' });
|
||||
createPairedTask(task);
|
||||
const turnIdentity = buildPairedTurnIdentity({
|
||||
taskId: task.id,
|
||||
taskUpdatedAt: task.updated_at,
|
||||
intentKind: 'owner-turn',
|
||||
role: 'owner',
|
||||
});
|
||||
|
||||
markPairedTurnRunning({
|
||||
turnIdentity,
|
||||
executorServiceId: CODEX_MAIN_SERVICE_ID,
|
||||
executorAgentType: 'codex',
|
||||
runId: 'codex-run-before-restart',
|
||||
});
|
||||
requireDatabase()
|
||||
.prepare(
|
||||
`
|
||||
INSERT INTO paired_task_execution_leases (
|
||||
task_id,
|
||||
chat_jid,
|
||||
role,
|
||||
turn_id,
|
||||
turn_attempt_id,
|
||||
turn_attempt_no,
|
||||
intent_kind,
|
||||
claimed_run_id,
|
||||
claimed_service_id,
|
||||
task_status,
|
||||
task_updated_at,
|
||||
claimed_at,
|
||||
updated_at,
|
||||
expires_at
|
||||
)
|
||||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||||
`,
|
||||
)
|
||||
.run(
|
||||
task.id,
|
||||
task.chat_jid,
|
||||
'owner',
|
||||
turnIdentity.turnId,
|
||||
`${turnIdentity.turnId}:attempt:1`,
|
||||
1,
|
||||
'owner-turn',
|
||||
'orchestrator-run-before-restart',
|
||||
CLAUDE_SERVICE_ID,
|
||||
task.status,
|
||||
task.updated_at,
|
||||
'2026-05-27T00:10:00.000Z',
|
||||
'2026-05-27T00:10:00.000Z',
|
||||
'2026-05-27T00:20:00.000Z',
|
||||
);
|
||||
|
||||
expect(
|
||||
recoverInterruptedPairedTurnAttemptsForService({
|
||||
serviceIds: [CLAUDE_SERVICE_ID, CODEX_MAIN_SERVICE_ID],
|
||||
now: '2026-05-27T00:11:00.000Z',
|
||||
}),
|
||||
).toEqual([
|
||||
expect.objectContaining({
|
||||
task_id: task.id,
|
||||
turn_id: turnIdentity.turnId,
|
||||
role: 'owner',
|
||||
intent_kind: 'owner-turn',
|
||||
}),
|
||||
]);
|
||||
expect(getPairedTurnAttempts(turnIdentity.turnId)).toEqual([
|
||||
expect.objectContaining({
|
||||
state: 'failed',
|
||||
active_run_id: null,
|
||||
}),
|
||||
]);
|
||||
});
|
||||
});
|
||||
|
||||
@@ -446,12 +446,18 @@ export function getPairedTurnAttemptIdFromDatabase(
|
||||
export function recoverInterruptedPairedTurnAttemptsForServiceInDatabase(
|
||||
database: Database,
|
||||
args: {
|
||||
serviceId: string;
|
||||
serviceIds: string[];
|
||||
now?: string;
|
||||
error?: string;
|
||||
},
|
||||
): InterruptedPairedTurnAttemptRecoveryCandidate[] {
|
||||
const serviceId = normalizeServiceId(args.serviceId);
|
||||
const serviceIds = Array.from(
|
||||
new Set(args.serviceIds.map((serviceId) => normalizeServiceId(serviceId))),
|
||||
).filter((serviceId) => serviceId.length > 0);
|
||||
if (serviceIds.length === 0) {
|
||||
return [];
|
||||
}
|
||||
const servicePlaceholders = serviceIds.map(() => '?').join(', ');
|
||||
const now = args.now ?? new Date().toISOString();
|
||||
const error =
|
||||
args.error ?? 'Interrupted by service restart before completion.';
|
||||
@@ -473,19 +479,22 @@ export function recoverInterruptedPairedTurnAttemptsForServiceInDatabase(
|
||||
JOIN paired_tasks tasks
|
||||
ON tasks.id = attempts.task_id
|
||||
WHERE attempts.state = 'running'
|
||||
AND attempts.executor_service_id = ?
|
||||
AND attempts.executor_service_id IN (${servicePlaceholders})
|
||||
AND tasks.status != 'completed'
|
||||
AND NOT EXISTS (
|
||||
SELECT 1
|
||||
FROM paired_task_execution_leases leases
|
||||
WHERE leases.task_id = attempts.task_id
|
||||
AND leases.turn_id = attempts.turn_id
|
||||
AND leases.claimed_service_id = attempts.executor_service_id
|
||||
AND leases.claimed_service_id NOT IN (${servicePlaceholders})
|
||||
)
|
||||
ORDER BY attempts.updated_at ASC, attempts.attempt_no ASC
|
||||
`,
|
||||
)
|
||||
.all(serviceId) as InterruptedPairedTurnAttemptRecoveryCandidate[];
|
||||
.all(
|
||||
...serviceIds,
|
||||
...serviceIds,
|
||||
) as InterruptedPairedTurnAttemptRecoveryCandidate[];
|
||||
|
||||
if (rows.length === 0) {
|
||||
return rows;
|
||||
@@ -501,11 +510,11 @@ export function recoverInterruptedPairedTurnAttemptsForServiceInDatabase(
|
||||
last_error = ?
|
||||
WHERE attempt_id = ?
|
||||
AND state = 'running'
|
||||
AND executor_service_id = ?
|
||||
AND executor_service_id IN (${servicePlaceholders})
|
||||
`,
|
||||
);
|
||||
for (const row of rows) {
|
||||
update.run(now, now, error, row.attempt_id, serviceId);
|
||||
update.run(now, now, error, row.attempt_id, ...serviceIds);
|
||||
}
|
||||
return rows;
|
||||
})();
|
||||
|
||||
@@ -242,7 +242,7 @@ export function getPairedTurnAttempts(
|
||||
}
|
||||
|
||||
export function recoverInterruptedPairedTurnAttemptsForService(args: {
|
||||
serviceId: string;
|
||||
serviceIds: string[];
|
||||
now?: string;
|
||||
error?: string;
|
||||
}): InterruptedPairedTurnAttemptRecoveryCandidate[] {
|
||||
|
||||
12
src/index.ts
12
src/index.ts
@@ -1,5 +1,9 @@
|
||||
import {
|
||||
ASSISTANT_NAME,
|
||||
ARBITER_SERVICE_ID,
|
||||
CLAUDE_SERVICE_ID,
|
||||
CODEX_MAIN_SERVICE_ID,
|
||||
CODEX_REVIEW_SERVICE_ID,
|
||||
IDLE_TIMEOUT,
|
||||
POLL_INTERVAL,
|
||||
SERVICE_ID,
|
||||
@@ -262,7 +266,13 @@ async function announceRestartRecovery(
|
||||
|
||||
function queueInterruptedPairedTurnAttemptRecovery(): void {
|
||||
for (const candidate of recoverInterruptedPairedTurnAttemptsForService({
|
||||
serviceId: SERVICE_ID,
|
||||
serviceIds: [
|
||||
SERVICE_ID,
|
||||
CLAUDE_SERVICE_ID,
|
||||
CODEX_MAIN_SERVICE_ID,
|
||||
CODEX_REVIEW_SERVICE_ID,
|
||||
ARBITER_SERVICE_ID,
|
||||
].filter((serviceId): serviceId is string => Boolean(serviceId)),
|
||||
})) {
|
||||
const binding = runtimeState.getRoomBindings()[candidate.chat_jid];
|
||||
if (!binding) {
|
||||
|
||||
Reference in New Issue
Block a user