refactor: complete config unification and split db helpers

This commit is contained in:
ejclaw
2026-04-08 19:05:19 +09:00
parent d996d5ff10
commit 8b9bebd29e
27 changed files with 4752 additions and 3652 deletions

View File

@@ -9,8 +9,11 @@ const { mockReadEnvFile, mockGetActiveCodexAuthPath } = vi.hoisted(() => ({
})); }));
vi.mock('./config.js', () => ({ vi.mock('./config.js', () => ({
ASSISTANT_NAME: 'Andy',
CODEX_REVIEW_SERVICE_ID: 'codex-review', CODEX_REVIEW_SERVICE_ID: 'codex-review',
GROUPS_DIR: '/tmp/ejclaw-test-groups', GROUPS_DIR: '/tmp/ejclaw-test-groups',
IS_TEST_ENV: true,
LOG_LEVEL: 'info',
SERVICE_ID: 'codex-main', SERVICE_ID: 'codex-main',
SERVICE_SESSION_SCOPE: 'codex-main', SERVICE_SESSION_SCOPE: 'codex-main',
TIMEZONE: 'Asia/Seoul', TIMEZONE: 'Asia/Seoul',

View File

@@ -13,6 +13,7 @@ vi.mock('./config.js', () => ({
DATA_DIR: '/tmp/ejclaw-test-data', DATA_DIR: '/tmp/ejclaw-test-data',
GROUPS_DIR: '/tmp/ejclaw-test-groups', GROUPS_DIR: '/tmp/ejclaw-test-groups',
IDLE_TIMEOUT: 1800000, // 30min IDLE_TIMEOUT: 1800000, // 30min
LOG_LEVEL: 'info',
SERVICE_ID: 'claude', SERVICE_ID: 'claude',
SERVICE_SESSION_SCOPE: 'claude', SERVICE_SESSION_SCOPE: 'claude',
TIMEZONE: 'America/Los_Angeles', TIMEZONE: 'America/Los_Angeles',

View File

@@ -11,6 +11,7 @@ import {
AGENT_MAX_OUTPUT_SIZE, AGENT_MAX_OUTPUT_SIZE,
AGENT_TIMEOUT, AGENT_TIMEOUT,
IDLE_TIMEOUT, IDLE_TIMEOUT,
LOG_LEVEL,
} from './config.js'; } from './config.js';
import { import {
prepareReadonlySessionEnvironment, prepareReadonlySessionEnvironment,
@@ -411,8 +412,7 @@ export async function runAgentProcess(
logsDir, logsDir,
`agent-${input.runId || 'adhoc'}-${timestamp}.log`, `agent-${input.runId || 'adhoc'}-${timestamp}.log`,
); );
const isVerbose = const isVerbose = LOG_LEVEL === 'debug' || LOG_LEVEL === 'trace';
process.env.LOG_LEVEL === 'debug' || process.env.LOG_LEVEL === 'trace';
const logLines = [ const logLines = [
`=== Agent Run Log ===`, `=== Agent Run Log ===`,

95
src/config.test.ts Normal file
View File

@@ -0,0 +1,95 @@
import fs from 'fs';
import path from 'path';
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
describe('config/env loading', () => {
let tempRoot: string;
let previousCwd: string;
let previousEnv: NodeJS.ProcessEnv;
beforeEach(() => {
tempRoot = fs.mkdtempSync(path.join('/tmp', 'ejclaw-config-'));
previousCwd = process.cwd();
previousEnv = { ...process.env };
process.chdir(tempRoot);
delete process.env.EJCLAW_STORE_DIR;
delete process.env.EJCLAW_GROUPS_DIR;
delete process.env.EJCLAW_DATA_DIR;
delete process.env.EJCLAW_CACHE_DIR;
delete process.env.AGENT_TIMEOUT;
delete process.env.LOG_LEVEL;
delete process.env.CLAUDE_MODEL;
delete process.env.CODEX_MODEL;
delete process.env.STATUS_CHANNEL_ID;
delete process.env.TZ;
vi.resetModules();
});
afterEach(() => {
process.chdir(previousCwd);
for (const key of Object.keys(process.env)) {
if (!(key in previousEnv)) delete process.env[key];
}
for (const [key, value] of Object.entries(previousEnv)) {
if (value === undefined) {
delete process.env[key];
} else {
process.env[key] = value;
}
}
fs.rmSync(tempRoot, { recursive: true, force: true });
});
it('preserves empty-string env values and still prefers process.env over .env', async () => {
fs.writeFileSync(
path.join(tempRoot, '.env'),
['FROM_FILE=file-value', 'EMPTY_IN_FILE=', 'PROCESS_EMPTY=file-fallback'].join(
'\n',
),
);
process.env.FROM_FILE = 'process-value';
process.env.PROCESS_EMPTY = '';
const env = await import('./env.js');
expect(env.getEnv('FROM_FILE')).toBe('process-value');
expect(env.getEnv('EMPTY_IN_FILE')).toBe('');
expect(env.getEnv('PROCESS_EMPTY')).toBe('');
expect(env.readEnvFile(['EMPTY_IN_FILE', 'PROCESS_EMPTY'])).toEqual({
EMPTY_IN_FILE: '',
PROCESS_EMPTY: 'file-fallback',
});
});
it('loads formerly process-only config values through the shared env path', async () => {
fs.writeFileSync(
path.join(tempRoot, '.env'),
[
'EJCLAW_STORE_DIR=./custom-store',
'EJCLAW_GROUPS_DIR=./custom-groups',
'EJCLAW_DATA_DIR=./custom-data',
'EJCLAW_CACHE_DIR=./custom-cache',
'AGENT_TIMEOUT=12345',
'LOG_LEVEL=trace',
'CLAUDE_MODEL=claude-opus-test',
'CODEX_MODEL=gpt-5.4-test',
'STATUS_CHANNEL_ID=status-room',
'TZ=UTC',
].join('\n'),
);
const config = await import('./config.js');
expect(config.STORE_DIR).toBe(path.resolve(tempRoot, 'custom-store'));
expect(config.GROUPS_DIR).toBe(path.resolve(tempRoot, 'custom-groups'));
expect(config.DATA_DIR).toBe(path.resolve(tempRoot, 'custom-data'));
expect(config.CACHE_DIR).toBe(path.resolve(tempRoot, 'custom-cache'));
expect(config.AGENT_TIMEOUT).toBe(12345);
expect(config.LOG_LEVEL).toBe('trace');
expect(config.DEFAULT_CLAUDE_MODEL).toBe('claude-opus-test');
expect(config.DEFAULT_CODEX_MODEL).toBe('gpt-5.4-test');
expect(config.STATUS_CHANNEL_ID).toBe('status-room');
expect(config.TIMEZONE).toBe('UTC');
});
});

View File

@@ -1,19 +1,18 @@
import os from 'os'; import { loadConfig } from './config/load-config.js';
import path from 'path'; import type { MoaConfig } from './moa.js';
import type { RoleModelConfig } from './config/schema.js';
import type { AgentType } from './types.js'; export type { AppConfig, RoleModelConfig } from './config/schema.js';
import { getEnv } from './env.js'; export { loadConfig } from './config/load-config.js';
export const ASSISTANT_NAME = getEnv('ASSISTANT_NAME') || 'Andy'; const CONFIG = loadConfig();
export const ASSISTANT_HAS_OWN_NUMBER =
getEnv('ASSISTANT_HAS_OWN_NUMBER') === 'true'; export const ASSISTANT_NAME = CONFIG.assistant.name;
const ASSISTANT_SLUG = ASSISTANT_NAME.trim().toLowerCase(); export const ASSISTANT_HAS_OWN_NUMBER = CONFIG.assistant.hasOwnNumber;
export const SERVICE_ID = getEnv('SERVICE_ID') || ASSISTANT_SLUG; export const SERVICE_ID = CONFIG.service.id;
export const CLAUDE_SERVICE_ID = getEnv('CLAUDE_SERVICE_ID') || 'claude'; export const CLAUDE_SERVICE_ID = CONFIG.service.claudeId;
export const CODEX_MAIN_SERVICE_ID = export const CODEX_MAIN_SERVICE_ID = CONFIG.service.codexMainId;
getEnv('CODEX_MAIN_SERVICE_ID') || 'codex-main'; export const CODEX_REVIEW_SERVICE_ID = CONFIG.service.codexReviewId;
export const CODEX_REVIEW_SERVICE_ID =
getEnv('CODEX_REVIEW_SERVICE_ID') || 'codex-review';
export function normalizeServiceId(serviceId: string): string { export function normalizeServiceId(serviceId: string): string {
if (serviceId === 'codex') { if (serviceId === 'codex') {
@@ -35,94 +34,52 @@ export function isCodexMainService(serviceId: string = SERVICE_ID): boolean {
export function isReviewService(serviceId: string = SERVICE_ID): boolean { export function isReviewService(serviceId: string = SERVICE_ID): boolean {
return normalizeServiceId(serviceId) === CODEX_REVIEW_SERVICE_ID; return normalizeServiceId(serviceId) === CODEX_REVIEW_SERVICE_ID;
} }
export const POLL_INTERVAL = 2000; export const POLL_INTERVAL = CONFIG.runtime.pollInterval;
export const SCHEDULER_POLL_INTERVAL = 60000; export const SCHEDULER_POLL_INTERVAL = CONFIG.runtime.schedulerPollInterval;
/** Minimum time (ms) a failover lease stays active before Claude can reclaim it. */ /** Minimum time (ms) a failover lease stays active before Claude can reclaim it. */
export const FAILOVER_MIN_DURATION_MS = 3 * 60 * 60 * 1000; // 3 hours export const FAILOVER_MIN_DURATION_MS = CONFIG.runtime.failoverMinDurationMs;
const PROJECT_ROOT = process.cwd(); export const SENDER_ALLOWLIST_PATH = CONFIG.paths.senderAllowlistPath;
const HOME_DIR = process.env.HOME || os.homedir(); export const STORE_DIR = CONFIG.paths.storeDir;
export const GROUPS_DIR = CONFIG.paths.groupsDir;
export const DATA_DIR = CONFIG.paths.dataDir;
export const CACHE_DIR = CONFIG.paths.cacheDir;
export const SENDER_ALLOWLIST_PATH = path.join( export const AGENT_TIMEOUT = CONFIG.runtime.agentTimeout;
HOME_DIR, export const AGENT_MAX_OUTPUT_SIZE = CONFIG.runtime.agentMaxOutputSize;
'.config', export const IPC_POLL_INTERVAL = CONFIG.runtime.ipcPollInterval;
'ejclaw', export const IDLE_TIMEOUT = CONFIG.runtime.idleTimeout;
'sender-allowlist.json', export const MAX_CONCURRENT_AGENTS = CONFIG.runtime.maxConcurrentAgents;
); export const RECOVERY_CONCURRENT_AGENTS =
export const STORE_DIR = path.resolve( CONFIG.runtime.recoveryConcurrentAgents;
process.env.EJCLAW_STORE_DIR || path.join(PROJECT_ROOT, 'store'), export const LOG_LEVEL = CONFIG.logging.level;
); export const IS_TEST_ENV = CONFIG.logging.isTestEnv;
export const GROUPS_DIR = path.resolve(
process.env.EJCLAW_GROUPS_DIR || path.join(PROJECT_ROOT, 'groups'),
);
export const DATA_DIR = path.resolve(
process.env.EJCLAW_DATA_DIR || path.join(PROJECT_ROOT, 'data'),
);
// Shared cache directory (same across both services for dedup)
export const CACHE_DIR = path.join(PROJECT_ROOT, 'cache');
export const AGENT_TIMEOUT = parseInt(
process.env.AGENT_TIMEOUT || '1800000',
10,
);
export const AGENT_MAX_OUTPUT_SIZE = parseInt(
process.env.AGENT_MAX_OUTPUT_SIZE || '10485760',
10,
); // 10MB default
export const IPC_POLL_INTERVAL = 1000;
export const IDLE_TIMEOUT = parseInt(process.env.IDLE_TIMEOUT || '1800000', 10); // 30min default — how long to keep agent alive after last result
export const MAX_CONCURRENT_AGENTS = Math.max(
1,
parseInt(process.env.MAX_CONCURRENT_AGENTS || '5', 10) || 5,
);
export const RECOVERY_CONCURRENT_AGENTS = parseInt(
getEnv('RECOVERY_CONCURRENT_AGENTS') || '3',
10,
);
// ── Paired review ───────────────────────────────────────────────── // ── Paired review ─────────────────────────────────────────────────
/** Owner agent type. Default: codex. Set OWNER_AGENT_TYPE=claude-code to use Claude as owner. */ /** Owner agent type. Default: codex. Set OWNER_AGENT_TYPE=claude-code to use Claude as owner. */
const rawOwnerAgentType = getEnv('OWNER_AGENT_TYPE'); export const OWNER_AGENT_TYPE = CONFIG.paired.ownerAgentType;
export const OWNER_AGENT_TYPE: AgentType =
rawOwnerAgentType === 'codex' || rawOwnerAgentType === 'claude-code'
? rawOwnerAgentType
: 'codex';
/** Reviewer agent type. Default: claude-code. Set REVIEWER_AGENT_TYPE=codex to use Codex as reviewer. */ /** Reviewer agent type. Default: claude-code. Set REVIEWER_AGENT_TYPE=codex to use Codex as reviewer. */
const rawReviewerAgentType = getEnv('REVIEWER_AGENT_TYPE'); export const REVIEWER_AGENT_TYPE = CONFIG.paired.reviewerAgentType;
export const REVIEWER_AGENT_TYPE: AgentType =
rawReviewerAgentType === 'codex' || rawReviewerAgentType === 'claude-code'
? rawReviewerAgentType
: 'claude-code';
/** Service ID for the reviewer based on agent type. */ /** Service ID for the reviewer based on agent type. */
export const REVIEWER_SERVICE_ID_FOR_TYPE = export const REVIEWER_SERVICE_ID_FOR_TYPE =
REVIEWER_AGENT_TYPE === 'claude-code' CONFIG.paired.reviewerServiceIdForType;
? CLAUDE_SERVICE_ID
: CODEX_REVIEW_SERVICE_ID;
/** Arbiter agent type. Disabled by default. Set ARBITER_AGENT_TYPE=codex or claude-code to enable. */ /** Arbiter agent type. Disabled by default. Set ARBITER_AGENT_TYPE=codex or claude-code to enable. */
const rawArbiterAgentType = getEnv('ARBITER_AGENT_TYPE'); export const ARBITER_AGENT_TYPE = CONFIG.paired.arbiterAgentType;
export const ARBITER_AGENT_TYPE: AgentType | undefined =
rawArbiterAgentType === 'codex' || rawArbiterAgentType === 'claude-code'
? rawArbiterAgentType
: undefined;
/** Service ID for the arbiter. Defaults to codex-review for internal routing when arbiter is enabled. */ /** Service ID for the arbiter. Defaults to codex-review for internal routing when arbiter is enabled. */
export const ARBITER_SERVICE_ID = ARBITER_AGENT_TYPE export const ARBITER_SERVICE_ID = CONFIG.paired.arbiterServiceId;
? getEnv('ARBITER_SERVICE_ID') || CODEX_REVIEW_SERVICE_ID
: null;
/** Language for agent responses. When set, a language instruction is appended to all paired room prompts. */ /** Language for agent responses. When set, a language instruction is appended to all paired room prompts. */
export const AGENT_LANGUAGE = getEnv('AGENT_LANGUAGE') || ''; export const AGENT_LANGUAGE = CONFIG.paired.agentLanguage;
/** Number of consecutive owner↔reviewer round trips before arbiter is auto-requested. */ /** Number of consecutive owner↔reviewer round trips before arbiter is auto-requested. */
export const ARBITER_DEADLOCK_THRESHOLD = parseInt( export const ARBITER_DEADLOCK_THRESHOLD =
getEnv('ARBITER_DEADLOCK_THRESHOLD') || '2', CONFIG.paired.arbiterDeadlockThreshold;
10,
);
export function isArbiterEnabled(): boolean { export function isArbiterEnabled(): boolean {
return ARBITER_AGENT_TYPE !== undefined; return ARBITER_AGENT_TYPE !== undefined;
@@ -130,26 +87,11 @@ export function isArbiterEnabled(): boolean {
// ── Per-role model configuration ───────────────────────────────── // ── Per-role model configuration ─────────────────────────────────
export interface RoleModelConfig { export const OWNER_MODEL_CONFIG = CONFIG.models.owner;
/** Model name override (e.g. 'claude-opus-4-6', 'gpt-5.4'). */ export const REVIEWER_MODEL_CONFIG = CONFIG.models.reviewer;
model?: string; export const ARBITER_MODEL_CONFIG = CONFIG.models.arbiter;
/** Effort level override. */ export const DEFAULT_CLAUDE_MODEL = CONFIG.providers.claudeDefaultModel;
effort?: string; export const DEFAULT_CODEX_MODEL = CONFIG.providers.codexDefaultModel;
/** Whether to fall back to codex when primary provider fails. Default: true. */
fallbackEnabled: boolean;
}
function buildRoleModelConfig(envPrefix: string): RoleModelConfig {
return {
model: getEnv(`${envPrefix}_MODEL`) || undefined,
effort: getEnv(`${envPrefix}_EFFORT`) || undefined,
fallbackEnabled: getEnv(`${envPrefix}_FALLBACK_ENABLED`) !== 'false',
};
}
export const OWNER_MODEL_CONFIG = buildRoleModelConfig('OWNER');
export const REVIEWER_MODEL_CONFIG = buildRoleModelConfig('REVIEWER');
export const ARBITER_MODEL_CONFIG = buildRoleModelConfig('ARBITER');
export function getRoleModelConfig( export function getRoleModelConfig(
role: 'owner' | 'reviewer' | 'arbiter', role: 'owner' | 'reviewer' | 'arbiter',
@@ -166,107 +108,38 @@ export function getRoleModelConfig(
// ── Mixture of Agents (MoA) ────────────────────────────────────── // ── Mixture of Agents (MoA) ──────────────────────────────────────
import type { MoaConfig, MoaModelConfig } from './moa.js';
/**
* Parse MOA reference models from env.
* Format: MOA_REF_MODELS=kimi,glm (comma-separated names)
* Each model: MOA_{NAME}_MODEL, MOA_{NAME}_BASE_URL, MOA_{NAME}_API_KEY
*/
function parseMoaReferenceModels(): MoaModelConfig[] {
const names = (getEnv('MOA_REF_MODELS') || '')
.split(',')
.map((s) => s.trim())
.filter(Boolean);
return names
.map((name) => {
const prefix = `MOA_${name.toUpperCase()}`;
const model = getEnv(`${prefix}_MODEL`) || '';
const baseUrl = getEnv(`${prefix}_BASE_URL`) || '';
const apiKey = getEnv(`${prefix}_API_KEY`) || '';
if (!model || !baseUrl || !apiKey) return null;
const rawFormat = getEnv(`${prefix}_API_FORMAT`) || '';
const apiFormat: 'openai' | 'anthropic' =
rawFormat === 'anthropic' ? 'anthropic' : 'openai';
return { name, model, baseUrl, apiKey, apiFormat };
})
.filter((m): m is MoaModelConfig => m !== null);
}
export function getMoaConfig(): MoaConfig { export function getMoaConfig(): MoaConfig {
const referenceModels = parseMoaReferenceModels(); return CONFIG.moa;
return {
enabled: getEnv('MOA_ENABLED') === 'true' && referenceModels.length > 0,
referenceModels,
};
} }
// Max owner↔reviewer round trips per task. 0 = unlimited. export const PAIRED_MAX_ROUND_TRIPS = CONFIG.paired.maxRoundTrips;
const rawMaxRoundTrips = getEnv('PAIRED_MAX_ROUND_TRIPS') || '1000';
export const PAIRED_MAX_ROUND_TRIPS =
rawMaxRoundTrips === '0' ? Infinity : parseInt(rawMaxRoundTrips, 10) || 1000;
export const RECOVERY_STAGGER_MS = parseInt( export const RECOVERY_STAGGER_MS = CONFIG.runtime.recoveryStaggerMs;
getEnv('RECOVERY_STAGGER_MS') || '2000', export const RECOVERY_DURATION_MS = CONFIG.runtime.recoveryDurationMs;
10,
);
export const RECOVERY_DURATION_MS = parseInt(
getEnv('RECOVERY_DURATION_MS') || '60000',
10,
);
function escapeRegex(str: string): string { export const TRIGGER_PATTERN = CONFIG.assistant.triggerPattern;
return str.replace(/[.*+?^${}()|[\]\\]/g, '\\$&');
}
export const TRIGGER_PATTERN = new RegExp(
`^@${escapeRegex(ASSISTANT_NAME)}\\b`,
'i',
);
// Status dashboard: Discord channel ID for live agent status updates // Status dashboard: Discord channel ID for live agent status updates
export const STATUS_CHANNEL_ID = process.env.STATUS_CHANNEL_ID || ''; export const STATUS_CHANNEL_ID = CONFIG.status.channelId;
export const STATUS_UPDATE_INTERVAL = 10000; // 10s export const STATUS_UPDATE_INTERVAL = CONFIG.status.updateInterval;
export const USAGE_UPDATE_INTERVAL = 300000; // 5 minutes export const USAGE_UPDATE_INTERVAL = CONFIG.status.usageUpdateInterval;
export const STATUS_SHOW_ROOMS = export const STATUS_SHOW_ROOMS = CONFIG.status.showRooms;
(getEnv('STATUS_SHOW_ROOMS') || 'true') !== 'false'; export const STATUS_SHOW_ROOM_DETAILS = CONFIG.status.showRoomDetails;
export const STATUS_SHOW_ROOM_DETAILS = export const USAGE_DASHBOARD_ENABLED = CONFIG.status.usageDashboardEnabled;
(getEnv('STATUS_SHOW_ROOM_DETAILS') || 'true') !== 'false';
export const USAGE_DASHBOARD_ENABLED = getEnv('USAGE_DASHBOARD') === 'true';
// Timezone for scheduled tasks (cron expressions, etc.) // Timezone for scheduled tasks (cron expressions, etc.)
// Uses system timezone by default // Uses system timezone by default
export const TIMEZONE = export const TIMEZONE = CONFIG.status.timezone;
process.env.TZ || Intl.DateTimeFormat().resolvedOptions().timeZone;
const rawSessionCommandAllowedSenders = const SESSION_COMMAND_ALLOWED_SENDERS = CONFIG.sessionCommands.allowedSenders;
getEnv('SESSION_COMMAND_ALLOWED_SENDERS') ||
getEnv('SESSION_COMMAND_USER_IDS') ||
'';
const SESSION_COMMAND_ALLOWED_SENDERS = new Set(
rawSessionCommandAllowedSenders
.split(',')
.map((value) => value.trim())
.filter(Boolean),
);
export function isSessionCommandSenderAllowed(sender: string): boolean { export function isSessionCommandSenderAllowed(sender: string): boolean {
return SESSION_COMMAND_ALLOWED_SENDERS.has(sender); return SESSION_COMMAND_ALLOWED_SENDERS.has(sender);
} }
// Delta handoff: cross-provider session continuity with probe + fallback // Delta handoff: cross-provider session continuity with probe + fallback
export const DELTA_HANDOFF_ENABLED = getEnv('DELTA_HANDOFF_ENABLED') === 'true'; export const DELTA_HANDOFF_ENABLED = CONFIG.deltaHandoff.enabled;
export const DELTA_HANDOFF_CANARY_GROUPS = CONFIG.deltaHandoff.canaryGroups;
// Comma-separated list of group folders for canary testing
const rawDeltaHandoffCanaryGroups = getEnv('DELTA_HANDOFF_CANARY_GROUPS') || '';
export const DELTA_HANDOFF_CANARY_GROUPS = new Set(
rawDeltaHandoffCanaryGroups
.split(',')
.map((v) => v.trim())
.filter(Boolean),
);
export function isDeltaHandoffEnabledForGroup(groupFolder: string): boolean { export function isDeltaHandoffEnabledForGroup(groupFolder: string): boolean {
if (!DELTA_HANDOFF_ENABLED) return false; if (!DELTA_HANDOFF_ENABLED) return false;

2
src/config/index.ts Normal file
View File

@@ -0,0 +1,2 @@
export type { AppConfig, RoleModelConfig } from './schema.js';
export { loadConfig } from './load-config.js';

237
src/config/load-config.ts Normal file
View File

@@ -0,0 +1,237 @@
import os from 'os';
import path from 'path';
import type { MoaConfig, MoaModelConfig } from '../moa.js';
import type { AgentType } from '../types.js';
import { getEnv } from '../env.js';
import type { AppConfig, RoleModelConfig } from './schema.js';
function readText(key: string): string | undefined {
return getEnv(key);
}
function readNonEmptyText(key: string): string | undefined {
const value = getEnv(key);
return value === '' ? undefined : value;
}
function readBoolean(key: string, fallback = false): boolean {
const value = readText(key);
if (value == null || value === '') return fallback;
return value === 'true';
}
function readBooleanUnlessFalse(key: string, fallback = true): boolean {
const value = readText(key);
if (value == null || value === '') return fallback;
return value !== 'false';
}
function readInteger(key: string, fallback: number): number {
const value = readNonEmptyText(key);
if (value == null) return fallback;
const parsed = parseInt(value, 10);
return Number.isFinite(parsed) ? parsed : fallback;
}
function readIntegerAtLeast(
key: string,
fallback: number,
minimum: number,
): number {
return Math.max(minimum, readInteger(key, fallback) || fallback);
}
function readAgentType(
key: string,
fallback?: AgentType,
): AgentType | undefined {
const value = readText(key);
if (value === 'codex' || value === 'claude-code') return value;
return fallback;
}
function buildRoleModelConfig(envPrefix: string): RoleModelConfig {
return {
model: readNonEmptyText(`${envPrefix}_MODEL`),
effort: readNonEmptyText(`${envPrefix}_EFFORT`),
fallbackEnabled: readText(`${envPrefix}_FALLBACK_ENABLED`) !== 'false',
};
}
function parseMoaReferenceModels(): MoaModelConfig[] {
const names = (readText('MOA_REF_MODELS') ?? '')
.split(',')
.map((value) => value.trim())
.filter(Boolean);
return names
.map((name) => {
const prefix = `MOA_${name.toUpperCase()}`;
const model = readNonEmptyText(`${prefix}_MODEL`) ?? '';
const baseUrl = readNonEmptyText(`${prefix}_BASE_URL`) ?? '';
const apiKey = readNonEmptyText(`${prefix}_API_KEY`) ?? '';
if (!model || !baseUrl || !apiKey) return null;
const rawFormat = readText(`${prefix}_API_FORMAT`) ?? '';
const apiFormat: 'openai' | 'anthropic' =
rawFormat === 'anthropic' ? 'anthropic' : 'openai';
return { name, model, baseUrl, apiKey, apiFormat };
})
.filter((value): value is MoaModelConfig => value !== null);
}
function buildMoaConfig(): MoaConfig {
const referenceModels = parseMoaReferenceModels();
return {
enabled: readText('MOA_ENABLED') === 'true' && referenceModels.length > 0,
referenceModels,
};
}
function escapeRegex(value: string): string {
return value.replace(/[.*+?^${}()|[\]\\]/g, '\\$&');
}
function normalizeServiceId(
serviceId: string,
codexMainServiceId: string,
): string {
return serviceId === 'codex' ? codexMainServiceId : serviceId;
}
export function loadConfig(): AppConfig {
const assistantName = readText('ASSISTANT_NAME') ?? 'Andy';
const assistantSlug = assistantName.trim().toLowerCase();
const claudeServiceId = readText('CLAUDE_SERVICE_ID') ?? 'claude';
const codexMainServiceId = readText('CODEX_MAIN_SERVICE_ID') ?? 'codex-main';
const codexReviewServiceId =
readText('CODEX_REVIEW_SERVICE_ID') ?? 'codex-review';
const serviceId = readText('SERVICE_ID') ?? assistantSlug;
const serviceSessionScope = normalizeServiceId(serviceId, codexMainServiceId);
const ownerAgentType = readAgentType('OWNER_AGENT_TYPE', 'codex') ?? 'codex';
const reviewerAgentType =
readAgentType('REVIEWER_AGENT_TYPE', 'claude-code') ?? 'claude-code';
const arbiterAgentType = readAgentType('ARBITER_AGENT_TYPE');
const projectRoot = process.cwd();
const homeDir = readNonEmptyText('HOME') ?? os.homedir();
const rawMaxRoundTrips = readText('PAIRED_MAX_ROUND_TRIPS');
const maxRoundTrips =
rawMaxRoundTrips === '0' ? Infinity : readInteger('PAIRED_MAX_ROUND_TRIPS', 1000);
return {
assistant: {
name: assistantName,
hasOwnNumber: readBoolean('ASSISTANT_HAS_OWN_NUMBER', false),
slug: assistantSlug,
triggerPattern: new RegExp(`^@${escapeRegex(assistantName)}\\b`, 'i'),
},
service: {
id: serviceId,
sessionScope: serviceSessionScope,
claudeId: claudeServiceId,
codexMainId: codexMainServiceId,
codexReviewId: codexReviewServiceId,
},
paths: {
projectRoot,
homeDir,
senderAllowlistPath: path.join(
homeDir,
'.config',
'ejclaw',
'sender-allowlist.json',
),
storeDir: path.resolve(
readNonEmptyText('EJCLAW_STORE_DIR') ?? path.join(projectRoot, 'store'),
),
groupsDir: path.resolve(
readNonEmptyText('EJCLAW_GROUPS_DIR') ??
path.join(projectRoot, 'groups'),
),
dataDir: path.resolve(
readNonEmptyText('EJCLAW_DATA_DIR') ?? path.join(projectRoot, 'data'),
),
cacheDir: path.resolve(
readNonEmptyText('EJCLAW_CACHE_DIR') ??
path.join(projectRoot, 'cache'),
),
},
runtime: {
pollInterval: 2000,
schedulerPollInterval: 60000,
failoverMinDurationMs: 3 * 60 * 60 * 1000,
agentTimeout: readInteger('AGENT_TIMEOUT', 1_800_000),
agentMaxOutputSize: readInteger('AGENT_MAX_OUTPUT_SIZE', 10_485_760),
ipcPollInterval: 1000,
idleTimeout: readInteger('IDLE_TIMEOUT', 1_800_000),
maxConcurrentAgents: readIntegerAtLeast('MAX_CONCURRENT_AGENTS', 5, 1),
recoveryConcurrentAgents: readInteger('RECOVERY_CONCURRENT_AGENTS', 3),
recoveryStaggerMs: readInteger('RECOVERY_STAGGER_MS', 2000),
recoveryDurationMs: readInteger('RECOVERY_DURATION_MS', 60000),
},
logging: {
level: readText('LOG_LEVEL') ?? 'info',
isTestEnv:
readText('VITEST') === 'true' || readText('NODE_ENV') === 'test',
},
paired: {
ownerAgentType,
reviewerAgentType,
reviewerServiceIdForType:
reviewerAgentType === 'claude-code'
? claudeServiceId
: codexReviewServiceId,
arbiterAgentType,
arbiterServiceId: arbiterAgentType
? (readText('ARBITER_SERVICE_ID') ?? codexReviewServiceId)
: null,
agentLanguage: readText('AGENT_LANGUAGE') ?? '',
arbiterDeadlockThreshold: readInteger('ARBITER_DEADLOCK_THRESHOLD', 2),
maxRoundTrips,
},
models: {
owner: buildRoleModelConfig('OWNER'),
reviewer: buildRoleModelConfig('REVIEWER'),
arbiter: buildRoleModelConfig('ARBITER'),
},
providers: {
claudeDefaultModel: readText('CLAUDE_MODEL') ?? 'claude',
codexDefaultModel: readText('CODEX_MODEL') ?? 'codex',
},
moa: buildMoaConfig(),
status: {
channelId: readText('STATUS_CHANNEL_ID') ?? '',
updateInterval: 10000,
usageUpdateInterval: 300000,
showRooms: readBooleanUnlessFalse('STATUS_SHOW_ROOMS', true),
showRoomDetails: readBooleanUnlessFalse('STATUS_SHOW_ROOM_DETAILS', true),
usageDashboardEnabled: readText('USAGE_DASHBOARD') === 'true',
timezone:
readText('TZ') ?? Intl.DateTimeFormat().resolvedOptions().timeZone,
},
sessionCommands: {
allowedSenders: new Set(
((readText('SESSION_COMMAND_ALLOWED_SENDERS') ??
readText('SESSION_COMMAND_USER_IDS')) ??
'')
.split(',')
.map((value) => value.trim())
.filter(Boolean),
),
},
deltaHandoff: {
enabled: readText('DELTA_HANDOFF_ENABLED') === 'true',
canaryGroups: new Set(
(readText('DELTA_HANDOFF_CANARY_GROUPS') ?? '')
.split(',')
.map((value) => value.trim())
.filter(Boolean),
),
},
};
}

86
src/config/schema.ts Normal file
View File

@@ -0,0 +1,86 @@
import type { MoaConfig } from '../moa.js';
import type { AgentType } from '../types.js';
export interface RoleModelConfig {
model?: string;
effort?: string;
fallbackEnabled: boolean;
}
export interface AppConfig {
assistant: {
name: string;
hasOwnNumber: boolean;
slug: string;
triggerPattern: RegExp;
};
service: {
id: string;
sessionScope: string;
claudeId: string;
codexMainId: string;
codexReviewId: string;
};
paths: {
projectRoot: string;
homeDir: string;
senderAllowlistPath: string;
storeDir: string;
groupsDir: string;
dataDir: string;
cacheDir: string;
};
runtime: {
pollInterval: number;
schedulerPollInterval: number;
failoverMinDurationMs: number;
agentTimeout: number;
agentMaxOutputSize: number;
ipcPollInterval: number;
idleTimeout: number;
maxConcurrentAgents: number;
recoveryConcurrentAgents: number;
recoveryStaggerMs: number;
recoveryDurationMs: number;
};
logging: {
level: string;
isTestEnv: boolean;
};
paired: {
ownerAgentType: AgentType;
reviewerAgentType: AgentType;
reviewerServiceIdForType: string;
arbiterAgentType?: AgentType;
arbiterServiceId: string | null;
agentLanguage: string;
arbiterDeadlockThreshold: number;
maxRoundTrips: number;
};
models: {
owner: RoleModelConfig;
reviewer: RoleModelConfig;
arbiter: RoleModelConfig;
};
providers: {
claudeDefaultModel: string;
codexDefaultModel: string;
};
moa: MoaConfig;
status: {
channelId: string;
updateInterval: number;
usageUpdateInterval: number;
showRooms: boolean;
showRoomDetails: boolean;
usageDashboardEnabled: boolean;
timezone: string;
};
sessionCommands: {
allowedSenders: Set<string>;
};
deltaHandoff: {
enabled: boolean;
canaryGroups: Set<string>;
};
}

View File

@@ -28,6 +28,7 @@ import {
getExplicitRoomMode, getExplicitRoomMode,
getLatestMessageSeqAtOrBefore, getLatestMessageSeqAtOrBefore,
getLatestPairedTaskForChat, getLatestPairedTaskForChat,
getLastRespondingAgentType,
getMessagesSinceSeq, getMessagesSinceSeq,
getNewMessagesBySeq, getNewMessagesBySeq,
getOpenWorkItem, getOpenWorkItem,
@@ -164,6 +165,31 @@ describe('storeMessage', () => {
expect(messages).toHaveLength(1); expect(messages).toHaveLength(1);
}); });
it('detects the most recent bot responder agent type', () => {
storeChatMetadata('group@g.us', '2024-01-01T00:00:00.000Z');
storeMessage({
id: 'bot-1',
chat_jid: 'group@g.us',
sender: 'claude-main',
sender_name: 'Claude',
content: 'first bot reply',
timestamp: '2024-01-01T00:00:01.000Z',
is_bot_message: true,
});
storeMessage({
id: 'bot-2',
chat_jid: 'group@g.us',
sender: 'codex-review',
sender_name: 'Codex',
content: 'second bot reply',
timestamp: '2024-01-01T00:00:02.000Z',
is_bot_message: true,
});
expect(getLastRespondingAgentType('group@g.us')).toBe('codex');
});
it('upserts on duplicate id+chat_jid', () => { it('upserts on duplicate id+chat_jid', () => {
storeChatMetadata('group@g.us', '2024-01-01T00:00:00.000Z'); storeChatMetadata('group@g.us', '2024-01-01T00:00:00.000Z');
@@ -361,6 +387,39 @@ describe('session accessors', () => {
expect(getSession('group-a')).toBeUndefined(); expect(getSession('group-a')).toBeUndefined();
}); });
it('migrates legacy sessions table rows into the composite primary key schema', () => {
const tempDir = fs.mkdtempSync('/tmp/ejclaw-session-schema-migration-');
const dbPath = path.join(tempDir, 'messages.db');
const legacyDb = new Database(dbPath);
legacyDb.exec(`
CREATE TABLE sessions (
group_folder TEXT PRIMARY KEY,
session_id TEXT NOT NULL
);
`);
legacyDb
.prepare(
`INSERT INTO sessions (group_folder, session_id)
VALUES (?, ?)`,
)
.run('group-legacy-schema', 'legacy-schema-session-123');
legacyDb.close();
_initTestDatabaseFromFile(dbPath);
expect(getSession('group-legacy-schema', 'claude-code')).toBe(
'legacy-schema-session-123',
);
const migratedDb = new Database(dbPath, { readonly: true });
const sessionColumns = migratedDb
.prepare(`PRAGMA table_info(sessions)`)
.all() as Array<{ name: string }>;
expect(sessionColumns.some((col) => col.name === 'agent_type')).toBe(true);
migratedDb.close();
});
it('backfills legacy service-scoped sessions into canonical agent sessions during init', () => { it('backfills legacy service-scoped sessions into canonical agent sessions during init', () => {
const tempDir = fs.mkdtempSync('/tmp/ejclaw-session-backfill-'); const tempDir = fs.mkdtempSync('/tmp/ejclaw-session-backfill-');
const dbPath = path.join(tempDir, 'messages.db'); const dbPath = path.join(tempDir, 'messages.db');
@@ -1522,6 +1581,34 @@ describe('room assignment writes', () => {
agentType: 'codex', agentType: 'codex',
}); });
}); });
it('updates room_settings-backed metadata across tribunal projection rows', () => {
assignRoom('dc:projection-room', {
name: 'Projection Room',
roomMode: 'tribunal',
ownerAgentType: 'codex',
folder: 'projection-room',
});
updateRegisteredGroupName('dc:projection-room', 'Projection Room Renamed');
expect(getStoredRoomSettings('dc:projection-room')).toMatchObject({
chatJid: 'dc:projection-room',
name: 'Projection Room Renamed',
roomMode: 'tribunal',
ownerAgentType: 'codex',
});
expect(getAllRegisteredGroups('claude-code')['dc:projection-room']).toMatchObject({
name: 'Projection Room Renamed',
folder: 'projection-room',
agentType: 'claude-code',
});
expect(getAllRegisteredGroups('codex')['dc:projection-room']).toMatchObject({
name: 'Projection Room Renamed',
folder: 'projection-room',
agentType: 'codex',
});
});
}); });
describe('paired room registration', () => { describe('paired room registration', () => {

3661
src/db.ts

File diff suppressed because it is too large Load Diff

261
src/db/base-schema.ts Normal file
View File

@@ -0,0 +1,261 @@
import { Database } from 'bun:sqlite';
export function applyBaseSchema(database: Database): void {
database.exec(`
CREATE TABLE IF NOT EXISTS chats (
jid TEXT PRIMARY KEY,
name TEXT,
last_message_time TEXT,
channel TEXT,
is_group INTEGER DEFAULT 0
);
CREATE TABLE IF NOT EXISTS messages (
id TEXT,
chat_jid TEXT,
sender TEXT,
sender_name TEXT,
content TEXT,
timestamp TEXT,
seq INTEGER,
is_from_me INTEGER,
is_bot_message INTEGER DEFAULT 0,
PRIMARY KEY (id, chat_jid),
FOREIGN KEY (chat_jid) REFERENCES chats(jid)
);
CREATE INDEX IF NOT EXISTS idx_timestamp ON messages(timestamp);
CREATE TABLE IF NOT EXISTS message_sequence (
id INTEGER PRIMARY KEY AUTOINCREMENT
);
CREATE TABLE IF NOT EXISTS work_items (
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)
);
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');
CREATE TABLE IF NOT EXISTS scheduled_tasks (
id TEXT PRIMARY KEY,
group_folder TEXT NOT NULL,
chat_jid TEXT NOT NULL,
agent_type TEXT,
ci_provider TEXT,
ci_metadata TEXT,
max_duration_ms INTEGER,
status_message_id TEXT,
status_started_at TEXT,
prompt TEXT NOT NULL,
schedule_type TEXT NOT NULL,
schedule_value TEXT NOT NULL,
next_run TEXT,
last_run TEXT,
last_result TEXT,
status TEXT DEFAULT 'active',
created_at TEXT NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_next_run ON scheduled_tasks(next_run);
CREATE INDEX IF NOT EXISTS idx_status ON scheduled_tasks(status);
CREATE TABLE IF NOT EXISTS task_run_logs (
id INTEGER PRIMARY KEY AUTOINCREMENT,
task_id TEXT NOT NULL,
run_at TEXT NOT NULL,
duration_ms INTEGER NOT NULL,
status TEXT NOT NULL,
result TEXT,
error TEXT,
FOREIGN KEY (task_id) REFERENCES scheduled_tasks(id)
);
CREATE INDEX IF NOT EXISTS idx_task_run_logs ON task_run_logs(task_id, run_at);
CREATE TABLE IF NOT EXISTS router_state (
key TEXT PRIMARY KEY,
value TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS sessions (
group_folder TEXT NOT NULL,
agent_type TEXT NOT NULL DEFAULT 'claude-code',
session_id TEXT NOT NULL,
PRIMARY KEY (group_folder, agent_type)
);
CREATE TABLE IF NOT EXISTS registered_groups (
jid TEXT NOT NULL,
name TEXT NOT NULL,
folder TEXT NOT NULL,
trigger_pattern TEXT NOT NULL,
added_at TEXT NOT NULL,
agent_config TEXT,
requires_trigger INTEGER DEFAULT 1,
is_main INTEGER DEFAULT 0,
agent_type TEXT NOT NULL DEFAULT 'claude-code',
work_dir TEXT,
PRIMARY KEY (jid, agent_type),
UNIQUE (folder, agent_type)
);
CREATE TABLE IF NOT EXISTS paired_projects (
chat_jid TEXT PRIMARY KEY,
group_folder TEXT NOT NULL,
canonical_work_dir TEXT NOT NULL,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS paired_tasks (
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)
);
CREATE INDEX IF NOT EXISTS idx_paired_tasks_chat_status
ON paired_tasks(chat_jid, status, updated_at);
CREATE TABLE IF NOT EXISTS paired_workspaces (
id TEXT PRIMARY KEY,
task_id TEXT NOT NULL,
role TEXT NOT NULL,
workspace_dir TEXT NOT NULL,
snapshot_source_dir TEXT,
snapshot_ref TEXT,
status TEXT NOT NULL DEFAULT 'ready',
snapshot_refreshed_at TEXT,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL,
CHECK (role IN ('owner', 'reviewer')),
CHECK (status IN ('ready', 'stale'))
);
CREATE UNIQUE INDEX IF NOT EXISTS idx_paired_workspaces_task_role
ON paired_workspaces(task_id, role);
CREATE TABLE IF NOT EXISTS paired_turn_outputs (
id INTEGER PRIMARY KEY AUTOINCREMENT,
task_id TEXT NOT NULL,
turn_number INTEGER NOT NULL,
role TEXT NOT NULL,
output_text TEXT NOT NULL,
created_at TEXT NOT NULL,
UNIQUE(task_id, turn_number, role)
);
CREATE INDEX IF NOT EXISTS idx_paired_turn_outputs_task
ON paired_turn_outputs(task_id, turn_number);
CREATE TABLE IF NOT EXISTS channel_owner (
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)
);
CREATE TABLE IF NOT EXISTS room_settings (
chat_jid TEXT PRIMARY KEY,
room_mode TEXT NOT NULL,
mode_source TEXT NOT NULL DEFAULT 'explicit',
name TEXT,
folder TEXT,
trigger_pattern TEXT,
requires_trigger INTEGER DEFAULT 1,
is_main INTEGER DEFAULT 0,
owner_agent_type TEXT,
work_dir TEXT,
updated_at TEXT NOT NULL,
CHECK (room_mode IN ('single', 'tribunal')),
CHECK (owner_agent_type IN ('claude-code', 'codex') OR owner_agent_type IS NULL)
);
CREATE TABLE IF NOT EXISTS service_handoffs (
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)
);
CREATE INDEX IF NOT EXISTS idx_service_handoffs_target
ON service_handoffs(status, target_role, target_agent_type, created_at);
CREATE TABLE IF NOT EXISTS memories (
id INTEGER PRIMARY KEY,
scope_kind TEXT NOT NULL,
scope_key TEXT NOT NULL,
content TEXT NOT NULL,
keywords_json TEXT NOT NULL DEFAULT '[]',
memory_kind TEXT,
source_kind TEXT NOT NULL,
source_ref TEXT,
created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
last_used_at TEXT,
archived_at TEXT
);
CREATE INDEX IF NOT EXISTS idx_memories_scope
ON memories(scope_kind, scope_key);
CREATE INDEX IF NOT EXISTS idx_memories_active
ON memories(scope_kind, scope_key, archived_at, created_at);
CREATE VIRTUAL TABLE IF NOT EXISTS memories_fts USING fts5(
content,
keywords,
content='',
tokenize='unicode61'
);
CREATE TRIGGER IF NOT EXISTS memories_ai AFTER INSERT ON memories BEGIN
INSERT INTO memories_fts(rowid, content, keywords)
VALUES (new.id, new.content, new.keywords_json);
END;
CREATE TRIGGER IF NOT EXISTS memories_ad AFTER DELETE ON memories BEGIN
INSERT INTO memories_fts(memories_fts, rowid, content, keywords)
VALUES ('delete', old.id, old.content, old.keywords_json);
END;
CREATE TRIGGER IF NOT EXISTS memories_au AFTER UPDATE ON memories BEGIN
INSERT INTO memories_fts(memories_fts, rowid, content, keywords)
VALUES ('delete', old.id, old.content, old.keywords_json);
INSERT INTO memories_fts(rowid, content, keywords)
VALUES (new.id, new.content, new.keywords_json);
END;
`);
}

104
src/db/bootstrap.ts Normal file
View File

@@ -0,0 +1,104 @@
import { Database } from 'bun:sqlite';
import fs from 'fs';
import path from 'path';
import { ASSISTANT_NAME, DATA_DIR, STORE_DIR } from '../config.js';
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';
export interface JsonStateMigrationHooks {
setRouterState(key: string, value: string): void;
setSession(groupFolder: string, sessionId: string): void;
writeLegacyRegisteredGroupAndSyncRoomSettings(
jid: string,
group: RegisteredGroup,
): void;
}
export function initializeDatabaseSchema(
database: Database,
hooks: SchemaMigrationHooks,
): void {
applyBaseSchema(database);
applySchemaMigrations(database, {
assistantName: ASSISTANT_NAME,
hooks,
});
}
export function openPersistentDatabase(): Database {
const dbPath = path.join(STORE_DIR, 'messages.db');
fs.mkdirSync(path.dirname(dbPath), { recursive: true });
const database = new Database(dbPath);
database.exec('PRAGMA journal_mode = WAL');
database.exec('PRAGMA busy_timeout = 5000');
return database;
}
export function openInMemoryDatabase(): Database {
return new Database(':memory:');
}
export function openDatabaseFromFile(dbPath: string): Database {
return new Database(dbPath);
}
export function migrateJsonStateFromFiles(
hooks: JsonStateMigrationHooks,
): void {
const migrateFile = (filename: string) => {
const filePath = path.join(DATA_DIR, filename);
const data = readJsonFile(filePath);
if (data === null) return null;
try {
fs.renameSync(filePath, `${filePath}.migrated`);
} catch {
/* best effort */
}
return data;
};
const routerState = migrateFile('router_state.json') as {
last_timestamp?: string;
last_agent_timestamp?: Record<string, string>;
} | null;
if (routerState) {
if (routerState.last_timestamp) {
hooks.setRouterState('last_timestamp', routerState.last_timestamp);
}
if (routerState.last_agent_timestamp) {
hooks.setRouterState(
'last_agent_timestamp',
JSON.stringify(routerState.last_agent_timestamp),
);
}
}
const sessions = migrateFile('sessions.json') as Record<string, string> | null;
if (sessions) {
for (const [folder, sessionId] of Object.entries(sessions)) {
hooks.setSession(folder, sessionId);
}
}
const groups = migrateFile('registered_groups.json') as Record<
string,
RegisteredGroup
> | null;
if (groups) {
for (const [jid, group] of Object.entries(groups)) {
try {
hooks.writeLegacyRegisteredGroupAndSyncRoomSettings(jid, group);
} catch (err) {
logger.warn(
{ jid, folder: group.folder, err },
'Skipping migrated registered group with invalid folder',
);
}
}
}
}

767
src/db/legacy-rebuilds.ts Normal file
View File

@@ -0,0 +1,767 @@
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 {
chat_jid: string;
group_folder: string;
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,
): AgentType | undefined {
const persistedOwnerAgentType = normalizeStoredAgentType(
task.owner_agent_type,
);
if (persistedOwnerAgentType) {
return persistedOwnerAgentType;
}
const stored = getStoredRoomSettingsRowFromDatabase(database, task.chat_jid);
if (stored?.ownerAgentType) {
return stored.ownerAgentType;
}
const jidAgentTypes = collectRegisteredAgentTypes(database, task.chat_jid);
if (jidAgentTypes.length > 0) {
return inferOwnerAgentTypeFromRegisteredAgentTypes(jidAgentTypes);
}
const folderAgentTypes = collectRegisteredAgentTypesForFolder(
database,
task.group_folder,
);
if (folderAgentTypes.length > 0) {
return inferOwnerAgentTypeFromRegisteredAgentTypes(folderAgentTypes);
}
return undefined;
}
export function resolveStableReviewerAgentType(
ownerAgentType: AgentType | undefined,
fallbackReviewerAgentType?: string | null,
): AgentType | null {
const persistedReviewerAgentType = normalizeStoredAgentType(
fallbackReviewerAgentType,
);
if (persistedReviewerAgentType) {
return persistedReviewerAgentType;
}
if (ownerAgentType) {
return REVIEWER_AGENT_TYPE !== ownerAgentType
? REVIEWER_AGENT_TYPE
: ownerAgentType;
}
return null;
}
export function resolveStableRoomRoleAgentType(
database: Database,
input: {
chatJid: string;
groupFolder: string;
role: PairedRoomRole;
},
): AgentType | null | undefined {
if (input.role === 'owner') {
return resolveStablePairedTaskOwnerAgentType(database, {
chat_jid: input.chatJid,
group_folder: input.groupFolder,
owner_agent_type: null,
});
}
if (input.role === 'reviewer') {
const ownerAgentType = resolveStablePairedTaskOwnerAgentType(database, {
chat_jid: input.chatJid,
group_folder: input.groupFolder,
owner_agent_type: null,
});
return resolveStableReviewerAgentType(ownerAgentType, null);
}
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);
`);
}

333
src/db/memories.ts Normal file
View File

@@ -0,0 +1,333 @@
import { Database } from 'bun:sqlite';
import { logger } from '../logger.js';
export type MemoryScopeKind = 'room' | 'user' | 'project' | 'global';
export type MemorySourceKind = 'compact' | 'explicit' | 'import' | 'system';
export interface MemoryRecord {
id: number;
scopeKind: MemoryScopeKind;
scopeKey: string;
content: string;
keywords: string[];
memoryKind: string | null;
sourceKind: MemorySourceKind;
sourceRef: string | null;
createdAt: string;
lastUsedAt: string | null;
archivedAt: string | null;
}
export interface RecallMemoryQuery {
scopeKind: MemoryScopeKind;
scopeKey: string;
text?: string;
keywords?: string[];
limit?: number;
}
interface MemoryDatabaseRow {
id: number;
scope_kind: string;
scope_key: string;
content: string;
keywords_json: string;
memory_kind: string | null;
source_kind: string;
source_ref: string | null;
created_at: string;
last_used_at: string | null;
archived_at: string | null;
}
const MEMORY_SCOPE_LIMITS: Record<MemoryScopeKind, number> = {
room: 300,
user: 100,
project: 200,
global: 100,
};
const COMPACT_MEMORY_TTL_DAYS = 30;
function normalizeMemoryKeywords(keywords?: string[]): string[] {
if (!Array.isArray(keywords)) return [];
return [
...new Set(
keywords.map((keyword) => keyword.trim().toLowerCase()).filter(Boolean),
),
];
}
function parseMemoryKeywords(raw: string | null | undefined): string[] {
if (!raw) return [];
try {
const parsed = JSON.parse(raw);
return normalizeMemoryKeywords(Array.isArray(parsed) ? parsed : []);
} catch {
return [];
}
}
function mapMemoryRow(row: MemoryDatabaseRow): MemoryRecord {
return {
id: row.id,
scopeKind: row.scope_kind as MemoryScopeKind,
scopeKey: row.scope_key,
content: row.content,
keywords: parseMemoryKeywords(row.keywords_json),
memoryKind: row.memory_kind,
sourceKind: row.source_kind as MemorySourceKind,
sourceRef: row.source_ref,
createdAt: row.created_at,
lastUsedAt: row.last_used_at,
archivedAt: row.archived_at,
};
}
function buildMemoryFtsQuery(query: RecallMemoryQuery): string | null {
const tokens = normalizeMemoryKeywords([
...(query.text?.match(/[\p{L}\p{N}_:-]+/gu) ?? []),
...(query.keywords ?? []),
]);
if (tokens.length === 0) return null;
return tokens.map((token) => `"${token.replaceAll('"', '""')}"`).join(' OR ');
}
function getMemoryRowsForScope(
database: Database,
scopeKind: MemoryScopeKind,
scopeKey: string,
): MemoryDatabaseRow[] {
return database
.prepare(
`SELECT *
FROM memories
WHERE scope_kind = ?
AND scope_key = ?
AND archived_at IS NULL
ORDER BY COALESCE(last_used_at, created_at) DESC, id DESC`,
)
.all(scopeKind, scopeKey) as MemoryDatabaseRow[];
}
function queryFtsRowOrder(
database: Database,
query: RecallMemoryQuery,
): Map<number, number> {
const ftsQuery = buildMemoryFtsQuery(query);
if (!ftsQuery) return new Map();
try {
const rows = database
.prepare(
`SELECT memories.id AS id
FROM memories_fts
JOIN memories ON memories.id = memories_fts.rowid
WHERE memories_fts MATCH ?
AND memories.scope_kind = ?
AND memories.scope_key = ?
AND memories.archived_at IS NULL
ORDER BY bm25(memories_fts), memories.created_at DESC
LIMIT ?`,
)
.all(
ftsQuery,
query.scopeKind,
query.scopeKey,
Math.max(25, (query.limit ?? 10) * 8),
) as Array<{ id: number }>;
return new Map(rows.map((row, index) => [row.id, rows.length - index]));
} catch (error) {
logger.warn(
{ query, error },
'Memory FTS query failed; falling back to scope-only recall',
);
return new Map();
}
}
export function touchMemoriesInDatabase(database: Database, ids: number[]): void {
const uniqueIds = [
...new Set(ids.filter((id) => Number.isInteger(id) && id > 0)),
];
if (uniqueIds.length === 0) return;
const now = new Date().toISOString();
const stmt = database.prepare(
`UPDATE memories
SET last_used_at = ?
WHERE id = ?`,
);
const tx = database.transaction(() => {
for (const id of uniqueIds) stmt.run(now, id);
});
tx();
}
export function archiveMemoryInDatabase(database: Database, id: number): void {
database
.prepare(
`UPDATE memories
SET archived_at = COALESCE(archived_at, ?)
WHERE id = ?`,
)
.run(new Date().toISOString(), id);
}
function buildCompactMemoryExpiryCutoff(nowIso: string): string {
return new Date(
new Date(nowIso).getTime() - COMPACT_MEMORY_TTL_DAYS * 24 * 60 * 60 * 1000,
).toISOString();
}
export function expireStaleMemoriesInDatabase(
database: Database,
args?: {
scopeKind?: MemoryScopeKind;
scopeKey?: string;
now?: string;
},
): number {
const nowIso = args?.now ?? new Date().toISOString();
const cutoff = buildCompactMemoryExpiryCutoff(nowIso);
const scopeClause =
args?.scopeKind && args?.scopeKey
? 'AND scope_kind = ? AND scope_key = ?'
: '';
const stmt = database.prepare(
`UPDATE memories
SET archived_at = COALESCE(archived_at, ?)
WHERE archived_at IS NULL
AND source_kind = 'compact'
AND COALESCE(last_used_at, created_at) < ?
${scopeClause}`,
);
const result = (
args?.scopeKind && args?.scopeKey
? stmt.run(nowIso, cutoff, args.scopeKind, args.scopeKey)
: stmt.run(nowIso, cutoff)
) as { changes?: number };
return result.changes ?? 0;
}
export function enforceMemoryBoundsInDatabase(
database: Database,
scopeKind: MemoryScopeKind,
scopeKey: string,
): void {
const limit = MEMORY_SCOPE_LIMITS[scopeKind];
const rows = database
.prepare(
`SELECT id
FROM memories
WHERE scope_kind = ?
AND scope_key = ?
AND archived_at IS NULL
ORDER BY COALESCE(last_used_at, created_at) DESC, id DESC
LIMIT -1 OFFSET ?`,
)
.all(scopeKind, scopeKey, limit) as Array<{ id: number }>;
if (rows.length === 0) return;
const now = new Date().toISOString();
const stmt = database.prepare(
`UPDATE memories
SET archived_at = ?
WHERE id = ?`,
);
const tx = database.transaction(() => {
for (const row of rows) stmt.run(now, row.id);
});
tx();
}
export function rememberMemoryInDatabase(
database: Database,
input: {
scopeKind: MemoryScopeKind;
scopeKey: string;
content: string;
keywords?: string[];
memoryKind?: string | null;
sourceKind: MemorySourceKind;
sourceRef?: string | null;
},
): number {
const normalizedContent = input.content.trim();
if (!normalizedContent) {
throw new Error('Memory content cannot be empty');
}
const normalizedKeywords = normalizeMemoryKeywords(input.keywords);
const createdAt = new Date().toISOString();
database
.prepare(
`INSERT INTO memories (
scope_kind,
scope_key,
content,
keywords_json,
memory_kind,
source_kind,
source_ref,
created_at
) VALUES (?, ?, ?, ?, ?, ?, ?, ?)`,
)
.run(
input.scopeKind,
input.scopeKey,
normalizedContent,
JSON.stringify(normalizedKeywords),
input.memoryKind ?? null,
input.sourceKind,
input.sourceRef ?? null,
createdAt,
);
const row = database.prepare('SELECT last_insert_rowid() AS id').get() as {
id: number;
};
expireStaleMemoriesInDatabase(database, {
scopeKind: input.scopeKind,
scopeKey: input.scopeKey,
});
enforceMemoryBoundsInDatabase(database, input.scopeKind, input.scopeKey);
return row.id;
}
export function recallMemoriesFromDatabase(
database: Database,
query: RecallMemoryQuery,
): MemoryRecord[] {
const limit = Math.max(1, query.limit ?? 6);
expireStaleMemoriesInDatabase(database, {
scopeKind: query.scopeKind,
scopeKey: query.scopeKey,
});
const rows = getMemoryRowsForScope(database, query.scopeKind, query.scopeKey);
if (rows.length === 0) return [];
const exactKeywords = new Set(normalizeMemoryKeywords(query.keywords));
const ftsOrder = queryFtsRowOrder(database, query);
const useQueryScoring = exactKeywords.size > 0 || Boolean(query.text?.trim());
const scored = rows
.map((row, index) => {
const keywords = parseMemoryKeywords(row.keywords_json);
const exactMatches = keywords.filter((keyword) =>
exactKeywords.has(keyword),
).length;
const ftsScore = ftsOrder.get(row.id) ?? 0;
const recencyScore = rows.length - index;
return {
row,
matched: exactMatches > 0 || ftsScore > 0,
score: exactMatches * 100 + ftsScore * 10 + recencyScore,
};
})
.filter((entry) => (useQueryScoring ? entry.matched : true))
.sort((a, b) => b.score - a.score || b.row.id - a.row.id)
.slice(0, limit);
const memories = scored.map((entry) => mapMemoryRow(entry.row));
touchMemoriesInDatabase(
database,
memories.map((memory) => memory.id),
);
return memories;
}

385
src/db/messages.ts Normal file
View File

@@ -0,0 +1,385 @@
import { Database } from 'bun:sqlite';
import { NewMessage } from '../types.js';
export interface ChatInfo {
jid: string;
name: string;
last_message_time: string;
channel: string;
is_group: number;
}
function normalizeMessageRow(
row: NewMessage & {
is_from_me?: boolean | number;
is_bot_message?: boolean | number;
},
): NewMessage {
return {
...row,
is_from_me: !!row.is_from_me,
is_bot_message: !!row.is_bot_message,
};
}
export function normalizeSeqCursor(
cursor: string | number | null | undefined,
): number {
if (typeof cursor === 'number') {
return Number.isFinite(cursor) && cursor > 0 ? cursor : 0;
}
if (!cursor) return 0;
const parsed = Number.parseInt(cursor, 10);
return Number.isFinite(parsed) && parsed > 0 ? parsed : 0;
}
export function storeChatMetadataInDatabase(
database: Database,
chatJid: string,
timestamp: string,
name?: string,
channel?: string,
isGroup?: boolean,
): void {
const ch = channel ?? null;
const group = isGroup === undefined ? null : isGroup ? 1 : 0;
if (name) {
database
.prepare(
`
INSERT INTO chats (jid, name, last_message_time, channel, is_group) VALUES (?, ?, ?, ?, ?)
ON CONFLICT(jid) DO UPDATE SET
name = excluded.name,
last_message_time = MAX(last_message_time, excluded.last_message_time),
channel = COALESCE(excluded.channel, channel),
is_group = COALESCE(excluded.is_group, is_group)
`,
)
.run(chatJid, name, timestamp, ch, group);
} else {
database
.prepare(
`
INSERT INTO chats (jid, name, last_message_time, channel, is_group) VALUES (?, ?, ?, ?, ?)
ON CONFLICT(jid) DO UPDATE SET
last_message_time = MAX(last_message_time, excluded.last_message_time),
channel = COALESCE(excluded.channel, channel),
is_group = COALESCE(excluded.is_group, is_group)
`,
)
.run(chatJid, chatJid, timestamp, ch, group);
}
}
export function getAllChatsFromDatabase(database: Database): ChatInfo[] {
return database
.prepare(
`
SELECT jid, name, last_message_time, channel, is_group
FROM chats
ORDER BY last_message_time DESC
`,
)
.all() as ChatInfo[];
}
export function storeMessageInDatabase(
database: Database,
msg: NewMessage,
): void {
const nextSeq = () => {
database.prepare('INSERT INTO message_sequence DEFAULT VALUES').run();
return (
database.prepare('SELECT last_insert_rowid() as id').get() as {
id: number;
}
).id;
};
database.transaction(() => {
const existing = database
.prepare('SELECT seq FROM messages WHERE id = ? AND chat_jid = ?')
.get(msg.id, msg.chat_jid) as { seq: number | null } | undefined;
const seq = existing?.seq ?? nextSeq();
database
.prepare(
`INSERT INTO messages (
id, chat_jid, sender, sender_name, content, timestamp, seq, is_from_me, is_bot_message
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(id, chat_jid) DO UPDATE SET
sender = excluded.sender,
sender_name = excluded.sender_name,
content = excluded.content,
timestamp = excluded.timestamp,
is_from_me = excluded.is_from_me,
is_bot_message = excluded.is_bot_message`,
)
.run(
msg.id,
msg.chat_jid,
msg.sender,
msg.sender_name,
msg.content,
msg.timestamp,
seq,
msg.is_from_me ? 1 : 0,
msg.is_bot_message ? 1 : 0,
);
})();
}
export function getNewMessagesFromDatabase(
database: Database,
jids: string[],
lastTimestamp: string,
botPrefix: string,
limit: number = 200,
): { messages: NewMessage[]; newTimestamp: string } {
if (jids.length === 0) return { messages: [], newTimestamp: lastTimestamp };
const placeholders = jids.map(() => '?').join(',');
const sql = `
SELECT * FROM (
SELECT id, chat_jid, sender, sender_name, content, timestamp, is_from_me, is_bot_message
FROM messages
WHERE timestamp > ? AND chat_jid IN (${placeholders})
AND content NOT LIKE ?
AND content != '' AND content IS NOT NULL
ORDER BY timestamp DESC
LIMIT ?
) ORDER BY timestamp
`;
const rows = database
.prepare(sql)
.all(lastTimestamp, ...jids, `${botPrefix}:%`, limit) as Array<
NewMessage & {
is_from_me?: boolean | number;
is_bot_message?: boolean | number;
}
>;
let newTimestamp = lastTimestamp;
for (const row of rows) {
if (row.timestamp > newTimestamp) newTimestamp = row.timestamp;
}
return { messages: rows.map(normalizeMessageRow), newTimestamp };
}
export function getMessagesSinceFromDatabase(
database: Database,
chatJid: string,
sinceTimestamp: string,
botPrefix: string,
limit: number = 200,
): NewMessage[] {
const sql = `
SELECT * FROM (
SELECT id, chat_jid, sender, sender_name, content, timestamp, is_from_me, is_bot_message
FROM messages
WHERE chat_jid = ? AND timestamp > ?
AND content NOT LIKE ?
AND content != '' AND content IS NOT NULL
ORDER BY timestamp DESC
LIMIT ?
) ORDER BY timestamp
`;
const rows = database
.prepare(sql)
.all(chatJid, sinceTimestamp, `${botPrefix}:%`, limit) as Array<
NewMessage & {
is_from_me?: boolean | number;
is_bot_message?: boolean | number;
}
>;
return rows.map(normalizeMessageRow);
}
export function getLatestMessageSeqAtOrBeforeFromDatabase(
database: Database,
timestamp: string,
chatJid?: string,
): number {
if (!timestamp) return 0;
const row = (
chatJid
? database
.prepare(
`SELECT COALESCE(MAX(seq), 0) AS maxSeq
FROM messages
WHERE chat_jid = ? AND timestamp <= ?`,
)
.get(chatJid, timestamp)
: database
.prepare(
`SELECT COALESCE(MAX(seq), 0) AS maxSeq
FROM messages
WHERE timestamp <= ?`,
)
.get(timestamp)
) as { maxSeq: number | null };
return row.maxSeq ?? 0;
}
export function getNewMessagesBySeqFromDatabase(
database: Database,
jids: string[],
lastSeqCursor: string | number,
botPrefix: string,
limit: number = 200,
): { messages: NewMessage[]; newSeqCursor: string } {
const sinceSeq = normalizeSeqCursor(lastSeqCursor);
if (jids.length === 0) {
return { messages: [], newSeqCursor: String(sinceSeq) };
}
const placeholders = jids.map(() => '?').join(',');
const sql = `
SELECT id, chat_jid, sender, sender_name, content, timestamp, seq, is_from_me, is_bot_message
FROM messages
WHERE seq > ? AND chat_jid IN (${placeholders})
AND content NOT LIKE ?
AND content != '' AND content IS NOT NULL
ORDER BY seq
LIMIT ?
`;
const rows = database
.prepare(sql)
.all(sinceSeq, ...jids, `${botPrefix}:%`, limit) as Array<
NewMessage & {
seq: number;
is_from_me?: boolean | number;
is_bot_message?: boolean | number;
}
>;
const lastSeq = rows.length > 0 ? rows[rows.length - 1].seq : sinceSeq;
return {
messages: rows.map(normalizeMessageRow),
newSeqCursor: String(lastSeq),
};
}
export function getMessagesSinceSeqFromDatabase(
database: Database,
chatJid: string,
sinceSeqCursor: string | number,
botPrefix: string,
limit: number = 200,
): NewMessage[] {
const sinceSeq = normalizeSeqCursor(sinceSeqCursor);
const sql = `
SELECT id, chat_jid, sender, sender_name, content, timestamp, seq, is_from_me, is_bot_message
FROM messages
WHERE chat_jid = ? AND seq > ?
AND content NOT LIKE ?
AND content != '' AND content IS NOT NULL
ORDER BY seq
LIMIT ?
`;
const rows = database
.prepare(sql)
.all(chatJid, sinceSeq, `${botPrefix}:%`, limit) as Array<
NewMessage & {
seq: number;
is_from_me?: boolean | number;
is_bot_message?: boolean | number;
}
>;
return rows.map(normalizeMessageRow);
}
export function getRecentChatMessagesFromDatabase(
database: Database,
chatJid: string,
limit: number = 20,
): NewMessage[] {
const sql = `
SELECT * FROM (
SELECT id, chat_jid, sender, sender_name, content, timestamp, is_from_me, is_bot_message
FROM messages
WHERE chat_jid = ?
AND content != '' AND content IS NOT NULL
ORDER BY timestamp DESC
LIMIT ?
) ORDER BY timestamp
`;
const rows = database.prepare(sql).all(chatJid, limit) as Array<
NewMessage & {
is_from_me?: boolean | number;
is_bot_message?: boolean | number;
}
>;
return rows.map(normalizeMessageRow);
}
export function getLastHumanMessageTimestampFromDatabase(
database: Database,
chatJid: string,
): string | null {
const row = database
.prepare(
`SELECT timestamp FROM messages
WHERE chat_jid = ? AND is_bot_message = 0 AND is_from_me = 0
AND content != '' AND content IS NOT NULL
ORDER BY timestamp DESC LIMIT 1`,
)
.get(chatJid) as { timestamp: string } | undefined;
return row?.timestamp ?? null;
}
export function getLastHumanMessageSenderFromDatabase(
database: Database,
chatJid: string,
): string | null {
const row = database
.prepare(
`SELECT sender FROM messages
WHERE chat_jid = ? AND is_bot_message = 0 AND is_from_me = 0
AND content != '' AND content IS NOT NULL
ORDER BY timestamp DESC, seq DESC LIMIT 1`,
)
.get(chatJid) as { sender: string } | undefined;
return row?.sender ?? null;
}
export function getLastHumanMessageContentFromDatabase(
database: Database,
chatJid: string,
): string | null {
const row = database
.prepare(
`SELECT content FROM messages
WHERE chat_jid = ? AND is_bot_message = 0 AND is_from_me = 0
AND content != '' AND content IS NOT NULL
ORDER BY timestamp DESC, seq DESC LIMIT 1`,
)
.get(chatJid) as { content: string } | undefined;
return row?.content ?? null;
}
export function hasRecentRestartAnnouncementInDatabase(
database: Database,
chatJid: string,
sinceTimestamp: string,
): boolean {
const row = database
.prepare(
`SELECT 1 FROM messages
WHERE chat_jid = ?
AND timestamp >= ?
AND is_bot_message = 1
AND (
content LIKE '재시작 완료.%'
OR content LIKE '재시작 감지.%'
OR content LIKE '서비스 재시작으로 이전 작업이 중단됐습니다.%'
)
LIMIT 1`,
)
.get(chatJid, sinceTimestamp) as { 1: number } | undefined;
return !!row;
}

678
src/db/room-registration.ts Normal file
View File

@@ -0,0 +1,678 @@
import { Database } from 'bun:sqlite';
import {
ASSISTANT_NAME,
OWNER_AGENT_TYPE,
REVIEWER_AGENT_TYPE,
} from '../config.js';
import { isValidGroupFolder } from '../group-folder.js';
import { logger } from '../logger.js';
import type { AgentType, RegisteredGroup, RoomMode } from '../types.js';
export type RoomModeSource = 'explicit' | 'inferred';
export interface StoredRoomSettings {
chatJid: string;
roomMode: RoomMode;
modeSource: RoomModeSource;
name?: string;
folder?: string;
trigger?: string;
requiresTrigger?: boolean;
isMain?: boolean;
ownerAgentType?: AgentType;
workDir?: string;
}
export interface RoomRegistrationSnapshot {
name: string;
folder: string;
triggerPattern: string;
requiresTrigger: boolean;
isMain: boolean;
ownerAgentType: AgentType;
workDir: string | null;
}
export interface RegisteredGroupDatabaseRow {
jid: string;
name: string;
folder: string;
trigger_pattern: string;
added_at: string;
agent_config: string | null;
requires_trigger: number | null;
is_main: number | null;
agent_type: string | null;
work_dir: string | null;
}
export function normalizeRoomModeSource(
source: string | null | undefined,
): RoomModeSource | undefined {
return source === 'explicit' || source === 'inferred' ? source : undefined;
}
export function normalizeStoredAgentType(
agentType: string | null | undefined,
): AgentType | undefined {
return agentType === 'claude-code' || agentType === 'codex'
? agentType
: undefined;
}
export function collectRegisteredAgentTypes(
database: Database,
jid: string,
): AgentType[] {
const rows = database
.prepare('SELECT agent_type FROM registered_groups WHERE jid = ?')
.all(jid) as Array<{ agent_type: string | null }>;
const types = new Set<AgentType>();
for (const row of rows) {
const agentType = normalizeStoredAgentType(row.agent_type);
if (agentType) {
types.add(agentType);
}
}
return [...types];
}
export function collectRegisteredAgentTypesForFolder(
database: Database,
folder: string,
): AgentType[] {
const rows = database
.prepare('SELECT agent_type FROM registered_groups WHERE folder = ?')
.all(folder) as Array<{ agent_type: string | null }>;
const types = new Set<AgentType>();
for (const row of rows) {
const agentType = normalizeStoredAgentType(row.agent_type);
if (agentType) {
types.add(agentType);
}
}
return [...types];
}
export function inferRoomModeFromRegisteredAgentTypes(
agentTypes: readonly AgentType[],
): RoomMode {
const types = new Set(agentTypes);
return types.has('claude-code') && types.has('codex') ? 'tribunal' : 'single';
}
export function inferOwnerAgentTypeFromRegisteredAgentTypes(
agentTypes: readonly AgentType[],
): AgentType {
const types = new Set(agentTypes);
if (types.has(OWNER_AGENT_TYPE)) return OWNER_AGENT_TYPE;
if (types.has('codex')) return 'codex';
return 'claude-code';
}
export function parseRegisteredGroupRow(
row: RegisteredGroupDatabaseRow | undefined,
): (RegisteredGroup & { jid: string }) | undefined {
if (!row) return undefined;
if (!isValidGroupFolder(row.folder)) {
logger.warn(
{ jid: row.jid, folder: row.folder },
'Skipping registered group with invalid folder',
);
return undefined;
}
return {
jid: row.jid,
name: row.name,
folder: row.folder,
trigger: row.trigger_pattern,
added_at: row.added_at,
agentConfig: row.agent_config ? JSON.parse(row.agent_config) : undefined,
requiresTrigger:
row.requires_trigger === null ? undefined : row.requires_trigger === 1,
isMain: row.is_main === 1 ? true : undefined,
agentType: normalizeStoredAgentType(row.agent_type),
workDir: row.work_dir || undefined,
};
}
export function getLegacyRegisteredGroupRows(
database: Database,
agentTypeFilter?: string,
): RegisteredGroupDatabaseRow[] {
return (
agentTypeFilter
? database
.prepare('SELECT * FROM registered_groups WHERE agent_type = ?')
.all(agentTypeFilter)
: database.prepare('SELECT * FROM registered_groups').all()
) as RegisteredGroupDatabaseRow[];
}
export function getLegacyRegisteredGroup(
database: Database,
jid: string,
agentType?: string,
): (RegisteredGroup & { jid: string }) | undefined {
const row = (
agentType
? database
.prepare(
'SELECT * FROM registered_groups WHERE jid = ? AND agent_type = ?',
)
.get(jid, agentType)
: database
.prepare('SELECT * FROM registered_groups WHERE jid = ?')
.get(jid)
) as RegisteredGroupDatabaseRow | undefined;
return parseRegisteredGroupRow(row);
}
export function getRegisteredGroupCapabilityMetadata(
database: Database,
jid: string,
preferredAgentType?: AgentType,
): Pick<RegisteredGroup, 'added_at' | 'agentConfig'> | undefined {
const row = (
preferredAgentType
? database
.prepare(
`SELECT added_at, agent_config
FROM registered_groups
WHERE jid = ? AND agent_type = ?
LIMIT 1`,
)
.get(jid, preferredAgentType)
: database
.prepare(
`SELECT added_at, agent_config
FROM registered_groups
WHERE jid = ?
ORDER BY CASE WHEN agent_type = ? THEN 0 ELSE 1 END, added_at
LIMIT 1`,
)
.get(jid, OWNER_AGENT_TYPE)
) as { added_at: string; agent_config: string | null } | undefined;
if (!row) return undefined;
return {
added_at: row.added_at,
agentConfig: row.agent_config ? JSON.parse(row.agent_config) : undefined,
};
}
export function getStoredRoomSettingsRowFromDatabase(
database: Database,
chatJid: string,
): StoredRoomSettings | undefined {
const row = database
.prepare(
`SELECT room_mode, mode_source, name, folder, trigger_pattern,
requires_trigger, is_main, owner_agent_type, work_dir
FROM room_settings
WHERE chat_jid = ?`,
)
.get(chatJid) as
| {
room_mode: string | null;
mode_source: string | null;
name: string | null;
folder: string | null;
trigger_pattern: string | null;
requires_trigger: number | null;
is_main: number | null;
owner_agent_type: string | null;
work_dir: string | null;
}
| undefined;
const roomMode =
row?.room_mode === 'single' || row?.room_mode === 'tribunal'
? row.room_mode
: undefined;
const source = normalizeRoomModeSource(row?.mode_source);
if (!row || !roomMode || !source) return undefined;
return {
chatJid,
roomMode,
modeSource: source,
name: row.name ?? undefined,
folder: row.folder ?? undefined,
trigger: row.trigger_pattern ?? undefined,
requiresTrigger:
row.requires_trigger === null ? undefined : row.requires_trigger === 1,
isMain: row.is_main === null ? undefined : row.is_main === 1,
ownerAgentType: normalizeStoredAgentType(row.owner_agent_type),
workDir: row.work_dir ?? undefined,
};
}
export function getStoredRoomRowsFromDatabase(
database: Database,
): StoredRoomSettings[] {
return database
.prepare(
`SELECT chat_jid
FROM room_settings
ORDER BY chat_jid`,
)
.all()
.map((row) =>
getStoredRoomSettingsRowFromDatabase(
database,
(row as { chat_jid: string }).chat_jid,
),
)
.filter((row): row is StoredRoomSettings => Boolean(row));
}
export function buildRegisteredGroupFromStoredSettings(
database: Database,
stored: StoredRoomSettings,
requestedAgentType?: AgentType,
): (RegisteredGroup & { jid: string }) | undefined {
const capabilityTypes = collectRegisteredAgentTypes(database, stored.chatJid);
const resolvedAgentType = requestedAgentType
? capabilityTypes.includes(requestedAgentType)
? requestedAgentType
: undefined
: stored.ownerAgentType ||
(capabilityTypes.length > 0
? inferOwnerAgentTypeFromRegisteredAgentTypes(capabilityTypes)
: undefined);
if (!resolvedAgentType) return undefined;
if (!stored.folder || !isValidGroupFolder(stored.folder)) {
logger.warn(
{ jid: stored.chatJid, folder: stored.folder ?? null },
'Skipping stored room with invalid folder',
);
return undefined;
}
const capabilityMetadata = getRegisteredGroupCapabilityMetadata(
database,
stored.chatJid,
resolvedAgentType,
);
return {
jid: stored.chatJid,
name: stored.name || stored.chatJid,
folder: stored.folder,
trigger: stored.trigger || `@${ASSISTANT_NAME}`,
added_at: capabilityMetadata?.added_at || new Date(0).toISOString(),
agentConfig: capabilityMetadata?.agentConfig,
requiresTrigger: stored.requiresTrigger,
isMain: stored.isMain ? true : undefined,
agentType: resolvedAgentType,
workDir: stored.workDir,
};
}
function detectChannelPrefixForFolder(chatJid: string): string {
if (chatJid.startsWith('dc:')) return 'discord';
if (chatJid.startsWith('tg:')) return 'telegram';
return 'whatsapp';
}
function slugifyGroupFolderSegment(value: string): string {
const slug = value
.trim()
.toLowerCase()
.replace(/[^a-z0-9]+/g, '-')
.replace(/^-+|-+$/g, '')
.replace(/-{2,}/g, '-');
return slug || 'room';
}
function collectReservedFolders(
database: Database,
exceptChatJid?: string,
): Set<string> {
const folders = new Set<string>();
const rows = database
.prepare(
`SELECT folder
FROM room_settings
WHERE folder IS NOT NULL
AND (? IS NULL OR chat_jid != ?)
UNION
SELECT folder
FROM registered_groups
WHERE ? IS NULL OR jid != ?`,
)
.all(
exceptChatJid ?? null,
exceptChatJid ?? null,
exceptChatJid ?? null,
exceptChatJid ?? null,
) as Array<{
folder: string | null;
}>;
for (const row of rows) {
if (row.folder) folders.add(row.folder);
}
return folders;
}
function buildGeneratedRoomFolder(
database: Database,
chatJid: string,
name: string,
): string {
const prefix = detectChannelPrefixForFolder(chatJid);
const slug = slugifyGroupFolderSegment(name);
const fallback = slugifyGroupFolderSegment(chatJid.replace(/[:@.]/g, '-'));
const baseCore = slug || fallback;
const maxBaseLength = 64 - (`grp_${prefix}_`.length + 4);
const truncatedCore = baseCore.slice(0, Math.max(8, maxBaseLength));
const candidateBase = `grp_${prefix}_${truncatedCore}`;
const reserved = collectReservedFolders(database, chatJid);
if (!reserved.has(candidateBase) && isValidGroupFolder(candidateBase)) {
return candidateBase;
}
for (let suffix = 2; suffix < 1000; suffix += 1) {
const candidate = `${candidateBase.slice(0, Math.max(1, 64 - `${suffix}`.length - 1))}-${suffix}`;
if (!reserved.has(candidate) && isValidGroupFolder(candidate)) {
return candidate;
}
}
throw new Error(`Unable to generate unique group folder for ${chatJid}`);
}
export function resolveAssignedRoomFolder(
database: Database,
chatJid: string,
name: string,
explicitFolder?: string,
): string {
const reserved = collectReservedFolders(database, chatJid);
if (explicitFolder) {
if (!isValidGroupFolder(explicitFolder)) {
throw new Error(
`Invalid group folder "${explicitFolder}" for JID ${chatJid}`,
);
}
if (reserved.has(explicitFolder)) {
throw new Error(`Group folder "${explicitFolder}" is already assigned`);
}
return explicitFolder;
}
const existingFolder = getStoredRoomSettingsRowFromDatabase(
database,
chatJid,
)?.folder;
if (existingFolder) return existingFolder;
return buildGeneratedRoomFolder(database, chatJid, name);
}
function getDesiredRegisteredAgentTypes(
roomMode: RoomMode,
ownerAgentType: AgentType,
): AgentType[] {
const types = new Set<AgentType>([ownerAgentType]);
if (roomMode === 'tribunal') {
types.add(REVIEWER_AGENT_TYPE);
}
return [...types];
}
export function materializeRegisteredGroupsForRoom(
database: Database,
chatJid: string,
snapshot: RoomRegistrationSnapshot,
roomMode: RoomMode,
ownerAgentType: AgentType,
ownerAgentConfig?: RegisteredGroup['agentConfig'],
addedAt?: string,
): void {
const now = new Date().toISOString();
const desiredTypes = getDesiredRegisteredAgentTypes(roomMode, ownerAgentType);
const existingRows = database
.prepare(
`SELECT agent_type, added_at, agent_config
FROM registered_groups
WHERE jid = ?`,
)
.all(chatJid) as Array<{
agent_type: string | null;
added_at: string;
agent_config: string | null;
}>;
const existingByType = new Map<
AgentType,
{ added_at: string; agent_config: string | null }
>();
for (const row of existingRows) {
const agentType = normalizeStoredAgentType(row.agent_type);
if (agentType) {
existingByType.set(agentType, row);
}
}
for (const agentType of desiredTypes) {
const existing = existingByType.get(agentType);
const agentConfig =
agentType === ownerAgentType
? (ownerAgentConfig ??
(existing?.agent_config
? JSON.parse(existing.agent_config)
: undefined))
: existing?.agent_config
? JSON.parse(existing.agent_config)
: undefined;
const rowAddedAt = existing?.added_at ?? addedAt ?? now;
database
.prepare(
`INSERT OR REPLACE INTO registered_groups (
jid,
name,
folder,
trigger_pattern,
added_at,
agent_config,
requires_trigger,
is_main,
agent_type,
work_dir
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
)
.run(
chatJid,
snapshot.name,
snapshot.folder,
snapshot.triggerPattern,
rowAddedAt,
agentConfig ? JSON.stringify(agentConfig) : null,
snapshot.requiresTrigger ? 1 : 0,
snapshot.isMain ? 1 : 0,
agentType,
snapshot.workDir,
);
}
const placeholders = desiredTypes.map(() => '?').join(',');
database
.prepare(
`DELETE FROM registered_groups
WHERE jid = ?
AND agent_type NOT IN (${placeholders})`,
)
.run(chatJid, ...desiredTypes);
}
export function collectRoomRegistrationSnapshot(
database: Database,
jid: string,
existingStored?: Pick<
StoredRoomSettings,
'modeSource' | 'ownerAgentType' | 'trigger'
>,
): RoomRegistrationSnapshot | undefined {
const rows = database
.prepare(
`SELECT name, folder, trigger_pattern, requires_trigger, is_main, agent_type, work_dir
FROM registered_groups
WHERE jid = ?
ORDER BY agent_type`,
)
.all(jid) as Array<{
name: string;
folder: string;
trigger_pattern: string;
requires_trigger: number | null;
is_main: number | null;
agent_type: string | null;
work_dir: string | null;
}>;
if (rows.length === 0) return undefined;
const first = rows[0];
const conflicts = new Set<string>();
for (const row of rows.slice(1)) {
if (row.name !== first.name) conflicts.add('name');
if (row.folder !== first.folder) conflicts.add('folder');
if ((row.requires_trigger ?? 1) !== (first.requires_trigger ?? 1)) {
conflicts.add('requires_trigger');
}
if ((row.is_main ?? 0) !== (first.is_main ?? 0)) {
conflicts.add('is_main');
}
if ((row.work_dir ?? null) !== (first.work_dir ?? null)) {
conflicts.add('work_dir');
}
}
if (conflicts.size > 0) {
throw new Error(
`Conflicting room-level registered_groups metadata for ${jid}: ${[
...conflicts,
].join(', ')}`,
);
}
const agentTypes = collectRegisteredAgentTypes(database, jid);
const inferredOwnerAgentType =
inferOwnerAgentTypeFromRegisteredAgentTypes(agentTypes);
const preferExplicitTrigger =
existingStored?.modeSource === 'explicit' && existingStored.trigger;
const preferExplicitOwner =
existingStored?.modeSource === 'explicit' && existingStored.ownerAgentType;
const preferredOwnerAgentType = preferExplicitOwner
? existingStored.ownerAgentType
: undefined;
const preferredOwnerRow = preferredOwnerAgentType
? rows.find(
(row) =>
normalizeStoredAgentType(row.agent_type) === preferredOwnerAgentType,
)
: undefined;
const inferredOwnerRow =
rows.find(
(row) =>
normalizeStoredAgentType(row.agent_type) === inferredOwnerAgentType,
) ?? rows[0];
const ownerAgentType = preferredOwnerAgentType
? preferredOwnerRow
? preferredOwnerAgentType
: preferredOwnerAgentType
: inferredOwnerAgentType;
const ownerRow = preferredOwnerRow ?? inferredOwnerRow;
return {
name: first.name,
folder: first.folder,
triggerPattern: preferExplicitTrigger
? existingStored.trigger!
: preferredOwnerRow != null
? preferredOwnerRow.trigger_pattern
: ownerRow.trigger_pattern,
requiresTrigger: (first.requires_trigger ?? 1) === 1,
isMain: (first.is_main ?? 0) === 1,
ownerAgentType,
workDir: first.work_dir ?? null,
};
}
export function insertStoredRoomSettings(
database: Database,
chatJid: string,
roomMode: RoomMode,
source: RoomModeSource,
snapshot: RoomRegistrationSnapshot,
): void {
database
.prepare(
`INSERT INTO room_settings (
chat_jid,
room_mode,
mode_source,
name,
folder,
trigger_pattern,
requires_trigger,
is_main,
owner_agent_type,
work_dir,
updated_at
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
)
.run(
chatJid,
roomMode,
source,
snapshot.name,
snapshot.folder,
snapshot.triggerPattern,
snapshot.requiresTrigger ? 1 : 0,
snapshot.isMain ? 1 : 0,
snapshot.ownerAgentType,
snapshot.workDir,
new Date().toISOString(),
);
}
export function updateStoredRoomMetadata(
database: Database,
chatJid: string,
snapshot: RoomRegistrationSnapshot,
): void {
database
.prepare(
`UPDATE room_settings
SET name = ?,
folder = ?,
trigger_pattern = ?,
requires_trigger = ?,
is_main = ?,
owner_agent_type = ?,
work_dir = ?,
updated_at = ?
WHERE chat_jid = ?`,
)
.run(
snapshot.name,
snapshot.folder,
snapshot.triggerPattern,
snapshot.requiresTrigger ? 1 : 0,
snapshot.isMain ? 1 : 0,
snapshot.ownerAgentType,
snapshot.workDir,
new Date().toISOString(),
chatJid,
);
}

87
src/db/router-state.ts Normal file
View File

@@ -0,0 +1,87 @@
import { Database } from 'bun:sqlite';
import { normalizeServiceId } from '../config.js';
import { AgentType } from '../types.js';
export function getRouterStateFromDatabase(
database: Database,
key: string,
currentServiceId: string,
): string | undefined {
const row = database
.prepare('SELECT value FROM router_state WHERE key = ?')
.get(key) as { value: string } | undefined;
if (row) return row.value;
const prefixedKey = `${normalizeServiceId(currentServiceId)}:${key}`;
const prefixedRow = database
.prepare('SELECT value FROM router_state WHERE key = ?')
.get(prefixedKey) as { value: string } | undefined;
if (!prefixedRow) return undefined;
database
.prepare('INSERT OR REPLACE INTO router_state (key, value) VALUES (?, ?)')
.run(key, prefixedRow.value);
database.prepare('DELETE FROM router_state WHERE key = ?').run(prefixedKey);
return prefixedRow.value;
}
export function getRouterStateForServiceFromDatabase(
database: Database,
key: string,
serviceId: string,
): string | undefined {
const prefixedKey = `${normalizeServiceId(serviceId)}:${key}`;
const row = database
.prepare('SELECT value FROM router_state WHERE key = ?')
.get(prefixedKey) as { value: string } | undefined;
if (row) return row.value;
const canonical = database
.prepare('SELECT value FROM router_state WHERE key = ?')
.get(key) as { value: string } | undefined;
return canonical?.value;
}
export function setRouterStateInDatabase(
database: Database,
key: string,
value: string,
): void {
database
.prepare('INSERT OR REPLACE INTO router_state (key, value) VALUES (?, ?)')
.run(key, value);
}
export function setRouterStateForServiceInDatabase(
database: Database,
key: string,
value: string,
serviceId: string,
): void {
const prefixedKey = `${normalizeServiceId(serviceId)}:${key}`;
database
.prepare('INSERT OR REPLACE INTO router_state (key, value) VALUES (?, ?)')
.run(prefixedKey, value);
}
export function getLastRespondingAgentTypeFromDatabase(
database: Database,
chatJid: string,
): AgentType | undefined {
const row = database
.prepare(
`SELECT sender FROM messages
WHERE chat_jid = ? AND is_bot_message = 1
ORDER BY timestamp DESC, seq DESC
LIMIT 1`,
)
.get(chatJid) as { sender: string } | undefined;
if (!row) return undefined;
const sender = row.sender.toLowerCase();
if (sender.includes('claude')) return 'claude-code';
if (sender.includes('codex')) return 'codex';
return undefined;
}

542
src/db/schema.ts Normal file
View File

@@ -0,0 +1,542 @@
import { Database } from 'bun:sqlite';
import { inferAgentTypeFromServiceShadow } from '../role-service-shadow.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<{
name: string;
}>
).map((column) => column.name);
}
export function tableHasColumn(
database: Database,
tableName: string,
columnName: string,
): boolean {
return getTableColumns(database, tableName).includes(columnName);
}
function tryExecMigration(database: Database, sql: string): void {
try {
database.exec(sql);
} catch {
/* column already exists */
}
}
export function applySchemaMigrations(
database: Database,
args: {
assistantName: string;
hooks: SchemaMigrationHooks;
},
): void {
const { assistantName, hooks } = args;
tryExecMigration(
database,
`ALTER TABLE scheduled_tasks ADD COLUMN context_mode TEXT DEFAULT 'isolated'`,
);
tryExecMigration(
database,
`ALTER TABLE scheduled_tasks ADD COLUMN agent_type TEXT`,
);
tryExecMigration(
database,
`ALTER TABLE scheduled_tasks ADD COLUMN ci_provider TEXT`,
);
tryExecMigration(
database,
`ALTER TABLE scheduled_tasks ADD COLUMN ci_metadata TEXT`,
);
tryExecMigration(
database,
`ALTER TABLE scheduled_tasks ADD COLUMN max_duration_ms INTEGER`,
);
tryExecMigration(
database,
`ALTER TABLE scheduled_tasks ADD COLUMN status_message_id TEXT`,
);
tryExecMigration(
database,
`ALTER TABLE scheduled_tasks ADD COLUMN status_started_at TEXT`,
);
tryExecMigration(
database,
`ALTER TABLE scheduled_tasks ADD COLUMN suspended_until TEXT`,
);
tryExecMigration(
database,
`ALTER TABLE room_settings ADD COLUMN mode_source TEXT NOT NULL DEFAULT 'explicit'`,
);
tryExecMigration(database, `ALTER TABLE room_settings ADD COLUMN name TEXT`);
tryExecMigration(database, `ALTER TABLE room_settings ADD COLUMN folder TEXT`);
tryExecMigration(
database,
`ALTER TABLE room_settings ADD COLUMN trigger_pattern TEXT`,
);
tryExecMigration(
database,
`ALTER TABLE room_settings ADD COLUMN requires_trigger INTEGER DEFAULT 1`,
);
tryExecMigration(
database,
`ALTER TABLE room_settings ADD COLUMN is_main INTEGER DEFAULT 0`,
);
tryExecMigration(
database,
`ALTER TABLE room_settings ADD COLUMN owner_agent_type TEXT`,
);
tryExecMigration(
database,
`ALTER TABLE room_settings ADD COLUMN work_dir TEXT`,
);
tryExecMigration(
database,
`ALTER TABLE service_handoffs ADD COLUMN intended_role TEXT`,
);
tryExecMigration(
database,
`ALTER TABLE service_handoffs ADD COLUMN source_role TEXT`,
);
tryExecMigration(
database,
`ALTER TABLE service_handoffs ADD COLUMN target_role TEXT`,
);
tryExecMigration(
database,
`ALTER TABLE service_handoffs ADD COLUMN source_agent_type TEXT`,
);
tryExecMigration(
database,
`ALTER TABLE paired_tasks ADD COLUMN owner_agent_type TEXT`,
);
tryExecMigration(
database,
`ALTER TABLE paired_tasks ADD COLUMN reviewer_agent_type TEXT`,
);
tryExecMigration(
database,
`ALTER TABLE paired_tasks ADD COLUMN arbiter_agent_type TEXT`,
);
tryExecMigration(
database,
`ALTER TABLE work_items ADD COLUMN delivery_role TEXT`,
);
database.exec(
`UPDATE service_handoffs
SET target_role = COALESCE(
target_role,
intended_role,
CASE
WHEN reason LIKE 'reviewer-%' THEN 'reviewer'
WHEN reason LIKE 'arbiter-%' THEN 'arbiter'
WHEN reason IS NOT NULL THEN 'owner'
ELSE NULL
END
)
WHERE target_role IS NULL`,
);
database.exec(
`UPDATE service_handoffs
SET source_role = COALESCE(source_role, target_role, intended_role)
WHERE source_role IS NULL`,
);
database.exec(
`UPDATE room_settings
SET mode_source = 'explicit'
WHERE COALESCE(mode_source, '') NOT IN ('explicit', 'inferred')`,
);
database.exec(`
UPDATE scheduled_tasks
SET agent_type = COALESCE(
(
SELECT CASE WHEN COUNT(*) = 1 THEN MIN(agent_type) ELSE NULL END
FROM registered_groups
WHERE jid = scheduled_tasks.chat_jid
AND folder = scheduled_tasks.group_folder
),
(
SELECT CASE WHEN COUNT(*) = 1 THEN MIN(agent_type) ELSE NULL END
FROM registered_groups
WHERE jid = scheduled_tasks.chat_jid
),
(
SELECT CASE WHEN COUNT(*) = 1 THEN MIN(agent_type) ELSE NULL END
FROM registered_groups
WHERE folder = scheduled_tasks.group_folder
)
)
WHERE agent_type IS NULL;
`);
try {
database.exec(
`ALTER TABLE messages ADD COLUMN is_bot_message INTEGER DEFAULT 0`,
);
database
.prepare(`UPDATE messages SET is_bot_message = 1 WHERE content LIKE ?`)
.run(`${assistantName}:%`);
} catch {
/* column already exists */
}
tryExecMigration(database, `ALTER TABLE messages ADD COLUMN seq INTEGER`);
hooks.backfillMessageSeq(database);
database.exec(`
CREATE UNIQUE INDEX IF NOT EXISTS idx_messages_seq ON messages(seq);
CREATE INDEX IF NOT EXISTS idx_messages_chat_jid_seq ON messages(chat_jid, seq);
`);
database.exec(`DROP INDEX IF EXISTS idx_work_items_group_agent;`);
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);
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');
`);
const registeredGroupsSql = (
database
.prepare(
`SELECT sql FROM sqlite_master WHERE type = 'table' AND name = 'registered_groups'`,
)
.get() as { sql?: string } | undefined
)?.sql;
if (
registeredGroupsSql &&
!registeredGroupsSql.includes('PRIMARY KEY (jid, agent_type)')
) {
const registeredGroupCols = database
.prepare('PRAGMA table_info(registered_groups)')
.all() as Array<{ name: string }>;
const hasIsMain = registeredGroupCols.some((col) => col.name === 'is_main');
const hasAgentType = registeredGroupCols.some(
(col) => col.name === 'agent_type',
);
const hasWorkDir = registeredGroupCols.some(
(col) => col.name === 'work_dir',
);
const hasAgentConfig = registeredGroupCols.some(
(col) => col.name === 'agent_config',
);
const hasContainerConfig = registeredGroupCols.some(
(col) => col.name === 'container_config',
);
database.exec(`
CREATE TABLE registered_groups_new (
jid TEXT NOT NULL,
name TEXT NOT NULL,
folder TEXT NOT NULL,
trigger_pattern TEXT NOT NULL,
added_at TEXT NOT NULL,
agent_config TEXT,
requires_trigger INTEGER DEFAULT 1,
is_main INTEGER DEFAULT 0,
agent_type TEXT NOT NULL DEFAULT 'claude-code',
work_dir TEXT,
PRIMARY KEY (jid, agent_type),
UNIQUE (folder, agent_type)
);
`);
database.exec(`
INSERT INTO registered_groups_new (
jid,
name,
folder,
trigger_pattern,
added_at,
agent_config,
requires_trigger,
is_main,
agent_type,
work_dir
)
SELECT
jid,
name,
folder,
trigger_pattern,
added_at,
${
hasAgentConfig
? 'agent_config'
: hasContainerConfig
? 'container_config'
: 'NULL'
},
requires_trigger,
${hasIsMain ? 'COALESCE(is_main, 0)' : "CASE WHEN folder = 'main' THEN 1 ELSE 0 END"},
${hasAgentType ? "COALESCE(agent_type, 'claude-code')" : "'claude-code'"},
${hasWorkDir ? 'work_dir' : 'NULL'}
FROM registered_groups;
`);
database.exec(`
DROP TABLE registered_groups;
ALTER TABLE registered_groups_new RENAME TO registered_groups;
`);
} else {
database.exec(
`UPDATE registered_groups SET is_main = 1 WHERE folder = 'main' AND COALESCE(is_main, 0) = 0`,
);
}
const registeredGroupCols = database
.prepare('PRAGMA table_info(registered_groups)')
.all() as Array<{ name: string }>;
const hasAgentConfig = registeredGroupCols.some(
(col) => col.name === 'agent_config',
);
const hasContainerConfig = registeredGroupCols.some(
(col) => col.name === 'container_config',
);
if (!hasAgentConfig) {
database.exec(`ALTER TABLE registered_groups ADD COLUMN agent_config TEXT`);
}
if (hasContainerConfig) {
database.exec(
`UPDATE registered_groups
SET agent_config = COALESCE(agent_config, container_config)
WHERE container_config IS NOT NULL`,
);
}
const pairedTasksSqlRow = database
.prepare(
`SELECT sql FROM sqlite_master WHERE type = 'table' AND name = 'paired_tasks'`,
)
.get() as { sql?: string } | undefined;
const pairedTasksSql = pairedTasksSqlRow?.sql || '';
const pairedTasksNeedsRebuild =
pairedTasksSql &&
(pairedTasksSql.includes('task_policy') ||
!pairedTasksSql.includes('round_trip_count'));
if (pairedTasksNeedsRebuild) {
database.exec(`DROP TABLE IF EXISTS paired_tasks`);
database.exec(`
CREATE TABLE IF NOT EXISTS paired_tasks (
id TEXT PRIMARY KEY,
chat_jid TEXT NOT NULL,
group_folder TEXT NOT NULL,
owner_agent_type TEXT,
reviewer_agent_type TEXT,
arbiter_agent_type TEXT,
title TEXT,
source_ref TEXT,
plan_notes TEXT,
review_requested_at TEXT,
round_trip_count INTEGER NOT NULL DEFAULT 0,
status TEXT NOT NULL DEFAULT 'active',
arbiter_verdict TEXT,
arbiter_requested_at TEXT,
completion_reason TEXT,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL,
CHECK (status IN ('active', 'review_ready', 'in_review', 'merge_ready', 'completed', 'arbiter_requested', 'in_arbitration')),
CHECK (owner_agent_type IN ('claude-code', 'codex') OR owner_agent_type IS NULL),
CHECK (reviewer_agent_type IN ('claude-code', 'codex') OR reviewer_agent_type IS NULL),
CHECK (arbiter_agent_type IN ('claude-code', 'codex') OR arbiter_agent_type IS NULL)
);
CREATE INDEX IF NOT EXISTS idx_paired_tasks_chat_status
ON paired_tasks(chat_jid, status, updated_at);
`);
}
{
const ptSqlRow = database
.prepare(
`SELECT sql FROM sqlite_master WHERE type = 'table' AND name = 'paired_tasks'`,
)
.get() as { sql?: string } | undefined;
const ptSql = ptSqlRow?.sql || '';
if (ptSql && !ptSql.includes('arbiter_requested')) {
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, 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, 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);
`);
}
}
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`,
);
for (const table of [
'paired_executions',
'paired_approvals',
'paired_artifacts',
'paired_events',
]) {
database.exec(`DROP TABLE IF EXISTS ${table}`);
}
const pairedWsSqlRow = database
.prepare(
`SELECT sql FROM sqlite_master WHERE type = 'table' AND name = 'paired_workspaces'`,
)
.get() as { sql?: string } | undefined;
const pairedWsSql = pairedWsSqlRow?.sql || '';
if (pairedWsSql && pairedWsSql.includes('snapshot_source_fingerprint')) {
database.exec(`DROP TABLE IF EXISTS paired_workspaces`);
database.exec(`
CREATE TABLE IF NOT EXISTS paired_workspaces (
id TEXT PRIMARY KEY,
task_id TEXT NOT NULL,
role TEXT NOT NULL,
workspace_dir TEXT NOT NULL,
snapshot_source_dir TEXT,
snapshot_ref TEXT,
status TEXT NOT NULL DEFAULT 'ready',
snapshot_refreshed_at TEXT,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL,
CHECK (role IN ('owner', 'reviewer')),
CHECK (status IN ('ready', 'stale'))
);
CREATE UNIQUE INDEX IF NOT EXISTS idx_paired_workspaces_task_role
ON paired_workspaces(task_id, role);
`);
}
const pairedProjSqlRow = database
.prepare(
`SELECT sql FROM sqlite_master WHERE type = 'table' AND name = 'paired_projects'`,
)
.get() as { sql?: string } | undefined;
const pairedProjSql = pairedProjSqlRow?.sql || '';
if (pairedProjSql && pairedProjSql.includes('workspace_topology')) {
database.exec(`DROP TABLE IF EXISTS paired_projects`);
database.exec(`
CREATE TABLE IF NOT EXISTS paired_projects (
chat_jid TEXT PRIMARY KEY,
group_folder TEXT NOT NULL,
canonical_work_dir TEXT NOT NULL,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL
);
`);
}
migrateSessionsTableToCompositePk(database, 'claude-code');
try {
database.exec(`ALTER TABLE chats ADD COLUMN channel TEXT`);
database.exec(`ALTER TABLE chats ADD COLUMN is_group INTEGER DEFAULT 0`);
database.exec(
`UPDATE chats SET channel = 'whatsapp', is_group = 1 WHERE jid LIKE '%@g.us'`,
);
database.exec(
`UPDATE chats SET channel = 'whatsapp', is_group = 0 WHERE jid LIKE '%@s.whatsapp.net'`,
);
database.exec(
`UPDATE chats SET channel = 'discord', is_group = 1 WHERE jid LIKE 'dc:%'`,
);
database.exec(
`UPDATE chats SET channel = 'telegram', is_group = 1 WHERE jid LIKE 'tg:%'`,
);
} catch {
/* 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);
}

163
src/db/sessions.ts Normal file
View File

@@ -0,0 +1,163 @@
import { Database } from 'bun:sqlite';
import { AgentType } from '../types.js';
export type ServiceShadowAgentTypeResolver = (
serviceId: string,
) => AgentType | undefined;
function hasLegacyServiceSessionsTable(database: Database): boolean {
return Boolean(
database
.prepare(
`SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = 'service_sessions'`,
)
.get(),
);
}
export function migrateSessionsTableToCompositePk(
database: Database,
defaultAgentType: AgentType,
): void {
const sessionCols = database
.prepare('PRAGMA table_info(sessions)')
.all() as Array<{ name: string }>;
if (sessionCols.some((col) => col.name === 'agent_type')) {
return;
}
database.exec(`
CREATE TABLE sessions_new (
group_folder TEXT NOT NULL,
agent_type TEXT NOT NULL DEFAULT '${defaultAgentType}',
session_id TEXT NOT NULL,
PRIMARY KEY (group_folder, agent_type)
);
`);
database
.prepare(
`INSERT INTO sessions_new (group_folder, agent_type, session_id)
SELECT group_folder, ?, session_id FROM sessions`,
)
.run(defaultAgentType);
database.exec(`
DROP TABLE sessions;
ALTER TABLE sessions_new RENAME TO sessions;
`);
}
export function getSessionFromDatabase(
database: Database,
groupFolder: string,
agentType: string,
): string | undefined {
const row = database
.prepare(
'SELECT session_id FROM sessions WHERE group_folder = ? AND agent_type = ?',
)
.get(groupFolder, agentType) as { session_id: string } | undefined;
return row?.session_id;
}
export function setSessionInDatabase(
database: Database,
groupFolder: string,
agentType: string,
sessionId: string,
): void {
database
.prepare(
'INSERT OR REPLACE INTO sessions (group_folder, agent_type, session_id) VALUES (?, ?, ?)',
)
.run(groupFolder, agentType, sessionId);
}
export function deleteSessionFromDatabase(
database: Database,
groupFolder: string,
agentType: string,
): void {
database
.prepare('DELETE FROM sessions WHERE group_folder = ? AND agent_type = ?')
.run(groupFolder, agentType);
}
export function deleteAllSessionsForGroupFromDatabase(
database: Database,
groupFolder: string,
): void {
database.prepare('DELETE FROM sessions WHERE group_folder = ?').run(groupFolder);
}
export function getAllSessionsForAgentTypeFromDatabase(
database: Database,
agentType: string,
): Record<string, string> {
const rows = database
.prepare(
'SELECT group_folder, session_id FROM sessions WHERE agent_type = ?',
)
.all(agentType) as Array<{
group_folder: string;
session_id: string;
}>;
const result: Record<string, string> = {};
for (const row of rows) {
result[row.group_folder] = row.session_id;
}
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`);
}

289
src/db/tasks.ts Normal file
View File

@@ -0,0 +1,289 @@
import { Database } from 'bun:sqlite';
import { AgentType, ScheduledTask, TaskRunLog } from '../types.js';
export type CreateScheduledTaskInput = Omit<
ScheduledTask,
| 'last_run'
| 'last_result'
| 'agent_type'
| 'ci_provider'
| 'ci_metadata'
| 'max_duration_ms'
| 'status_message_id'
| 'status_started_at'
> & {
agent_type?: AgentType | null;
ci_provider?: ScheduledTask['ci_provider'];
ci_metadata?: string | null;
max_duration_ms?: number | null;
status_message_id?: string | null;
status_started_at?: string | null;
};
export type ScheduledTaskUpdates = Partial<
Pick<
ScheduledTask,
| 'prompt'
| 'schedule_type'
| 'schedule_value'
| 'next_run'
| 'status'
| 'suspended_until'
| 'ci_metadata'
>
>;
export type ScheduledTaskStatusTrackingUpdates = Partial<
Pick<ScheduledTask, 'status_message_id' | 'status_started_at'>
>;
export function createTaskInDatabase(
database: Database,
task: CreateScheduledTaskInput,
): void {
database
.prepare(
`
INSERT INTO scheduled_tasks (id, group_folder, chat_jid, agent_type, ci_provider, ci_metadata, max_duration_ms, status_message_id, status_started_at, prompt, schedule_type, schedule_value, context_mode, next_run, status, created_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
`,
)
.run(
task.id,
task.group_folder,
task.chat_jid,
task.agent_type || 'claude-code',
task.ci_provider ?? null,
task.ci_metadata ?? null,
task.max_duration_ms ?? null,
task.status_message_id || null,
task.status_started_at || null,
task.prompt,
task.schedule_type,
task.schedule_value,
task.context_mode || 'isolated',
task.next_run,
task.status,
task.created_at,
);
}
export function getTaskByIdFromDatabase(
database: Database,
id: string,
): ScheduledTask | undefined {
return database.prepare('SELECT * FROM scheduled_tasks WHERE id = ?').get(id) as
| ScheduledTask
| undefined;
}
export function findDuplicateCiWatcherInDatabase(
database: Database,
chatJid: string,
ciProvider: string,
ciMetadata: string,
): ScheduledTask | undefined {
return database
.prepare(
`SELECT * FROM scheduled_tasks
WHERE chat_jid = ? AND ci_provider = ? AND ci_metadata = ?
AND status IN ('active', 'paused')
LIMIT 1`,
)
.get(chatJid, ciProvider, ciMetadata) as ScheduledTask | undefined;
}
export function getTasksForGroupFromDatabase(
database: Database,
groupFolder: string,
agentType?: AgentType,
): ScheduledTask[] {
if (agentType) {
return database
.prepare(
'SELECT * FROM scheduled_tasks WHERE group_folder = ? AND agent_type = ? ORDER BY created_at DESC',
)
.all(groupFolder, agentType) as ScheduledTask[];
}
return database
.prepare(
'SELECT * FROM scheduled_tasks WHERE group_folder = ? ORDER BY created_at DESC',
)
.all(groupFolder) as ScheduledTask[];
}
export function getAllTasksFromDatabase(
database: Database,
agentType?: AgentType,
): ScheduledTask[] {
if (agentType) {
return database
.prepare(
'SELECT * FROM scheduled_tasks WHERE agent_type = ? ORDER BY created_at DESC',
)
.all(agentType) as ScheduledTask[];
}
return database
.prepare('SELECT * FROM scheduled_tasks ORDER BY created_at DESC')
.all() as ScheduledTask[];
}
export function updateTaskInDatabase(
database: Database,
id: string,
updates: ScheduledTaskUpdates,
): void {
const fields: string[] = [];
const values: (string | number | null)[] = [];
if (updates.prompt !== undefined) {
fields.push('prompt = ?');
values.push(updates.prompt);
}
if (updates.schedule_type !== undefined) {
fields.push('schedule_type = ?');
values.push(updates.schedule_type);
}
if (updates.schedule_value !== undefined) {
fields.push('schedule_value = ?');
values.push(updates.schedule_value);
}
if (updates.next_run !== undefined) {
fields.push('next_run = ?');
values.push(updates.next_run);
}
if (updates.status !== undefined) {
fields.push('status = ?');
values.push(updates.status);
}
if (updates.suspended_until !== undefined) {
fields.push('suspended_until = ?');
values.push(updates.suspended_until);
}
if (updates.ci_metadata !== undefined) {
fields.push('ci_metadata = ?');
values.push(updates.ci_metadata);
}
if (fields.length === 0) return;
values.push(id);
database
.prepare(`UPDATE scheduled_tasks SET ${fields.join(', ')} WHERE id = ?`)
.run(...values);
}
export function updateTaskStatusTrackingInDatabase(
database: Database,
id: string,
updates: ScheduledTaskStatusTrackingUpdates,
): void {
const fields: string[] = [];
const values: (string | number | null)[] = [];
if (updates.status_message_id !== undefined) {
fields.push('status_message_id = ?');
values.push(updates.status_message_id);
}
if (updates.status_started_at !== undefined) {
fields.push('status_started_at = ?');
values.push(updates.status_started_at);
}
if (fields.length === 0) return;
values.push(id);
database
.prepare(`UPDATE scheduled_tasks SET ${fields.join(', ')} WHERE id = ?`)
.run(...values);
}
export function hasActiveCiWatcherForChatInDatabase(
database: Database,
chatJid: string,
): boolean {
const row = database
.prepare(
`SELECT 1 FROM scheduled_tasks
WHERE chat_jid = ? AND status = 'active' AND prompt LIKE '[BACKGROUND CI WATCH]%'
LIMIT 1`,
)
.get(chatJid);
return !!row;
}
export function getDueTasksFromDatabase(database: Database): ScheduledTask[] {
const now = new Date().toISOString();
return database
.prepare(
`
SELECT * FROM scheduled_tasks
WHERE status = 'active' AND next_run IS NOT NULL AND next_run <= ?
AND (suspended_until IS NULL OR suspended_until <= ?)
ORDER BY next_run
`,
)
.all(now, now) as ScheduledTask[];
}
export function updateTaskAfterRunInDatabase(
database: Database,
id: string,
nextRun: string | null,
lastResult: string,
): void {
const now = new Date().toISOString();
database
.prepare(
`
UPDATE scheduled_tasks
SET next_run = ?, last_run = ?, last_result = ?, status = CASE WHEN ? IS NULL THEN 'completed' ELSE status END
WHERE id = ?
`,
)
.run(nextRun, now, lastResult, nextRun, id);
}
export function logTaskRunInDatabase(
database: Database,
log: TaskRunLog,
): void {
database
.prepare(
`
INSERT INTO task_run_logs (task_id, run_at, duration_ms, status, result, error)
VALUES (?, ?, ?, ?, ?, ?)
`,
)
.run(
log.task_id,
log.run_at,
log.duration_ms,
log.status,
log.result,
log.error,
);
}
export function getRecentConsecutiveErrorsFromDatabase(
database: Database,
taskId: string,
limit: number = 5,
): string[] {
const rows = database
.prepare(
`SELECT status, error FROM task_run_logs
WHERE task_id = ? ORDER BY run_at DESC LIMIT ?`,
)
.all(taskId, limit) as Array<{ status: string; error: string | null }>;
const errors: string[] = [];
for (const row of rows) {
if (row.status !== 'error' || !row.error) break;
errors.push(row.error);
}
return errors;
}

182
src/db/work-items.ts Normal file
View File

@@ -0,0 +1,182 @@
import { Database } from 'bun:sqlite';
import { SERVICE_SESSION_SCOPE } from '../config.js';
import { inferRoleFromServiceShadow } from '../role-service-shadow.js';
import { AgentType, PairedRoomRole } from '../types.js';
import { resolveWorkItemServiceShadow } from './legacy-rebuilds.js';
export interface WorkItem {
id: number;
group_folder: string;
chat_jid: string;
agent_type: AgentType;
service_id: string;
delivery_role?: PairedRoomRole | null;
status: 'produced' | 'delivery_retry' | 'delivered';
start_seq: number | null;
end_seq: number | null;
result_payload: string;
delivery_attempts: number;
delivery_message_id: string | null;
last_error: string | null;
created_at: string;
updated_at: string;
delivered_at: string | null;
}
interface StoredWorkItemRow extends Omit<
WorkItem,
'service_id' | 'agent_type'
> {
agent_type: string;
}
export interface CreateProducedWorkItemInput {
group_folder: string;
chat_jid: string;
agent_type?: AgentType;
delivery_role?: PairedRoomRole | null;
start_seq: number | null;
end_seq: number | null;
result_payload: string;
}
function normalizeStoredAgentType(
agentType: string | null | undefined,
): AgentType | undefined {
if (agentType === 'claude-code' || agentType === 'codex') return agentType;
return undefined;
}
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),
};
}
export function getOpenWorkItemFromDatabase(
database: Database,
chatJid: string,
agentType: AgentType = 'claude-code',
serviceId: string = SERVICE_SESSION_SCOPE,
): WorkItem | undefined {
const preferredRole = inferRoleFromServiceShadow(agentType, serviceId);
const row = database
.prepare(
`SELECT *
FROM work_items
WHERE chat_jid = ? AND agent_type = ?
AND status IN ('produced', 'delivery_retry')
ORDER BY
CASE
WHEN ? IS NOT NULL AND delivery_role = ? THEN 0
WHEN delivery_role IS NULL THEN 1
ELSE 2
END,
id ASC
LIMIT 1`,
)
.get(chatJid, agentType, preferredRole, preferredRole) as
| StoredWorkItemRow
| undefined;
return row ? hydrateWorkItemRow(row) : undefined;
}
export function getOpenWorkItemForChatFromDatabase(
database: Database,
chatJid: string,
): WorkItem | undefined {
const row = database
.prepare(
`SELECT *
FROM work_items
WHERE chat_jid = ?
AND status IN ('produced', 'delivery_retry')
ORDER BY id ASC
LIMIT 1`,
)
.get(chatJid) as StoredWorkItemRow | undefined;
return row ? hydrateWorkItemRow(row) : undefined;
}
export function createProducedWorkItemInDatabase(
database: Database,
input: CreateProducedWorkItemInput,
): WorkItem {
const now = new Date().toISOString();
const agentType = input.agent_type || 'claude-code';
database
.prepare(
`INSERT INTO work_items (
group_folder,
chat_jid,
agent_type,
delivery_role,
status,
start_seq,
end_seq,
result_payload,
delivery_attempts,
created_at,
updated_at
) VALUES (?, ?, ?, ?, 'produced', ?, ?, ?, 0, ?, ?)`,
)
.run(
input.group_folder,
input.chat_jid,
agentType,
input.delivery_role ?? null,
input.start_seq,
input.end_seq,
input.result_payload,
now,
now,
);
const lastId = (
database.prepare('SELECT last_insert_rowid() as id').get() as { id: number }
).id;
return hydrateWorkItemRow(
database.prepare('SELECT * FROM work_items WHERE id = ?').get(lastId) as
| StoredWorkItemRow,
);
}
export function markWorkItemDeliveredInDatabase(
database: Database,
id: number,
deliveryMessageId?: string | null,
): void {
const now = new Date().toISOString();
database
.prepare(
`UPDATE work_items
SET status = 'delivered',
delivered_at = ?,
delivery_message_id = ?,
updated_at = ?
WHERE id = ?`,
)
.run(now, deliveryMessageId || null, now, id);
}
export function markWorkItemDeliveryRetryInDatabase(
database: Database,
id: number,
error: string,
): void {
const now = new Date().toISOString();
database
.prepare(
`UPDATE work_items
SET status = 'delivery_retry',
delivery_attempts = delivery_attempts + 1,
last_error = ?,
updated_at = ?
WHERE id = ?`,
)
.run(error, now, id);
}

View File

@@ -1,6 +1,5 @@
import fs from 'fs'; import fs from 'fs';
import path from 'path'; import path from 'path';
import { logger } from './logger.js';
// ── Internal cache ────────────────────────────────────────────── // ── Internal cache ──────────────────────────────────────────────
@@ -12,8 +11,7 @@ function parseEnvFile(): Record<string, string> {
let content: string; let content: string;
try { try {
content = fs.readFileSync(envFile, 'utf-8'); content = fs.readFileSync(envFile, 'utf-8');
} catch (err) { } catch {
logger.debug({ err }, '.env file not found, using defaults');
return {}; return {};
} }
@@ -31,7 +29,7 @@ function parseEnvFile(): Record<string, string> {
) { ) {
value = value.slice(1, -1); value = value.slice(1, -1);
} }
if (value) result[key] = value; result[key] = value;
} }
return result; return result;
} }
@@ -49,13 +47,13 @@ export function loadEnvFile(): void {
*/ */
export function getEnv(key: string): string | undefined { export function getEnv(key: string): string | undefined {
if (!_cache) loadEnvFile(); if (!_cache) loadEnvFile();
return process.env[key] || _cache![key] || undefined; if (Object.prototype.hasOwnProperty.call(process.env, key)) {
} return process.env[key];
}
/** Force-reload the .env file (e.g. after token refresh writes new values). */ if (Object.prototype.hasOwnProperty.call(_cache!, key)) {
export function reloadEnvFile(): void { return _cache![key];
_cache = null; }
loadEnvFile(); return undefined;
} }
/** /**

View File

@@ -6,7 +6,10 @@ import { GroupQueue, type GroupRunContext } from './group-queue.js';
// Mock config to control concurrency limit // Mock config to control concurrency limit
vi.mock('./config.js', () => ({ vi.mock('./config.js', () => ({
ASSISTANT_NAME: 'Andy',
DATA_DIR: '/tmp/ejclaw-test-data', DATA_DIR: '/tmp/ejclaw-test-data',
IS_TEST_ENV: true,
LOG_LEVEL: 'info',
MAX_CONCURRENT_AGENTS: 2, MAX_CONCURRENT_AGENTS: 2,
})); }));

View File

@@ -1,8 +1,7 @@
import pino, { type Logger } from 'pino'; import pino, { type Logger } from 'pino';
import { ASSISTANT_NAME, IS_TEST_ENV, LOG_LEVEL } from './config.js';
const serviceName = (process.env.ASSISTANT_NAME || 'claude').toLowerCase(); const serviceName = ASSISTANT_NAME.toLowerCase();
const isTestEnv =
process.env.VITEST === 'true' || process.env.NODE_ENV === 'test';
type LoggerGlobalState = typeof globalThis & { type LoggerGlobalState = typeof globalThis & {
__ejclawLogger?: Logger; __ejclawLogger?: Logger;
@@ -13,11 +12,11 @@ const globalState = globalThis as LoggerGlobalState;
function createRootLogger(): Logger { function createRootLogger(): Logger {
const baseOptions = { const baseOptions = {
level: process.env.LOG_LEVEL || 'info', level: LOG_LEVEL,
name: serviceName, name: serviceName,
}; };
if (isTestEnv) { if (IS_TEST_ENV) {
return pino(baseOptions); return pino(baseOptions);
} }

View File

@@ -1,4 +1,14 @@
import { describe, expect, it } from 'vitest'; import { describe, expect, it, vi } from 'vitest';
vi.mock('./config.js', async () => {
const actual =
await vi.importActual<typeof import('./config.js')>('./config.js');
return {
...actual,
REVIEWER_AGENT_TYPE: actual.REVIEWER_AGENT_TYPE ?? 'claude-code',
ARBITER_AGENT_TYPE: undefined,
};
});
import { import {
resolveFollowUpDispatch, resolveFollowUpDispatch,
@@ -228,6 +238,46 @@ describe('message-runtime-rules', () => {
expect(resolution.effectiveServiceId).toBe(resolution.reviewerServiceId); expect(resolution.effectiveServiceId).toBe(resolution.reviewerServiceId);
}); });
it('resolves merge_ready execution target back to the owner/finalize path', () => {
const resolution = resolveExecutionTarget({
lease: baseLease,
pairedTaskStatus: 'merge_ready',
groupFolder: 'group-1',
groupAgentType: 'claude-code',
});
expect(resolution).toMatchObject({
inferredRole: 'owner',
activeRole: 'owner',
configuredAgentType: 'claude-code',
effectiveAgentType: 'claude-code',
sessionFolder: 'group-1',
});
expect(resolution.effectiveServiceId).toBe(
resolveLeaseServiceId(baseLease, 'owner'),
);
});
it('resolves arbiter execution target from arbiter_requested task status', () => {
const resolution = resolveExecutionTarget({
lease: baseLease,
pairedTaskStatus: 'arbiter_requested',
groupFolder: 'group-1',
groupAgentType: 'claude-code',
});
expect(resolution).toMatchObject({
inferredRole: 'arbiter',
activeRole: 'arbiter',
configuredAgentType: 'claude-code',
effectiveAgentType: 'claude-code',
sessionFolder: 'group-1:arbiter',
});
expect(resolution.effectiveServiceId).toBe(
resolveLeaseServiceId(baseLease, 'arbiter'),
);
});
it('honors an arbiter forced role only when arbiter is configured', () => { it('honors an arbiter forced role only when arbiter is configured', () => {
const resolution = resolveExecutionTarget({ const resolution = resolveExecutionTarget({
lease: { lease: {

View File

@@ -346,6 +346,108 @@ describe('task scheduler', () => {
expect(enqueueTask.mock.calls[0][1]).toBe('task-single-tick'); expect(enqueueTask.mock.calls[0][1]).toBe('task-single-tick');
}); });
it('marks one-off tasks completed after a successful scheduler run', async () => {
const dueAt = new Date(Date.now() - 60_000).toISOString();
createTask({
id: 'task-once-success-finalize',
group_folder: 'shared-group',
chat_jid: 'shared@g.us',
agent_type: 'codex',
prompt: 'finalize once task',
schedule_type: 'once',
schedule_value: dueAt,
context_mode: 'isolated',
next_run: dueAt,
status: 'active',
created_at: '2026-02-22T00:00:00.000Z',
});
let execution: Promise<void> | null = null;
const enqueueTask = vi.fn(
(_groupJid: string, _taskId: string, fn: () => Promise<void>) => {
execution = fn();
return execution;
},
);
await runSchedulerTickOnce({
serviceAgentType: 'codex',
registeredGroups: () => ({
'shared@g.us': {
name: 'Shared',
folder: 'shared-group',
trigger: '@Codex',
added_at: '2026-02-22T00:00:00.000Z',
agentType: 'codex',
},
}),
getSessions: () => ({}),
queue: { enqueueTask } as any,
onProcess: () => {},
sendMessage: async () => {},
});
await execution;
const task = getTaskById('task-once-success-finalize');
expect(task?.status).toBe('completed');
expect(task?.next_run).toBeNull();
expect(task?.last_result).toBe('done');
});
it('keeps interval tasks active and reschedules them after an execution error', async () => {
const dueAt = new Date(Date.now() - 60_000).toISOString();
createTask({
id: 'task-interval-error-retry',
group_folder: 'shared-group',
chat_jid: 'shared@g.us',
agent_type: 'codex',
prompt: 'retry interval task',
schedule_type: 'interval',
schedule_value: '60000',
context_mode: 'isolated',
next_run: dueAt,
status: 'active',
created_at: '2026-02-22T00:00:00.000Z',
});
(runAgentProcessMock as any).mockResolvedValueOnce({
status: 'error',
error: 'scheduler boom',
});
let execution: Promise<void> | null = null;
const enqueueTask = vi.fn(
(_groupJid: string, _taskId: string, fn: () => Promise<void>) => {
execution = fn();
return execution;
},
);
await runSchedulerTickOnce({
serviceAgentType: 'codex',
registeredGroups: () => ({
'shared@g.us': {
name: 'Shared',
folder: 'shared-group',
trigger: '@Codex',
added_at: '2026-02-22T00:00:00.000Z',
agentType: 'codex',
},
}),
getSessions: () => ({}),
queue: { enqueueTask } as any,
onProcess: () => {},
sendMessage: async () => {},
});
await execution;
const task = getTaskById('task-interval-error-retry');
expect(task?.status).toBe('active');
expect(task?.last_result).toBe('Error: scheduler boom');
expect(task?.next_run).not.toBe(dueAt);
expect(task?.next_run).not.toBeNull();
});
it('uses the reviewer bot identity for paired-room scheduled output', async () => { it('uses the reviewer bot identity for paired-room scheduled output', async () => {
const dueAt = new Date(Date.now() - 60_000).toISOString(); const dueAt = new Date(Date.now() - 60_000).toISOString();
createTask({ createTask({

View File

@@ -5,6 +5,8 @@ import os from 'os';
import { import {
ARBITER_AGENT_TYPE, ARBITER_AGENT_TYPE,
ARBITER_MODEL_CONFIG, ARBITER_MODEL_CONFIG,
DEFAULT_CLAUDE_MODEL,
DEFAULT_CODEX_MODEL,
OWNER_AGENT_TYPE, OWNER_AGENT_TYPE,
OWNER_MODEL_CONFIG, OWNER_MODEL_CONFIG,
REVIEWER_AGENT_TYPE, REVIEWER_AGENT_TYPE,
@@ -597,15 +599,13 @@ function buildModelConfigSection(): string {
if (!role.agentType && role.label === 'Arbiter') continue; if (!role.agentType && role.label === 'Arbiter') continue;
const type = role.agentType || '—'; const type = role.agentType || '—';
const defaultModel = const defaultModel =
type === 'codex' type === 'codex' ? DEFAULT_CODEX_MODEL : DEFAULT_CLAUDE_MODEL;
? process.env.CODEX_MODEL || 'codex'
: process.env.CLAUDE_MODEL || 'claude';
const model = role.model || defaultModel; const model = role.model || defaultModel;
// Show fallback status for claude-code roles when global failover is active // Show fallback status for claude-code roles when global failover is active
const isFallback = failover.active && type === 'claude-code'; const isFallback = failover.active && type === 'claude-code';
if (isFallback) { if (isFallback) {
const fallbackModel = process.env.CODEX_MODEL || 'codex'; const fallbackModel = DEFAULT_CODEX_MODEL;
lines.push(` **${role.label}** — codex \`${fallbackModel}\` (fallback)`); lines.push(` **${role.label}** — codex \`${fallbackModel}\` (fallback)`);
} else { } else {
lines.push(` **${role.label}** — ${type} \`${model}\``); lines.push(` **${role.label}** — ${type} \`${model}\``);