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

refactor: split paired db helpers

Browse Source
This commit is contained in:
ejclaw
2026-04-08 21:09:24 +09:00
parent 1fedd36ec7
commit e4ec22b35b
5 changed files with 678 additions and 440 deletions
Download Patch File Download Diff File Expand all files Collapse all files

45
src/db.test.ts
View File

@@ -28,6 +28,7 @@ import {
getExplicitRoomMode,
getLatestMessageSeqAtOrBefore,
getLatestPairedTaskForChat,
getLatestTurnNumber,
getLastRespondingAgentType,
getMessagesSinceSeq,
getNewMessagesBySeq,
@@ -40,11 +41,13 @@ import {
getNewMessages,
getPairedProject,
getPairedTaskById,
getPairedTurnOutputs,
getPairedWorkspace,
getRouterState,
getSession,
getStoredRoomSettings,
getTaskById,
insertPairedTurnOutput,
listPairedWorkspacesForTask,
markWorkItemDelivered,
markWorkItemDeliveryRetry,
@@ -771,6 +774,48 @@ describe('paired task state', () => {
).toEqual(['/tmp/reviewer-v2']);
});
it('stores paired turn outputs in order and truncates oversized text', () => {
createPairedTask({
id: 'paired-task-turn-output',
chat_jid: 'dc:paired',
group_folder: 'paired-room',
owner_service_id: 'codex-main',
reviewer_service_id: 'codex-review',
title: null,
source_ref: null,
plan_notes: null,
round_trip_count: 0,
review_requested_at: null,
status: 'active',
arbiter_verdict: null,
arbiter_requested_at: null,
completion_reason: null,
created_at: '2026-03-28T00:00:00.000Z',
updated_at: '2026-03-28T00:00:00.000Z',
});
insertPairedTurnOutput(
'paired-task-turn-output',
2,
'reviewer',
'review turn',
);
insertPairedTurnOutput(
'paired-task-turn-output',
1,
'owner',
'x'.repeat(60_000),
);
const outputs = getPairedTurnOutputs('paired-task-turn-output');
expect(outputs.map((output) => output.turn_number)).toEqual([1, 2]);
expect(outputs[0].role).toBe('owner');
expect(outputs[0].output_text).toHaveLength(50_000);
expect(outputs[1].output_text).toBe('review turn');
expect(getLatestTurnNumber('paired-task-turn-output')).toBe(2);
});
it('normalizes paired task service shadow from persisted role agent types during init', () => {
const tempDir = fs.mkdtempSync('/tmp/ejclaw-paired-task-shadow-');
const dbPath = path.join(tempDir, 'messages.db');

489
src/db.ts
View File

@@ -49,6 +49,14 @@ import {
openInMemoryDatabase,
openPersistentDatabase,
} from './db/bootstrap.js';
import {
clearChannelOwnerLeaseInDatabase,
getAllChannelOwnerLeasesFromDatabase,
getChannelOwnerLeaseFromDatabase,
setChannelOwnerLeaseInDatabase,
type ChannelOwnerLeaseRow,
type SetChannelOwnerLeaseInput,
} from './db/channel-owner-leases.js';
import {
type MemoryRecord,
type MemoryScopeKind,
@@ -87,6 +95,25 @@ import {
markWorkItemDeliveredInDatabase,
markWorkItemDeliveryRetryInDatabase,
} from './db/work-items.js';
import {
type PairedTaskUpdates,
createPairedTaskInDatabase,
getLastBotFinalMessageFromDatabase,
getLatestOpenPairedTaskForChatFromDatabase,
getLatestPairedTaskForChatFromDatabase,
getPairedProjectFromDatabase,
getPairedTaskByIdFromDatabase,
getPairedWorkspaceFromDatabase,
listPairedWorkspacesForTaskFromDatabase,
updatePairedTaskInDatabase,
upsertPairedProjectInDatabase,
upsertPairedWorkspaceInDatabase,
} from './db/paired-state.js';
import {
getLatestTurnNumberFromDatabase,
getPairedTurnOutputsFromDatabase,
insertPairedTurnOutputInDatabase,
} from './db/paired-turn-outputs.js';
import {
getLastRespondingAgentTypeFromDatabase,
getRouterStateForServiceFromDatabase,
@@ -105,7 +132,6 @@ import {
rebuildServiceHandoffsCanonicalSchema,
rebuildWorkItemsCanonicalSchema,
resolveStablePairedTaskOwnerAgentType,
resolveStableReviewerAgentType,
resolveStableRoomRoleAgentType,
} from './db/legacy-rebuilds.js';
import {
@@ -181,18 +207,7 @@ export type {
} from './db/memories.js';
export type { ChatInfo } from './db/messages.js';
export type { WorkItem } from './db/work-items.js';
export interface ChannelOwnerLeaseRow {
chat_jid: string;
owner_service_id: string;
reviewer_service_id: string | null;
arbiter_service_id: string | null;
owner_agent_type?: AgentType | null;
reviewer_agent_type?: AgentType | null;
arbiter_agent_type?: AgentType | null;
activated_at: string | null;
reason: string | null;
}
export type { ChannelOwnerLeaseRow } from './db/channel-owner-leases.js';
export interface ServiceHandoff {
id: number;
@@ -216,21 +231,6 @@ export interface ServiceHandoff {
last_error: string | null;
}
interface StoredChannelOwnerLeaseRow {
chat_jid: string;
owner_agent_type?: string | null;
reviewer_agent_type?: string | null;
arbiter_agent_type?: string | null;
activated_at: string | null;
reason: string | null;
}
interface LegacyChannelOwnerLeaseServiceRow extends StoredChannelOwnerLeaseRow {
owner_service_id?: string | null;
reviewer_service_id?: string | null;
arbiter_service_id?: string | null;
}
interface StoredServiceHandoffRow extends Omit<
ServiceHandoff,
| 'source_service_id'
@@ -247,24 +247,6 @@ interface LegacyServiceHandoffServiceRow extends StoredServiceHandoffRow {
target_service_id?: string | null;
}
interface StoredPairedTaskRow extends Omit<
PairedTask,
| 'owner_service_id'
| 'reviewer_service_id'
| 'owner_agent_type'
| 'reviewer_agent_type'
| 'arbiter_agent_type'
> {
owner_agent_type?: string | null;
reviewer_agent_type?: string | null;
arbiter_agent_type?: string | null;
}
interface LegacyPairedTaskServiceRow extends StoredPairedTaskRow {
owner_service_id?: string | null;
reviewer_service_id?: string | null;
}
function backfillMessageSeq(database: Database): void {
const rows = database
.prepare(
@@ -318,68 +300,6 @@ function getSchemaMigrationHooks(): SchemaMigrationHooks {
};
}
function hydrateChannelOwnerLeaseRow(
row: StoredChannelOwnerLeaseRow,
): ChannelOwnerLeaseRow {
const ownerAgentType =
normalizeStoredAgentType(row.owner_agent_type) ?? OWNER_AGENT_TYPE;
const reviewerAgentType =
row.reviewer_agent_type == null
? null
: (normalizeStoredAgentType(row.reviewer_agent_type) ??
resolveStableReviewerAgentType(ownerAgentType, null));
const arbiterAgentType =
row.arbiter_agent_type == null
? null
: (normalizeStoredAgentType(row.arbiter_agent_type) ??
ARBITER_AGENT_TYPE ??
null);
return {
chat_jid: row.chat_jid,
owner_service_id:
resolveRoleServiceShadow('owner', ownerAgentType) ?? CLAUDE_SERVICE_ID,
reviewer_service_id:
reviewerAgentType == null
? null
: resolveRoleServiceShadow('reviewer', reviewerAgentType),
arbiter_service_id:
arbiterAgentType == null
? null
: resolveRoleServiceShadow('arbiter', arbiterAgentType),
owner_agent_type: ownerAgentType,
reviewer_agent_type: reviewerAgentType,
arbiter_agent_type: arbiterAgentType,
activated_at: row.activated_at,
reason: row.reason,
};
}
function hydratePairedTaskRow(row: StoredPairedTaskRow): PairedTask {
const ownerAgentType = resolveStablePairedTaskOwnerAgentType(db, row);
const reviewerAgentType = resolveStableReviewerAgentType(
ownerAgentType,
row.reviewer_agent_type ?? null,
);
const arbiterAgentType =
normalizeStoredAgentType(row.arbiter_agent_type) ??
ARBITER_AGENT_TYPE ??
null;
return {
...row,
owner_service_id:
resolveRoleServiceShadow('owner', ownerAgentType) ??
CODEX_MAIN_SERVICE_ID,
reviewer_service_id:
resolveRoleServiceShadow('reviewer', reviewerAgentType) ??
CODEX_REVIEW_SERVICE_ID,
owner_agent_type: ownerAgentType ?? null,
reviewer_agent_type: reviewerAgentType ?? null,
arbiter_agent_type: arbiterAgentType,
};
}
function hydrateServiceHandoffRow(
row: StoredServiceHandoffRow,
): ServiceHandoff {
@@ -1198,244 +1118,50 @@ export function getEffectiveRuntimeRoomMode(chatJid: string): RoomMode {
// --- Paired task/project/workspace state ---
export function upsertPairedProject(project: PairedProject): void {
db.prepare(
`
INSERT INTO paired_projects (
chat_jid,
group_folder,
canonical_work_dir,
created_at,
updated_at
)
VALUES (?, ?, ?, ?, ?)
ON CONFLICT(chat_jid) DO UPDATE SET
group_folder = excluded.group_folder,
canonical_work_dir = excluded.canonical_work_dir,
updated_at = excluded.updated_at
`,
).run(
project.chat_jid,
project.group_folder,
project.canonical_work_dir,
project.created_at,
project.updated_at,
);
upsertPairedProjectInDatabase(db, project);
}
export function getPairedProject(chatJid: string): PairedProject | undefined {
return db
.prepare('SELECT * FROM paired_projects WHERE chat_jid = ?')
.get(chatJid) as PairedProject | undefined;
return getPairedProjectFromDatabase(db, chatJid);
}
export function createPairedTask(task: PairedTask): void {
db.prepare(
`
INSERT INTO paired_tasks (
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,
created_at,
updated_at
)
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,
task.title,
task.source_ref,
task.plan_notes,
task.review_requested_at,
task.round_trip_count,
task.status,
task.arbiter_verdict,
task.arbiter_requested_at,
task.created_at,
task.updated_at,
);
createPairedTaskInDatabase(db, task);
}
export function getPairedTaskById(id: string): PairedTask | undefined {
const row = db.prepare('SELECT * FROM paired_tasks WHERE id = ?').get(id) as
| StoredPairedTaskRow
| undefined;
return row ? hydratePairedTaskRow(row) : undefined;
return getPairedTaskByIdFromDatabase(db, id);
}
export function getLatestPairedTaskForChat(
chatJid: string,
): PairedTask | undefined {
const row = db
.prepare(
`
SELECT *
FROM paired_tasks
WHERE chat_jid = ?
ORDER BY updated_at DESC
LIMIT 1
`,
)
.get(chatJid) as StoredPairedTaskRow | undefined;
return row ? hydratePairedTaskRow(row) : undefined;
return getLatestPairedTaskForChatFromDatabase(db, chatJid);
}
export function getLatestOpenPairedTaskForChat(
chatJid: string,
): PairedTask | undefined {
const row = db
.prepare(
`
SELECT *
FROM paired_tasks
WHERE chat_jid = ?
AND status NOT IN ('completed')
ORDER BY updated_at DESC
LIMIT 1
`,
)
.get(chatJid) as StoredPairedTaskRow | undefined;
return row ? hydratePairedTaskRow(row) : undefined;
return getLatestOpenPairedTaskForChatFromDatabase(db, chatJid);
}
export function updatePairedTask(
id: string,
updates: Partial<
Pick<
PairedTask,
| 'title'
| 'source_ref'
| 'plan_notes'
| 'review_requested_at'
| 'round_trip_count'
| 'status'
| 'arbiter_verdict'
| 'arbiter_requested_at'
| 'completion_reason'
| 'updated_at'
>
>,
): void {
const fields: string[] = [];
const values: (string | number | null)[] = [];
if (updates.title !== undefined) {
fields.push('title = ?');
values.push(updates.title);
}
if (updates.source_ref !== undefined) {
fields.push('source_ref = ?');
values.push(updates.source_ref);
}
if (updates.plan_notes !== undefined) {
fields.push('plan_notes = ?');
values.push(updates.plan_notes);
}
if (updates.review_requested_at !== undefined) {
fields.push('review_requested_at = ?');
values.push(updates.review_requested_at);
}
if (updates.round_trip_count !== undefined) {
fields.push('round_trip_count = ?');
values.push(updates.round_trip_count);
}
if (updates.status !== undefined) {
fields.push('status = ?');
values.push(updates.status);
}
if (updates.arbiter_verdict !== undefined) {
fields.push('arbiter_verdict = ?');
values.push(updates.arbiter_verdict);
}
if (updates.arbiter_requested_at !== undefined) {
fields.push('arbiter_requested_at = ?');
values.push(updates.arbiter_requested_at);
}
if (updates.completion_reason !== undefined) {
fields.push('completion_reason = ?');
values.push(updates.completion_reason);
}
if (updates.updated_at !== undefined) {
fields.push('updated_at = ?');
values.push(updates.updated_at);
}
if (fields.length === 0) return;
values.push(id);
db.prepare(`UPDATE paired_tasks SET ${fields.join(', ')} WHERE id = ?`).run(
...values,
);
export function updatePairedTask(id: string, updates: PairedTaskUpdates): void {
updatePairedTaskInDatabase(db, id, updates);
}
export function upsertPairedWorkspace(workspace: PairedWorkspace): void {
db.prepare(
`
INSERT INTO paired_workspaces (
id,
task_id,
role,
workspace_dir,
snapshot_source_dir,
snapshot_ref,
status,
snapshot_refreshed_at,
created_at,
updated_at
)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(id) DO UPDATE SET
workspace_dir = excluded.workspace_dir,
snapshot_source_dir = excluded.snapshot_source_dir,
snapshot_ref = excluded.snapshot_ref,
status = excluded.status,
snapshot_refreshed_at = excluded.snapshot_refreshed_at,
updated_at = excluded.updated_at
`,
).run(
workspace.id,
workspace.task_id,
workspace.role,
workspace.workspace_dir,
workspace.snapshot_source_dir,
workspace.snapshot_ref,
workspace.status,
workspace.snapshot_refreshed_at,
workspace.created_at,
workspace.updated_at,
);
upsertPairedWorkspaceInDatabase(db, workspace);
}
export function getPairedWorkspace(
taskId: string,
role: PairedWorkspace['role'],
): PairedWorkspace | undefined {
return db
.prepare('SELECT * FROM paired_workspaces WHERE task_id = ? AND role = ?')
.get(taskId, role) as PairedWorkspace | undefined;
return getPairedWorkspaceFromDatabase(db, taskId, role);
}
export function listPairedWorkspacesForTask(taskId: string): PairedWorkspace[] {
return db
.prepare(
'SELECT * FROM paired_workspaces WHERE task_id = ? ORDER BY created_at',
)
.all(taskId) as PairedWorkspace[];
return listPairedWorkspacesForTaskFromDatabase(db, taskId);
}
/**
@@ -1444,19 +1170,10 @@ export function listPairedWorkspacesForTask(taskId: string): PairedWorkspace[] {
*/
export function getLastBotFinalMessage(
chatJid: string,
_agentType: AgentType = 'claude-code',
agentType: AgentType = 'claude-code',
limit: number = 1,
): Array<{ content: string; timestamp: string }> {
const rows = db
.prepare(
`SELECT content, timestamp
FROM messages
WHERE chat_jid = ? AND is_bot_message = 1
ORDER BY timestamp DESC, seq DESC
LIMIT ?`,
)
.all(chatJid, limit) as Array<{ content: string; timestamp: string }>;
return rows;
return getLastBotFinalMessageFromDatabase(db, chatJid, agentType, limit);
}
// --- Channel owner lease accessors ---
@@ -1464,88 +1181,19 @@ export function getLastBotFinalMessage(
export function getChannelOwnerLease(
chatJid: string,
): ChannelOwnerLeaseRow | undefined {
const row = db
.prepare(
`SELECT
chat_jid,
owner_agent_type,
reviewer_agent_type,
arbiter_agent_type,
activated_at,
reason
FROM channel_owner
WHERE chat_jid = ?`,
)
.get(chatJid) as StoredChannelOwnerLeaseRow | undefined;
return row ? hydrateChannelOwnerLeaseRow(row) : undefined;
return getChannelOwnerLeaseFromDatabase(db, chatJid);
}
export function getAllChannelOwnerLeases(): ChannelOwnerLeaseRow[] {
const rows = db
.prepare(
`SELECT
chat_jid,
owner_agent_type,
reviewer_agent_type,
arbiter_agent_type,
activated_at,
reason
FROM channel_owner`,
)
.all() as StoredChannelOwnerLeaseRow[];
return rows.map(hydrateChannelOwnerLeaseRow);
return getAllChannelOwnerLeasesFromDatabase(db);
}
export function setChannelOwnerLease(input: {
chat_jid: string;
owner_service_id?: string;
reviewer_service_id?: string | null;
arbiter_service_id?: string | null;
owner_agent_type?: AgentType | null;
reviewer_agent_type?: AgentType | null;
arbiter_agent_type?: AgentType | null;
activated_at?: string | null;
reason?: string | null;
}): void {
const ownerAgentType =
normalizeStoredAgentType(input.owner_agent_type) ??
inferAgentTypeFromServiceShadow(input.owner_service_id) ??
OWNER_AGENT_TYPE;
const reviewerAgentType =
input.reviewer_service_id == null && input.reviewer_agent_type == null
? null
: (normalizeStoredAgentType(input.reviewer_agent_type) ??
inferAgentTypeFromServiceShadow(input.reviewer_service_id ?? null) ??
resolveStableReviewerAgentType(ownerAgentType, null));
const arbiterAgentType =
input.arbiter_service_id == null && input.arbiter_agent_type == null
? null
: (normalizeStoredAgentType(input.arbiter_agent_type) ??
inferAgentTypeFromServiceShadow(input.arbiter_service_id ?? null) ??
ARBITER_AGENT_TYPE ??
null);
db.prepare(
`INSERT OR REPLACE INTO channel_owner (
chat_jid,
owner_agent_type,
reviewer_agent_type,
arbiter_agent_type,
activated_at,
reason
) VALUES (?, ?, ?, ?, ?, ?)`,
).run(
input.chat_jid,
ownerAgentType ?? null,
reviewerAgentType ?? null,
arbiterAgentType ?? null,
input.activated_at ?? new Date().toISOString(),
input.reason ?? null,
);
export function setChannelOwnerLease(input: SetChannelOwnerLeaseInput): void {
setChannelOwnerLeaseInDatabase(db, input);
}
export function clearChannelOwnerLease(chatJid: string): void {
db.prepare('DELETE FROM channel_owner WHERE chat_jid = ?').run(chatJid);
clearChannelOwnerLeaseInDatabase(db, chatJid);
}
// --- Cross-service handoff accessors ---
@@ -1762,58 +1410,19 @@ export function completeServiceHandoffAndAdvanceTargetCursor(input: {
})();
}
// ── Paired turn outputs ──────────────────────────────────────────
const MAX_TURN_OUTPUT_CHARS = 50_000;
export function insertPairedTurnOutput(
taskId: string,
turnNumber: number,
role: PairedRoomRole,
outputText: string,
): void {
if (outputText.length > MAX_TURN_OUTPUT_CHARS) {
logger.warn(
{
taskId,
turnNumber,
role,
originalLen: outputText.length,
maxLen: MAX_TURN_OUTPUT_CHARS,
},
'Paired turn output truncated — agent output exceeds storage limit',
);
}
db.prepare(
`INSERT OR REPLACE INTO paired_turn_outputs
(task_id, turn_number, role, output_text, created_at)
VALUES (?, ?, ?, ?, ?)`,
).run(
taskId,
turnNumber,
role,
outputText.slice(0, MAX_TURN_OUTPUT_CHARS),
new Date().toISOString(),
);
insertPairedTurnOutputInDatabase(db, taskId, turnNumber, role, outputText);
}
export function getPairedTurnOutputs(taskId: string): PairedTurnOutput[] {
return db
.prepare(
`SELECT * FROM paired_turn_outputs
WHERE task_id = ?
ORDER BY turn_number ASC`,
)
.all(taskId) as PairedTurnOutput[];
return getPairedTurnOutputsFromDatabase(db, taskId);
}
export function getLatestTurnNumber(taskId: string): number {
const row = db
.prepare(
`SELECT MAX(turn_number) as max_turn
FROM paired_turn_outputs
WHERE task_id = ?`,
)
.get(taskId) as { max_turn: number | null } | undefined;
return row?.max_turn ?? 0;
return getLatestTurnNumberFromDatabase(db, taskId);
}

172
src/db/channel-owner-leases.ts Normal file
View File

@@ -0,0 +1,172 @@
import { Database } from 'bun:sqlite';
import {
ARBITER_AGENT_TYPE,
CLAUDE_SERVICE_ID,
OWNER_AGENT_TYPE,
} from '../config.js';
import {
inferAgentTypeFromServiceShadow,
resolveRoleServiceShadow,
} from '../role-service-shadow.js';
import { AgentType } from '../types.js';
import { resolveStableReviewerAgentType } from './legacy-rebuilds.js';
import { normalizeStoredAgentType } from './room-registration.js';
export interface ChannelOwnerLeaseRow {
chat_jid: string;
owner_service_id: string;
reviewer_service_id: string | null;
arbiter_service_id: string | null;
owner_agent_type?: AgentType | null;
reviewer_agent_type?: AgentType | null;
arbiter_agent_type?: AgentType | null;
activated_at: string | null;
reason: string | null;
}
interface StoredChannelOwnerLeaseRow {
chat_jid: string;
owner_agent_type?: string | null;
reviewer_agent_type?: string | null;
arbiter_agent_type?: string | null;
activated_at: string | null;
reason: string | null;
}
export interface SetChannelOwnerLeaseInput {
chat_jid: string;
owner_service_id?: string;
reviewer_service_id?: string | null;
arbiter_service_id?: string | null;
owner_agent_type?: AgentType | null;
reviewer_agent_type?: AgentType | null;
arbiter_agent_type?: AgentType | null;
activated_at?: string | null;
reason?: string | null;
}
function hydrateChannelOwnerLeaseRow(
row: StoredChannelOwnerLeaseRow,
): ChannelOwnerLeaseRow {
const ownerAgentType =
normalizeStoredAgentType(row.owner_agent_type) ?? OWNER_AGENT_TYPE;
const reviewerAgentType =
row.reviewer_agent_type == null
? null
: (normalizeStoredAgentType(row.reviewer_agent_type) ??
resolveStableReviewerAgentType(ownerAgentType, null));
const arbiterAgentType =
row.arbiter_agent_type == null
? null
: (normalizeStoredAgentType(row.arbiter_agent_type) ??
ARBITER_AGENT_TYPE ??
null);
return {
chat_jid: row.chat_jid,
owner_service_id:
resolveRoleServiceShadow('owner', ownerAgentType) ?? CLAUDE_SERVICE_ID,
reviewer_service_id:
reviewerAgentType == null
? null
: resolveRoleServiceShadow('reviewer', reviewerAgentType),
arbiter_service_id:
arbiterAgentType == null
? null
: resolveRoleServiceShadow('arbiter', arbiterAgentType),
owner_agent_type: ownerAgentType,
reviewer_agent_type: reviewerAgentType,
arbiter_agent_type: arbiterAgentType,
activated_at: row.activated_at,
reason: row.reason,
};
}
export function getChannelOwnerLeaseFromDatabase(
database: Database,
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 = ?`,
)
.get(chatJid) as StoredChannelOwnerLeaseRow | undefined;
return row ? hydrateChannelOwnerLeaseRow(row) : undefined;
}
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`,
)
.all() as StoredChannelOwnerLeaseRow[];
return rows.map(hydrateChannelOwnerLeaseRow);
}
export function setChannelOwnerLeaseInDatabase(
database: Database,
input: SetChannelOwnerLeaseInput,
): void {
const ownerAgentType =
normalizeStoredAgentType(input.owner_agent_type) ??
inferAgentTypeFromServiceShadow(input.owner_service_id) ??
OWNER_AGENT_TYPE;
const reviewerAgentType =
input.reviewer_service_id == null && input.reviewer_agent_type == null
? null
: (normalizeStoredAgentType(input.reviewer_agent_type) ??
inferAgentTypeFromServiceShadow(input.reviewer_service_id ?? null) ??
resolveStableReviewerAgentType(ownerAgentType, null));
const arbiterAgentType =
input.arbiter_service_id == null && input.arbiter_agent_type == null
? null
: (normalizeStoredAgentType(input.arbiter_agent_type) ??
inferAgentTypeFromServiceShadow(input.arbiter_service_id ?? null) ??
ARBITER_AGENT_TYPE ??
null);
database
.prepare(
`INSERT OR REPLACE INTO channel_owner (
chat_jid,
owner_agent_type,
reviewer_agent_type,
arbiter_agent_type,
activated_at,
reason
) VALUES (?, ?, ?, ?, ?, ?)`,
)
.run(
input.chat_jid,
ownerAgentType ?? null,
reviewerAgentType ?? null,
arbiterAgentType ?? null,
input.activated_at ?? new Date().toISOString(),
input.reason ?? null,
);
}
export function clearChannelOwnerLeaseInDatabase(
database: Database,
chatJid: string,
): void {
database.prepare('DELETE FROM channel_owner WHERE chat_jid = ?').run(chatJid);
}

344
src/db/paired-state.ts Normal file
View File

@@ -0,0 +1,344 @@
import { Database } from 'bun:sqlite';
import {
ARBITER_AGENT_TYPE,
CODEX_MAIN_SERVICE_ID,
CODEX_REVIEW_SERVICE_ID,
} from '../config.js';
import {
resolveStablePairedTaskOwnerAgentType,
resolveStableReviewerAgentType,
} from './legacy-rebuilds.js';
import { normalizeStoredAgentType } from './room-registration.js';
import { resolveRoleServiceShadow } from '../role-service-shadow.js';
import {
AgentType,
PairedProject,
PairedTask,
PairedWorkspace,
} from '../types.js';
interface StoredPairedTaskRow extends Omit<
PairedTask,
'owner_service_id' | 'reviewer_service_id' | 'owner_agent_type' | 'reviewer_agent_type' | 'arbiter_agent_type'
> {
owner_agent_type?: string | null;
reviewer_agent_type?: string | null;
arbiter_agent_type?: string | null;
}
export type PairedTaskUpdates = Partial<
Pick<
PairedTask,
| 'title'
| 'source_ref'
| 'plan_notes'
| 'review_requested_at'
| 'round_trip_count'
| 'status'
| 'arbiter_verdict'
| 'arbiter_requested_at'
| 'completion_reason'
| 'updated_at'
>
>;
function hydratePairedTaskRow(
database: Database,
row: StoredPairedTaskRow,
): PairedTask {
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;
return {
...row,
owner_service_id:
resolveRoleServiceShadow('owner', ownerAgentType) ??
CODEX_MAIN_SERVICE_ID,
reviewer_service_id:
resolveRoleServiceShadow('reviewer', reviewerAgentType) ??
CODEX_REVIEW_SERVICE_ID,
owner_agent_type: ownerAgentType ?? null,
reviewer_agent_type: reviewerAgentType ?? null,
arbiter_agent_type: arbiterAgentType,
};
}
export function upsertPairedProjectInDatabase(
database: Database,
project: PairedProject,
): void {
database
.prepare(
`
INSERT INTO paired_projects (
chat_jid,
group_folder,
canonical_work_dir,
created_at,
updated_at
)
VALUES (?, ?, ?, ?, ?)
ON CONFLICT(chat_jid) DO UPDATE SET
group_folder = excluded.group_folder,
canonical_work_dir = excluded.canonical_work_dir,
updated_at = excluded.updated_at
`,
)
.run(
project.chat_jid,
project.group_folder,
project.canonical_work_dir,
project.created_at,
project.updated_at,
);
}
export function getPairedProjectFromDatabase(
database: Database,
chatJid: string,
): PairedProject | undefined {
return database
.prepare('SELECT * FROM paired_projects WHERE chat_jid = ?')
.get(chatJid) as PairedProject | undefined;
}
export function createPairedTaskInDatabase(
database: Database,
task: PairedTask,
): void {
database
.prepare(
`
INSERT INTO paired_tasks (
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,
created_at,
updated_at
)
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,
task.title,
task.source_ref,
task.plan_notes,
task.review_requested_at,
task.round_trip_count,
task.status,
task.arbiter_verdict,
task.arbiter_requested_at,
task.created_at,
task.updated_at,
);
}
export function getPairedTaskByIdFromDatabase(
database: Database,
id: string,
): PairedTask | undefined {
const row = database
.prepare('SELECT * FROM paired_tasks WHERE id = ?')
.get(id) as StoredPairedTaskRow | undefined;
return row ? hydratePairedTaskRow(database, row) : undefined;
}
export function getLatestPairedTaskForChatFromDatabase(
database: Database,
chatJid: string,
): PairedTask | undefined {
const row = database
.prepare(
`
SELECT *
FROM paired_tasks
WHERE chat_jid = ?
ORDER BY updated_at DESC
LIMIT 1
`,
)
.get(chatJid) as StoredPairedTaskRow | undefined;
return row ? hydratePairedTaskRow(database, row) : undefined;
}
export function getLatestOpenPairedTaskForChatFromDatabase(
database: Database,
chatJid: string,
): PairedTask | undefined {
const row = database
.prepare(
`
SELECT *
FROM paired_tasks
WHERE chat_jid = ?
AND status NOT IN ('completed')
ORDER BY updated_at DESC
LIMIT 1
`,
)
.get(chatJid) as StoredPairedTaskRow | undefined;
return row ? hydratePairedTaskRow(database, row) : undefined;
}
export function updatePairedTaskInDatabase(
database: Database,
id: string,
updates: PairedTaskUpdates,
): void {
const fields: string[] = [];
const values: (string | number | null)[] = [];
if (updates.title !== undefined) {
fields.push('title = ?');
values.push(updates.title);
}
if (updates.source_ref !== undefined) {
fields.push('source_ref = ?');
values.push(updates.source_ref);
}
if (updates.plan_notes !== undefined) {
fields.push('plan_notes = ?');
values.push(updates.plan_notes);
}
if (updates.review_requested_at !== undefined) {
fields.push('review_requested_at = ?');
values.push(updates.review_requested_at);
}
if (updates.round_trip_count !== undefined) {
fields.push('round_trip_count = ?');
values.push(updates.round_trip_count);
}
if (updates.status !== undefined) {
fields.push('status = ?');
values.push(updates.status);
}
if (updates.arbiter_verdict !== undefined) {
fields.push('arbiter_verdict = ?');
values.push(updates.arbiter_verdict);
}
if (updates.arbiter_requested_at !== undefined) {
fields.push('arbiter_requested_at = ?');
values.push(updates.arbiter_requested_at);
}
if (updates.completion_reason !== undefined) {
fields.push('completion_reason = ?');
values.push(updates.completion_reason);
}
if (updates.updated_at !== undefined) {
fields.push('updated_at = ?');
values.push(updates.updated_at);
}
if (fields.length === 0) return;
values.push(id);
database
.prepare(`UPDATE paired_tasks SET ${fields.join(', ')} WHERE id = ?`)
.run(...values);
}
export function upsertPairedWorkspaceInDatabase(
database: Database,
workspace: PairedWorkspace,
): void {
database
.prepare(
`
INSERT INTO paired_workspaces (
id,
task_id,
role,
workspace_dir,
snapshot_source_dir,
snapshot_ref,
status,
snapshot_refreshed_at,
created_at,
updated_at
)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(id) DO UPDATE SET
workspace_dir = excluded.workspace_dir,
snapshot_source_dir = excluded.snapshot_source_dir,
snapshot_ref = excluded.snapshot_ref,
status = excluded.status,
snapshot_refreshed_at = excluded.snapshot_refreshed_at,
updated_at = excluded.updated_at
`,
)
.run(
workspace.id,
workspace.task_id,
workspace.role,
workspace.workspace_dir,
workspace.snapshot_source_dir,
workspace.snapshot_ref,
workspace.status,
workspace.snapshot_refreshed_at,
workspace.created_at,
workspace.updated_at,
);
}
export function getPairedWorkspaceFromDatabase(
database: Database,
taskId: string,
role: PairedWorkspace['role'],
): PairedWorkspace | undefined {
return database
.prepare('SELECT * FROM paired_workspaces WHERE task_id = ? AND role = ?')
.get(taskId, role) as PairedWorkspace | undefined;
}
export function listPairedWorkspacesForTaskFromDatabase(
database: Database,
taskId: string,
): PairedWorkspace[] {
return database
.prepare(
'SELECT * FROM paired_workspaces WHERE task_id = ? ORDER BY created_at',
)
.all(taskId) as PairedWorkspace[];
}
export function getLastBotFinalMessageFromDatabase(
database: Database,
chatJid: string,
_agentType: AgentType = 'claude-code',
limit: number = 1,
): Array<{ content: string; timestamp: string }> {
return database
.prepare(
`SELECT content, timestamp
FROM messages
WHERE chat_jid = ? AND is_bot_message = 1
ORDER BY timestamp DESC, seq DESC
LIMIT ?`,
)
.all(chatJid, limit) as Array<{ content: string; timestamp: string }>;
}

68
src/db/paired-turn-outputs.ts Normal file
View File

@@ -0,0 +1,68 @@
import { Database } from 'bun:sqlite';
import { logger } from '../logger.js';
import { PairedRoomRole, PairedTurnOutput } from '../types.js';
const MAX_TURN_OUTPUT_CHARS = 50_000;
export function insertPairedTurnOutputInDatabase(
database: Database,
taskId: string,
turnNumber: number,
role: PairedRoomRole,
outputText: string,
): void {
if (outputText.length > MAX_TURN_OUTPUT_CHARS) {
logger.warn(
{
taskId,
turnNumber,
role,
originalLen: outputText.length,
maxLen: MAX_TURN_OUTPUT_CHARS,
},
'Paired turn output truncated — agent output exceeds storage limit',
);
}
database
.prepare(
`INSERT OR REPLACE INTO paired_turn_outputs
(task_id, turn_number, role, output_text, created_at)
VALUES (?, ?, ?, ?, ?)`,
)
.run(
taskId,
turnNumber,
role,
outputText.slice(0, MAX_TURN_OUTPUT_CHARS),
new Date().toISOString(),
);
}
export function getPairedTurnOutputsFromDatabase(
database: Database,
taskId: string,
): PairedTurnOutput[] {
return database
.prepare(
`SELECT * FROM paired_turn_outputs
WHERE task_id = ?
ORDER BY turn_number ASC`,
)
.all(taskId) as PairedTurnOutput[];
}
export function getLatestTurnNumberFromDatabase(
database: Database,
taskId: string,
): number {
const row = database
.prepare(
`SELECT MAX(turn_number) as max_turn
FROM paired_turn_outputs
WHERE task_id = ?`,
)
.get(taskId) as { max_turn: number | null } | undefined;
return row?.max_turn ?? 0;
}