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

Merge branch 'codex/owner/ejclaw'

Browse Source
This commit is contained in:
ejclaw
2026-04-09 22:49:45 +09:00
parent a48dd1cc20 5714348897
commit 79f842cf29
17 changed files with 1911 additions and 1387 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1405
src/db.test.ts
View File

File diff suppressed because it is too large Load Diff

99
src/db.ts
View File

@@ -144,15 +144,6 @@ import {
setRouterStateInDatabase,
} from './db/router-state.js';
import {
backfillChannelOwnerRoleMetadata,
backfillPairedTaskRoleMetadata,
backfillServiceHandoffServiceShadows,
backfillStoredRoomSettings,
backfillWorkItemServiceShadows,
rebuildChannelOwnerCanonicalSchema,
rebuildPairedTasksCanonicalSchema,
rebuildServiceHandoffsCanonicalSchema,
rebuildWorkItemsCanonicalSchema,
resolveStablePairedTaskOwnerAgentType,
resolveStableRoomRoleAgentType,
} from './db/legacy-rebuilds.js';
@@ -163,7 +154,7 @@ import {
getSessionFromDatabase,
setSessionInDatabase,
} from './db/sessions.js';
import { type SchemaMigrationHooks, tableHasColumn } from './db/schema.js';
import { tableHasColumn } from './db/schema.js';
import {
type CreateScheduledTaskInput,
type ScheduledTaskStatusTrackingUpdates,
@@ -233,62 +224,10 @@ export type { WorkItem } from './db/work-items.js';
export type { ChannelOwnerLeaseRow } from './db/channel-owner-leases.js';
export type { ServiceHandoff } from './db/service-handoffs.js';
function backfillMessageSeq(database: Database): void {
const rows = database
.prepare(
`SELECT rowid, seq
FROM messages
ORDER BY CASE WHEN seq IS NULL THEN 1 ELSE 0 END, seq, timestamp, rowid`,
)
.all() as Array<{ rowid: number; seq: number | null }>;
if (rows.length === 0) {
return;
}
let nextSeq = 1;
const assignSeq = database.prepare(
'UPDATE messages SET seq = ? WHERE rowid = ? AND seq IS NULL',
);
const tx = database.transaction(() => {
for (const row of rows) {
if (row.seq === null) {
assignSeq.run(nextSeq, row.rowid);
}
nextSeq = Math.max(nextSeq, (row.seq ?? nextSeq) + 1);
}
});
tx();
const maxSeqRow = database
.prepare('SELECT MAX(seq) AS maxSeq FROM messages')
.get() as { maxSeq: number | null };
const maxSeq = maxSeqRow.maxSeq ?? 0;
if (maxSeq > 0) {
database
.prepare('INSERT OR IGNORE INTO message_sequence (id) VALUES (?)')
.run(maxSeq);
}
}
function getSchemaMigrationHooks(): SchemaMigrationHooks {
return {
backfillMessageSeq,
backfillStoredRoomSettings,
backfillChannelOwnerRoleMetadata,
backfillWorkItemServiceShadows,
backfillServiceHandoffServiceShadows,
backfillPairedTaskRoleMetadata,
rebuildWorkItemsCanonicalSchema,
rebuildChannelOwnerCanonicalSchema,
rebuildPairedTasksCanonicalSchema,
rebuildServiceHandoffsCanonicalSchema,
};
}
export function initDatabase(): void {
db = openPersistentDatabase();
initializeDatabaseSchema(db, getSchemaMigrationHooks());
initializeDatabaseSchema(db);
syncLegacyRegisteredGroupsIntoStoredRooms();
clearPairedTaskExecutionLeasesForServiceInDatabase(
db,
normalizeServiceId(SERVICE_ID),
@@ -304,7 +243,8 @@ export function initDatabase(): void {
/** @internal - for tests only. Creates a fresh in-memory database. */
export function _initTestDatabase(): void {
db = openInMemoryDatabase();
initializeDatabaseSchema(db, getSchemaMigrationHooks());
initializeDatabaseSchema(db);
syncLegacyRegisteredGroupsIntoStoredRooms();
clearPairedTaskExecutionLeasesForServiceInDatabase(
db,
normalizeServiceId(SERVICE_ID),
@@ -315,7 +255,8 @@ export function _initTestDatabase(): void {
/** @internal - for tests only. Opens an existing database file and runs schema/migrations. */
export function _initTestDatabaseFromFile(dbPath: string): void {
db = openDatabaseFromFile(dbPath);
initializeDatabaseSchema(db, getSchemaMigrationHooks());
initializeDatabaseSchema(db);
syncLegacyRegisteredGroupsIntoStoredRooms();
clearPairedTaskExecutionLeasesForServiceInDatabase(
db,
normalizeServiceId(SERVICE_ID),
@@ -540,8 +481,11 @@ export function getOpenWorkItem(
return getOpenWorkItemFromDatabase(db, chatJid, agentType, serviceId);
}
export function getOpenWorkItemForChat(chatJid: string): WorkItem | undefined {
return getOpenWorkItemForChatFromDatabase(db, chatJid);
export function getOpenWorkItemForChat(
chatJid: string,
serviceId: string = SERVICE_SESSION_SCOPE,
): WorkItem | undefined {
return getOpenWorkItemForChatFromDatabase(db, chatJid, serviceId);
}
export function createProducedWorkItem(
@@ -1053,6 +997,25 @@ function syncStoredRoomRegistrationSnapshotForJid(chatJid: string): void {
updateStoredRoomMetadata(db, chatJid, snapshot);
}
function syncLegacyRegisteredGroupsIntoStoredRooms(): void {
const rows = db
.prepare(
`SELECT DISTINCT jid
FROM registered_groups
ORDER BY jid`,
)
.all() as Array<{ jid: string }>;
if (rows.length === 0) {
return;
}
db.transaction((registeredGroupRows: Array<{ jid: string }>) => {
for (const row of registeredGroupRows) {
syncStoredRoomRegistrationSnapshotForJid(row.jid);
}
})(rows);
}
function buildRoomRegistrationSnapshotFromStoredRoom(
stored: StoredRoomSettings,
overrides?: Partial<Pick<RoomRegistrationSnapshot, 'name'>>,

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

@@ -32,6 +32,7 @@ export function applyBaseSchema(database: Database): void {
group_folder TEXT NOT NULL,
chat_jid TEXT NOT NULL,
agent_type TEXT NOT NULL,
service_id TEXT,
delivery_role TEXT,
status TEXT NOT NULL DEFAULT 'produced',
start_seq INTEGER,
@@ -121,6 +122,8 @@ export function applyBaseSchema(database: Database): void {
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,
@@ -223,6 +226,9 @@ export function applyBaseSchema(database: Database): void {
ON paired_task_execution_leases(expires_at);
CREATE TABLE IF NOT EXISTS channel_owner (
chat_jid TEXT PRIMARY KEY,
owner_service_id TEXT NOT NULL,
reviewer_service_id TEXT,
arbiter_service_id TEXT,
owner_agent_type TEXT,
reviewer_agent_type TEXT,
arbiter_agent_type TEXT,
@@ -251,6 +257,8 @@ export function applyBaseSchema(database: Database): void {
id INTEGER PRIMARY KEY AUTOINCREMENT,
chat_jid TEXT NOT NULL,
group_folder TEXT NOT NULL,
source_service_id TEXT NOT NULL,
target_service_id TEXT NOT NULL,
source_role TEXT,
source_agent_type TEXT,
target_role TEXT,

8
src/db/bootstrap.ts
View File

@@ -7,7 +7,7 @@ import { logger } from '../logger.js';
import type { RegisteredGroup } from '../types.js';
import { readJsonFile } from '../utils.js';
import { applyBaseSchema } from './base-schema.js';
import { type SchemaMigrationHooks, applySchemaMigrations } from './schema.js';
import { applySchemaMigrations } from './schema.js';
export interface JsonStateMigrationHooks {
setRouterState(key: string, value: string): void;
@@ -18,14 +18,10 @@ export interface JsonStateMigrationHooks {
): void;
}
export function initializeDatabaseSchema(
database: Database,
hooks: SchemaMigrationHooks,
): void {
export function initializeDatabaseSchema(database: Database): void {
applyBaseSchema(database);
applySchemaMigrations(database, {
assistantName: ASSISTANT_NAME,
hooks,
});
}

95
src/db/channel-owner-leases.ts
View File

@@ -27,6 +27,9 @@ export interface ChannelOwnerLeaseRow {
interface StoredChannelOwnerLeaseRow {
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;
@@ -49,32 +52,55 @@ export interface SetChannelOwnerLeaseInput {
function hydrateChannelOwnerLeaseRow(
row: StoredChannelOwnerLeaseRow,
): ChannelOwnerLeaseRow {
const persistedOwnerAgentType = normalizeStoredAgentType(
row.owner_agent_type,
);
const persistedReviewerAgentType = normalizeStoredAgentType(
row.reviewer_agent_type,
);
const persistedArbiterAgentType = normalizeStoredAgentType(
row.arbiter_agent_type,
);
const ownerAgentType =
normalizeStoredAgentType(row.owner_agent_type) ?? OWNER_AGENT_TYPE;
persistedOwnerAgentType ??
(row.owner_service_id
? inferAgentTypeFromServiceShadow(row.owner_service_id)
: undefined) ??
OWNER_AGENT_TYPE;
const reviewerAgentType =
row.reviewer_agent_type == null
row.reviewer_agent_type == null && row.reviewer_service_id == null
? null
: (normalizeStoredAgentType(row.reviewer_agent_type) ??
: (persistedReviewerAgentType ??
(row.reviewer_service_id
? inferAgentTypeFromServiceShadow(row.reviewer_service_id)
: null) ??
resolveStableReviewerAgentType(ownerAgentType, null));
const arbiterAgentType =
row.arbiter_agent_type == null
row.arbiter_agent_type == null && row.arbiter_service_id == null
? null
: (normalizeStoredAgentType(row.arbiter_agent_type) ??
: (persistedArbiterAgentType ??
(row.arbiter_service_id
? inferAgentTypeFromServiceShadow(row.arbiter_service_id)
: undefined) ??
ARBITER_AGENT_TYPE ??
null);
return {
chat_jid: row.chat_jid,
owner_service_id:
resolveRoleServiceShadow('owner', ownerAgentType) ?? CLAUDE_SERVICE_ID,
row.owner_service_id ??
resolveRoleServiceShadow('owner', ownerAgentType) ??
CLAUDE_SERVICE_ID,
reviewer_service_id:
reviewerAgentType == null
row.reviewer_service_id ??
(reviewerAgentType == null
? null
: resolveRoleServiceShadow('reviewer', reviewerAgentType),
: resolveRoleServiceShadow('reviewer', reviewerAgentType)),
arbiter_service_id:
arbiterAgentType == null
row.arbiter_service_id ??
(arbiterAgentType == null
? null
: resolveRoleServiceShadow('arbiter', arbiterAgentType),
: resolveRoleServiceShadow('arbiter', arbiterAgentType)),
owner_agent_type: ownerAgentType,
reviewer_agent_type: reviewerAgentType,
arbiter_agent_type: arbiterAgentType,
@@ -88,17 +114,7 @@ export function getChannelOwnerLeaseFromDatabase(
chatJid: string,
): ChannelOwnerLeaseRow | undefined {
const row = database
.prepare(
`SELECT
chat_jid,
owner_agent_type,
reviewer_agent_type,
arbiter_agent_type,
activated_at,
reason
FROM channel_owner
WHERE chat_jid = ?`,
)
.prepare(`SELECT * FROM channel_owner WHERE chat_jid = ?`)
.get(chatJid) as StoredChannelOwnerLeaseRow | undefined;
return row ? hydrateChannelOwnerLeaseRow(row) : undefined;
}
@@ -107,16 +123,7 @@ export function getAllChannelOwnerLeasesFromDatabase(
database: Database,
): ChannelOwnerLeaseRow[] {
const rows = database
.prepare(
`SELECT
chat_jid,
owner_agent_type,
reviewer_agent_type,
arbiter_agent_type,
activated_at,
reason
FROM channel_owner`,
)
.prepare(`SELECT * FROM channel_owner`)
.all() as StoredChannelOwnerLeaseRow[];
return rows.map(hydrateChannelOwnerLeaseRow);
}
@@ -142,20 +149,44 @@ export function setChannelOwnerLeaseInDatabase(
inferAgentTypeFromServiceShadow(input.arbiter_service_id ?? null) ??
ARBITER_AGENT_TYPE ??
null);
const ownerServiceId =
input.owner_service_id ??
resolveRoleServiceShadow('owner', ownerAgentType) ??
CLAUDE_SERVICE_ID;
const reviewerServiceId =
input.reviewer_service_id == null && reviewerAgentType == null
? null
: (input.reviewer_service_id ??
(reviewerAgentType == null
? null
: resolveRoleServiceShadow('reviewer', reviewerAgentType)));
const arbiterServiceId =
input.arbiter_service_id == null && arbiterAgentType == null
? null
: (input.arbiter_service_id ??
(arbiterAgentType == null
? null
: resolveRoleServiceShadow('arbiter', arbiterAgentType)));
database
.prepare(
`INSERT OR REPLACE INTO channel_owner (
chat_jid,
owner_service_id,
reviewer_service_id,
arbiter_service_id,
owner_agent_type,
reviewer_agent_type,
arbiter_agent_type,
activated_at,
reason
) VALUES (?, ?, ?, ?, ?, ?)`,
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`,
)
.run(
input.chat_jid,
ownerServiceId,
reviewerServiceId,
arbiterServiceId,
ownerAgentType ?? null,
reviewerAgentType ?? null,
arbiterAgentType ?? null,

668
src/db/legacy-rebuilds.ts
View File

@@ -2,26 +2,15 @@ import { Database } from 'bun:sqlite';
import {
ARBITER_AGENT_TYPE,
OWNER_AGENT_TYPE,
REVIEWER_AGENT_TYPE,
SERVICE_SESSION_SCOPE,
} from '../config.js';
import {
collectRegisteredAgentTypes,
collectRegisteredAgentTypesForFolder,
collectRoomRegistrationSnapshot,
getStoredRoomSettingsRowFromDatabase,
inferOwnerAgentTypeFromRegisteredAgentTypes,
inferRoomModeFromRegisteredAgentTypes,
insertStoredRoomSettings,
normalizeStoredAgentType,
updateStoredRoomMetadata,
} from './room-registration.js';
import { tableHasColumn } from './schema.js';
import {
inferAgentTypeFromServiceShadow,
resolveRoleServiceShadow,
} from '../role-service-shadow.js';
import type { AgentType, PairedRoomRole } from '../types.js';
interface StablePairedTaskRowInput {
@@ -30,81 +19,6 @@ interface StablePairedTaskRowInput {
owner_agent_type?: string | null;
}
interface StableLeaseRoleRowInput {
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 WorkItemServiceShadowRow {
id: number;
agent_type: string;
service_id: string;
delivery_role: PairedRoomRole | null;
}
interface LegacyServiceHandoffServiceRow {
id: number;
chat_jid: string;
group_folder: string;
source_service_id?: string | null;
target_service_id?: string | null;
source_role: PairedRoomRole | null;
source_agent_type?: string | null;
target_role: PairedRoomRole | null;
target_agent_type?: string | null;
}
interface LegacyPairedTaskServiceRow {
id: string;
chat_jid: string;
group_folder: string;
owner_service_id: string;
reviewer_service_id: string;
owner_agent_type?: string | null;
reviewer_agent_type?: string | null;
arbiter_agent_type?: string | null;
}
export function backfillStoredRoomSettings(database: Database): void {
const rows = database
.prepare(`SELECT DISTINCT jid FROM registered_groups`)
.all() as Array<{ jid: string }>;
if (rows.length === 0) return;
const tx = database.transaction((registeredRows: Array<{ jid: string }>) => {
for (const row of registeredRows) {
const existing = getStoredRoomSettingsRowFromDatabase(database, row.jid);
const snapshot = collectRoomRegistrationSnapshot(
database,
row.jid,
existing,
);
if (!snapshot) continue;
if (existing) {
updateStoredRoomMetadata(database, row.jid, snapshot);
continue;
}
insertStoredRoomSettings(
database,
row.jid,
inferRoomModeFromRegisteredAgentTypes(
collectRegisteredAgentTypes(database, row.jid),
),
'inferred',
snapshot,
);
}
});
tx(rows);
}
export function resolveStablePairedTaskOwnerAgentType(
database: Database,
task: StablePairedTaskRowInput,
@@ -183,585 +97,3 @@ export function resolveStableRoomRoleAgentType(
return ARBITER_AGENT_TYPE ?? null;
}
function resolveStableLeaseOwnerAgentType(
database: Database,
row: StableLeaseRoleRowInput,
): AgentType | undefined {
const persisted = normalizeStoredAgentType(row.owner_agent_type);
if (persisted) {
return persisted;
}
const stored = getStoredRoomSettingsRowFromDatabase(database, row.chat_jid);
if (stored?.ownerAgentType) {
return stored.ownerAgentType;
}
return inferAgentTypeFromServiceShadow(row.owner_service_id);
}
function resolveStableLeaseRoleAgentType(
database: Database,
row: StableLeaseRoleRowInput,
role: PairedRoomRole,
): AgentType | null | undefined {
if (role === 'owner') {
return resolveStableLeaseOwnerAgentType(database, row);
}
if (role === 'reviewer') {
if (row.reviewer_service_id == null) {
return null;
}
return (
normalizeStoredAgentType(row.reviewer_agent_type) ??
inferAgentTypeFromServiceShadow(row.reviewer_service_id) ??
resolveStableReviewerAgentType(
resolveStableLeaseOwnerAgentType(database, row),
null,
)
);
}
return (
normalizeStoredAgentType(row.arbiter_agent_type) ??
(row.arbiter_service_id
? inferAgentTypeFromServiceShadow(row.arbiter_service_id)
: undefined) ??
ARBITER_AGENT_TYPE ??
null
);
}
export function backfillChannelOwnerRoleMetadata(database: Database): void {
const rows = database
.prepare(
`SELECT
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 StableLeaseRoleRowInput[];
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 chat_jid = ?`,
);
const tx = database.transaction((leaseRows: StableLeaseRoleRowInput[]) => {
for (const row of leaseRows) {
const ownerAgentType = resolveStableLeaseRoleAgentType(
database,
row,
'owner',
);
const reviewerAgentType = resolveStableLeaseRoleAgentType(
database,
row,
'reviewer',
);
const arbiterAgentType = resolveStableLeaseRoleAgentType(
database,
row,
'arbiter',
);
const ownerServiceId =
resolveRoleServiceShadow('owner', ownerAgentType) ??
row.owner_service_id ??
null;
const reviewerServiceId =
row.reviewer_service_id == null
? null
: (resolveRoleServiceShadow('reviewer', reviewerAgentType) ??
row.reviewer_service_id);
const arbiterServiceId =
row.arbiter_service_id == null
? null
: (resolveRoleServiceShadow('arbiter', arbiterAgentType) ??
row.arbiter_service_id);
if (
ownerServiceId === row.owner_service_id &&
reviewerServiceId === row.reviewer_service_id &&
arbiterServiceId === row.arbiter_service_id &&
(ownerAgentType ?? null) === (row.owner_agent_type ?? null) &&
(reviewerAgentType ?? null) === (row.reviewer_agent_type ?? null) &&
(arbiterAgentType ?? null) === (row.arbiter_agent_type ?? null)
) {
continue;
}
update.run(
ownerServiceId,
reviewerServiceId,
arbiterServiceId,
ownerAgentType ?? null,
reviewerAgentType ?? null,
arbiterAgentType ?? null,
row.chat_jid,
);
}
});
tx(rows);
}
export function resolveWorkItemServiceShadow(
agentType: AgentType,
deliveryRole?: PairedRoomRole | null,
): string {
return (
resolveRoleServiceShadow(deliveryRole ?? 'owner', agentType) ??
SERVICE_SESSION_SCOPE
);
}
export function backfillWorkItemServiceShadows(database: Database): void {
const rows = database
.prepare(
`SELECT id, agent_type, service_id, delivery_role
FROM work_items`,
)
.all() as WorkItemServiceShadowRow[];
const update = database.prepare(
`UPDATE work_items
SET service_id = ?
WHERE id = ?`,
);
const tx = database.transaction(
(workItemRows: WorkItemServiceShadowRow[]) => {
for (const row of workItemRows) {
const agentType = normalizeStoredAgentType(row.agent_type);
if (!agentType) {
continue;
}
const normalizedServiceId = resolveWorkItemServiceShadow(
agentType,
row.delivery_role,
);
if (normalizedServiceId === row.service_id) {
continue;
}
update.run(normalizedServiceId, row.id);
}
},
);
tx(rows);
}
export function backfillServiceHandoffServiceShadows(database: Database): void {
const rows = database
.prepare(
`SELECT
id,
chat_jid,
group_folder,
source_service_id,
target_service_id,
source_role,
source_agent_type,
target_role,
target_agent_type
FROM service_handoffs`,
)
.all() as LegacyServiceHandoffServiceRow[];
const update = database.prepare(
`UPDATE service_handoffs
SET source_service_id = ?,
target_service_id = ?,
source_agent_type = ?,
target_agent_type = ?
WHERE id = ?`,
);
const tx = database.transaction(
(handoffRows: Array<LegacyServiceHandoffServiceRow>) => {
for (const row of handoffRows) {
const sourceAgentType =
normalizeStoredAgentType(row.source_agent_type) ??
(row.source_role
? resolveStableRoomRoleAgentType(database, {
chatJid: row.chat_jid,
groupFolder: row.group_folder,
role: row.source_role,
})
: null) ??
inferAgentTypeFromServiceShadow(row.source_service_id);
const targetAgentType =
normalizeStoredAgentType(row.target_agent_type) ??
(row.target_role
? resolveStableRoomRoleAgentType(database, {
chatJid: row.chat_jid,
groupFolder: row.group_folder,
role: row.target_role,
})
: null);
const normalizedSourceServiceId =
row.source_role != null
? (resolveRoleServiceShadow(row.source_role, sourceAgentType) ??
row.source_service_id)
: row.source_service_id;
const normalizedTargetServiceId =
row.target_role != null
? (resolveRoleServiceShadow(row.target_role, targetAgentType) ??
row.target_service_id)
: row.target_service_id;
if (
normalizedSourceServiceId === row.source_service_id &&
normalizedTargetServiceId === row.target_service_id &&
(sourceAgentType ?? null) === (row.source_agent_type ?? null) &&
(targetAgentType ?? null) === (row.target_agent_type ?? null)
) {
continue;
}
update.run(
normalizedSourceServiceId ?? SERVICE_SESSION_SCOPE,
normalizedTargetServiceId ?? SERVICE_SESSION_SCOPE,
sourceAgentType ?? null,
targetAgentType ?? null,
row.id,
);
}
},
);
tx(rows);
}
export function backfillPairedTaskRoleMetadata(database: Database): void {
const rows = database
.prepare(
`SELECT
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 LegacyPairedTaskServiceRow[];
const tx = database.transaction((taskRows: LegacyPairedTaskServiceRow[]) => {
const update = database.prepare(
`UPDATE paired_tasks
SET owner_service_id = ?,
reviewer_service_id = ?,
owner_agent_type = ?,
reviewer_agent_type = ?,
arbiter_agent_type = ?
WHERE id = ?`,
);
for (const row of taskRows) {
const ownerAgentType = resolveStablePairedTaskOwnerAgentType(
database,
row,
);
const reviewerAgentType = resolveStableReviewerAgentType(
ownerAgentType,
row.reviewer_agent_type ?? null,
);
const arbiterAgentType =
normalizeStoredAgentType(row.arbiter_agent_type) ??
ARBITER_AGENT_TYPE ??
null;
const ownerServiceId =
resolveRoleServiceShadow('owner', ownerAgentType) ??
row.owner_service_id;
const reviewerServiceId =
resolveRoleServiceShadow('reviewer', reviewerAgentType) ??
row.reviewer_service_id;
update.run(
ownerServiceId,
reviewerServiceId,
ownerAgentType ?? null,
reviewerAgentType ?? null,
arbiterAgentType,
row.id,
);
}
});
tx(rows);
}
export function rebuildWorkItemsCanonicalSchema(database: Database): void {
if (!tableHasColumn(database, 'work_items', 'service_id')) {
return;
}
database.exec(`
CREATE TABLE work_items_new (
id INTEGER PRIMARY KEY AUTOINCREMENT,
group_folder TEXT NOT NULL,
chat_jid TEXT NOT NULL,
agent_type TEXT NOT NULL,
delivery_role TEXT,
status TEXT NOT NULL DEFAULT 'produced',
start_seq INTEGER,
end_seq INTEGER,
result_payload TEXT NOT NULL,
delivery_attempts INTEGER NOT NULL DEFAULT 0,
delivery_message_id TEXT,
last_error TEXT,
created_at TEXT NOT NULL DEFAULT (datetime('now')),
updated_at TEXT NOT NULL DEFAULT (datetime('now')),
delivered_at TEXT,
CHECK (status IN ('produced', 'delivery_retry', 'delivered')),
CHECK (delivery_role IN ('owner', 'reviewer', 'arbiter') OR delivery_role IS NULL)
);
INSERT INTO work_items_new (
id,
group_folder,
chat_jid,
agent_type,
delivery_role,
status,
start_seq,
end_seq,
result_payload,
delivery_attempts,
delivery_message_id,
last_error,
created_at,
updated_at,
delivered_at
)
SELECT
id,
group_folder,
chat_jid,
agent_type,
delivery_role,
status,
start_seq,
end_seq,
result_payload,
delivery_attempts,
delivery_message_id,
last_error,
created_at,
updated_at,
delivered_at
FROM work_items;
DROP TABLE work_items;
ALTER TABLE work_items_new RENAME TO work_items;
CREATE INDEX IF NOT EXISTS idx_work_items_status
ON work_items(status, updated_at);
CREATE INDEX IF NOT EXISTS idx_work_items_group_agent
ON work_items(chat_jid, agent_type, delivery_role, status);
CREATE UNIQUE INDEX IF NOT EXISTS idx_work_items_open
ON work_items(chat_jid, agent_type, IFNULL(delivery_role, ''))
WHERE status IN ('produced', 'delivery_retry');
`);
}
export function rebuildChannelOwnerCanonicalSchema(database: Database): void {
if (!tableHasColumn(database, 'channel_owner', 'owner_service_id')) {
return;
}
database.exec(`
CREATE TABLE channel_owner_new (
chat_jid TEXT PRIMARY KEY,
owner_agent_type TEXT,
reviewer_agent_type TEXT,
arbiter_agent_type TEXT,
activated_at TEXT,
reason TEXT,
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 channel_owner_new (
chat_jid,
owner_agent_type,
reviewer_agent_type,
arbiter_agent_type,
activated_at,
reason
)
SELECT
chat_jid,
owner_agent_type,
reviewer_agent_type,
arbiter_agent_type,
activated_at,
reason
FROM channel_owner;
DROP TABLE channel_owner;
ALTER TABLE channel_owner_new RENAME TO channel_owner;
`);
}
export function rebuildPairedTasksCanonicalSchema(database: Database): void {
if (!tableHasColumn(database, 'paired_tasks', 'owner_service_id')) {
return;
}
database.exec(`
CREATE TABLE paired_tasks_new (
id TEXT PRIMARY KEY,
chat_jid TEXT NOT NULL,
group_folder 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)
);
INSERT INTO paired_tasks_new (
id,
chat_jid,
group_folder,
owner_agent_type,
reviewer_agent_type,
arbiter_agent_type,
title,
source_ref,
plan_notes,
review_requested_at,
round_trip_count,
status,
arbiter_verdict,
arbiter_requested_at,
completion_reason,
created_at,
updated_at
)
SELECT
id,
chat_jid,
group_folder,
owner_agent_type,
reviewer_agent_type,
arbiter_agent_type,
title,
source_ref,
plan_notes,
review_requested_at,
round_trip_count,
status,
arbiter_verdict,
arbiter_requested_at,
completion_reason,
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);
`);
}
export function rebuildServiceHandoffsCanonicalSchema(
database: Database,
): void {
if (!tableHasColumn(database, 'service_handoffs', 'source_service_id')) {
return;
}
database.exec(`
CREATE TABLE service_handoffs_new (
id INTEGER PRIMARY KEY AUTOINCREMENT,
chat_jid TEXT NOT NULL,
group_folder TEXT NOT NULL,
source_role TEXT,
source_agent_type TEXT,
target_role TEXT,
target_agent_type TEXT NOT NULL,
prompt TEXT NOT NULL,
status TEXT NOT NULL DEFAULT 'pending',
start_seq INTEGER,
end_seq INTEGER,
reason TEXT,
intended_role TEXT,
created_at TEXT NOT NULL DEFAULT (datetime('now')),
claimed_at TEXT,
completed_at TEXT,
last_error TEXT,
CHECK (status IN ('pending', 'claimed', 'completed', 'failed')),
CHECK (intended_role IN ('owner', 'reviewer', 'arbiter') OR intended_role IS NULL),
CHECK (source_role IN ('owner', 'reviewer', 'arbiter') OR source_role IS NULL),
CHECK (target_role IN ('owner', 'reviewer', 'arbiter') OR target_role IS NULL),
CHECK (source_agent_type IN ('claude-code', 'codex') OR source_agent_type IS NULL)
);
INSERT INTO service_handoffs_new (
id,
chat_jid,
group_folder,
source_role,
source_agent_type,
target_role,
target_agent_type,
prompt,
status,
start_seq,
end_seq,
reason,
intended_role,
created_at,
claimed_at,
completed_at,
last_error
)
SELECT
id,
chat_jid,
group_folder,
source_role,
source_agent_type,
target_role,
target_agent_type,
prompt,
status,
start_seq,
end_seq,
reason,
intended_role,
created_at,
claimed_at,
completed_at,
last_error
FROM service_handoffs;
DROP TABLE service_handoffs;
ALTER TABLE service_handoffs_new RENAME TO service_handoffs;
CREATE INDEX IF NOT EXISTS idx_service_handoffs_target
ON service_handoffs(status, target_role, target_agent_type, created_at);
`);
}

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

@@ -12,7 +12,10 @@ import {
resolveStableReviewerAgentType,
} from './legacy-rebuilds.js';
import { normalizeStoredAgentType } from './room-registration.js';
import { resolveRoleServiceShadow } from '../role-service-shadow.js';
import {
inferAgentTypeFromServiceShadow,
resolveRoleServiceShadow,
} from '../role-service-shadow.js';
import {
AgentType,
PairedProject,
@@ -31,6 +34,8 @@ interface StoredPairedTaskRow extends Omit<
| 'reviewer_agent_type'
| 'arbiter_agent_type'
> {
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;
@@ -81,11 +86,32 @@ function hydratePairedTaskRow(
database: Database,
row: StoredPairedTaskRow,
): PairedTask {
const ownerAgentType = resolveStablePairedTaskOwnerAgentType(database, row);
const reviewerAgentType = resolveStableReviewerAgentType(
ownerAgentType,
const persistedOwnerAgentType = normalizeStoredAgentType(
row.owner_agent_type ?? null,
);
const persistedReviewerAgentType = normalizeStoredAgentType(
row.reviewer_agent_type ?? null,
);
const stableOwnerAgentType = resolveStablePairedTaskOwnerAgentType(
database,
row,
);
const ownerAgentType =
stableOwnerAgentType ??
(row.owner_service_id
? inferAgentTypeFromServiceShadow(row.owner_service_id)
: undefined);
const stableReviewerAgentType = resolveStableReviewerAgentType(
stableOwnerAgentType,
row.reviewer_agent_type ?? null,
);
const reviewerAgentType =
persistedReviewerAgentType ??
stableReviewerAgentType ??
(row.reviewer_service_id
? inferAgentTypeFromServiceShadow(row.reviewer_service_id)
: null) ??
resolveStableReviewerAgentType(ownerAgentType, null);
const arbiterAgentType =
normalizeStoredAgentType(row.arbiter_agent_type) ??
ARBITER_AGENT_TYPE ??
@@ -94,9 +120,11 @@ function hydratePairedTaskRow(
return {
...row,
owner_service_id:
row.owner_service_id ??
resolveRoleServiceShadow('owner', ownerAgentType) ??
CODEX_MAIN_SERVICE_ID,
reviewer_service_id:
row.reviewer_service_id ??
resolveRoleServiceShadow('reviewer', reviewerAgentType) ??
CODEX_REVIEW_SERVICE_ID,
owner_agent_type: ownerAgentType ?? null,
@@ -148,6 +176,27 @@ export function createPairedTaskInDatabase(
database: Database,
task: PairedTask,
): void {
const ownerAgentType =
normalizeStoredAgentType(task.owner_agent_type) ??
inferAgentTypeFromServiceShadow(task.owner_service_id) ??
null;
const reviewerAgentType =
normalizeStoredAgentType(task.reviewer_agent_type) ??
inferAgentTypeFromServiceShadow(task.reviewer_service_id) ??
resolveStableReviewerAgentType(ownerAgentType ?? undefined, null);
const arbiterAgentType =
normalizeStoredAgentType(task.arbiter_agent_type) ??
ARBITER_AGENT_TYPE ??
null;
const ownerServiceId =
task.owner_service_id ??
resolveRoleServiceShadow('owner', ownerAgentType) ??
CODEX_MAIN_SERVICE_ID;
const reviewerServiceId =
task.reviewer_service_id ??
resolveRoleServiceShadow('reviewer', reviewerAgentType) ??
CODEX_REVIEW_SERVICE_ID;
database
.prepare(
`
@@ -155,6 +204,8 @@ export function createPairedTaskInDatabase(
id,
chat_jid,
group_folder,
owner_service_id,
reviewer_service_id,
owner_agent_type,
reviewer_agent_type,
arbiter_agent_type,
@@ -166,19 +217,22 @@ export function createPairedTaskInDatabase(
status,
arbiter_verdict,
arbiter_requested_at,
completion_reason,
created_at,
updated_at
)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
`,
)
.run(
task.id,
task.chat_jid,
task.group_folder,
task.owner_agent_type ?? null,
task.reviewer_agent_type ?? null,
task.arbiter_agent_type ?? null,
ownerServiceId,
reviewerServiceId,
ownerAgentType,
reviewerAgentType,
arbiterAgentType,
task.title,
task.source_ref,
task.plan_notes,
@@ -187,6 +241,7 @@ export function createPairedTaskInDatabase(
task.status,
task.arbiter_verdict,
task.arbiter_requested_at,
task.completion_reason,
task.created_at,
task.updated_at,
);

549
src/db/schema.ts
View File

@@ -1,29 +1,30 @@
import { Database } from 'bun:sqlite';
import { normalizeServiceId } from '../config.js';
import {
ARBITER_AGENT_TYPE,
CLAUDE_SERVICE_ID,
CODEX_MAIN_SERVICE_ID,
CODEX_REVIEW_SERVICE_ID,
OWNER_AGENT_TYPE,
SERVICE_SESSION_SCOPE,
normalizeServiceId,
} from '../config.js';
import {
inferAgentTypeFromServiceShadow,
resolveRoleServiceShadow,
} from '../role-service-shadow.js';
import {
resolveStablePairedTaskOwnerAgentType,
resolveStableReviewerAgentType,
resolveStableRoomRoleAgentType,
} from './legacy-rebuilds.js';
import { normalizeStoredAgentType } from './room-registration.js';
import {
backfillLegacyServiceSessions,
dropLegacyServiceSessionsTable,
migrateSessionsTableToCompositePk,
} from './sessions.js';
export interface SchemaMigrationHooks {
backfillMessageSeq(database: Database): void;
backfillStoredRoomSettings(database: Database): void;
backfillChannelOwnerRoleMetadata(database: Database): void;
backfillWorkItemServiceShadows(database: Database): void;
backfillServiceHandoffServiceShadows(database: Database): void;
backfillPairedTaskRoleMetadata(database: Database): void;
rebuildWorkItemsCanonicalSchema(database: Database): void;
rebuildChannelOwnerCanonicalSchema(database: Database): void;
rebuildPairedTasksCanonicalSchema(database: Database): void;
rebuildServiceHandoffsCanonicalSchema(database: Database): void;
}
function getTableColumns(database: Database, tableName: string): string[] {
return (
database.prepare(`PRAGMA table_info(${tableName})`).all() as Array<{
@@ -48,6 +49,43 @@ function tryExecMigration(database: Database, sql: string): void {
}
}
function backfillMessageSeq(database: Database): void {
const rows = database
.prepare(
`SELECT rowid, seq
FROM messages
ORDER BY CASE WHEN seq IS NULL THEN 1 ELSE 0 END, seq, timestamp, rowid`,
)
.all() as Array<{ rowid: number; seq: number | null }>;
if (rows.length === 0) {
return;
}
let nextSeq = 1;
const assignSeq = database.prepare(
'UPDATE messages SET seq = ? WHERE rowid = ? AND seq IS NULL',
);
const tx = database.transaction(() => {
for (const row of rows) {
if (row.seq === null) {
assignSeq.run(nextSeq, row.rowid);
}
nextSeq = Math.max(nextSeq, (row.seq ?? nextSeq) + 1);
}
});
tx();
const maxSeqRow = database
.prepare('SELECT MAX(seq) AS maxSeq FROM messages')
.get() as { maxSeq: number | null };
const maxSeq = maxSeqRow.maxSeq ?? 0;
if (maxSeq > 0) {
database
.prepare('INSERT OR IGNORE INTO message_sequence (id) VALUES (?)')
.run(maxSeq);
}
}
interface LegacyExecutionLeaseServiceRow {
rowid: number;
role: string;
@@ -59,13 +97,53 @@ interface LegacyExecutionLeaseServiceRow {
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;
interface StoredPairedTaskServiceRow {
rowid: number;
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;
}
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 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(
@@ -73,7 +151,7 @@ function inferLegacyExecutionLeaseServiceId(
): string | null {
switch (row.role) {
case 'owner': {
const ownerAgentType = normalizePairedAgentType(row.owner_agent_type);
const ownerAgentType = normalizeStoredAgentType(row.owner_agent_type);
return (
(row.owner_service_id
? normalizeServiceId(row.owner_service_id)
@@ -81,7 +159,7 @@ function inferLegacyExecutionLeaseServiceId(
);
}
case 'reviewer': {
const reviewerAgentType = normalizePairedAgentType(
const reviewerAgentType = normalizeStoredAgentType(
row.reviewer_agent_type,
);
return (
@@ -91,7 +169,7 @@ function inferLegacyExecutionLeaseServiceId(
);
}
case 'arbiter': {
const arbiterAgentType = normalizePairedAgentType(row.arbiter_agent_type);
const arbiterAgentType = normalizeStoredAgentType(row.arbiter_agent_type);
return (
(row.arbiter_service_id
? normalizeServiceId(row.arbiter_service_id)
@@ -181,14 +259,344 @@ 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,
chat_jid,
group_folder,
owner_service_id,
reviewer_service_id,
owner_agent_type,
reviewer_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 = ?
WHERE rowid = ?
`,
);
const tx = database.transaction((taskRows: StoredPairedTaskServiceRow[]) => {
for (const row of taskRows) {
const persistedOwnerAgentType = normalizeStoredAgentType(
row.owner_agent_type,
);
const persistedReviewerAgentType = normalizeStoredAgentType(
row.reviewer_agent_type,
);
const stableOwnerAgentType = resolveStablePairedTaskOwnerAgentType(
database,
row,
);
const ownerAgentType =
stableOwnerAgentType ??
(row.owner_service_id
? inferAgentTypeFromServiceShadow(row.owner_service_id)
: undefined);
const stableReviewerAgentType = resolveStableReviewerAgentType(
stableOwnerAgentType,
row.reviewer_agent_type ?? null,
);
const reviewerAgentType =
persistedReviewerAgentType ??
stableReviewerAgentType ??
(row.reviewer_service_id
? inferAgentTypeFromServiceShadow(row.reviewer_service_id)
: null) ??
resolveStableReviewerAgentType(ownerAgentType, null);
const ownerServiceId =
(persistedOwnerAgentType
? resolveRoleServiceShadow('owner', ownerAgentType)
: null) ??
row.owner_service_id ??
resolveRoleServiceShadow('owner', ownerAgentType) ??
CODEX_MAIN_SERVICE_ID;
const reviewerServiceId =
(persistedReviewerAgentType
? resolveRoleServiceShadow('reviewer', reviewerAgentType)
: null) ??
row.reviewer_service_id ??
resolveRoleServiceShadow('reviewer', reviewerAgentType) ??
CODEX_REVIEW_SERVICE_ID;
if (
row.owner_service_id === ownerServiceId &&
row.reviewer_service_id === reviewerServiceId
) {
continue;
}
update.run(ownerServiceId, reviewerServiceId, 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 = ?
WHERE rowid = ?
`,
);
const tx = database.transaction(
(leaseRows: StoredChannelOwnerLeaseServiceRow[]) => {
for (const row of leaseRows) {
const persistedOwnerAgentType = normalizeStoredAgentType(
row.owner_agent_type,
);
const persistedReviewerAgentType = normalizeStoredAgentType(
row.reviewer_agent_type,
);
const persistedArbiterAgentType = normalizeStoredAgentType(
row.arbiter_agent_type,
);
const ownerAgentType =
persistedOwnerAgentType ??
(row.owner_service_id
? inferAgentTypeFromServiceShadow(row.owner_service_id)
: undefined) ??
OWNER_AGENT_TYPE;
const reviewerAgentType =
row.reviewer_agent_type == null && row.reviewer_service_id == null
? null
: (persistedReviewerAgentType ??
(row.reviewer_service_id
? inferAgentTypeFromServiceShadow(row.reviewer_service_id)
: null) ??
resolveStableReviewerAgentType(ownerAgentType, null));
const arbiterAgentType =
row.arbiter_agent_type == null && row.arbiter_service_id == null
? null
: (persistedArbiterAgentType ??
(row.arbiter_service_id
? inferAgentTypeFromServiceShadow(row.arbiter_service_id)
: undefined) ??
ARBITER_AGENT_TYPE ??
null);
const ownerServiceId =
row.owner_service_id ??
resolveRoleServiceShadow('owner', ownerAgentType) ??
CLAUDE_SERVICE_ID;
const reviewerServiceId =
row.reviewer_service_id ??
(reviewerAgentType == null
? null
: resolveRoleServiceShadow('reviewer', reviewerAgentType));
const arbiterServiceId =
row.arbiter_service_id ??
(arbiterAgentType == null
? null
: resolveRoleServiceShadow('arbiter', arbiterAgentType));
if (
row.owner_service_id === ownerServiceId &&
row.reviewer_service_id === reviewerServiceId &&
row.arbiter_service_id === arbiterServiceId
) {
continue;
}
update.run(
ownerServiceId,
reviewerServiceId,
arbiterServiceId,
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 = ?
WHERE id = ?
`,
);
const tx = database.transaction(
(handoffRows: StoredServiceHandoffServiceRow[]) => {
for (const row of handoffRows) {
const sourceRole =
normalizePairedRole(row.source_role) ??
normalizePairedRole(row.intended_role);
const targetRole =
normalizePairedRole(row.target_role) ??
normalizePairedRole(row.intended_role);
const sourceAgentType =
normalizeStoredAgentType(row.source_agent_type) ??
(sourceRole
? resolveStableRoomRoleAgentType(database, {
chatJid: row.chat_jid,
groupFolder: row.group_folder,
role: sourceRole,
})
: null);
const targetAgentType =
normalizeStoredAgentType(row.target_agent_type) ??
(targetRole
? resolveStableRoomRoleAgentType(database, {
chatJid: row.chat_jid,
groupFolder: row.group_folder,
role: targetRole,
})
: null) ??
'claude-code';
const sourceServiceId =
row.source_service_id ??
(sourceRole != null
? (resolveRoleServiceShadow(sourceRole, sourceAgentType) ??
SERVICE_SESSION_SCOPE)
: SERVICE_SESSION_SCOPE);
const targetServiceId =
row.target_service_id ??
(targetRole != null
? (resolveRoleServiceShadow(targetRole, targetAgentType) ??
SERVICE_SESSION_SCOPE)
: SERVICE_SESSION_SCOPE);
if (
row.source_service_id === sourceServiceId &&
row.target_service_id === targetServiceId
) {
continue;
}
update.run(sourceServiceId, targetServiceId, 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 applySchemaMigrations(
database: Database,
args: {
assistantName: string;
hooks: SchemaMigrationHooks;
},
): void {
const { assistantName, hooks } = args;
const { assistantName } = args;
tryExecMigration(
database,
@@ -269,6 +677,14 @@ export function applySchemaMigrations(
database,
`ALTER TABLE service_handoffs ADD COLUMN source_agent_type TEXT`,
);
tryExecMigration(
database,
`ALTER TABLE service_handoffs ADD COLUMN source_service_id TEXT`,
);
tryExecMigration(
database,
`ALTER TABLE service_handoffs ADD COLUMN target_service_id TEXT`,
);
tryExecMigration(
database,
@@ -282,6 +698,14 @@ export function applySchemaMigrations(
database,
`ALTER TABLE paired_tasks ADD COLUMN arbiter_agent_type TEXT`,
);
tryExecMigration(
database,
`ALTER TABLE paired_tasks ADD COLUMN owner_service_id TEXT`,
);
tryExecMigration(
database,
`ALTER TABLE paired_tasks ADD COLUMN reviewer_service_id TEXT`,
);
tryExecMigration(
database,
`ALTER TABLE paired_task_execution_leases ADD COLUMN expires_at TEXT`,
@@ -311,6 +735,7 @@ export function applySchemaMigrations(
database,
`ALTER TABLE work_items ADD COLUMN delivery_role TEXT`,
);
tryExecMigration(database, `ALTER TABLE work_items ADD COLUMN service_id TEXT`);
database.exec(
`UPDATE service_handoffs
@@ -333,6 +758,22 @@ export function applySchemaMigrations(
WHERE source_role IS NULL`,
);
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);
database.exec(
`UPDATE room_settings
SET mode_source = 'explicit'
@@ -375,7 +816,7 @@ export function applySchemaMigrations(
tryExecMigration(database, `ALTER TABLE messages ADD COLUMN seq INTEGER`);
hooks.backfillMessageSeq(database);
backfillMessageSeq(database);
database.exec(`
CREATE UNIQUE INDEX IF NOT EXISTS idx_messages_seq ON messages(seq);
@@ -385,9 +826,9 @@ export function applySchemaMigrations(
database.exec(`DROP INDEX IF EXISTS idx_work_items_open;`);
database.exec(`
CREATE INDEX IF NOT EXISTS idx_work_items_group_agent
ON work_items(chat_jid, agent_type, delivery_role, status);
ON work_items(chat_jid, agent_type, service_id, delivery_role, status);
CREATE UNIQUE INDEX IF NOT EXISTS idx_work_items_open
ON work_items(chat_jid, agent_type, IFNULL(delivery_role, ''))
ON work_items(chat_jid, agent_type, IFNULL(service_id, ''), IFNULL(delivery_role, ''))
WHERE status IN ('produced', 'delivery_retry');
`);
@@ -516,6 +957,8 @@ export function applySchemaMigrations(
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,
@@ -548,11 +991,18 @@ export function applySchemaMigrations(
.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,
@@ -573,12 +1023,16 @@ export function applySchemaMigrations(
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_agent_type, reviewer_agent_type,
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, owner_agent_type, reviewer_agent_type,
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;
@@ -590,17 +1044,6 @@ export function applySchemaMigrations(
}
}
for (const column of [
'owner_agent_type',
'reviewer_agent_type',
'arbiter_agent_type',
]) {
tryExecMigration(
database,
`ALTER TABLE channel_owner ADD COLUMN ${column} TEXT`,
);
}
tryExecMigration(
database,
`ALTER TABLE paired_tasks ADD COLUMN completion_reason TEXT`,
@@ -663,6 +1106,8 @@ export function applySchemaMigrations(
}
migrateSessionsTableToCompositePk(database, 'claude-code');
backfillLegacyServiceSessions(database, inferAgentTypeFromServiceShadow);
dropLegacyServiceSessionsTable(database);
try {
database.exec(`ALTER TABLE chats ADD COLUMN channel TEXT`);
@@ -683,24 +1128,4 @@ export function applySchemaMigrations(
/* columns already exist */
}
hooks.backfillStoredRoomSettings(database);
if (tableHasColumn(database, 'channel_owner', 'owner_service_id')) {
hooks.backfillChannelOwnerRoleMetadata(database);
}
backfillLegacyServiceSessions(database, inferAgentTypeFromServiceShadow);
if (tableHasColumn(database, 'work_items', 'service_id')) {
hooks.backfillWorkItemServiceShadows(database);
}
if (tableHasColumn(database, 'service_handoffs', 'source_service_id')) {
hooks.backfillServiceHandoffServiceShadows(database);
}
if (tableHasColumn(database, 'paired_tasks', 'owner_service_id')) {
hooks.backfillPairedTaskRoleMetadata(database);
}
hooks.rebuildWorkItemsCanonicalSchema(database);
dropLegacyServiceSessionsTable(database);
hooks.rebuildChannelOwnerCanonicalSchema(database);
hooks.rebuildPairedTasksCanonicalSchema(database);
hooks.rebuildServiceHandoffsCanonicalSchema(database);
}

30
src/db/service-handoffs.ts
View File

@@ -70,6 +70,8 @@ interface StoredServiceHandoffRow extends Omit<
| 'source_agent_type'
| 'target_agent_type'
> {
source_service_id?: string | null;
target_service_id?: string | null;
source_agent_type?: string | null;
target_agent_type: string;
}
@@ -103,15 +105,17 @@ function hydrateServiceHandoffRow(
source_agent_type: sourceAgentType ?? null,
target_agent_type: targetAgentType,
source_service_id:
row.source_role != null
row.source_service_id ??
(row.source_role != null
? (resolveRoleServiceShadow(row.source_role, sourceAgentType) ??
SERVICE_SESSION_SCOPE)
: SERVICE_SESSION_SCOPE,
: SERVICE_SESSION_SCOPE),
target_service_id:
row.target_role != null
row.target_service_id ??
(row.target_role != null
? (resolveRoleServiceShadow(row.target_role, targetAgentType) ??
SERVICE_SESSION_SCOPE)
: SERVICE_SESSION_SCOPE,
: SERVICE_SESSION_SCOPE),
};
}
@@ -191,12 +195,26 @@ export function createServiceHandoffInDatabase(
role: sourceRole,
})
: null);
const sourceServiceId =
input.source_service_id ??
(sourceRole != null
? (resolveRoleServiceShadow(sourceRole, sourceAgentType) ?? null)
: null) ??
SERVICE_SESSION_SCOPE;
const targetServiceId =
input.target_service_id ??
(targetRole != null
? (resolveRoleServiceShadow(targetRole, input.target_agent_type) ?? null)
: null) ??
SERVICE_SESSION_SCOPE;
database
.prepare(
`INSERT INTO service_handoffs (
chat_jid,
group_folder,
source_service_id,
target_service_id,
source_role,
source_agent_type,
target_role,
@@ -206,11 +224,13 @@ export function createServiceHandoffInDatabase(
end_seq,
reason,
intended_role
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
)
.run(
input.chat_jid,
input.group_folder,
sourceServiceId,
targetServiceId,
sourceRole,
sourceAgentType ?? null,
targetRole,

106
src/db/sessions.ts
View File

@@ -47,6 +47,59 @@ export function migrateSessionsTableToCompositePk(
`);
}
export function backfillLegacyServiceSessions(
database: Database,
resolveAgentTypeFromServiceId: ServiceShadowAgentTypeResolver,
): void {
if (!hasLegacyServiceSessionsTable(database)) {
return;
}
const rows = database
.prepare(
`SELECT group_folder, service_id, session_id
FROM service_sessions`,
)
.all() as Array<{
group_folder: string;
service_id: string;
session_id: string;
}>;
const upsert = database.prepare(
`INSERT OR IGNORE INTO sessions (group_folder, agent_type, session_id)
VALUES (?, ?, ?)`,
);
const tx = database.transaction(
(
sessionRows: Array<{
group_folder: string;
service_id: string;
session_id: string;
}>,
) => {
for (const row of sessionRows) {
const agentType = resolveAgentTypeFromServiceId(row.service_id);
if (!agentType) {
continue;
}
upsert.run(row.group_folder, agentType, row.session_id);
}
},
);
tx(rows);
}
export function dropLegacyServiceSessionsTable(database: Database): void {
if (!hasLegacyServiceSessionsTable(database)) {
return;
}
database.exec(`DROP TABLE service_sessions`);
}
export function getSessionFromDatabase(
database: Database,
groupFolder: string,
@@ -110,56 +163,3 @@ export function getAllSessionsForAgentTypeFromDatabase(
}
return result;
}
export function backfillLegacyServiceSessions(
database: Database,
resolveAgentTypeFromServiceId: ServiceShadowAgentTypeResolver,
): void {
if (!hasLegacyServiceSessionsTable(database)) {
return;
}
const rows = database
.prepare(
`SELECT group_folder, service_id, session_id
FROM service_sessions`,
)
.all() as Array<{
group_folder: string;
service_id: string;
session_id: string;
}>;
const upsert = database.prepare(
`INSERT OR IGNORE INTO sessions (group_folder, agent_type, session_id)
VALUES (?, ?, ?)`,
);
const tx = database.transaction(
(
sessionRows: Array<{
group_folder: string;
service_id: string;
session_id: string;
}>,
) => {
for (const row of sessionRows) {
const agentType = resolveAgentTypeFromServiceId(row.service_id);
if (!agentType) {
continue;
}
upsert.run(row.group_folder, agentType, row.session_id);
}
},
);
tx(rows);
}
export function dropLegacyServiceSessionsTable(database: Database): void {
if (!hasLegacyServiceSessionsTable(database)) {
return;
}
database.exec(`DROP TABLE service_sessions`);
}

114
src/db/work-items.ts
View File

@@ -1,9 +1,12 @@
import { Database } from 'bun:sqlite';
import { SERVICE_SESSION_SCOPE } from '../config.js';
import { inferRoleFromServiceShadow } from '../role-service-shadow.js';
import { SERVICE_SESSION_SCOPE, normalizeServiceId } from '../config.js';
import {
inferAgentTypeFromServiceShadow,
inferRoleFromServiceShadow,
resolveRoleServiceShadow,
} from '../role-service-shadow.js';
import { AgentType, PairedRoomRole } from '../types.js';
import { resolveWorkItemServiceShadow } from './legacy-rebuilds.js';
export interface WorkItem {
id: number;
@@ -26,15 +29,17 @@ export interface WorkItem {
interface StoredWorkItemRow extends Omit<
WorkItem,
'service_id' | 'agent_type'
'agent_type' | 'service_id'
> {
agent_type: string;
service_id?: string | null;
}
export interface CreateProducedWorkItemInput {
group_folder: string;
chat_jid: string;
agent_type?: AgentType;
service_id?: string;
delivery_role?: PairedRoomRole | null;
start_seq: number | null;
end_seq: number | null;
@@ -48,38 +53,85 @@ function normalizeStoredAgentType(
return undefined;
}
function resolveStoredWorkItemServiceId(args: {
agentType: AgentType;
deliveryRole?: PairedRoomRole | null;
serviceId?: string | null;
}): string {
return (
(args.serviceId ? normalizeServiceId(args.serviceId) : null) ??
resolveRoleServiceShadow(args.deliveryRole ?? 'owner', args.agentType) ??
SERVICE_SESSION_SCOPE
);
}
function hydrateWorkItemRow(row: StoredWorkItemRow): WorkItem {
const agentType = normalizeStoredAgentType(row.agent_type) ?? 'claude-code';
return {
...row,
agent_type: agentType,
service_id: resolveWorkItemServiceShadow(agentType, row.delivery_role),
service_id: resolveStoredWorkItemServiceId({
agentType,
deliveryRole: row.delivery_role,
serviceId: row.service_id,
}),
};
}
function resolvePreferredWorkItemRole(
serviceId: string | null | undefined,
): PairedRoomRole | null {
const normalizedServiceId = serviceId ? normalizeServiceId(serviceId) : null;
if (!normalizedServiceId) {
return null;
}
const inferredAgentType = inferAgentTypeFromServiceShadow(normalizedServiceId);
return inferRoleFromServiceShadow(inferredAgentType, normalizedServiceId);
}
export function getOpenWorkItemFromDatabase(
database: Database,
chatJid: string,
agentType: AgentType = 'claude-code',
serviceId: string = SERVICE_SESSION_SCOPE,
): WorkItem | undefined {
const preferredRole = inferRoleFromServiceShadow(agentType, serviceId);
const normalizedServiceId = normalizeServiceId(serviceId);
const preferredRole = inferRoleFromServiceShadow(
agentType,
normalizedServiceId,
);
const row = database
.prepare(
`SELECT *
FROM work_items
WHERE chat_jid = ? AND agent_type = ?
AND status IN ('produced', 'delivery_retry')
AND (
service_id = ?
OR (? IS NOT NULL AND delivery_role = ?)
OR (? IS NULL AND delivery_role IS NULL)
)
ORDER BY
CASE
WHEN ? IS NOT NULL AND delivery_role = ? THEN 0
WHEN delivery_role IS NULL THEN 1
WHEN service_id = ? THEN 0
WHEN ? IS NOT NULL AND delivery_role = ? THEN 1
ELSE 2
END,
id ASC
LIMIT 1`,
)
.get(chatJid, agentType, preferredRole, preferredRole) as
.get(
chatJid,
agentType,
normalizedServiceId,
preferredRole,
preferredRole,
preferredRole,
normalizedServiceId,
preferredRole,
preferredRole,
) as
| StoredWorkItemRow
| undefined;
return row ? hydrateWorkItemRow(row) : undefined;
@@ -88,17 +140,40 @@ export function getOpenWorkItemFromDatabase(
export function getOpenWorkItemForChatFromDatabase(
database: Database,
chatJid: string,
serviceId: string = SERVICE_SESSION_SCOPE,
): WorkItem | undefined {
const normalizedServiceId = normalizeServiceId(serviceId);
const preferredRole = resolvePreferredWorkItemRole(normalizedServiceId);
const row = database
.prepare(
`SELECT *
FROM work_items
WHERE chat_jid = ?
AND status IN ('produced', 'delivery_retry')
ORDER BY id ASC
AND (
service_id = ?
OR (? IS NOT NULL AND delivery_role = ?)
OR (? IS NULL AND delivery_role IS NULL)
)
ORDER BY
CASE
WHEN service_id = ? THEN 0
WHEN ? IS NOT NULL AND delivery_role = ? THEN 1
ELSE 2
END,
id ASC
LIMIT 1`,
)
.get(chatJid) as StoredWorkItemRow | undefined;
.get(
chatJid,
normalizedServiceId,
preferredRole,
preferredRole,
preferredRole,
normalizedServiceId,
preferredRole,
preferredRole,
) as StoredWorkItemRow | undefined;
return row ? hydrateWorkItemRow(row) : undefined;
}
@@ -108,26 +183,33 @@ export function createProducedWorkItemInDatabase(
): WorkItem {
const now = new Date().toISOString();
const agentType = input.agent_type || 'claude-code';
const serviceId = resolveStoredWorkItemServiceId({
agentType,
deliveryRole: input.delivery_role,
serviceId: input.service_id,
});
database
.prepare(
`INSERT INTO work_items (
group_folder,
chat_jid,
agent_type,
service_id,
delivery_role,
status,
start_seq,
end_seq,
result_payload,
delivery_attempts,
created_at,
updated_at
) VALUES (?, ?, ?, ?, 'produced', ?, ?, ?, 0, ?, ?)`,
result_payload,
delivery_attempts,
created_at,
updated_at
) VALUES (?, ?, ?, ?, ?, 'produced', ?, ?, ?, 0, ?, ?)`,
)
.run(
input.group_folder,
input.chat_jid,
agentType,
serviceId,
input.delivery_role ?? null,
input.start_seq,
input.end_seq,

6
src/message-runtime-loop.ts
View File

@@ -10,6 +10,7 @@ import {
advanceLastAgentCursor,
getProcessableMessages,
} from './message-runtime-rules.js';
import { SERVICE_SESSION_SCOPE } from './config.js';
import type { ExecuteTurnFn } from './message-runtime-types.js';
import type { ScheduledPairedFollowUpIntentKind } from './paired-follow-up-scheduler.js';
import { findChannel, formatMessages } from './router.js';
@@ -134,7 +135,10 @@ export function recoverPendingMessages(args: {
}): void {
const registeredGroups = args.getRegisteredGroups();
for (const [chatJid, group] of Object.entries(registeredGroups)) {
const openWorkItem = getOpenWorkItemForChat(chatJid);
const openWorkItem = getOpenWorkItemForChat(
chatJid,
SERVICE_SESSION_SCOPE,
);
if (openWorkItem) {
logger.info(
{ chatJid, group: group.name, workItemId: openWorkItem.id },

15
src/message-runtime.test.ts
View File

@@ -1012,7 +1012,7 @@ describe('createMessageRuntime', () => {
});
expect(result).toBe(true);
expect(db.getOpenWorkItemForChat).toHaveBeenCalledWith(chatJid);
expect(db.getOpenWorkItemForChat).toHaveBeenCalledWith(chatJid, 'claude');
expect(reviewerChannel.sendMessage).toHaveBeenCalledWith(
chatJid,
'fallback reviewer final retry',
@@ -1137,7 +1137,10 @@ describe('createMessageRuntime', () => {
expect(lastAgentTimestamps[`${chatJid}:reviewer`]).toBe('41');
expect(saveState).toHaveBeenCalled();
expect(db.createProducedWorkItem).toHaveBeenCalledWith(
expect.objectContaining({ delivery_role: 'reviewer' }),
expect.objectContaining({
delivery_role: 'reviewer',
service_id: 'codex-main',
}),
);
expect(reviewerChannel.sendMessage).toHaveBeenCalledWith(
chatJid,
@@ -1324,6 +1327,7 @@ describe('createMessageRuntime', () => {
expect(db.createProducedWorkItem).toHaveBeenCalledWith(
expect.objectContaining({
delivery_role: 'reviewer',
service_id: 'codex-main',
result_payload: '**DONE**\n\n리뷰 승인',
}),
);
@@ -1514,7 +1518,10 @@ If your first line is DONE_WITH_CONCERNS, the system will reopen review instead
expect(result).toBe(true);
expect(agentRunner.runAgentProcess).toHaveBeenCalledTimes(1);
expect(db.createProducedWorkItem).toHaveBeenCalledWith(
expect.objectContaining({ delivery_role: 'owner' }),
expect.objectContaining({
delivery_role: 'owner',
service_id: 'claude',
}),
);
expect(channel.sendMessage).toHaveBeenCalledWith(chatJid, '최종 정리 완료');
});
@@ -5078,7 +5085,7 @@ If your first line is DONE_WITH_CONCERNS, the system will reopen review instead
runtime.recoverPendingMessages();
expect(db.getOpenWorkItemForChat).toHaveBeenCalledWith(chatJid);
expect(db.getOpenWorkItemForChat).toHaveBeenCalledWith(chatJid, 'claude');
expect(enqueueMessageCheck).toHaveBeenCalledWith(
chatJid,
resolveGroupIpcPath(group.folder),

18
src/message-runtime.ts
View File

@@ -7,7 +7,7 @@ import {
getLastBotFinalMessage,
getLatestOpenPairedTaskForChat,
} from './db.js';
import { isSessionCommandSenderAllowed } from './config.js';
import { isSessionCommandSenderAllowed, SERVICE_SESSION_SCOPE } from './config.js';
import { GroupQueue, GroupRunContext } from './group-queue.js';
import {
findChannel,
@@ -59,7 +59,11 @@ import {
type PairedTask,
} from './types.js';
import { createScopedLogger, logger } from './logger.js';
import { hasReviewerLease } from './service-routing.js';
import {
getEffectiveChannelLease,
hasReviewerLease,
resolveLeaseServiceId,
} from './service-routing.js';
export {
resolveHandoffCursorKey,
resolveHandoffRoleOverride,
@@ -270,6 +274,10 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): {
deliverFinalText: async (text) => {
try {
const persistedDeliveryRole = resolvedDeliveryRole;
const persistedDeliveryServiceId = resolveLeaseServiceId(
getEffectiveChannelLease(chatJid),
persistedDeliveryRole ?? 'owner',
);
if (
(persistedDeliveryRole === 'reviewer' ||
persistedDeliveryRole === 'arbiter') &&
@@ -294,6 +302,7 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): {
chat_jid: chatJid,
agent_type:
args.forcedAgentType ?? group.agentType ?? 'claude-code',
service_id: persistedDeliveryServiceId ?? undefined,
delivery_role: persistedDeliveryRole,
start_seq: startSeq,
end_seq: endSeq,
@@ -471,7 +480,10 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): {
// Delivery retries can come from forced fallback runs whose agent_type
// differs from the room owner's registered agent type.
const openWorkItem = getOpenWorkItemForChat(chatJid);
const openWorkItem = getOpenWorkItemForChat(
chatJid,
SERVICE_SESSION_SCOPE,
);
const openWorkItemOutcome = await processOpenWorkItemDelivery({
chatJid,
runId,

8
src/room-role-context.test.ts
View File

@@ -28,7 +28,7 @@ describe('buildRoomRoleContext', () => {
});
});
it('prefers the canonical reviewer shadow over a stale compatibility field', () => {
it('uses the stored reviewer service id from the lease row as-is', () => {
expect(
buildRoomRoleContext(
{
@@ -43,13 +43,13 @@ describe('buildRoomRoleContext', () => {
reason: null,
explicit: true,
},
'codex-review',
'stale-reviewer-shadow',
),
).toEqual({
serviceId: 'codex-review',
serviceId: 'stale-reviewer-shadow',
role: 'reviewer',
ownerServiceId: 'claude',
reviewerServiceId: 'codex-review',
reviewerServiceId: 'stale-reviewer-shadow',
ownerAgentType: 'claude-code',
reviewerAgentType: 'codex',
failoverOwner: false,

32
src/service-routing.test.ts
View File

@@ -4,6 +4,7 @@ import {
_setRegisteredGroupForTests,
_setStoredRoomOwnerAgentTypeForTests,
_initTestDatabase,
setChannelOwnerLease,
setExplicitRoomMode,
} from './db.js';
import {
@@ -246,7 +247,7 @@ describe('service-routing global failover', () => {
});
describe('resolveLeaseServiceId', () => {
it('prefers canonical reviewer shadow over a stale compatibility field', () => {
it('trusts the stored reviewer service id when a lease row already persisted it', () => {
expect(
resolveLeaseServiceId(
{
@@ -260,7 +261,7 @@ describe('resolveLeaseServiceId', () => {
},
'reviewer',
),
).toBe('codex-review');
).toBe('stale-reviewer-shadow');
});
it('keeps the explicit owner failover service instead of recomputing the stable shadow', () => {
@@ -280,3 +281,30 @@ describe('resolveLeaseServiceId', () => {
).toBe('codex-review');
});
});
describe('stored lease ids as SSOT', () => {
it('preserves a stored reviewer service id instead of recomputing it from role metadata', () => {
setChannelOwnerLease({
chat_jid: 'dc:stored-reviewer-ssot',
owner_service_id: 'claude',
reviewer_service_id: 'stale-reviewer-shadow',
owner_agent_type: 'claude-code',
reviewer_agent_type: 'codex',
activated_at: '2026-04-09T00:00:00.000Z',
reason: 'ssot-test',
});
refreshChannelOwnerCache(true);
const lease = getEffectiveChannelLease('dc:stored-reviewer-ssot');
expect(lease).toMatchObject({
owner_service_id: 'claude',
reviewer_service_id: 'stale-reviewer-shadow',
owner_agent_type: 'claude-code',
reviewer_agent_type: 'codex',
explicit: true,
});
expect(resolveLeaseServiceId(lease, 'reviewer')).toBe(
'stale-reviewer-shadow',
);
});
});

66
src/service-routing.ts
View File

@@ -76,22 +76,24 @@ function normalizeLeaseRow(
reviewer_agent_type: reviewerAgentType,
arbiter_agent_type: arbiterAgentType,
owner_service_id:
(row.owner_service_id ? normalizeServiceId(row.owner_service_id) : null) ??
resolveRoleServiceShadow('owner', ownerAgentType) ??
normalizeServiceId(row.owner_service_id),
reviewer_service_id: reviewerAgentType
? (resolveRoleServiceShadow('reviewer', reviewerAgentType) ??
normalizeServiceId(row.reviewer_service_id!))
: null,
arbiter_service_id: arbiterAgentType
? (resolveRoleServiceShadow('arbiter', arbiterAgentType) ??
(row.arbiter_service_id
? normalizeServiceId(row.arbiter_service_id)
: null))
: row.arbiter_service_id
normalizeServiceId(SERVICE_ID),
reviewer_service_id:
(row.reviewer_service_id
? normalizeServiceId(row.reviewer_service_id)
: null) ??
(reviewerAgentType
? resolveRoleServiceShadow('reviewer', reviewerAgentType)
: null),
arbiter_service_id:
(row.arbiter_service_id
? normalizeServiceId(row.arbiter_service_id)
: isArbiterEnabled()
? ARBITER_SERVICE_ID
: null,
: null) ??
(arbiterAgentType
? resolveRoleServiceShadow('arbiter', arbiterAgentType)
: null) ??
(isArbiterEnabled() ? ARBITER_SERVICE_ID : null),
owner_failover_active: false,
activated_at: row.activated_at,
reason: row.reason,
@@ -211,31 +213,31 @@ export function resolveLeaseServiceId(
): string | null {
switch (role) {
case 'owner':
if (lease.owner_failover_active) {
return normalizeServiceId(lease.owner_service_id);
}
return (
(lease.owner_service_id
? normalizeServiceId(lease.owner_service_id)
: null) ??
resolveRoleServiceShadow('owner', lease.owner_agent_type) ??
normalizeServiceId(lease.owner_service_id)
null
);
case 'reviewer':
return lease.reviewer_agent_type
? (resolveRoleServiceShadow('reviewer', lease.reviewer_agent_type) ??
(lease.reviewer_service_id
? normalizeServiceId(lease.reviewer_service_id)
: null))
: lease.reviewer_service_id
return (
(lease.reviewer_service_id
? normalizeServiceId(lease.reviewer_service_id)
: null;
: null) ??
(lease.reviewer_agent_type
? resolveRoleServiceShadow('reviewer', lease.reviewer_agent_type)
: null)
);
case 'arbiter':
return lease.arbiter_agent_type
? (resolveRoleServiceShadow('arbiter', lease.arbiter_agent_type) ??
(lease.arbiter_service_id
? normalizeServiceId(lease.arbiter_service_id)
: null))
: lease.arbiter_service_id
return (
(lease.arbiter_service_id
? normalizeServiceId(lease.arbiter_service_id)
: null;
: null) ??
(lease.arbiter_agent_type
? resolveRoleServiceShadow('arbiter', lease.arbiter_agent_type)
: null)
);
}
}