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

db: extract paired schema cleanup migrations

Browse Source
This commit is contained in:
ejclaw
2026-04-11 06:57:41 +09:00
parent 0e12daed8b
commit eed2d5124f
6 changed files with 571 additions and 548 deletions
Download Patch File Download Diff File Expand all files Collapse all files

551
src/db/schema.ts
View File

@@ -1,13 +1,6 @@
import { Database } from 'bun:sqlite';
import { SERVICE_SESSION_SCOPE, normalizeServiceId } from '../config.js';
import { resolveRoleServiceShadow } from '../role-service-shadow.js';
import {
fillCanonicalChannelOwnerLeaseMetadata,
fillCanonicalPairedTaskMetadata,
fillCanonicalServiceHandoffMetadata,
inferExecutionLeaseServiceIdFromCanonicalMetadata,
} from './canonical-role-metadata.js';
import { inferExecutionLeaseServiceIdFromCanonicalMetadata } from './canonical-role-metadata.js';
import {
backfillPairedTurnAttemptsFromTurns,
buildPairedTurnAttemptId,
@@ -17,7 +10,6 @@ import {
tableHasColumn,
tryExecMigration,
} from './migrations/helpers.js';
import { normalizeStoredAgentType } from './room-registration.js';
// Legacy monolithic migration bundle kept for compatibility with databases
// that predate ordered schema migration tracking. New schema changes belong in
@@ -167,42 +159,6 @@ interface LegacyExecutionLeaseServiceRow {
arbiter_agent_type?: string | null;
}
interface StoredPairedTaskServiceRow {
rowid: number;
id: string;
chat_jid: string;
group_folder: string;
owner_service_id?: string | null;
reviewer_service_id?: string | null;
owner_agent_type?: string | null;
reviewer_agent_type?: string | null;
arbiter_agent_type?: string | null;
}
interface StoredChannelOwnerLeaseServiceRow {
rowid: number;
chat_jid: 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;
}
interface StoredServiceHandoffServiceRow {
id: number;
chat_jid: string;
group_folder: string;
source_service_id?: string | null;
target_service_id?: string | null;
source_role?: string | null;
target_role?: string | null;
source_agent_type?: string | null;
target_agent_type?: string | null;
intended_role?: string | null;
}
interface PairedTurnAttemptTimingRow {
turn_id: string;
attempt_no: number;
@@ -2113,27 +2069,6 @@ function applyPairedTurnAttemptProvenanceConstraints(database: Database): void {
}
}
interface StoredWorkItemServiceRow {
id: number;
agent_type?: string | null;
service_id?: string | null;
delivery_role?: string | null;
}
function normalizePairedRole(
role: string | null | undefined,
): 'owner' | 'reviewer' | 'arbiter' | null {
return role === 'owner' || role === 'reviewer' || role === 'arbiter'
? role
: null;
}
function inferLegacyExecutionLeaseServiceId(
row: LegacyExecutionLeaseServiceRow,
): string | null {
return inferExecutionLeaseServiceIdFromCanonicalMetadata(row);
}
function backfillLegacyExecutionLeaseServiceShadows(database: Database): void {
if (
!tableHasColumn(
@@ -2200,7 +2135,8 @@ function backfillLegacyExecutionLeaseServiceShadows(database: Database): void {
const tx = database.transaction(
(leaseRows: LegacyExecutionLeaseServiceRow[]) => {
for (const row of leaseRows) {
const claimedServiceId = inferLegacyExecutionLeaseServiceId(row);
const claimedServiceId =
inferExecutionLeaseServiceIdFromCanonicalMetadata(row);
if (!claimedServiceId) {
continue;
}
@@ -2212,302 +2148,6 @@ function backfillLegacyExecutionLeaseServiceShadows(database: Database): void {
tx(rows);
}
function backfillCanonicalPairedTaskServiceIds(database: Database): void {
if (
!tableHasColumn(database, 'paired_tasks', 'owner_service_id') ||
!tableHasColumn(database, 'paired_tasks', 'reviewer_service_id')
) {
return;
}
const rows = database
.prepare(
`
SELECT
rowid,
id,
chat_jid,
group_folder,
owner_service_id,
reviewer_service_id,
owner_agent_type,
reviewer_agent_type,
arbiter_agent_type
FROM paired_tasks
`,
)
.all() as StoredPairedTaskServiceRow[];
if (rows.length === 0) {
return;
}
const update = database.prepare(
`
UPDATE paired_tasks
SET owner_service_id = ?,
reviewer_service_id = ?,
owner_agent_type = ?,
reviewer_agent_type = ?,
arbiter_agent_type = ?
WHERE rowid = ?
`,
);
const tx = database.transaction((taskRows: StoredPairedTaskServiceRow[]) => {
for (const row of taskRows) {
const {
ownerAgentType,
reviewerAgentType,
arbiterAgentType,
ownerServiceId,
reviewerServiceId,
} = fillCanonicalPairedTaskMetadata({
id: row.id,
owner_service_id: row.owner_service_id,
reviewer_service_id: row.reviewer_service_id,
owner_agent_type: row.owner_agent_type,
reviewer_agent_type: row.reviewer_agent_type,
arbiter_agent_type: row.arbiter_agent_type,
});
if (
row.owner_service_id === ownerServiceId &&
row.reviewer_service_id === reviewerServiceId &&
row.owner_agent_type === ownerAgentType &&
row.reviewer_agent_type === reviewerAgentType &&
row.arbiter_agent_type === arbiterAgentType
) {
continue;
}
update.run(
ownerServiceId,
reviewerServiceId,
ownerAgentType,
reviewerAgentType,
arbiterAgentType,
row.rowid,
);
}
});
tx(rows);
}
function backfillCanonicalChannelOwnerServiceIds(database: Database): void {
if (!tableHasColumn(database, 'channel_owner', 'owner_service_id')) {
return;
}
const rows = database
.prepare(
`
SELECT
rowid,
chat_jid,
owner_service_id,
reviewer_service_id,
arbiter_service_id,
owner_agent_type,
reviewer_agent_type,
arbiter_agent_type
FROM channel_owner
`,
)
.all() as StoredChannelOwnerLeaseServiceRow[];
if (rows.length === 0) {
return;
}
const update = database.prepare(
`
UPDATE channel_owner
SET owner_service_id = ?,
reviewer_service_id = ?,
arbiter_service_id = ?,
owner_agent_type = ?,
reviewer_agent_type = ?,
arbiter_agent_type = ?
WHERE rowid = ?
`,
);
const tx = database.transaction(
(leaseRows: StoredChannelOwnerLeaseServiceRow[]) => {
for (const row of leaseRows) {
const {
ownerAgentType,
reviewerAgentType,
arbiterAgentType,
ownerServiceId,
reviewerServiceId,
arbiterServiceId,
} = fillCanonicalChannelOwnerLeaseMetadata({
chat_jid: row.chat_jid,
owner_service_id: row.owner_service_id,
reviewer_service_id: row.reviewer_service_id,
arbiter_service_id: row.arbiter_service_id,
owner_agent_type: row.owner_agent_type,
reviewer_agent_type: row.reviewer_agent_type,
arbiter_agent_type: row.arbiter_agent_type,
});
if (
row.owner_service_id === ownerServiceId &&
row.reviewer_service_id === reviewerServiceId &&
row.arbiter_service_id === arbiterServiceId &&
row.owner_agent_type === ownerAgentType &&
row.reviewer_agent_type === reviewerAgentType &&
row.arbiter_agent_type === arbiterAgentType
) {
continue;
}
update.run(
ownerServiceId,
reviewerServiceId,
arbiterServiceId,
ownerAgentType,
reviewerAgentType,
arbiterAgentType,
row.rowid,
);
}
},
);
tx(rows);
}
function backfillCanonicalServiceHandoffServiceIds(database: Database): void {
if (!tableHasColumn(database, 'service_handoffs', 'source_service_id')) {
return;
}
const rows = database
.prepare(
`
SELECT
id,
chat_jid,
group_folder,
source_service_id,
target_service_id,
source_role,
target_role,
source_agent_type,
target_agent_type,
intended_role
FROM service_handoffs
`,
)
.all() as StoredServiceHandoffServiceRow[];
if (rows.length === 0) {
return;
}
const update = database.prepare(
`
UPDATE service_handoffs
SET source_service_id = ?,
target_service_id = ?,
source_agent_type = ?,
target_agent_type = ?,
source_role = ?,
target_role = ?
WHERE id = ?
`,
);
const tx = database.transaction(
(handoffRows: StoredServiceHandoffServiceRow[]) => {
for (const row of handoffRows) {
const {
sourceRole,
targetRole,
sourceAgentType,
targetAgentType,
sourceServiceId,
targetServiceId,
} = fillCanonicalServiceHandoffMetadata({
id: row.id,
chat_jid: row.chat_jid,
source_service_id: row.source_service_id,
target_service_id: row.target_service_id,
source_role: row.source_role,
target_role: row.target_role,
intended_role: row.intended_role,
source_agent_type: row.source_agent_type,
target_agent_type: row.target_agent_type,
});
if (
row.source_service_id === sourceServiceId &&
row.target_service_id === targetServiceId &&
row.source_agent_type === (sourceAgentType ?? null) &&
row.target_agent_type === targetAgentType &&
row.source_role === sourceRole &&
row.target_role === targetRole
) {
continue;
}
update.run(
sourceServiceId,
targetServiceId,
sourceAgentType,
targetAgentType,
sourceRole,
targetRole,
row.id,
);
}
},
);
tx(rows);
}
function backfillCanonicalWorkItemServiceIds(database: Database): void {
if (!tableHasColumn(database, 'work_items', 'service_id')) {
return;
}
const rows = database
.prepare(
`SELECT id, agent_type, service_id, delivery_role
FROM work_items`,
)
.all() as StoredWorkItemServiceRow[];
if (rows.length === 0) {
return;
}
const update = database.prepare(
`UPDATE work_items
SET service_id = ?
WHERE id = ?`,
);
const tx = database.transaction(
(workItemRows: StoredWorkItemServiceRow[]) => {
for (const row of workItemRows) {
const agentType =
normalizeStoredAgentType(row.agent_type) ?? 'claude-code';
const deliveryRole = normalizePairedRole(row.delivery_role) ?? 'owner';
const serviceId =
(row.service_id ? normalizeServiceId(row.service_id) : null) ??
resolveRoleServiceShadow(deliveryRole, agentType) ??
SERVICE_SESSION_SCOPE;
if (row.service_id === serviceId) {
continue;
}
update.run(serviceId, row.id);
}
},
);
tx(rows);
}
export function applyLegacySchemaMigrations(
database: Database,
_args: {
@@ -2651,7 +2291,6 @@ export function applyLegacySchemaMigrations(
CREATE INDEX IF NOT EXISTS idx_paired_task_execution_leases_expires_at
ON paired_task_execution_leases(expires_at)
`);
tryExecMigration(
database,
`ALTER TABLE work_items ADD COLUMN delivery_role TEXT`,
@@ -2763,190 +2402,6 @@ export function applyLegacySchemaMigrations(
rebuildPairedTurnsWithoutLegacyScratchColumns(database);
rebuildPairedTurnAttemptForeignKeyTables(database);
applyPairedTurnAttemptProvenanceConstraints(database);
for (const column of [
'owner_service_id',
'reviewer_service_id',
'arbiter_service_id',
'owner_agent_type',
'reviewer_agent_type',
'arbiter_agent_type',
]) {
tryExecMigration(
database,
`ALTER TABLE channel_owner ADD COLUMN ${column} TEXT`,
);
}
backfillCanonicalPairedTaskServiceIds(database);
backfillCanonicalChannelOwnerServiceIds(database);
backfillCanonicalServiceHandoffServiceIds(database);
backfillCanonicalWorkItemServiceIds(database);
const pairedTasksSqlRow = database
.prepare(
`SELECT sql FROM sqlite_master WHERE type = 'table' AND name = 'paired_tasks'`,
)
.get() as { sql?: string } | undefined;
const pairedTasksSql = pairedTasksSqlRow?.sql || '';
const pairedTasksNeedsRebuild =
pairedTasksSql &&
(pairedTasksSql.includes('task_policy') ||
!pairedTasksSql.includes('round_trip_count'));
if (pairedTasksNeedsRebuild) {
database.exec(`DROP TABLE IF EXISTS paired_tasks`);
database.exec(`
CREATE TABLE IF NOT EXISTS paired_tasks (
id TEXT PRIMARY KEY,
chat_jid TEXT NOT NULL,
group_folder TEXT NOT NULL,
owner_service_id TEXT NOT NULL,
reviewer_service_id TEXT NOT NULL,
owner_agent_type TEXT,
reviewer_agent_type TEXT,
arbiter_agent_type TEXT,
title TEXT,
source_ref TEXT,
plan_notes TEXT,
review_requested_at TEXT,
round_trip_count INTEGER NOT NULL DEFAULT 0,
status TEXT NOT NULL DEFAULT 'active',
arbiter_verdict TEXT,
arbiter_requested_at TEXT,
completion_reason TEXT,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL,
CHECK (status IN ('active', 'review_ready', 'in_review', 'merge_ready', 'completed', 'arbiter_requested', 'in_arbitration')),
CHECK (owner_agent_type IN ('claude-code', 'codex') OR owner_agent_type IS NULL),
CHECK (reviewer_agent_type IN ('claude-code', 'codex') OR reviewer_agent_type IS NULL),
CHECK (arbiter_agent_type IN ('claude-code', 'codex') OR arbiter_agent_type IS NULL)
);
CREATE INDEX IF NOT EXISTS idx_paired_tasks_chat_status
ON paired_tasks(chat_jid, status, updated_at);
`);
}
{
const ptSqlRow = database
.prepare(
`SELECT sql FROM sqlite_master WHERE type = 'table' AND name = 'paired_tasks'`,
)
.get() as { sql?: string } | undefined;
const ptSql = ptSqlRow?.sql || '';
if (ptSql && !ptSql.includes('arbiter_requested')) {
const pairedTaskCols = getTableColumns(database, 'paired_tasks');
const selectPairedTaskColumn = (columnName: string): string =>
pairedTaskCols.includes(columnName)
? columnName
: `NULL AS ${columnName}`;
database.exec(`
CREATE TABLE paired_tasks_new (
id TEXT PRIMARY KEY,
chat_jid TEXT NOT NULL,
group_folder TEXT NOT NULL,
owner_service_id TEXT,
reviewer_service_id TEXT,
owner_agent_type TEXT,
reviewer_agent_type TEXT,
arbiter_agent_type TEXT,
title TEXT,
source_ref TEXT,
plan_notes TEXT,
review_requested_at TEXT,
round_trip_count INTEGER NOT NULL DEFAULT 0,
status TEXT NOT NULL DEFAULT 'active',
arbiter_verdict TEXT,
arbiter_requested_at TEXT,
completion_reason TEXT,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL,
CHECK (status IN ('active', 'review_ready', 'in_review', 'merge_ready', 'completed', 'arbiter_requested', 'in_arbitration')),
CHECK (owner_agent_type IN ('claude-code', 'codex') OR owner_agent_type IS NULL),
CHECK (reviewer_agent_type IN ('claude-code', 'codex') OR reviewer_agent_type IS NULL),
CHECK (arbiter_agent_type IN ('claude-code', 'codex') OR arbiter_agent_type IS NULL)
);
INSERT INTO paired_tasks_new (
id, chat_jid, group_folder, owner_service_id, reviewer_service_id,
owner_agent_type, reviewer_agent_type,
arbiter_agent_type, title, source_ref, plan_notes, review_requested_at,
round_trip_count, status, created_at, updated_at
)
SELECT
id, chat_jid, group_folder,
${selectPairedTaskColumn('owner_service_id')},
${selectPairedTaskColumn('reviewer_service_id')},
owner_agent_type, reviewer_agent_type,
arbiter_agent_type, title, source_ref, plan_notes, review_requested_at,
round_trip_count, status, created_at, updated_at
FROM paired_tasks;
DROP TABLE paired_tasks;
ALTER TABLE paired_tasks_new RENAME TO paired_tasks;
CREATE INDEX IF NOT EXISTS idx_paired_tasks_chat_status
ON paired_tasks(chat_jid, status, updated_at);
`);
}
}
tryExecMigration(
database,
`ALTER TABLE paired_tasks ADD COLUMN completion_reason TEXT`,
);
for (const table of [
'paired_executions',
'paired_approvals',
'paired_artifacts',
'paired_events',
]) {
database.exec(`DROP TABLE IF EXISTS ${table}`);
}
const pairedWsSqlRow = database
.prepare(
`SELECT sql FROM sqlite_master WHERE type = 'table' AND name = 'paired_workspaces'`,
)
.get() as { sql?: string } | undefined;
const pairedWsSql = pairedWsSqlRow?.sql || '';
if (pairedWsSql && pairedWsSql.includes('snapshot_source_fingerprint')) {
database.exec(`DROP TABLE IF EXISTS paired_workspaces`);
database.exec(`
CREATE TABLE IF NOT EXISTS paired_workspaces (
id TEXT PRIMARY KEY,
task_id TEXT NOT NULL,
role TEXT NOT NULL,
workspace_dir TEXT NOT NULL,
snapshot_source_dir TEXT,
snapshot_ref TEXT,
status TEXT NOT NULL DEFAULT 'ready',
snapshot_refreshed_at TEXT,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL,
CHECK (role IN ('owner', 'reviewer')),
CHECK (status IN ('ready', 'stale'))
);
CREATE UNIQUE INDEX IF NOT EXISTS idx_paired_workspaces_task_role
ON paired_workspaces(task_id, role);
`);
}
const pairedProjSqlRow = database
.prepare(
`SELECT sql FROM sqlite_master WHERE type = 'table' AND name = 'paired_projects'`,
)
.get() as { sql?: string } | undefined;
const pairedProjSql = pairedProjSqlRow?.sql || '';
if (pairedProjSql && pairedProjSql.includes('workspace_topology')) {
database.exec(`DROP TABLE IF EXISTS paired_projects`);
database.exec(`
CREATE TABLE IF NOT EXISTS paired_projects (
chat_jid TEXT PRIMARY KEY,
group_folder TEXT NOT NULL,
canonical_work_dir TEXT NOT NULL,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL
);
`);
}
}
/** @deprecated New schema changes should be added under src/db/migrations/*. */