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

fix: stabilize paired runtime turn identity

Browse Source
This commit is contained in:
ejclaw
2026-04-09 16:31:01 +09:00
parent 4f557a3b85
commit 1e46b05b03
10 changed files with 716 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
src/db/base-schema.ts
View File

@@ -202,6 +202,7 @@ export function applyBaseSchema(database: Database): void {
role TEXT NOT NULL,
intent_kind TEXT NOT NULL,
claimed_run_id TEXT NOT NULL,
claimed_service_id TEXT NOT NULL,
task_status TEXT NOT NULL,
task_updated_at TEXT NOT NULL,
claimed_at TEXT NOT NULL,

43
src/db/paired-state.ts
View File

@@ -4,6 +4,8 @@ import {
ARBITER_AGENT_TYPE,
CODEX_MAIN_SERVICE_ID,
CODEX_REVIEW_SERVICE_ID,
SERVICE_ID,
normalizeServiceId,
} from '../config.js';
import {
resolveStablePairedTaskOwnerAgentType,
@@ -51,6 +53,7 @@ export type PairedTaskUpdates = Partial<
>;
export const PAIRED_TASK_EXECUTION_LEASE_TTL_MS = 10 * 60_000;
const CURRENT_SERVICE_ID = normalizeServiceId(SERVICE_ID);
function computeExecutionLeaseExpiry(now: string): string {
return new Date(
@@ -414,13 +417,12 @@ export function claimPairedTurnReservationInDatabase(
taskStatus: PairedTaskStatus;
roundTripCount: number;
taskUpdatedAt: string;
nextTaskUpdatedAt: string;
intentKind: PairedTurnReservationIntentKind;
runId: string;
},
): boolean {
const tx = database.transaction(() => {
const now = args.nextTaskUpdatedAt;
const now = new Date().toISOString();
const expiresAt = computeExecutionLeaseExpiry(now);
const leaseRole = resolveExecutionLeaseRole(args.intentKind);
const existingLease = database
@@ -434,6 +436,7 @@ export function claimPairedTurnReservationInDatabase(
.get(args.taskId) as
| {
claimed_run_id: string;
claimed_service_id: string;
updated_at: string;
expires_at: string;
}
@@ -449,13 +452,14 @@ export function claimPairedTurnReservationInDatabase(
role,
intent_kind,
claimed_run_id,
claimed_service_id,
task_status,
task_updated_at,
claimed_at,
updated_at,
expires_at
)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
`,
)
.run(
@@ -464,6 +468,7 @@ export function claimPairedTurnReservationInDatabase(
leaseRole,
args.intentKind,
args.runId,
CURRENT_SERVICE_ID,
args.taskStatus,
args.taskUpdatedAt,
now,
@@ -485,6 +490,7 @@ export function claimPairedTurnReservationInDatabase(
role = ?,
intent_kind = ?,
claimed_run_id = ?,
claimed_service_id = ?,
task_status = ?,
task_updated_at = ?,
claimed_at = ?,
@@ -501,6 +507,7 @@ export function claimPairedTurnReservationInDatabase(
leaseRole,
args.intentKind,
args.runId,
CURRENT_SERVICE_ID,
args.taskStatus,
args.taskUpdatedAt,
now,
@@ -546,14 +553,13 @@ export function claimPairedTurnReservationInDatabase(
.prepare(
`
UPDATE paired_tasks
SET updated_at = ?
SET updated_at = updated_at
WHERE id = ?
AND updated_at = ?
AND status = ?
`,
)
.run(
args.nextTaskUpdatedAt,
args.taskId,
args.taskUpdatedAt,
args.taskStatus,
@@ -593,10 +599,11 @@ export function releasePairedTaskExecutionLeaseInDatabase(
`
DELETE FROM paired_task_execution_leases
WHERE task_id = ?
AND claimed_service_id = ?
AND claimed_run_id = ?
`,
)
.run(args.taskId, args.runId);
.run(args.taskId, CURRENT_SERVICE_ID, args.runId);
}
export function refreshPairedTaskExecutionLeaseInDatabase(
@@ -615,14 +622,36 @@ export function refreshPairedTaskExecutionLeaseInDatabase(
SET updated_at = ?,
expires_at = ?
WHERE task_id = ?
AND claimed_service_id = ?
AND claimed_run_id = ?
AND expires_at >= ?
`,
)
.run(now, computeExecutionLeaseExpiry(now), args.taskId, args.runId, now);
.run(
now,
computeExecutionLeaseExpiry(now),
args.taskId,
CURRENT_SERVICE_ID,
args.runId,
now,
);
return result.changes > 0;
}
export function clearPairedTaskExecutionLeasesForServiceInDatabase(
database: Database,
serviceId: string = CURRENT_SERVICE_ID,
): number {
return database
.prepare(
`
DELETE FROM paired_task_execution_leases
WHERE claimed_service_id = ?
`,
)
.run(serviceId).changes;
}
export function clearExpiredPairedTaskExecutionLeasesInDatabase(
database: Database,
now: string = new Date().toISOString(),

143
src/db/schema.ts
View File

@@ -1,6 +1,10 @@
import { Database } from 'bun:sqlite';
import { inferAgentTypeFromServiceShadow } from '../role-service-shadow.js';
import { normalizeServiceId } from '../config.js';
import {
inferAgentTypeFromServiceShadow,
resolveRoleServiceShadow,
} from '../role-service-shadow.js';
import {
backfillLegacyServiceSessions,
dropLegacyServiceSessionsTable,
@@ -44,6 +48,134 @@ function tryExecMigration(database: Database, sql: string): void {
}
}
interface LegacyExecutionLeaseServiceRow {
rowid: number;
role: string;
owner_service_id?: string | null;
reviewer_service_id?: string | null;
arbiter_service_id?: string | null;
owner_agent_type?: string | null;
reviewer_agent_type?: string | null;
arbiter_agent_type?: string | null;
}
function normalizePairedAgentType(
agentType: string | null | undefined,
): 'codex' | 'claude-code' | null {
if (agentType === 'codex' || agentType === 'claude-code') {
return agentType;
}
return null;
}
function inferLegacyExecutionLeaseServiceId(
row: LegacyExecutionLeaseServiceRow,
): string | null {
switch (row.role) {
case 'owner': {
const ownerAgentType = normalizePairedAgentType(row.owner_agent_type);
return (
(row.owner_service_id ? normalizeServiceId(row.owner_service_id) : null) ??
resolveRoleServiceShadow('owner', ownerAgentType)
);
}
case 'reviewer': {
const reviewerAgentType = normalizePairedAgentType(
row.reviewer_agent_type,
);
return (
(row.reviewer_service_id
? normalizeServiceId(row.reviewer_service_id)
: null) ??
resolveRoleServiceShadow('reviewer', reviewerAgentType)
);
}
case 'arbiter': {
const arbiterAgentType = normalizePairedAgentType(row.arbiter_agent_type);
return (
(row.arbiter_service_id
? normalizeServiceId(row.arbiter_service_id)
: null) ??
resolveRoleServiceShadow('arbiter', arbiterAgentType)
);
}
default:
return null;
}
}
function backfillLegacyExecutionLeaseServiceShadows(database: Database): void {
if (
!tableHasColumn(database, 'paired_task_execution_leases', 'claimed_service_id')
) {
return;
}
const pairedTasksTable = database
.prepare(
`
SELECT 1
FROM sqlite_master
WHERE type = 'table'
AND name = 'paired_tasks'
`,
)
.get();
if (!pairedTasksTable) {
return;
}
const pairedTaskColumns = getTableColumns(database, 'paired_tasks');
const selectPairedTaskColumn = (columnName: string): string =>
pairedTaskColumns.includes(columnName)
? `paired_tasks.${columnName} AS ${columnName}`
: `NULL AS ${columnName}`;
const rows = database
.prepare(
`
SELECT
paired_task_execution_leases.rowid AS rowid,
paired_task_execution_leases.role AS role,
${selectPairedTaskColumn('owner_service_id')},
${selectPairedTaskColumn('reviewer_service_id')},
${selectPairedTaskColumn('arbiter_service_id')},
${selectPairedTaskColumn('owner_agent_type')},
${selectPairedTaskColumn('reviewer_agent_type')},
${selectPairedTaskColumn('arbiter_agent_type')}
FROM paired_task_execution_leases
LEFT JOIN paired_tasks
ON paired_tasks.id = paired_task_execution_leases.task_id
WHERE paired_task_execution_leases.claimed_service_id IS NULL
`,
)
.all() as LegacyExecutionLeaseServiceRow[];
if (rows.length === 0) {
return;
}
const update = database.prepare(
`
UPDATE paired_task_execution_leases
SET claimed_service_id = ?
WHERE rowid = ?
AND claimed_service_id IS NULL
`,
);
const tx = database.transaction((leaseRows: LegacyExecutionLeaseServiceRow[]) => {
for (const row of leaseRows) {
const claimedServiceId = inferLegacyExecutionLeaseServiceId(row);
if (!claimedServiceId) {
continue;
}
update.run(claimedServiceId, row.rowid);
}
});
tx(rows);
}
export function applySchemaMigrations(
database: Database,
args: {
@@ -149,6 +281,10 @@ export function applySchemaMigrations(
database,
`ALTER TABLE paired_task_execution_leases ADD COLUMN expires_at TEXT`,
);
tryExecMigration(
database,
`ALTER TABLE paired_task_execution_leases ADD COLUMN claimed_service_id TEXT`,
);
database.exec(`
UPDATE paired_task_execution_leases
SET expires_at = COALESCE(
@@ -156,6 +292,11 @@ export function applySchemaMigrations(
strftime('%Y-%m-%dT%H:%M:%fZ', 'now', '+10 minutes')
)
`);
// Legacy lease rows predate claimed_service_id. Infer the original service
// from paired-task role metadata when possible instead of blanket-marking
// everything as the current service, which would let startup cleanup delete
// leases that belong to another runtime.
backfillLegacyExecutionLeaseServiceShadows(database);
database.exec(`
CREATE INDEX IF NOT EXISTS idx_paired_task_execution_leases_expires_at
ON paired_task_execution_leases(expires_at)