Merge remote-tracking branch 'origin/main' into codex/owner/ejclaw

# Conflicts:
#	prompts/owner-common-platform.md
#	runners/shared/src/agent-protocol.ts
#	runners/shared/test/agent-protocol.test.ts
#	src/db/bootstrap.test.ts
#	src/db/migrations/index.ts
#	src/message-turn-controller.test.ts
#	src/message-turn-controller.ts
#	src/outbound-attachments.ts
#	src/paired-execution-context.test.ts
This commit is contained in:
ejclaw
2026-04-27 19:51:18 +09:00
80 changed files with 19587 additions and 181 deletions

View File

@@ -56,6 +56,19 @@ function createFakeAccounts(homeDir: string, count: number): void {
}
}
function createDefaultCodexAuth(homeDir: string): string {
const authPath = path.join(homeDir, '.codex', 'auth.json');
fs.mkdirSync(path.dirname(authPath), { recursive: true });
fs.writeFileSync(
authPath,
JSON.stringify({
auth_mode: 'chatgpt',
tokens: { account_id: 'default-acct', access_token: 'default-token' },
}),
);
return authPath;
}
describe('codex-token-rotation d7 ≥ 100% auto-skip', () => {
let tempHome: string;
@@ -142,4 +155,46 @@ describe('codex-token-rotation d7 ≥ 100% auto-skip', () => {
'Failed to persist Codex rotation state',
);
});
it('does not append ~/.codex/auth.json fallback when numbered accounts exist', async () => {
const fallbackAuthPath = createDefaultCodexAuth(tempHome);
const mod = await import('./codex-token-rotation.js');
mod.initCodexTokenRotation();
expect(mod.getCodexAccountCount()).toBe(4);
expect(mod.getActiveCodexAuthPath()).not.toBe(fallbackAuthPath);
});
});
describe('codex-token-rotation single-account fallback', () => {
let tempHome: string;
beforeEach(() => {
vi.resetModules();
tempHome = fs.mkdtempSync(path.join('/tmp', 'ejclaw-codex-fallback-'));
process.env.CODEX_ROT_TEST_HOME = tempHome;
});
afterEach(() => {
delete process.env.CODEX_ROT_TEST_HOME;
fs.rmSync(tempHome, { recursive: true, force: true });
});
it('uses ~/.codex/auth.json fallback when ~/.codex-accounts is absent', async () => {
const fallbackAuthPath = createDefaultCodexAuth(tempHome);
const mod = await import('./codex-token-rotation.js');
mod.initCodexTokenRotation();
expect(mod.getCodexAccountCount()).toBe(1);
expect(mod.getActiveCodexAuthPath()).toBe(fallbackAuthPath);
expect(mod.getAllCodexAccounts()).toEqual([
expect.objectContaining({
index: 0,
accountId: 'default-acct',
isActive: true,
}),
]);
});
});

View File

@@ -78,46 +78,63 @@ let currentIndex = 0;
let initialized = false;
const ACCOUNTS_DIR = path.join(os.homedir(), '.codex-accounts');
const DEFAULT_CODEX_DIR = path.join(os.homedir(), '.codex');
const DEFAULT_AUTH_PATH = path.join(DEFAULT_CODEX_DIR, 'auth.json');
function loadCodexAccount(
authPath: string,
fallbackAccountId: string,
): boolean {
const data = readJsonFile<{
tokens?: { account_id?: string; id_token?: string };
}>(authPath);
if (!data) {
logger.warn({ authPath }, 'Failed to parse codex account auth.json');
return false;
}
const accountId = data?.tokens?.account_id || fallbackAccountId;
const jwt = parseJwtAuth(data?.tokens?.id_token || '');
accounts.push({
index: accounts.length,
authPath,
accountId,
planType: jwt.planType,
subscriptionUntil: jwt.expiresAt,
rateLimitedUntil: null,
});
return true;
}
export function initCodexTokenRotation(): void {
if (initialized) return;
initialized = true;
if (!fs.existsSync(ACCOUNTS_DIR)) {
logger.info(
{ dir: ACCOUNTS_DIR },
'Codex accounts dir not found, skipping',
);
return;
}
const hasAccountsDir = fs.existsSync(ACCOUNTS_DIR);
const dirs = hasAccountsDir
? fs
.readdirSync(ACCOUNTS_DIR)
.filter((d) => /^\d+$/.test(d))
.sort((a, b) => parseInt(a) - parseInt(b))
: [];
const dirs = fs
.readdirSync(ACCOUNTS_DIR)
.filter((d) => /^\d+$/.test(d))
.sort((a, b) => parseInt(a) - parseInt(b));
if (!hasAccountsDir) {
logger.info({ dir: ACCOUNTS_DIR }, 'Codex accounts dir not found');
}
for (const dir of dirs) {
const authPath = path.join(ACCOUNTS_DIR, dir, 'auth.json');
if (!fs.existsSync(authPath)) continue;
loadCodexAccount(authPath, `account-${dir}`);
}
const data = readJsonFile<{
tokens?: { account_id?: string; id_token?: string };
}>(authPath);
if (!data) {
logger.warn({ authPath }, 'Failed to parse codex account auth.json');
continue;
if (dirs.length === 0 && fs.existsSync(DEFAULT_AUTH_PATH)) {
if (loadCodexAccount(DEFAULT_AUTH_PATH, 'default-account')) {
logger.info(
{ authPath: DEFAULT_AUTH_PATH },
'Codex accounts dir absent/empty; using ~/.codex/auth.json fallback',
);
}
const accountId = data?.tokens?.account_id || `account-${dir}`;
const jwt = parseJwtAuth(data?.tokens?.id_token || '');
const planType = jwt.planType;
accounts.push({
index: accounts.length,
authPath,
accountId,
planType,
subscriptionUntil: jwt.expiresAt,
rateLimitedUntil: null,
});
}
if (accounts.length > 1) loadCodexState();
@@ -236,8 +253,14 @@ export function reloadCodexStateFromDisk(): void {
/** Get the auth.json path for the current active account. */
export function getActiveCodexAuthPath(): string | null {
return getCodexAuthPath();
}
export function getCodexAuthPath(
accountIndex: number = currentIndex,
): string | null {
if (accounts.length === 0) return null;
return accounts[currentIndex]?.authPath ?? null;
return accounts[accountIndex]?.authPath ?? null;
}
export function detectCodexRotationTrigger(

View File

@@ -0,0 +1,242 @@
import { EventEmitter } from 'events';
import fs from 'fs';
import path from 'path';
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
vi.mock('child_process', () => ({
spawn: vi.fn(),
}));
vi.mock('./logger.js', () => ({
logger: {
info: vi.fn(),
warn: vi.fn(),
error: vi.fn(),
debug: vi.fn(),
},
}));
vi.mock('./config.js', () => ({
DATA_DIR: '/tmp/ejclaw-codex-usage-data',
}));
vi.mock('./utils.js', async () => {
const actual =
await vi.importActual<typeof import('./utils.js')>('./utils.js');
return {
...actual,
writeJsonFile: vi.fn(),
};
});
vi.mock('os', async () => {
const actual = await vi.importActual<typeof import('os')>('os');
return {
...actual,
default: {
...actual,
homedir: () => process.env.CODEX_USAGE_TEST_HOME || '/tmp',
},
homedir: () => process.env.CODEX_USAGE_TEST_HOME || '/tmp',
};
});
function createDefaultCodexAuth(homeDir: string): string {
const authPath = path.join(homeDir, '.codex', 'auth.json');
fs.mkdirSync(path.dirname(authPath), { recursive: true });
fs.writeFileSync(
authPath,
JSON.stringify({
auth_mode: 'chatgpt',
tokens: { account_id: 'default-acct', access_token: 'default-token' },
}),
);
return authPath;
}
function createFakeChildProcess(rateLimitsByLimitId: Record<string, unknown>) {
const proc = new EventEmitter() as EventEmitter & {
stdout: EventEmitter;
stdin: { write: ReturnType<typeof vi.fn> };
kill: ReturnType<typeof vi.fn>;
};
const stdout = new EventEmitter();
proc.stdout = stdout;
proc.stdin = {
write: vi.fn((payload: string) => {
const message = JSON.parse(payload.trim()) as {
id: number;
method?: string;
};
if (message.id === 1) {
setImmediate(() => {
stdout.emit(
'data',
Buffer.from(`${JSON.stringify({ id: 1, result: {} })}\n`),
);
});
}
if (message.id === 2 && message.method === 'account/rateLimits/read') {
setImmediate(() => {
stdout.emit(
'data',
Buffer.from(
`${JSON.stringify({
id: 2,
result: { rateLimitsByLimitId },
})}\n`,
),
);
});
}
return true;
}),
};
proc.kill = vi.fn();
return proc;
}
describe('codex-usage-collector fallback account usage', () => {
let tempHome: string;
beforeEach(() => {
vi.resetModules();
vi.clearAllMocks();
tempHome = fs.mkdtempSync(path.join('/tmp', 'ejclaw-codex-usage-'));
process.env.CODEX_USAGE_TEST_HOME = tempHome;
});
afterEach(() => {
delete process.env.CODEX_USAGE_TEST_HOME;
fs.rmSync(tempHome, { recursive: true, force: true });
});
it('refreshes usage via ~/.codex/auth.json when ~/.codex-accounts is missing', async () => {
createDefaultCodexAuth(tempHome);
const childProcess = await import('child_process');
const fallbackCodexHome = path.join(tempHome, '.codex');
let capturedCodexHome: string | undefined;
vi.mocked(childProcess.spawn).mockImplementation(((
_cmd: string,
_args: readonly string[] | undefined,
opts?: { env?: Record<string, string> },
) => {
capturedCodexHome = opts?.env?.CODEX_HOME;
return createFakeChildProcess({
codex: {
limitName: 'Codex',
primary: {
usedPercent: 12.4,
resetsAt: new Date(Date.now() + 3_600_000).toISOString(),
},
secondary: {
usedPercent: 67.6,
resetsAt: new Date(Date.now() + 86_400_000).toISOString(),
},
},
}) as never;
}) as unknown as typeof childProcess.spawn);
const rotation = await import('./codex-token-rotation.js');
const usage = await import('./codex-usage-collector.js');
rotation.initCodexTokenRotation();
const result = await usage.refreshActiveCodexUsage();
expect(rotation.getCodexAccountCount()).toBe(1);
expect(capturedCodexHome).toBe(fallbackCodexHome);
expect(result.fetchedAt).toEqual(expect.any(String));
expect(result.rows).toEqual([
expect.objectContaining({
name: 'Codex',
h5pct: 12,
d7pct: 68,
}),
]);
});
it('finds codex via ~/.hermes/node/bin when running under bun', async () => {
createDefaultCodexAuth(tempHome);
const childProcess = await import('child_process');
const bunExecPath = path.join(tempHome, '.bun', 'bin', 'bun');
const hermesBin = path.join(tempHome, '.hermes', 'node', 'bin');
const originalExecPath = process.execPath;
const originalPath = process.env.PATH;
fs.mkdirSync(hermesBin, { recursive: true });
process.env.PATH = '/usr/local/bin:/usr/bin:/bin';
Object.defineProperty(process, 'execPath', {
configurable: true,
writable: true,
value: bunExecPath,
});
vi.mocked(childProcess.spawn).mockImplementation(((
cmd: string,
_args: readonly string[] | undefined,
opts?: { env?: Record<string, string> },
) => {
const spawnPathEntries = (opts?.env?.PATH || '').split(path.delimiter);
expect(cmd).toBe('codex');
expect(spawnPathEntries).toContain(path.dirname(bunExecPath));
expect(spawnPathEntries).toContain(hermesBin);
if (!spawnPathEntries.includes(hermesBin)) {
const proc = new EventEmitter() as EventEmitter & {
stdout: EventEmitter;
stdin: { write: ReturnType<typeof vi.fn> };
kill: ReturnType<typeof vi.fn>;
};
proc.stdout = new EventEmitter();
proc.stdin = { write: vi.fn() };
proc.kill = vi.fn();
setImmediate(() => proc.emit('error', new Error('spawn codex ENOENT')));
return proc as never;
}
return createFakeChildProcess({
codex: {
limitName: 'Codex',
primary: {
usedPercent: 34.2,
resetsAt: new Date(Date.now() + 7_200_000).toISOString(),
},
secondary: {
usedPercent: 56.1,
resetsAt: new Date(Date.now() + 172_800_000).toISOString(),
},
},
}) as never;
}) as unknown as typeof childProcess.spawn);
try {
const rotation = await import('./codex-token-rotation.js');
const usage = await import('./codex-usage-collector.js');
rotation.initCodexTokenRotation();
const result = await usage.refreshActiveCodexUsage();
expect(result.fetchedAt).toEqual(expect.any(String));
expect(result.rows).toEqual([
expect.objectContaining({
name: 'Codex',
h5pct: 34,
d7pct: 56,
}),
]);
} finally {
Object.defineProperty(process, 'execPath', {
configurable: true,
writable: true,
value: originalExecPath,
});
if (originalPath === undefined) {
delete process.env.PATH;
} else {
process.env.PATH = originalPath;
}
}
});
});

View File

@@ -5,6 +5,7 @@ import path from 'path';
import {
getAllCodexAccounts,
getCodexAuthPath,
updateCodexAccountUsage,
} from './codex-token-rotation.js';
import { formatResetRemaining, type UsageRow } from './dashboard-usage-rows.js';
@@ -27,11 +28,26 @@ export interface CodexUsageRefreshResult {
fetchedAt: string | null;
}
const CODEX_ACCOUNTS_DIR = path.join(os.homedir(), '.codex-accounts');
/** Full scan interval — exported so the orchestrator can schedule it. */
export const CODEX_FULL_SCAN_INTERVAL = 3_600_000; // 1 hour
function getPreferredCodexPathEntries(): string[] {
const entries = [
path.dirname(process.execPath),
path.join(os.homedir(), '.npm-global', 'bin'),
];
if (process.versions.bun || path.basename(process.execPath) === 'bun') {
entries.push(path.join(os.homedir(), '.hermes', 'node', 'bin'));
}
return [...new Set(entries)];
}
function getCodexHomeForAccount(accountIndex?: number): string | null {
const authPath = getCodexAuthPath(accountIndex);
if (!authPath || !fs.existsSync(authPath)) return null;
return path.dirname(authPath);
}
export async function fetchCodexUsage(
codexHomeOverride?: string,
): Promise<CodexRateLimit[] | null> {
@@ -59,11 +75,9 @@ export async function fetchCodexUsage(
const spawnEnv: Record<string, string> = {
...(process.env as Record<string, string>),
PATH: [
path.dirname(process.execPath),
path.join(os.homedir(), '.npm-global', 'bin'),
process.env.PATH || '',
].join(':'),
PATH: [...getPreferredCodexPathEntries(), process.env.PATH || '']
.filter(Boolean)
.join(path.delimiter),
};
if (codexHomeOverride) {
spawnEnv.CODEX_HOME = codexHomeOverride;
@@ -243,8 +257,8 @@ export async function refreshAllCodexAccountUsage(): Promise<CodexUsageRefreshRe
let anySuccess = false;
for (const acct of codexAccounts) {
const accountDir = path.join(CODEX_ACCOUNTS_DIR, String(acct.index + 1));
if (!fs.existsSync(accountDir)) continue;
const accountDir = getCodexHomeForAccount(acct.index);
if (!accountDir) continue;
try {
const usage = await fetchCodexUsage(accountDir);
@@ -281,8 +295,8 @@ export async function refreshActiveCodexUsage(): Promise<CodexUsageRefreshResult
return { rows: buildCodexUsageRowsFromState(), fetchedAt: null };
}
const accountDir = path.join(CODEX_ACCOUNTS_DIR, String(active.index + 1));
if (!fs.existsSync(accountDir)) {
const accountDir = getCodexHomeForAccount(active.index);
if (!accountDir) {
return { rows: buildCodexUsageRowsFromState(), fetchedAt: null };
}

330
src/codex-warmup.test.ts Normal file
View File

@@ -0,0 +1,330 @@
import { EventEmitter } from 'events';
import fs from 'fs';
import path from 'path';
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
vi.mock('child_process', () => ({
spawn: vi.fn(),
}));
vi.mock('./codex-token-rotation.js', () => ({
getAllCodexAccounts: vi.fn(),
getCodexAuthPath: vi.fn(),
}));
vi.mock('./logger.js', () => ({
logger: {
info: vi.fn(),
warn: vi.fn(),
error: vi.fn(),
debug: vi.fn(),
},
}));
type FakeProcess = EventEmitter & {
stdout: EventEmitter;
stderr: EventEmitter;
stdin: { write: ReturnType<typeof vi.fn>; end: ReturnType<typeof vi.fn> };
kill: ReturnType<typeof vi.fn>;
};
function createFakeCodexProcess(exitCode = 0): FakeProcess {
const proc = new EventEmitter() as FakeProcess;
proc.stdout = new EventEmitter();
proc.stderr = new EventEmitter();
proc.stdin = { write: vi.fn(), end: vi.fn() };
proc.kill = vi.fn();
setImmediate(() => {
proc.stdout.emit('data', Buffer.from('OK\n'));
proc.emit('close', exitCode, null);
});
return proc;
}
function authPathFor(tempHome: string, accountIndex: number): string {
return path.join(
tempHome,
'.codex-accounts',
String(accountIndex + 1),
'auth.json',
);
}
describe('Codex warm-up scheduler', () => {
let tempHome: string;
let statePath: string;
beforeEach(() => {
vi.resetModules();
vi.clearAllMocks();
tempHome = fs.mkdtempSync(path.join('/tmp', 'ejclaw-codex-warmup-'));
statePath = path.join(tempHome, 'codex-warmup-state.json');
for (let i = 0; i < 3; i++) {
const p = authPathFor(tempHome, i);
fs.mkdirSync(path.dirname(p), { recursive: true });
fs.writeFileSync(p, '{}');
}
});
afterEach(() => {
fs.rmSync(tempHome, { recursive: true, force: true });
});
it('warms one eligible zero-usage account with a real codex exec and persists account cooldown state', async () => {
const childProcess = await import('child_process');
const rotation = await import('./codex-token-rotation.js');
const { runCodexWarmupCycle } = await import('./codex-warmup.js');
const now = new Date('2026-04-24T09:00:00Z').getTime();
let capturedEnv: Record<string, string> | undefined;
let capturedArgs: readonly string[] | undefined;
vi.mocked(rotation.getAllCodexAccounts).mockReturnValue([
{
index: 0,
accountId: 'busy-account',
planType: 'pro',
isActive: true,
isRateLimited: false,
cachedUsagePct: 14,
cachedUsageD7Pct: 30,
},
{
index: 1,
accountId: 'rate-limited-account',
planType: 'pro',
isActive: false,
isRateLimited: true,
cachedUsagePct: 0,
cachedUsageD7Pct: 0,
},
{
index: 2,
accountId: 'fresh-account',
planType: 'team',
isActive: false,
isRateLimited: false,
cachedUsagePct: 0,
cachedUsageD7Pct: 0,
resetAt: '2026-04-24T14:00:00.000Z',
resetD7At: '2026-05-01T09:00:00.000Z',
},
]);
vi.mocked(rotation.getCodexAuthPath).mockImplementation(
(accountIndex = 0) => authPathFor(tempHome, accountIndex),
);
vi.mocked(childProcess.spawn).mockImplementation(((
_cmd: string,
args?: readonly string[],
opts?: { env?: Record<string, string> },
) => {
capturedArgs = args;
capturedEnv = opts?.env;
return createFakeCodexProcess(0) as never;
}) as unknown as typeof childProcess.spawn);
const result = await runCodexWarmupCycle(
{
enabled: true,
prompt: 'Reply exactly OK. Do not run tools.',
model: 'gpt-5.5',
intervalMs: 300_000,
minIntervalMs: 18_300_000,
staggerMs: 1_800_000,
maxUsagePct: 0,
maxD7UsagePct: 0,
commandTimeoutMs: 120_000,
failureCooldownMs: 21_600_000,
maxConsecutiveFailures: 2,
},
{ nowMs: now, statePath },
);
expect(result).toEqual({ status: 'warmed', accountIndex: 2 });
expect(childProcess.spawn).toHaveBeenCalledTimes(1);
expect(capturedArgs).toEqual(
expect.arrayContaining([
'exec',
'--ephemeral',
'--ignore-rules',
'--skip-git-repo-check',
'--sandbox',
'read-only',
'-m',
'gpt-5.5',
'Reply exactly OK. Do not run tools.',
]),
);
expect(capturedEnv?.CODEX_HOME).toBe(
path.dirname(authPathFor(tempHome, 2)),
);
const state = JSON.parse(fs.readFileSync(statePath, 'utf8'));
expect(state.lastWarmupAt).toBe('2026-04-24T09:00:00.000Z');
expect(state.accounts['2'].lastWarmupAt).toBe('2026-04-24T09:00:00.000Z');
expect(state.accounts['2'].zeroUsageWarmupUntil).toBe(
'2026-05-01T09:00:00.000Z',
);
expect(state.consecutiveFailures).toBe(0);
});
it('does not repeat warm-up while the same zero-usage quota window is already marked warmed', async () => {
const childProcess = await import('child_process');
const rotation = await import('./codex-token-rotation.js');
const { runCodexWarmupCycle } = await import('./codex-warmup.js');
const now = new Date('2026-04-24T15:00:00Z').getTime();
fs.writeFileSync(
statePath,
JSON.stringify({
lastWarmupAt: '2026-04-24T09:00:00.000Z',
consecutiveFailures: 0,
accounts: {
'0': {
lastWarmupAt: '2026-04-24T09:00:00.000Z',
zeroUsageWarmupUntil: '2026-05-01T09:00:00.000Z',
},
},
}),
);
vi.mocked(rotation.getAllCodexAccounts).mockReturnValue([
{
index: 0,
accountId: 'fresh-account-still-rounded-zero',
planType: 'pro',
isActive: false,
isRateLimited: false,
cachedUsagePct: 0,
cachedUsageD7Pct: 0,
resetAt: '2026-04-24T14:00:00.000Z',
resetD7At: '2026-05-01T09:00:00.000Z',
},
]);
const result = await runCodexWarmupCycle(
{
enabled: true,
prompt: 'Reply exactly OK. Do not run tools.',
model: 'gpt-5.5',
intervalMs: 300_000,
minIntervalMs: 18_300_000,
staggerMs: 0,
maxUsagePct: 0,
maxD7UsagePct: 0,
commandTimeoutMs: 120_000,
failureCooldownMs: 21_600_000,
maxConsecutiveFailures: 2,
},
{ nowMs: now, statePath },
);
expect(result).toEqual({
status: 'skipped',
reason: 'no_eligible_accounts',
});
expect(childProcess.spawn).not.toHaveBeenCalled();
});
it('auto-backs off after repeated codex exec failures so OpenAI-side blocking does not hammer accounts', async () => {
const childProcess = await import('child_process');
const rotation = await import('./codex-token-rotation.js');
const { runCodexWarmupCycle } = await import('./codex-warmup.js');
const now = new Date('2026-04-24T09:00:00Z').getTime();
vi.mocked(rotation.getAllCodexAccounts).mockReturnValue([
{
index: 0,
accountId: 'fresh-account',
planType: 'pro',
isActive: false,
isRateLimited: false,
cachedUsagePct: 0,
cachedUsageD7Pct: 0,
},
]);
vi.mocked(rotation.getCodexAuthPath).mockImplementation(
(accountIndex = 0) => authPathFor(tempHome, accountIndex),
);
vi.mocked(childProcess.spawn).mockImplementation(
() => createFakeCodexProcess(1) as never,
);
const config = {
enabled: true,
prompt: 'Reply exactly OK. Do not run tools.',
model: 'gpt-5.5',
intervalMs: 300_000,
minIntervalMs: 18_300_000,
staggerMs: 0,
maxUsagePct: 0,
maxD7UsagePct: 0,
commandTimeoutMs: 120_000,
failureCooldownMs: 21_600_000,
maxConsecutiveFailures: 1,
};
const failed = await runCodexWarmupCycle(config, { nowMs: now, statePath });
expect(failed.status).toBe('failed');
const backedOff = await runCodexWarmupCycle(config, {
nowMs: now + 60_000,
statePath,
});
expect(backedOff).toEqual({
status: 'skipped',
reason: 'disabled_cooldown',
});
expect(childProcess.spawn).toHaveBeenCalledTimes(1);
const state = JSON.parse(fs.readFileSync(statePath, 'utf8'));
expect(state.disabledUntil).toBe('2026-04-24T15:00:00.000Z');
expect(state.consecutiveFailures).toBe(1);
});
it('respects global stagger and does not warm another account too soon', async () => {
const childProcess = await import('child_process');
const rotation = await import('./codex-token-rotation.js');
const { runCodexWarmupCycle } = await import('./codex-warmup.js');
const now = new Date('2026-04-24T09:20:00Z').getTime();
fs.writeFileSync(
statePath,
JSON.stringify({
lastWarmupAt: '2026-04-24T09:00:00.000Z',
consecutiveFailures: 0,
accounts: { '0': { lastWarmupAt: '2026-04-24T09:00:00.000Z' } },
}),
);
vi.mocked(rotation.getAllCodexAccounts).mockReturnValue([
{
index: 1,
accountId: 'another-fresh-account',
planType: 'pro',
isActive: false,
isRateLimited: false,
cachedUsagePct: 0,
cachedUsageD7Pct: 0,
},
]);
const result = await runCodexWarmupCycle(
{
enabled: true,
prompt: '.',
model: 'gpt-5.5',
intervalMs: 300_000,
minIntervalMs: 18_300_000,
staggerMs: 1_800_000,
maxUsagePct: 0,
maxD7UsagePct: 0,
commandTimeoutMs: 120_000,
failureCooldownMs: 21_600_000,
maxConsecutiveFailures: 2,
},
{ nowMs: now, statePath },
);
expect(result).toEqual({ status: 'skipped', reason: 'stagger_wait' });
expect(childProcess.spawn).not.toHaveBeenCalled();
});
});

315
src/codex-warmup.ts Normal file
View File

@@ -0,0 +1,315 @@
import { ChildProcess, spawn } from 'child_process';
import fs from 'fs';
import os from 'os';
import path from 'path';
import { DATA_DIR } from './config.js';
import type { AppConfig } from './config/schema.js';
import {
getAllCodexAccounts,
getCodexAuthPath,
} from './codex-token-rotation.js';
import { logger } from './logger.js';
export type CodexWarmupConfig = AppConfig['codexWarmup'];
interface CodexWarmupAccountState {
lastWarmupAt?: string;
lastAttemptAt?: string;
lastErrorAt?: string;
zeroUsageWarmupUntil?: string;
failures?: number;
}
interface CodexWarmupState {
lastWarmupAt?: string;
lastAttemptAt?: string;
disabledUntil?: string;
consecutiveFailures?: number;
accounts?: Record<string, CodexWarmupAccountState>;
}
interface CodexWarmupRuntimeOptions {
nowMs?: number;
statePath?: string;
shouldSkip?: () => boolean;
}
export type CodexWarmupCycleResult =
| { status: 'disabled' }
| { status: 'skipped'; reason: string }
| { status: 'warmed'; accountIndex: number }
| { status: 'failed'; accountIndex: number; reason: string };
const DEFAULT_STATE_FILE = path.join(DATA_DIR, 'codex-warmup-state.json');
const DEFAULT_ZERO_USAGE_WARMUP_WINDOW_MS = 7 * 24 * 60 * 60 * 1000;
function parseTimestamp(value?: string): number | null {
if (!value) return null;
const parsed = Date.parse(value);
return Number.isFinite(parsed) ? parsed : null;
}
function readWarmupState(statePath: string): CodexWarmupState {
try {
if (!fs.existsSync(statePath)) return { accounts: {} };
const parsed = JSON.parse(
fs.readFileSync(statePath, 'utf8'),
) as CodexWarmupState;
if (!parsed.accounts || typeof parsed.accounts !== 'object') {
parsed.accounts = {};
}
return parsed;
} catch (err) {
logger.warn(
{ err, statePath },
'Failed to read Codex warm-up state; starting fresh',
);
return { accounts: {} };
}
}
function writeWarmupState(statePath: string, state: CodexWarmupState): void {
fs.mkdirSync(path.dirname(statePath), { recursive: true });
fs.writeFileSync(`${statePath}.tmp`, `${JSON.stringify(state, null, 2)}\n`);
fs.renameSync(`${statePath}.tmp`, statePath);
}
function getPreferredCodexPathEntries(): string[] {
const entries = [
path.dirname(process.execPath),
path.join(os.homedir(), '.npm-global', 'bin'),
];
if (process.versions.bun || path.basename(process.execPath) === 'bun') {
entries.push(path.join(os.homedir(), '.hermes', 'node', 'bin'));
}
return [...new Set(entries)];
}
function resolveCodexBinary(): string {
const npmGlobalBin = path.join(os.homedir(), '.npm-global', 'bin', 'codex');
return fs.existsSync(npmGlobalBin) ? npmGlobalBin : 'codex';
}
function ensureAccountState(
state: CodexWarmupState,
accountIndex: number,
): CodexWarmupAccountState {
state.accounts ??= {};
const key = String(accountIndex);
state.accounts[key] ??= {};
return state.accounts[key];
}
function selectWarmupCandidate(
config: CodexWarmupConfig,
state: CodexWarmupState,
nowMs: number,
): { accountIndex: number; zeroUsageWarmupUntil: string } | { reason: string } {
const disabledUntilMs = parseTimestamp(state.disabledUntil);
if (disabledUntilMs != null && disabledUntilMs > nowMs) {
return { reason: 'disabled_cooldown' };
}
const lastGlobalWarmupMs = parseTimestamp(state.lastWarmupAt);
if (
lastGlobalWarmupMs != null &&
config.staggerMs > 0 &&
nowMs - lastGlobalWarmupMs < config.staggerMs
) {
return { reason: 'stagger_wait' };
}
const accounts = getAllCodexAccounts();
if (accounts.length === 0) return { reason: 'no_accounts' };
for (const account of accounts) {
if (account.isRateLimited) continue;
if (typeof account.cachedUsagePct !== 'number') continue;
if (typeof account.cachedUsageD7Pct !== 'number') continue;
if (account.cachedUsagePct > config.maxUsagePct) continue;
if (account.cachedUsageD7Pct > config.maxD7UsagePct) continue;
const accountState = state.accounts?.[String(account.index)];
const zeroUsageWarmupUntilMs = parseTimestamp(
accountState?.zeroUsageWarmupUntil,
);
if (zeroUsageWarmupUntilMs != null && zeroUsageWarmupUntilMs > nowMs) {
continue;
}
const lastWarmupMs = parseTimestamp(accountState?.lastWarmupAt);
if (lastWarmupMs != null && nowMs - lastWarmupMs < config.minIntervalMs) {
continue;
}
const resetD7Ms = parseTimestamp(account.resetD7At);
const zeroUsageWarmupUntilMsForState =
resetD7Ms != null && resetD7Ms > nowMs
? resetD7Ms
: nowMs + DEFAULT_ZERO_USAGE_WARMUP_WINDOW_MS;
return {
accountIndex: account.index,
zeroUsageWarmupUntil: new Date(
zeroUsageWarmupUntilMsForState,
).toISOString(),
};
}
return { reason: 'no_eligible_accounts' };
}
function runCodexWarmupCommand(
accountDir: string,
config: CodexWarmupConfig,
): Promise<{ ok: boolean; reason: string }> {
const args = [
'exec',
'--ephemeral',
'--ignore-rules',
'--skip-git-repo-check',
'--sandbox',
'read-only',
'-C',
os.tmpdir(),
'-m',
config.model,
config.prompt,
];
const spawnEnv: Record<string, string> = {
...(process.env as Record<string, string>),
CODEX_HOME: accountDir,
PATH: [...getPreferredCodexPathEntries(), process.env.PATH || '']
.filter(Boolean)
.join(path.delimiter),
};
return new Promise((resolve) => {
let done = false;
let proc: ChildProcess | null = null;
let stderr = '';
const finish = (result: { ok: boolean; reason: string }) => {
if (done) return;
done = true;
clearTimeout(timer);
if (proc && result.reason === 'timeout') {
try {
proc.kill();
} catch {
/* ignore */
}
}
resolve(result);
};
const timer = setTimeout(
() => finish({ ok: false, reason: 'timeout' }),
config.commandTimeoutMs,
);
try {
proc = spawn(resolveCodexBinary(), args, {
stdio: ['ignore', 'pipe', 'pipe'],
env: spawnEnv,
});
} catch (err) {
logger.warn({ err }, 'Failed to spawn Codex warm-up command');
finish({ ok: false, reason: 'spawn_error' });
return;
}
proc.stderr?.on('data', (chunk: Buffer) => {
stderr += chunk.toString();
if (stderr.length > 2000) stderr = stderr.slice(-2000);
});
proc.on('error', (err) => {
logger.warn({ err }, 'Codex warm-up process error');
finish({ ok: false, reason: 'process_error' });
});
proc.on('close', (code) => {
if (code === 0) {
finish({ ok: true, reason: 'ok' });
return;
}
logger.warn(
{ exitCode: code, stderr: stderr.trim() || undefined },
'Codex warm-up command failed',
);
finish({ ok: false, reason: `exit_${code ?? 'unknown'}` });
});
});
}
export async function runCodexWarmupCycle(
config: CodexWarmupConfig,
runtime: CodexWarmupRuntimeOptions = {},
): Promise<CodexWarmupCycleResult> {
if (!config.enabled) return { status: 'disabled' };
if (runtime.shouldSkip?.())
return { status: 'skipped', reason: 'runtime_busy' };
const nowMs = runtime.nowMs ?? Date.now();
const nowIso = new Date(nowMs).toISOString();
const statePath = runtime.statePath ?? DEFAULT_STATE_FILE;
const state = readWarmupState(statePath);
const selected = selectWarmupCandidate(config, state, nowMs);
if ('reason' in selected) {
return { status: 'skipped', reason: selected.reason };
}
const authPath = getCodexAuthPath(selected.accountIndex);
if (!authPath || !fs.existsSync(authPath)) {
return { status: 'skipped', reason: 'missing_auth' };
}
const accountDir = path.dirname(authPath);
const accountState = ensureAccountState(state, selected.accountIndex);
state.lastAttemptAt = nowIso;
accountState.lastAttemptAt = nowIso;
logger.info(
{ account: selected.accountIndex + 1, model: config.model },
'Starting Codex warm-up prompt',
);
const result = await runCodexWarmupCommand(accountDir, config);
if (result.ok) {
state.lastWarmupAt = nowIso;
state.consecutiveFailures = 0;
delete state.disabledUntil;
accountState.lastWarmupAt = nowIso;
accountState.zeroUsageWarmupUntil = selected.zeroUsageWarmupUntil;
accountState.failures = 0;
writeWarmupState(statePath, state);
logger.info(
{ account: selected.accountIndex + 1 },
'Codex warm-up prompt completed',
);
return { status: 'warmed', accountIndex: selected.accountIndex };
}
state.consecutiveFailures = (state.consecutiveFailures ?? 0) + 1;
accountState.lastErrorAt = nowIso;
accountState.failures = (accountState.failures ?? 0) + 1;
if (state.consecutiveFailures >= config.maxConsecutiveFailures) {
state.disabledUntil = new Date(
nowMs + config.failureCooldownMs,
).toISOString();
}
writeWarmupState(statePath, state);
logger.warn(
{
account: selected.accountIndex + 1,
reason: result.reason,
consecutiveFailures: state.consecutiveFailures,
disabledUntil: state.disabledUntil,
},
'Codex warm-up prompt failed',
);
return {
status: 'failed',
accountIndex: selected.accountIndex,
reason: result.reason,
};
}

View File

@@ -31,6 +31,21 @@ describe('config/env loading', () => {
delete process.env.DISCORD_CODEX_REVIEW_BOT_TOKEN;
delete process.env.PAIRED_CARRY_FORWARD_LATEST_OWNER_FINAL;
delete process.env.PAIRED_FORCE_FRESH_CLAUDE_REVIEWER_SESSION;
delete process.env.CODEX_WARMUP_ENABLED;
delete process.env.CODEX_WARMUP_PROMPT;
delete process.env.CODEX_WARMUP_MODEL;
delete process.env.CODEX_WARMUP_INTERVAL_MS;
delete process.env.CODEX_WARMUP_MIN_INTERVAL_MS;
delete process.env.CODEX_WARMUP_STAGGER_MS;
delete process.env.CODEX_WARMUP_MAX_USAGE_PCT;
delete process.env.CODEX_WARMUP_MAX_D7_USAGE_PCT;
delete process.env.CODEX_WARMUP_COMMAND_TIMEOUT_MS;
delete process.env.CODEX_WARMUP_FAILURE_COOLDOWN_MS;
delete process.env.CODEX_WARMUP_MAX_CONSECUTIVE_FAILURES;
delete process.env.WEB_DASHBOARD_ENABLED;
delete process.env.WEB_DASHBOARD_HOST;
delete process.env.WEB_DASHBOARD_PORT;
delete process.env.WEB_DASHBOARD_STATIC_DIR;
delete process.env.SESSION_COMMAND_USER_IDS;
vi.resetModules();
});
@@ -124,8 +139,60 @@ describe('config/env loading', () => {
expect(config.PAIRED_FORCE_FRESH_CLAUDE_REVIEWER_SESSION).toBe(true);
});
it('keeps Codex warm-up disabled by default and exposes conservative opt-in env config', async () => {
let config = await import('./config.js');
expect(config.CODEX_WARMUP_CONFIG.enabled).toBe(false);
expect(config.CODEX_WARMUP_CONFIG.maxUsagePct).toBe(0);
expect(config.CODEX_WARMUP_CONFIG.maxD7UsagePct).toBe(0);
expect(
config.CODEX_WARMUP_CONFIG.maxConsecutiveFailures,
).toBeGreaterThanOrEqual(1);
vi.resetModules();
process.env.CODEX_MODEL = 'gpt-5.5';
process.env.CODEX_WARMUP_ENABLED = 'true';
process.env.CODEX_WARMUP_PROMPT = '.';
process.env.CODEX_WARMUP_STAGGER_MS = '600000';
process.env.CODEX_WARMUP_MAX_CONSECUTIVE_FAILURES = '1';
config = await import('./config.js');
expect(config.CODEX_WARMUP_CONFIG).toEqual(
expect.objectContaining({
enabled: true,
prompt: '.',
model: 'gpt-5.5',
staggerMs: 600000,
maxConsecutiveFailures: 1,
}),
);
});
it('keeps the web dashboard disabled by default and loads explicit bind/static settings', async () => {
let config = await import('./config.js');
expect(config.WEB_DASHBOARD.enabled).toBe(false);
expect(config.WEB_DASHBOARD.host).toBe('127.0.0.1');
expect(config.WEB_DASHBOARD.port).toBe(8734);
expect(config.WEB_DASHBOARD.staticDir).toBe(
path.resolve(tempRoot, 'apps', 'dashboard', 'dist'),
);
vi.resetModules();
process.env.WEB_DASHBOARD_ENABLED = 'true';
process.env.WEB_DASHBOARD_HOST = '0.0.0.0';
process.env.WEB_DASHBOARD_PORT = '9001';
process.env.WEB_DASHBOARD_STATIC_DIR = './custom-dashboard-dist';
config = await import('./config.js');
expect(config.WEB_DASHBOARD).toEqual({
enabled: true,
host: '0.0.0.0',
port: 9001,
staticDir: path.resolve(tempRoot, 'custom-dashboard-dist'),
});
});
it('fails fast when a legacy Discord token alias is configured', async () => {
process.env.DISCORD_BOT_TOKEN = 'legacy-owner-token';
process.env.DISCORD_BOT_TOKEN = 'legacy...oken';
const { loadConfig } = await import('./config/load-config.js');

View File

@@ -140,6 +140,8 @@ export const USAGE_UPDATE_INTERVAL = CONFIG.status.usageUpdateInterval;
export const STATUS_SHOW_ROOMS = CONFIG.status.showRooms;
export const STATUS_SHOW_ROOM_DETAILS = CONFIG.status.showRoomDetails;
export const USAGE_DASHBOARD_ENABLED = CONFIG.status.usageDashboardEnabled;
export const CODEX_WARMUP_CONFIG = CONFIG.codexWarmup;
export const WEB_DASHBOARD = CONFIG.webDashboard;
// Timezone for scheduled tasks (cron expressions, etc.)
// Uses system timezone by default

View File

@@ -53,6 +53,11 @@ function readIntegerAtLeast(
return Math.max(minimum, readInteger(key, fallback) || fallback);
}
function readPercent(key: string, fallback: number): number {
const value = readInteger(key, fallback);
return Math.min(100, Math.max(0, value));
}
function readAgentType(
key: string,
fallback?: AgentType,
@@ -255,6 +260,49 @@ export function loadConfig(): AppConfig {
timezone:
readText('TZ') ?? Intl.DateTimeFormat().resolvedOptions().timeZone,
},
webDashboard: {
enabled: readBoolean('WEB_DASHBOARD_ENABLED', false),
host: readNonEmptyText('WEB_DASHBOARD_HOST') ?? '127.0.0.1',
port: readIntegerAtLeast('WEB_DASHBOARD_PORT', 8734, 1),
staticDir: path.resolve(
readNonEmptyText('WEB_DASHBOARD_STATIC_DIR') ??
path.join(projectRoot, 'apps', 'dashboard', 'dist'),
),
},
codexWarmup: {
enabled: readBoolean('CODEX_WARMUP_ENABLED', false),
prompt:
readNonEmptyText('CODEX_WARMUP_PROMPT') ??
'Reply exactly OK. Do not run tools.',
model:
readNonEmptyText('CODEX_WARMUP_MODEL') ??
readNonEmptyText('CODEX_MODEL') ??
'codex',
intervalMs: readIntegerAtLeast('CODEX_WARMUP_INTERVAL_MS', 300000, 60000),
minIntervalMs: readIntegerAtLeast(
'CODEX_WARMUP_MIN_INTERVAL_MS',
18300000,
60000,
),
staggerMs: Math.max(0, readInteger('CODEX_WARMUP_STAGGER_MS', 1800000)),
maxUsagePct: readPercent('CODEX_WARMUP_MAX_USAGE_PCT', 0),
maxD7UsagePct: readPercent('CODEX_WARMUP_MAX_D7_USAGE_PCT', 0),
commandTimeoutMs: readIntegerAtLeast(
'CODEX_WARMUP_COMMAND_TIMEOUT_MS',
120000,
10000,
),
failureCooldownMs: readIntegerAtLeast(
'CODEX_WARMUP_FAILURE_COOLDOWN_MS',
21600000,
60000,
),
maxConsecutiveFailures: readIntegerAtLeast(
'CODEX_WARMUP_MAX_CONSECUTIVE_FAILURES',
2,
1,
),
},
sessionCommands: {
allowedSenders: new Set(
(readText('SESSION_COMMAND_ALLOWED_SENDERS') ?? '')

View File

@@ -78,6 +78,25 @@ export interface AppConfig {
usageDashboardEnabled: boolean;
timezone: string;
};
webDashboard: {
enabled: boolean;
host: string;
port: number;
staticDir: string;
};
codexWarmup: {
enabled: boolean;
prompt: string;
model: string;
intervalMs: number;
minIntervalMs: number;
staggerMs: number;
maxUsagePct: number;
maxD7UsagePct: number;
commandTimeoutMs: number;
failureCooldownMs: number;
maxConsecutiveFailures: number;
};
sessionCommands: {
allowedSenders: Set<string>;
};

View File

@@ -0,0 +1,43 @@
import fs from 'fs';
import os from 'os';
import path from 'path';
import { afterEach, describe, expect, it } from 'vitest';
import { listUnexpectedDataStateFiles } from './data-state-files.js';
const tempDirs: string[] = [];
function makeTempDir(): string {
const tempDir = fs.mkdtempSync(path.join(os.tmpdir(), 'ejclaw-data-state-'));
tempDirs.push(tempDir);
return tempDir;
}
afterEach(() => {
for (const tempDir of tempDirs.splice(0)) {
fs.rmSync(tempDir, { force: true, recursive: true });
}
});
describe('listUnexpectedDataStateFiles', () => {
it('allows known runtime state files', () => {
const dataDir = makeTempDir();
fs.writeFileSync(path.join(dataDir, 'token-rotation-state.json'), '{}');
fs.writeFileSync(path.join(dataDir, 'codex-rotation-state.json'), '{}');
fs.writeFileSync(path.join(dataDir, 'codex-warmup-state.json'), '{}');
fs.writeFileSync(path.join(dataDir, 'claude-usage-cache.json'), '{}');
fs.writeFileSync(path.join(dataDir, 'restart-context.abc.json'), '{}');
expect(listUnexpectedDataStateFiles(dataDir)).toEqual([]);
});
it('reports unsupported legacy state files', () => {
const dataDir = makeTempDir();
fs.writeFileSync(path.join(dataDir, 'router_state.json'), '{}');
expect(listUnexpectedDataStateFiles(dataDir)).toEqual([
'router_state.json',
]);
});
});

View File

@@ -4,6 +4,7 @@ import path from 'path';
const ALLOWED_DATA_JSON_PATTERNS = [
/^token-rotation-state\.json$/,
/^codex-rotation-state\.json$/,
/^codex-warmup-state\.json$/,
/^claude-usage-cache\.json$/,
/^restart-context\..+\.json$/,
];

View File

@@ -0,0 +1,157 @@
import { Database } from 'bun:sqlite';
import fs from 'fs';
import os from 'os';
import path from 'path';
import { afterEach, beforeEach, describe, expect, it } from 'vitest';
import {
_initTestDatabase,
_initTestDatabaseFromFile,
getMessagesSinceSeq,
getRecentChatMessages,
storeChatMetadata,
storeMessage,
} from './db.js';
describe('message_source_kind persistence', () => {
let tempDir: string | null = null;
beforeEach(() => {
_initTestDatabase();
});
afterEach(() => {
if (tempDir) {
fs.rmSync(tempDir, { recursive: true, force: true });
tempDir = null;
}
});
it('persists explicit message source kind through store and read paths', () => {
storeChatMetadata('room@g.us', '2026-04-24T00:00:00.000Z', 'Room');
storeMessage({
id: 'ipc-1',
chat_jid: 'room@g.us',
sender: 'hermes',
sender_name: 'Hermes',
content: 'wake up',
timestamp: '2026-04-24T00:00:01.000Z',
is_from_me: false,
is_bot_message: false,
message_source_kind: 'trusted_external_bot',
});
const [recent] = getRecentChatMessages('room@g.us', 1);
expect(recent).toMatchObject({
id: 'ipc-1',
is_bot_message: false,
message_source_kind: 'trusted_external_bot',
});
storeMessage({
id: 'ipc-1',
chat_jid: 'room@g.us',
sender: 'hermes',
sender_name: 'Hermes',
content: 'updated',
timestamp: '2026-04-24T00:00:02.000Z',
is_from_me: false,
is_bot_message: true,
message_source_kind: 'ipc_injected_bot',
});
const [updated] = getMessagesSinceSeq('room@g.us', 0, 'EJClaw', 10);
expect(updated).toMatchObject({
id: 'ipc-1',
content: 'updated',
is_bot_message: true,
message_source_kind: 'ipc_injected_bot',
});
});
it('defaults missing message source kind from is_bot_message', () => {
storeChatMetadata('room@g.us', '2026-04-24T00:00:00.000Z', 'Room');
storeMessage({
id: 'human-1',
chat_jid: 'room@g.us',
sender: 'user',
sender_name: 'User',
content: 'hello',
timestamp: '2026-04-24T00:00:01.000Z',
is_from_me: false,
is_bot_message: false,
});
storeMessage({
id: 'bot-1',
chat_jid: 'room@g.us',
sender: 'bot',
sender_name: 'Bot',
content: 'done',
timestamp: '2026-04-24T00:00:02.000Z',
is_from_me: false,
is_bot_message: true,
});
expect(getRecentChatMessages('room@g.us', 2)).toEqual([
expect.objectContaining({ id: 'human-1', message_source_kind: 'human' }),
expect.objectContaining({ id: 'bot-1', message_source_kind: 'bot' }),
]);
});
it('migrates legacy message tables and backfills source kind', () => {
tempDir = fs.mkdtempSync(path.join(os.tmpdir(), 'ejclaw-msg-source-'));
const dbPath = path.join(tempDir, 'messages.db');
const legacyDb = new Database(dbPath);
legacyDb.exec(`
CREATE TABLE chats (
jid TEXT PRIMARY KEY,
name TEXT,
last_message_time TEXT,
channel TEXT,
is_group INTEGER DEFAULT 0
);
CREATE TABLE 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)
);
CREATE TABLE schema_migrations (
version INTEGER PRIMARY KEY,
name TEXT NOT NULL,
applied_at TEXT NOT NULL DEFAULT (datetime('now'))
);
INSERT INTO chats (jid, name, last_message_time, channel, is_group)
VALUES ('room@g.us', 'Room', '2026-04-24T00:00:02.000Z', 'discord', 1);
INSERT INTO messages (
id, chat_jid, sender, sender_name, content, timestamp, seq, is_from_me, is_bot_message
) VALUES
('human-legacy', 'room@g.us', 'user', 'User', 'hello', '2026-04-24T00:00:01.000Z', 1, 0, 0),
('bot-legacy', 'room@g.us', 'bot', 'Bot', 'done', '2026-04-24T00:00:02.000Z', 2, 0, 1);
`);
const insertMigration = legacyDb.prepare(
'INSERT INTO schema_migrations (version, name) VALUES (?, ?)',
);
for (let version = 1; version <= 12; version += 1) {
insertMigration.run(version, `legacy-${version}`);
}
legacyDb.close();
_initTestDatabaseFromFile(dbPath);
expect(getRecentChatMessages('room@g.us', 2)).toEqual([
expect.objectContaining({
id: 'human-legacy',
message_source_kind: 'human',
}),
expect.objectContaining({ id: 'bot-legacy', message_source_kind: 'bot' }),
]);
});
});

View File

@@ -22,6 +22,8 @@ export {
getOpenWorkItem,
getOpenWorkItemForChat,
getRecentChatMessages,
getRecentChatMessagesBatch,
hasMessage,
hasRecentRestartAnnouncement,
markWorkItemDelivered,
markWorkItemDeliveryRetry,
@@ -86,6 +88,7 @@ export {
createServiceHandoff,
failPairedTurn,
failServiceHandoff,
getAllOpenPairedTasks,
getAllChannelOwnerLeases,
getAllPendingServiceHandoffs,
getChannelOwnerLease,
@@ -99,7 +102,10 @@ export {
getPairedTurnAttempts,
getPairedTurnById,
getPairedTurnOutputs,
getRecentPairedTurnOutputsForChat,
getPairedTurnsForTask,
getLatestPairedTurnForTask,
updatePairedTurnProgressText,
getPairedWorkspace,
getPendingServiceHandoffs,
insertPairedTurnOutput,

View File

@@ -19,10 +19,12 @@ export function applyBaseSchema(database: Database): void {
seq INTEGER,
is_from_me INTEGER,
is_bot_message INTEGER DEFAULT 0,
message_source_kind TEXT NOT NULL DEFAULT 'human',
PRIMARY KEY (id, chat_jid),
FOREIGN KEY (chat_jid) REFERENCES chats(jid)
);
CREATE INDEX IF NOT EXISTS idx_timestamp ON messages(timestamp);
CREATE INDEX IF NOT EXISTS idx_messages_chat_timestamp ON messages(chat_jid, timestamp DESC);
CREATE TABLE IF NOT EXISTS message_sequence (
id INTEGER PRIMARY KEY AUTOINCREMENT
);

View File

@@ -37,6 +37,9 @@ function getExpectedSchemaMigrations(): Array<{
{ version: 10, name: 'paired_turn_provenance_upgrade' },
{ version: 11, name: 'owner_failure_count' },
{ version: 12, name: 'paired_verdict_and_step_telemetry' },
{ version: 13, name: 'message_source_kind' },
{ version: 14, name: 'work_item_attachments' },
{ version: 15, name: 'turn_progress_text' },
];
}

View File

@@ -1,5 +1,9 @@
import { Database } from 'bun:sqlite';
import {
inferMessageSourceKindFromBotFlag,
normalizeMessageSourceKind,
} from '../message-source.js';
import { NewMessage } from '../types.js';
export interface ChatInfo {
@@ -14,12 +18,18 @@ function normalizeMessageRow(
row: NewMessage & {
is_from_me?: boolean | number;
is_bot_message?: boolean | number;
message_source_kind?: unknown;
},
): NewMessage {
const isBotMessage = !!row.is_bot_message;
return {
...row,
is_from_me: !!row.is_from_me,
is_bot_message: !!row.is_bot_message,
is_bot_message: isBotMessage,
message_source_kind: normalizeMessageSourceKind(
row.message_source_kind,
inferMessageSourceKindFromBotFlag(isBotMessage),
),
};
}
@@ -85,6 +95,17 @@ export function getAllChatsFromDatabase(database: Database): ChatInfo[] {
.all() as ChatInfo[];
}
export function hasMessageInDatabase(
database: Database,
chatJid: string,
id: string,
): boolean {
const row = database
.prepare('SELECT 1 FROM messages WHERE chat_jid = ? AND id = ? LIMIT 1')
.get(chatJid, id);
return !!row;
}
export function storeMessageInDatabase(
database: Database,
msg: NewMessage,
@@ -106,15 +127,16 @@ export function storeMessageInDatabase(
database
.prepare(
`INSERT INTO messages (
id, chat_jid, sender, sender_name, content, timestamp, seq, is_from_me, is_bot_message
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
id, chat_jid, sender, sender_name, content, timestamp, seq, is_from_me, is_bot_message, message_source_kind
) 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`,
is_bot_message = excluded.is_bot_message,
message_source_kind = excluded.message_source_kind`,
)
.run(
msg.id,
@@ -126,6 +148,10 @@ export function storeMessageInDatabase(
seq,
msg.is_from_me ? 1 : 0,
msg.is_bot_message ? 1 : 0,
normalizeMessageSourceKind(
msg.message_source_kind,
inferMessageSourceKindFromBotFlag(msg.is_bot_message),
),
);
})();
}
@@ -142,7 +168,7 @@ export function getNewMessagesFromDatabase(
const placeholders = jids.map(() => '?').join(',');
const sql = `
SELECT * FROM (
SELECT id, chat_jid, sender, sender_name, content, timestamp, is_from_me, is_bot_message
SELECT id, chat_jid, sender, sender_name, content, timestamp, is_from_me, is_bot_message, message_source_kind
FROM messages
WHERE timestamp > ? AND chat_jid IN (${placeholders})
AND content NOT LIKE ?
@@ -178,7 +204,7 @@ export function getMessagesSinceFromDatabase(
): NewMessage[] {
const sql = `
SELECT * FROM (
SELECT id, chat_jid, sender, sender_name, content, timestamp, is_from_me, is_bot_message
SELECT id, chat_jid, sender, sender_name, content, timestamp, is_from_me, is_bot_message, message_source_kind
FROM messages
WHERE chat_jid = ? AND timestamp > ?
AND content NOT LIKE ?
@@ -238,7 +264,7 @@ export function getNewMessagesBySeqFromDatabase(
const placeholders = jids.map(() => '?').join(',');
const sql = `
SELECT id, chat_jid, sender, sender_name, content, timestamp, seq, is_from_me, is_bot_message
SELECT id, chat_jid, sender, sender_name, content, timestamp, seq, is_from_me, is_bot_message, message_source_kind
FROM messages
WHERE seq > ? AND chat_jid IN (${placeholders})
AND content NOT LIKE ?
@@ -273,7 +299,7 @@ export function getMessagesSinceSeqFromDatabase(
): NewMessage[] {
const sinceSeq = normalizeSeqCursor(sinceSeqCursor);
const sql = `
SELECT id, chat_jid, sender, sender_name, content, timestamp, seq, is_from_me, is_bot_message
SELECT id, chat_jid, sender, sender_name, content, timestamp, seq, is_from_me, is_bot_message, message_source_kind
FROM messages
WHERE chat_jid = ? AND seq > ?
AND content NOT LIKE ?
@@ -293,22 +319,31 @@ export function getMessagesSinceSeqFromDatabase(
return rows.map(normalizeMessageRow);
}
const recentChatMessagesStmtCache = new WeakMap<
Database,
ReturnType<Database['prepare']>
>();
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<
let stmt = recentChatMessagesStmtCache.get(database);
if (!stmt) {
stmt = database.prepare(`
SELECT * FROM (
SELECT id, chat_jid, sender, sender_name, content, timestamp, is_from_me, is_bot_message, message_source_kind
FROM messages
WHERE chat_jid = ?
AND content != '' AND content IS NOT NULL
ORDER BY timestamp DESC
LIMIT ?
) ORDER BY timestamp
`);
recentChatMessagesStmtCache.set(database, stmt);
}
const rows = stmt.all(chatJid, limit) as Array<
NewMessage & {
is_from_me?: boolean | number;
is_bot_message?: boolean | number;
@@ -317,6 +352,44 @@ export function getRecentChatMessagesFromDatabase(
return rows.map(normalizeMessageRow);
}
export function getRecentChatMessagesBatchFromDatabase(
database: Database,
chatJids: string[],
limit: number = 8,
): Map<string, NewMessage[]> {
const out = new Map<string, NewMessage[]>();
if (chatJids.length === 0) return out;
const placeholders = chatJids.map(() => '?').join(',');
const sql = `
WITH ranked AS (
SELECT id, chat_jid, sender, sender_name, content, timestamp,
is_from_me, is_bot_message, message_source_kind,
ROW_NUMBER() OVER (PARTITION BY chat_jid ORDER BY timestamp DESC) AS rn
FROM messages
WHERE chat_jid IN (${placeholders})
AND content != '' AND content IS NOT NULL
)
SELECT id, chat_jid, sender, sender_name, content, timestamp,
is_from_me, is_bot_message, message_source_kind
FROM ranked
WHERE rn <= ?
ORDER BY chat_jid, timestamp ASC
`;
const rows = database.prepare(sql).all(...chatJids, limit) as Array<
NewMessage & {
is_from_me?: boolean | number;
is_bot_message?: boolean | number;
}
>;
for (const row of rows) {
const normalized = normalizeMessageRow(row);
const existing = out.get(normalized.chat_jid);
if (existing) existing.push(normalized);
else out.set(normalized.chat_jid, [normalized]);
}
return out;
}
export function getLastHumanMessageTimestampFromDatabase(
database: Database,
chatJid: string,

View File

@@ -0,0 +1,43 @@
import type { Database } from 'bun:sqlite';
import { tableHasColumn } from './helpers.js';
import type { SchemaMigrationDefinition } from './types.js';
export const MESSAGE_SOURCE_KIND_MIGRATION: SchemaMigrationDefinition = {
version: 13,
name: 'message_source_kind',
apply(database: Database) {
const addedColumn = !tableHasColumn(
database,
'messages',
'message_source_kind',
);
if (addedColumn) {
database.exec(`
ALTER TABLE messages
ADD COLUMN message_source_kind TEXT NOT NULL DEFAULT 'human'
`);
}
const invalidOnlyWhere = `
WHERE message_source_kind IS NULL
OR message_source_kind = ''
OR message_source_kind NOT IN (
'human',
'bot',
'trusted_external_bot',
'ipc_injected_human',
'ipc_injected_bot'
)
`;
database.exec(`
UPDATE messages
SET message_source_kind = CASE
WHEN is_bot_message = 1 THEN 'bot'
ELSE 'human'
END
${addedColumn ? '' : invalidOnlyWhere}
`);
},
};

View File

@@ -0,0 +1,17 @@
import type { Database } from 'bun:sqlite';
import { tableHasColumn } from './helpers.js';
import type { SchemaMigrationDefinition } from './types.js';
export const WORK_ITEM_ATTACHMENTS_MIGRATION: SchemaMigrationDefinition = {
version: 14,
name: 'work_item_attachments',
apply(database: Database) {
if (!tableHasColumn(database, 'work_items', 'attachment_payload')) {
database.exec(`
ALTER TABLE work_items
ADD COLUMN attachment_payload TEXT
`);
}
},
};

View File

@@ -0,0 +1,19 @@
import type { Database } from 'bun:sqlite';
import { tableHasColumn } from './helpers.js';
import type { SchemaMigrationDefinition } from './types.js';
export const TURN_PROGRESS_TEXT_MIGRATION: SchemaMigrationDefinition = {
version: 15,
name: 'turn_progress_text',
apply(database: Database) {
if (!tableHasColumn(database, 'paired_turns', 'progress_text')) {
database.exec(`ALTER TABLE paired_turns ADD COLUMN progress_text TEXT`);
}
if (!tableHasColumn(database, 'paired_turns', 'progress_updated_at')) {
database.exec(
`ALTER TABLE paired_turns ADD COLUMN progress_updated_at TEXT`,
);
}
},
};

View File

@@ -12,7 +12,9 @@ import { PAIRED_WORKSPACE_PROJECT_SCHEMA_CLEANUP_MIGRATION } from './009_paired-
import { PAIRED_TURN_PROVENANCE_UPGRADE_MIGRATION } from './010_paired-turn-provenance-upgrade.js';
import { OWNER_FAILURE_COUNT_MIGRATION } from './011_owner-failure-count.js';
import { PAIRED_VERDICT_AND_STEP_TELEMETRY_MIGRATION } from './012_paired-verdict-and-step-telemetry.js';
import { WORK_ITEM_ATTACHMENTS_MIGRATION } from './013_work-item-attachments.js';
import { MESSAGE_SOURCE_KIND_MIGRATION } from './013_message-source-kind.js';
import { WORK_ITEM_ATTACHMENTS_MIGRATION } from './014_work-item-attachments.js';
import { TURN_PROGRESS_TEXT_MIGRATION } from './015_turn-progress-text.js';
import type {
SchemaMigrationArgs,
SchemaMigrationDefinition,
@@ -33,7 +35,9 @@ const ORDERED_SCHEMA_MIGRATIONS: readonly SchemaMigrationDefinition[] = [
PAIRED_TURN_PROVENANCE_UPGRADE_MIGRATION,
OWNER_FAILURE_COUNT_MIGRATION,
PAIRED_VERDICT_AND_STEP_TELEMETRY_MIGRATION,
MESSAGE_SOURCE_KIND_MIGRATION,
WORK_ITEM_ATTACHMENTS_MIGRATION,
TURN_PROGRESS_TEXT_MIGRATION,
];
function ensureSchemaMigrationsTable(database: Database): void {

View File

@@ -229,21 +229,27 @@ export function getPairedTaskByIdFromDatabase(
return row ? hydratePairedTaskRow(database, row) : undefined;
}
const latestPairedTaskStmtCache = new WeakMap<
Database,
ReturnType<Database['prepare']>
>();
export function getLatestPairedTaskForChatFromDatabase(
database: Database,
chatJid: string,
): PairedTask | undefined {
const row = database
.prepare(
`
SELECT *
FROM paired_tasks
WHERE chat_jid = ?
ORDER BY updated_at DESC
LIMIT 1
`,
)
.get(chatJid) as StoredPairedTaskRow | undefined;
let stmt = latestPairedTaskStmtCache.get(database);
if (!stmt) {
stmt = database.prepare(`
SELECT *
FROM paired_tasks
WHERE chat_jid = ?
ORDER BY updated_at DESC
LIMIT 1
`);
latestPairedTaskStmtCache.set(database, stmt);
}
const row = stmt.get(chatJid) as StoredPairedTaskRow | undefined;
return row ? hydratePairedTaskRow(database, row) : undefined;
}
@@ -266,6 +272,22 @@ export function getLatestOpenPairedTaskForChatFromDatabase(
return row ? hydratePairedTaskRow(database, row) : undefined;
}
export function getAllOpenPairedTasksFromDatabase(
database: Database,
): PairedTask[] {
const rows = database
.prepare(
`
SELECT *
FROM paired_tasks
WHERE status NOT IN ('completed')
ORDER BY updated_at DESC, created_at DESC
`,
)
.all() as StoredPairedTaskRow[];
return rows.map((row) => hydratePairedTaskRow(database, row));
}
export function getLatestPreviousPairedTaskForChatFromDatabase(
database: Database,
chatJid: string,

View File

@@ -56,6 +56,32 @@ export function getPairedTurnOutputsFromDatabase(
.all(taskId) as PairedTurnOutput[];
}
const recentOutputsForChatStmtCache = new WeakMap<
Database,
ReturnType<Database['prepare']>
>();
export function getRecentPairedTurnOutputsForChatFromDatabase(
database: Database,
chatJid: string,
limit: number = 8,
): PairedTurnOutput[] {
let stmt = recentOutputsForChatStmtCache.get(database);
if (!stmt) {
stmt = database.prepare(`
SELECT o.*
FROM paired_turn_outputs o
INNER JOIN paired_tasks t ON o.task_id = t.id
WHERE t.chat_jid = ?
ORDER BY o.created_at DESC
LIMIT ?
`);
recentOutputsForChatStmtCache.set(database, stmt);
}
const rows = stmt.all(chatJid, limit) as PairedTurnOutput[];
return rows.reverse();
}
export function getLatestTurnNumberFromDatabase(
database: Database,
taskId: string,

View File

@@ -33,6 +33,8 @@ export interface PairedTurnRecord {
updated_at: string;
completed_at: string | null;
last_error: string | null;
progress_text?: string | null;
progress_updated_at?: string | null;
}
interface StoredPairedTurnRow {
@@ -43,6 +45,8 @@ interface StoredPairedTurnRow {
intent_kind: PairedTurnIdentity['intentKind'];
created_at: string;
updated_at: string;
progress_text?: string | null;
progress_updated_at?: string | null;
}
function hydratePairedTurnRecord(
@@ -493,6 +497,56 @@ export function getPairedTurnByIdFromDatabase(
);
}
const updateProgressTextStmtCache = new WeakMap<
Database,
ReturnType<Database['prepare']>
>();
export function updatePairedTurnProgressTextFromDatabase(
database: Database,
turnId: string,
progressText: string | null,
): void {
let stmt = updateProgressTextStmtCache.get(database);
if (!stmt) {
stmt = database.prepare(`
UPDATE paired_turns
SET progress_text = ?, progress_updated_at = ?
WHERE turn_id = ?
`);
updateProgressTextStmtCache.set(database, stmt);
}
stmt.run(progressText, new Date().toISOString(), turnId);
}
const latestPairedTurnStmtCache = new WeakMap<
Database,
ReturnType<Database['prepare']>
>();
export function getLatestPairedTurnForTaskFromDatabase(
database: Database,
taskId: string,
): PairedTurnRecord | null {
let stmt = latestPairedTurnStmtCache.get(database);
if (!stmt) {
stmt = database.prepare(`
SELECT *
FROM paired_turns
WHERE task_id = ?
ORDER BY updated_at DESC, turn_id DESC
LIMIT 1
`);
latestPairedTurnStmtCache.set(database, stmt);
}
const row = stmt.get(taskId) as StoredPairedTurnRow | undefined;
if (!row) return null;
return hydratePairedTurnRecord(
row,
getCurrentPairedTurnAttemptForTurnFromDatabase(database, row.turn_id),
);
}
export function getPairedTurnsForTaskFromDatabase(
database: Database,
taskId: string,

View File

@@ -26,6 +26,8 @@ import {
getNewMessagesBySeqFromDatabase,
getNewMessagesFromDatabase,
getRecentChatMessagesFromDatabase,
getRecentChatMessagesBatchFromDatabase,
hasMessageInDatabase,
hasRecentRestartAnnouncementInDatabase,
storeChatMetadataInDatabase,
storeMessageInDatabase,
@@ -141,6 +143,10 @@ export function storeMessage(msg: NewMessage): void {
storeMessageInDatabase(requireDatabase(), msg);
}
export function hasMessage(chatJid: string, id: string): boolean {
return hasMessageInDatabase(requireDatabase(), chatJid, id);
}
export function getNewMessages(
jids: string[],
lastTimestamp: string,
@@ -219,6 +225,17 @@ export function getRecentChatMessages(
return getRecentChatMessagesFromDatabase(requireDatabase(), chatJid, limit);
}
export function getRecentChatMessagesBatch(
chatJids: string[],
limit: number = 8,
): Map<string, NewMessage[]> {
return getRecentChatMessagesBatchFromDatabase(
requireDatabase(),
chatJids,
limit,
);
}
export function getLastHumanMessageTimestamp(chatJid: string): string | null {
return getLastHumanMessageTimestampFromDatabase(requireDatabase(), chatJid);
}

View File

@@ -24,6 +24,7 @@ import {
clearPairedTurnReservationsInDatabase,
type PairedTaskUpdates,
createPairedTaskInDatabase,
getAllOpenPairedTasksFromDatabase,
getLastBotFinalMessageFromDatabase,
getLatestOpenPairedTaskForChatFromDatabase,
getLatestPreviousPairedTaskForChatFromDatabase,
@@ -48,6 +49,7 @@ import {
import {
getLatestTurnNumberFromDatabase,
getPairedTurnOutputsFromDatabase,
getRecentPairedTurnOutputsForChatFromDatabase,
insertPairedTurnOutputInDatabase,
} from './paired-turn-outputs.js';
import {
@@ -57,6 +59,8 @@ import {
failPairedTurnInDatabase,
getPairedTurnByIdFromDatabase,
getPairedTurnsForTaskFromDatabase,
getLatestPairedTurnForTaskFromDatabase,
updatePairedTurnProgressTextFromDatabase,
markPairedTurnRunningInDatabase,
type PairedTurnRecord,
} from './paired-turns.js';
@@ -113,6 +117,10 @@ export function getLatestPreviousPairedTaskForChat(
);
}
export function getAllOpenPairedTasks(): PairedTask[] {
return getAllOpenPairedTasksFromDatabase(requireDatabase());
}
export function updatePairedTask(id: string, updates: PairedTaskUpdates): void {
updatePairedTaskInDatabase(requireDatabase(), id, updates);
}
@@ -206,6 +214,23 @@ export function getPairedTurnsForTask(taskId: string): PairedTurnRecord[] {
return getPairedTurnsForTaskFromDatabase(requireDatabase(), taskId);
}
export function getLatestPairedTurnForTask(
taskId: string,
): PairedTurnRecord | null {
return getLatestPairedTurnForTaskFromDatabase(requireDatabase(), taskId);
}
export function updatePairedTurnProgressText(
turnId: string,
progressText: string | null,
): void {
updatePairedTurnProgressTextFromDatabase(
requireDatabase(),
turnId,
progressText,
);
}
export function getPairedTurnAttempts(
turnId: string,
): PairedTurnAttemptRecord[] {
@@ -328,6 +353,17 @@ export function getPairedTurnOutputs(taskId: string): PairedTurnOutput[] {
return getPairedTurnOutputsFromDatabase(requireDatabase(), taskId);
}
export function getRecentPairedTurnOutputsForChat(
chatJid: string,
limit: number = 8,
): PairedTurnOutput[] {
return getRecentPairedTurnOutputsForChatFromDatabase(
requireDatabase(),
chatJid,
limit,
);
}
export function getLatestTurnNumber(taskId: string): number {
return getLatestTurnNumberFromDatabase(requireDatabase(), taskId);
}

View File

@@ -10,6 +10,7 @@ import {
TIMEZONE,
TRIGGER_PATTERN,
USAGE_UPDATE_INTERVAL,
WEB_DASHBOARD,
} from './config.js';
import './channels/index.js';
import {
@@ -54,6 +55,7 @@ import {
import { createMessageRuntime } from './message-runtime.js';
import { nudgeSchedulerLoop, startSchedulerLoop } from './task-scheduler.js';
import { startUnifiedDashboard } from './unified-dashboard.js';
import { startWebDashboardServer } from './web-dashboard-server.js';
import { Channel, NewMessage, RegisteredGroup } from './types.js';
import { logger } from './logger.js';
import { initCodexTokenRotation } from './codex-token-rotation.js';
@@ -61,6 +63,10 @@ import {
hasAvailableClaudeToken,
initTokenRotation,
} from './token-rotation.js';
import {
isBotMessageSourceKind,
resolveInjectedMessageSourceKind,
} from './message-source.js';
import { parseVisibleVerdict } from './paired-execution-context-shared.js';
export function isTerminalStatusMessage(text: string): boolean {
@@ -244,6 +250,7 @@ async function main(): Promise<void> {
// Graceful shutdown handlers
let leaseRecoveryTimer: ReturnType<typeof setInterval> | null = null;
let webDashboardServer: ReturnType<typeof startWebDashboardServer> = null;
const shutdown = async (signal: string) => {
logger.info({ signal }, 'Shutdown signal received');
stopTokenRefreshLoop();
@@ -251,6 +258,8 @@ async function main(): Promise<void> {
clearInterval(leaseRecoveryTimer);
leaseRecoveryTimer = null;
}
webDashboardServer?.stop();
webDashboardServer = null;
const roomBindings = runtimeState.getRoomBindings();
const interruptedGroups = queue
.getStatuses(Object.keys(roomBindings))
@@ -412,6 +421,51 @@ async function main(): Promise<void> {
queue.noteDirectTerminalDelivery(jid, senderRole, text);
}
},
injectInboundMessage: async (payload) => {
const jid = payload.chatJid;
const binding = runtimeState.getRoomBindings()[jid];
if (!binding) {
logger.warn(
{ chatJid: jid, sender: payload.sender ?? null },
'inject_inbound_message: no room binding, dropping',
);
return;
}
const ts = payload.timestamp || new Date().toISOString();
const msgId =
payload.messageId ||
`ipc-inject-${Date.now()}-${Math.random().toString(36).slice(2, 8)}`;
const treatAsHuman = payload.treatAsHuman === true;
const messageSourceKind = resolveInjectedMessageSourceKind({
treatAsHuman,
sourceKind: payload.sourceKind,
});
storeChatMetadata(jid, ts, binding.name, 'discord', true);
storeMessage({
id: msgId,
chat_jid: jid,
sender: payload.sender || 'ipc-inject',
sender_name: payload.senderName || payload.sender || 'IPC Inject',
content: payload.text,
timestamp: ts,
is_from_me: false,
is_bot_message: isBotMessageSourceKind(messageSourceKind),
message_source_kind: messageSourceKind,
});
queue.enqueueMessageCheck(jid, resolveGroupIpcPath(binding.folder));
logger.info(
{
chatJid: jid,
sender: payload.sender ?? null,
senderName: payload.senderName ?? null,
treatAsHuman,
messageSourceKind,
messageId: msgId,
groupFolder: binding.folder,
},
'Injected inbound message via IPC',
);
},
nudgeScheduler: nudgeSchedulerLoop,
roomBindings: runtimeState.getRoomBindings,
assignRoom: runtimeState.assignRoomForIpc,
@@ -467,6 +521,13 @@ async function main(): Promise<void> {
},
purgeOnStart: true,
});
webDashboardServer = startWebDashboardServer({
...WEB_DASHBOARD,
getRoomBindings: runtimeState.getRoomBindings,
enqueueMessageCheck: (chatJid, groupFolder) =>
queue.enqueueMessageCheck(chatJid, resolveGroupIpcPath(groupFolder)),
nudgeScheduler: nudgeSchedulerLoop,
});
leaseRecoveryTimer = setInterval(() => {
const failover = getGlobalFailoverInfo();

View File

@@ -119,6 +119,7 @@ interface CreateExecuteTurnDeps {
deliveryRole: PairedRoomRole | null;
pairedRoom: boolean;
}) => Promise<void>;
recordTurnProgress: (turnId: string, progressText: string) => void;
}
export function createRunAgent(deps: {
@@ -190,6 +191,8 @@ export function createExecuteTurn(deps: CreateExecuteTurnDeps): ExecuteTurnFn {
deliveryRole: resolvedDeliveryRole,
deliveryServiceId: resolvedDeliveryServiceId,
pairedTurnIdentity: args.pairedTurnIdentity ?? null,
recordTurnProgress: (turnId, progressText) =>
deps.recordTurnProgress(turnId, progressText),
canDeliverFinalText: () => {
if (!args.pairedTurnIdentity) {
return true;

View File

@@ -6,6 +6,7 @@ import {
getLatestOpenPairedTaskForChat,
markWorkItemDelivered,
getPairedTaskById,
updatePairedTurnProgressText,
} from './db.js';
import {
isSessionCommandSenderAllowed,
@@ -262,6 +263,9 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): {
continuationTracker.open(targetChatJid),
});
},
recordTurnProgress: (turnId, progressText) => {
updatePairedTurnProgressText(turnId, progressText);
},
afterDeliverySuccess: async ({
chatJid,
runId,

View File

@@ -0,0 +1,43 @@
import { describe, expect, it } from 'vitest';
import {
isBotMessageSourceKind,
normalizeMessageSourceKind,
resolveInjectedMessageSourceKind,
} from './message-source.js';
describe('message source helpers', () => {
it('defaults IPC injected messages to trusted human-equivalent provenance', () => {
expect(resolveInjectedMessageSourceKind({ treatAsHuman: true })).toBe(
'trusted_external_bot',
);
expect(
isBotMessageSourceKind(
resolveInjectedMessageSourceKind({ treatAsHuman: true }),
),
).toBe(false);
});
it('defaults non-human IPC injected messages to bot-equivalent provenance', () => {
expect(resolveInjectedMessageSourceKind({ treatAsHuman: false })).toBe(
'ipc_injected_bot',
);
expect(
isBotMessageSourceKind(
resolveInjectedMessageSourceKind({ treatAsHuman: false }),
),
).toBe(true);
});
it('honors valid explicit source kinds and normalizes invalid values', () => {
expect(
resolveInjectedMessageSourceKind({
treatAsHuman: true,
sourceKind: 'ipc_injected_human',
}),
).toBe('ipc_injected_human');
expect(normalizeMessageSourceKind('bot', 'human')).toBe('bot');
expect(normalizeMessageSourceKind('not-real', 'human')).toBe('human');
});
});

39
src/message-source.ts Normal file
View File

@@ -0,0 +1,39 @@
import type { MessageSourceKind } from './types.js';
const MESSAGE_SOURCE_KINDS = new Set<MessageSourceKind>([
'human',
'bot',
'trusted_external_bot',
'ipc_injected_human',
'ipc_injected_bot',
]);
export function normalizeMessageSourceKind(
value: unknown,
fallback: MessageSourceKind = 'human',
): MessageSourceKind {
return typeof value === 'string' &&
MESSAGE_SOURCE_KINDS.has(value as MessageSourceKind)
? (value as MessageSourceKind)
: fallback;
}
export function isBotMessageSourceKind(kind: MessageSourceKind): boolean {
return kind === 'bot' || kind === 'ipc_injected_bot';
}
export function inferMessageSourceKindFromBotFlag(
isBotMessage: boolean | number | null | undefined,
): MessageSourceKind {
return isBotMessage ? 'bot' : 'human';
}
export function resolveInjectedMessageSourceKind(args: {
treatAsHuman: boolean;
sourceKind?: unknown;
}): MessageSourceKind {
return normalizeMessageSourceKind(
args.sourceKind,
args.treatAsHuman ? 'trusted_external_bot' : 'ipc_injected_bot',
);
}

View File

@@ -370,6 +370,71 @@ describe('MessageTurnController outbound audit logging', () => {
);
});
it('sends final attachments as a fresh final message instead of replacing text-only progress', async () => {
const channel = {
...makeChannel(),
name: 'discord',
} satisfies Channel;
const deliverFinalText = vi.fn().mockResolvedValue(true);
const attachments = [
{
path: '/tmp/ejclaw-discord-image-final.png',
name: 'final.png',
mime: 'image/png',
},
];
const controller = new MessageTurnController({
chatJid: 'dc:test-room',
group: makeGroup(),
runId: 'run-owner-final-attachment-send',
channel,
idleTimeout: 1_000,
failureFinalText: '실패',
isClaudeCodeAgent: true,
clearSession: vi.fn(),
requestClose: vi.fn(),
deliverFinalText,
deliveryRole: 'owner',
pairedTurnIdentity: {
turnId: 'task-1:2026-04-10T14:22:00.000Z:owner-turn',
taskId: 'task-1',
taskUpdatedAt: '2026-04-10T14:22:00.000Z',
intentKind: 'finalize-owner-turn',
role: 'owner',
},
});
await controller.start();
await controller.handleOutput({
status: 'success',
phase: 'progress',
result: '이미지 렌더링 중',
} as any);
await controller.handleOutput({
status: 'success',
phase: 'progress',
result: '이미지 파일 쓰는 중',
} as any);
await flushAsync();
await controller.handleOutput({
status: 'success',
phase: 'final',
result: '이미지 렌더 완료.',
output: {
visibility: 'public',
text: '이미지 렌더 완료.',
attachments,
},
} as any);
await controller.finish('success');
expect(channel.sendAndTrack).toHaveBeenCalledTimes(1);
expect(deliverFinalText).toHaveBeenCalledWith('이미지 렌더 완료.', {
replaceMessageId: null,
attachments,
});
});
it('replaces the tracked progress message when an owner final arrives', async () => {
const channel = {
...makeChannel(),

View File

@@ -49,6 +49,7 @@ interface MessageTurnControllerOptions {
deliveryRole?: PairedRoomRole | null;
deliveryServiceId?: string | null;
pairedTurnIdentity?: PairedTurnIdentity | null;
recordTurnProgress?: (turnId: string, progressText: string) => void;
}
export class MessageTurnController {
@@ -398,6 +399,52 @@ export class MessageTurnController {
return this.visiblePhase === 'final';
}
private composeProgressBody(text: string): string {
if (this.subagents.size > 1) {
const lines: string[] = [];
for (const [, track] of this.subagents) {
const latest = track.activities[track.activities.length - 1];
lines.push(latest ? `${track.label} · ${latest}` : track.label);
}
return lines.join('\n');
}
if (this.subagents.size === 1) {
const [, track] = this.subagents.entries().next().value!;
const lines: string[] = [track.label];
for (let i = 0; i < track.activities.length; i++) {
const isLast = i === track.activities.length - 1;
lines.push(`${isLast ? '└' : '├'} ${track.activities[i]}`);
}
return lines.join('\n');
}
const activityLines =
this.toolActivities.length > 0
? '\n' +
this.toolActivities
.map((a, i) => {
const isLast = i === this.toolActivities.length - 1;
const connector = isLast ? '└' : '├';
const isSummary = a.startsWith('📋');
return isSummary ? `${connector} ${a}` : `${connector} ${a}`;
})
.join('\n')
: '';
return text + activityLines;
}
private persistProgressBody(body: string): void {
const turnId = this.options.pairedTurnIdentity?.turnId;
if (!turnId || !this.options.recordTurnProgress) return;
try {
this.options.recordTurnProgress(turnId, body);
} catch (err) {
this.log.warn(
{ err, turnId, bodyLength: body.length },
'Failed to persist progress body',
);
}
}
private renderProgressMessage(text: string): string {
const elapsedMs =
this.progressStartedAt === null
@@ -405,41 +452,9 @@ export class MessageTurnController {
: Math.floor((Date.now() - this.progressStartedAt) / 5_000) * 5000;
const suffix = `\n\n${formatElapsedKorean(elapsedMs)}`;
let body: string;
const body = this.composeProgressBody(text);
if (this.subagents.size > 1) {
// Compact: one line per subagent with latest activity
const lines: string[] = [];
for (const [, track] of this.subagents) {
const latest = track.activities[track.activities.length - 1];
lines.push(latest ? `${track.label} · ${latest}` : track.label);
}
body = lines.join('\n');
} else if (this.subagents.size === 1) {
// Single subagent: detailed view with activity sub-lines
const [, track] = this.subagents.entries().next().value!;
const lines: string[] = [track.label];
for (let i = 0; i < track.activities.length; i++) {
const isLast = i === track.activities.length - 1;
lines.push(`${isLast ? '└' : '├'} ${track.activities[i]}`);
}
body = lines.join('\n');
} else {
// Single agent rendering
const activityLines =
this.toolActivities.length > 0
? '\n' +
this.toolActivities
.map((a, i) => {
const isLast = i === this.toolActivities.length - 1;
const connector = isLast ? '└' : '├';
const isSummary = a.startsWith('📋');
return isSummary ? `${connector} ${a}` : `${connector} ${a}`;
})
.join('\n')
: '';
body = text + activityLines;
}
this.persistProgressBody(body);
const maxBody = 2000 - TASK_STATUS_MESSAGE_PREFIX.length - suffix.length;
const truncated =
@@ -465,6 +480,14 @@ export class MessageTurnController {
this.progressMessageId = null;
this.progressStartedAt = null;
this.progressEditFailCount = 0;
const turnId = this.options.pairedTurnIdentity?.turnId;
if (turnId && this.options.recordTurnProgress) {
try {
this.options.recordTurnProgress(turnId, '');
} catch {
/* clearing progress is best-effort */
}
}
}
/**
@@ -666,7 +689,10 @@ export class MessageTurnController {
await this.flushPendingProgress(options.flushPendingText);
}
const replaceMessageId = this.consumeProgressForFinalDelivery();
const hasAttachments = (options?.attachments?.length ?? 0) > 0;
const replaceMessageId = hasAttachments
? this.discardProgressForAttachmentFinalDelivery()
: this.consumeProgressForFinalDelivery();
await this.deliverFinalText(text, {
...(options?.attachments?.length
? { attachments: options.attachments }
@@ -690,6 +716,20 @@ export class MessageTurnController {
return replaceMessageId;
}
private discardProgressForAttachmentFinalDelivery(): null {
this.log.info(
{
progressMessageId: this.progressMessageId,
latestProgressText: this.latestProgressText,
},
this.progressMessageId
? 'Discarding tracked progress replacement for final attachment delivery'
: 'Delivering final attachment output without a tracked progress message to replace',
);
this.resetProgressState();
return null;
}
private async deliverFinalText(
text: string,
options?: {

View File

@@ -1,4 +1,5 @@
import fs from 'fs';
import os from 'os';
import path from 'path';
import { DATA_DIR } from './config.js';
@@ -16,6 +17,10 @@ export interface ValidateOutboundAttachmentsResult {
const MAX_ATTACHMENT_BYTES = 8 * 1024 * 1024;
const IMAGE_EXTS = /\.(png|jpe?g|gif|webp|bmp)$/i;
const DEFAULT_TEMP_ATTACHMENT_DIR_PREFIXES = [
'ejclaw-attachment-',
'ejclaw-discord-image-',
] as const;
function unique(values: Array<string | null | undefined>): string[] {
return [
@@ -38,8 +43,9 @@ export function getDefaultAttachmentBaseDirs(): string[] {
process.env.CODEX_HOME || (home ? path.join(home, '.codex') : null);
// Keep defaults narrow. Runtime-specific workspaces must be passed via
// attachmentBaseDirs so one room cannot attach another room's files by path.
// Ad-hoc generated files under os.tmpdir()/ejclaw-attachment-* are handled
// separately after resolving symlinks, without allowlisting all of /tmp.
return unique([
'/tmp',
path.join(DATA_DIR, 'attachments'),
codexHome ? path.join(codexHome, 'generated_images') : null,
])
@@ -59,6 +65,24 @@ function matchesAllowedBaseDir(realPath: string, baseDirs: string[]): boolean {
return baseDirs.some((baseDir) => isWithinBaseDir(realPath, baseDir));
}
function isWithinDefaultTempAttachmentDir(
realPath: string,
tempDir: string | null,
): boolean {
if (!tempDir) return false;
const relative = path.relative(tempDir, realPath);
if (!relative || relative.startsWith('..') || path.isAbsolute(relative)) {
return false;
}
const [firstSegment, ...rest] = relative.split(path.sep);
return (
rest.length > 0 &&
DEFAULT_TEMP_ATTACHMENT_DIR_PREFIXES.some((prefix) =>
firstSegment.startsWith(prefix),
)
);
}
function detectImageMime(filePath: string): string | null {
const handle = fs.openSync(filePath, 'r');
try {
@@ -114,6 +138,7 @@ export function validateOutboundAttachments(
...getDefaultAttachmentBaseDirs(),
...(options.baseDirs ?? []).map(resolveExistingDir),
]).filter((dir): dir is string => Boolean(dir));
const defaultTempDir = resolveExistingDir(os.tmpdir());
const files: ValidatedOutboundAttachment[] = [];
const rejected: Array<{ path: string; reason: string }> = [];
const seen = new Set<string>();
@@ -144,7 +169,10 @@ export function validateOutboundAttachments(
rejected.push({ path: requestedPath, reason: 'too-large' });
continue;
}
if (!matchesAllowedBaseDir(realPath, baseDirs)) {
if (
!matchesAllowedBaseDir(realPath, baseDirs) &&
!isWithinDefaultTempAttachmentDir(realPath, defaultTempDir)
) {
rejected.push({ path: requestedPath, reason: 'outside-allowed-dirs' });
continue;
}

View File

@@ -802,6 +802,42 @@ describe('paired execution context', () => {
).not.toHaveBeenCalled();
});
it('re-anchors the owner workspace before handling a successful owner completion', () => {
vi.mocked(db.getPairedTaskById).mockReturnValue(
buildPairedTask({
status: 'active',
round_trip_count: 1,
}),
);
vi.mocked(pairedWorkspaceManager.markPairedTaskReviewReady).mockReturnValue(
{
ownerWorkspace: buildWorkspace('owner', '/tmp/paired/task-1/owner'),
reviewerWorkspace: buildWorkspace(
'reviewer',
'/tmp/paired/task-1/reviewer',
),
},
);
completePairedExecutionContext({
taskId: 'task-1',
role: 'owner',
status: 'succeeded',
summary: 'STEP_DONE\n1단계 완료, 후속 작업 계속',
});
expect(
pairedWorkspaceManager.provisionOwnerWorkspaceForPairedTask,
).toHaveBeenCalledWith('task-1');
const reanchorCallOrder = vi.mocked(
pairedWorkspaceManager.provisionOwnerWorkspaceForPairedTask,
).mock.invocationCallOrder[0];
const reviewReadyCallOrder = vi.mocked(
pairedWorkspaceManager.markPairedTaskReviewReady,
).mock.invocationCallOrder[0];
expect(reanchorCallOrder).toBeLessThan(reviewReadyCallOrder);
});
it('requests review when the owner reports STEP_DONE in active mode', () => {
vi.mocked(db.getPairedTaskById).mockReturnValue(
buildPairedTask({

View File

@@ -676,6 +676,27 @@ export function completePairedExecutionContext(args: {
}
if (role === 'owner') {
try {
provisionOwnerWorkspaceForPairedTask(taskId);
} catch (error) {
if (isOwnerWorkspaceRepairNeededError(error)) {
logger.warn(
{
taskId,
role,
repairMessage: error.blockMessage || error.message,
},
'Owner workspace post-run guard blocked completion handling',
);
handleFailedOwnerExecution({
task,
taskId,
summary: error.blockMessage || error.message,
});
return;
}
throw error;
}
handleOwnerCompletion({ task, taskId, summary: args.summary });
return;
}

View File

@@ -1145,7 +1145,7 @@ describe('paired workspace manager', () => {
).toBe(ownerBranchName('named-existing-room'));
});
it('blocks provisioning when a named owner workspace branch mismatch has local changes', async () => {
it('re-anchors a dirty named owner workspace without touching local changes', async () => {
const { db, manager } = await loadModules();
db._initTestDatabase();
@@ -1156,6 +1156,10 @@ describe('paired workspace manager', () => {
groupFolder: 'dirty-named-room',
});
runGit(
['branch', ownerBranchName('dirty-named-room'), 'HEAD'],
canonicalDir,
);
const workspaceDir = ownerWorkspacePath('dirty-named-room');
fs.mkdirSync(path.dirname(workspaceDir), { recursive: true });
runGit(
@@ -1173,14 +1177,90 @@ describe('paired workspace manager', () => {
path.join(workspaceDir, 'README.md'),
'dirty named branch\n',
);
fs.writeFileSync(path.join(workspaceDir, 'NOTES.md'), 'untracked keep\n');
const dirtyBefore = runGit(['status', '--short'], workspaceDir);
expect(() =>
manager.provisionOwnerWorkspaceForPairedTask(
'paired-task-dirty-named-branch',
const ownerWorkspace = manager.provisionOwnerWorkspaceForPairedTask(
'paired-task-dirty-named-branch',
);
expect(
runGit(['branch', '--show-current'], ownerWorkspace.workspace_dir),
).toBe(ownerBranchName('dirty-named-room'));
expect(runGit(['status', '--short'], ownerWorkspace.workspace_dir)).toBe(
dirtyBefore,
);
expect(
fs.readFileSync(
path.join(ownerWorkspace.workspace_dir, 'README.md'),
'utf-8',
),
).toThrow(/Owner workspace needs repair/);
expect(runGit(['branch', '--show-current'], workspaceDir)).toBe(
'codex/owner/dirty-named-room-sync',
).toBe('dirty named branch\n');
expect(
fs.readFileSync(
path.join(ownerWorkspace.workspace_dir, 'NOTES.md'),
'utf-8',
),
).toBe('untracked keep\n');
expect(
runGit(['branch', '--list', 'backup/dirty-named-room-*'], canonicalDir),
).toContain('backup/dirty-named-room-current-pre-reanchor-');
});
it('re-anchors a clean divergent named owner workspace to the current feature branch head', async () => {
const { db, manager } = await loadModules();
db._initTestDatabase();
const canonicalDir = path.join(tempRoot, 'canonical');
initCanonicalRepo(canonicalDir);
seedPairedTask(db, canonicalDir, {
taskId: 'paired-task-divergent-named-branch',
groupFolder: 'divergent-named-room',
});
runGit(
['branch', ownerBranchName('divergent-named-room'), 'HEAD'],
canonicalDir,
);
const workspaceDir = ownerWorkspacePath('divergent-named-room');
fs.mkdirSync(path.dirname(workspaceDir), { recursive: true });
runGit(
[
'worktree',
'add',
'-b',
'codex/feature-divergent-named-room',
workspaceDir,
'HEAD',
],
canonicalDir,
);
fs.writeFileSync(path.join(workspaceDir, 'README.md'), 'feature head\n');
runGit(['add', 'README.md'], workspaceDir);
runGit(['commit', '-m', 'feature work'], workspaceDir);
const featureHead = runGit(['rev-parse', 'HEAD'], workspaceDir);
const ownerWorkspace = manager.provisionOwnerWorkspaceForPairedTask(
'paired-task-divergent-named-branch',
);
expect(
runGit(['branch', '--show-current'], ownerWorkspace.workspace_dir),
).toBe(ownerBranchName('divergent-named-room'));
expect(
runGit(
['rev-parse', ownerBranchName('divergent-named-room')],
canonicalDir,
),
).toBe(featureHead);
expect(
fs.readFileSync(
path.join(ownerWorkspace.workspace_dir, 'README.md'),
'utf-8',
),
).toBe('feature head\n');
expect(runGit(['status', '--short'], ownerWorkspace.workspace_dir)).toBe(
'',
);
});

View File

@@ -152,6 +152,74 @@ Workspace: \`${args.workspaceDir}\`
Repair the owner workspace branch, then retry the task.`;
}
function buildOwnerReanchorBackupPrefix(targetBranch: string): string {
const groupFolder = targetBranch.startsWith('codex/owner/')
? targetBranch.slice('codex/owner/'.length)
: targetBranch.replace(/\//g, '-');
return `backup/${groupFolder}`;
}
function buildOwnerReanchorBackupSuffix(): string {
const timestamp = new Date()
.toISOString()
.replace(/[-:.TZ]/g, '')
.slice(0, 14);
return `${timestamp}-${crypto.randomBytes(3).toString('hex')}`;
}
function reanchorNamedOwnerWorkspaceBranch(args: {
canonicalWorkDir: string;
workspaceDir: string;
currentBranch: string;
targetBranch: string;
targetBranchCommit: string | null;
currentHeadCommit: string;
reason: string;
}): boolean {
const {
canonicalWorkDir,
workspaceDir,
currentBranch,
targetBranch,
targetBranchCommit,
currentHeadCommit,
reason,
} = args;
ensureBranchNotCheckedOutElsewhere(
canonicalWorkDir,
workspaceDir,
targetBranch,
);
const backupPrefix = buildOwnerReanchorBackupPrefix(targetBranch);
const backupSuffix = buildOwnerReanchorBackupSuffix();
const currentBackupBranch = `${backupPrefix}-current-pre-reanchor-${backupSuffix}`;
const targetBackupBranch = `${backupPrefix}-target-pre-reanchor-${backupSuffix}`;
runGit(['branch', currentBackupBranch, currentBranch], workspaceDir);
if (targetBranchCommit) {
runGit(['branch', targetBackupBranch, targetBranch], workspaceDir);
}
runGit(['branch', '-f', targetBranch, currentHeadCommit], workspaceDir);
runGit(['symbolic-ref', 'HEAD', branchRefName(targetBranch)], workspaceDir);
logger.warn(
{
workspaceDir,
previousBranch: currentBranch,
targetBranch,
currentHeadCommit,
targetBranchCommit,
currentBackupBranch,
targetBackupBranch: targetBranchCommit ? targetBackupBranch : null,
reason,
},
'Re-anchored owner workspace branch mismatch while preserving worktree state',
);
return true;
}
function maybeRepairNamedOwnerWorkspaceBranch(args: {
canonicalWorkDir: string;
workspaceDir: string;
@@ -174,32 +242,39 @@ function maybeRepairNamedOwnerWorkspaceBranch(args: {
}
if (!isGitWorktreeClean(workspaceDir)) {
throw new OwnerWorkspaceRepairNeededError(
buildOwnerWorkspaceRepairBlockMessage({
workspaceDir,
currentBranch,
targetBranch,
reason:
'Automatic repair was skipped because the workspace has local changes.',
}),
);
return reanchorNamedOwnerWorkspaceBranch({
canonicalWorkDir,
workspaceDir,
currentBranch,
targetBranch,
targetBranchCommit,
currentHeadCommit,
reason: 'workspace has local changes',
});
}
if (!targetBranchCommit) {
runGit(['switch', '-c', targetBranch], workspaceDir);
return true;
return reanchorNamedOwnerWorkspaceBranch({
canonicalWorkDir,
workspaceDir,
currentBranch,
targetBranch,
targetBranchCommit,
currentHeadCommit,
reason: 'expected branch does not exist yet',
});
}
if (targetBranchCommit !== currentHeadCommit) {
throw new OwnerWorkspaceRepairNeededError(
buildOwnerWorkspaceRepairBlockMessage({
workspaceDir,
currentBranch,
targetBranch,
reason:
'Automatic repair was skipped because the expected branch points at a different commit.',
}),
);
return reanchorNamedOwnerWorkspaceBranch({
canonicalWorkDir,
workspaceDir,
currentBranch,
targetBranch,
targetBranchCommit,
currentHeadCommit,
reason: 'expected branch points at a different commit',
});
}
ensureBranchNotCheckedOutElsewhere(

330
src/settings-store.ts Normal file
View File

@@ -0,0 +1,330 @@
/**
* Settings store for the web dashboard:
* - lists Claude / Codex accounts (safe metadata only, no tokens)
* - reads/writes model role configuration via .env
* - removes account directories on the filesystem
*
* Account directory layout (existing convention):
* Claude default: ~/.claude/.credentials.json
* Claude index N≥1: ~/.claude-accounts/{N}/.credentials.json
* Codex default: ~/.codex/auth.json
* Codex index N≥1: ~/.codex-accounts/{N}/auth.json
*/
import fs from 'node:fs';
import os from 'node:os';
import path from 'node:path';
export interface ClaudeAccountSummary {
index: number;
expiresAt: number | null;
scopes: string[];
subscriptionType?: string;
rateLimitTier?: string;
exists: boolean;
}
export interface CodexAccountSummary {
index: number;
accountId: string | null;
planType: string | null;
subscriptionUntil: string | null;
exists: boolean;
}
export interface ModelRoleConfig {
model: string;
effort: string;
}
export interface ModelConfigSnapshot {
owner: ModelRoleConfig;
reviewer: ModelRoleConfig;
arbiter: ModelRoleConfig;
}
const ROLE_KEYS = ['OWNER', 'REVIEWER', 'ARBITER'] as const;
type RoleKey = (typeof ROLE_KEYS)[number];
function envFilePath(): string {
return path.join(process.cwd(), '.env');
}
function readJson<T>(file: string): T | null {
try {
if (!fs.existsSync(file)) return null;
return JSON.parse(fs.readFileSync(file, 'utf-8')) as T;
} catch {
return null;
}
}
function claudeCredsPath(index: number): string {
if (index === 0) {
return path.join(os.homedir(), '.claude', '.credentials.json');
}
return path.join(
os.homedir(),
'.claude-accounts',
String(index),
'.credentials.json',
);
}
function codexAuthPath(index: number): string {
if (index === 0) {
return path.join(os.homedir(), '.codex', 'auth.json');
}
return path.join(os.homedir(), '.codex-accounts', String(index), 'auth.json');
}
function readClaudeAccount(index: number): ClaudeAccountSummary | null {
const file = claudeCredsPath(index);
if (!fs.existsSync(file)) return null;
const data = readJson<{ claudeAiOauth?: Record<string, unknown> }>(file);
const oauth = data?.claudeAiOauth ?? {};
return {
index,
expiresAt: typeof oauth.expiresAt === 'number' ? oauth.expiresAt : null,
scopes: Array.isArray(oauth.scopes) ? (oauth.scopes as string[]) : [],
subscriptionType:
typeof oauth.subscriptionType === 'string'
? oauth.subscriptionType
: undefined,
rateLimitTier:
typeof oauth.rateLimitTier === 'string' ? oauth.rateLimitTier : undefined,
exists: true,
};
}
function readCodexAccount(index: number): CodexAccountSummary | null {
const file = codexAuthPath(index);
if (!fs.existsSync(file)) return null;
const data = readJson<{
OPENAI_API_KEY?: string;
tokens?: { id_token?: string; access_token?: string };
}>(file);
let accountId: string | null = null;
let planType: string | null = null;
let subscriptionUntil: string | null = null;
const idToken = data?.tokens?.id_token;
if (idToken && typeof idToken === 'string') {
const parts = idToken.split('.');
if (parts.length >= 2) {
try {
const payload = JSON.parse(
Buffer.from(parts[1], 'base64').toString('utf-8'),
) as Record<string, unknown>;
if (typeof payload.sub === 'string') accountId = payload.sub;
const auth = payload['https://api.openai.com/auth'] as
| Record<string, unknown>
| undefined;
if (auth) {
if (typeof auth.chatgpt_plan_type === 'string') {
planType = auth.chatgpt_plan_type;
}
if (typeof auth.chatgpt_subscription_active_until === 'string') {
subscriptionUntil = auth.chatgpt_subscription_active_until;
}
}
} catch {
/* ignore parse errors */
}
}
}
return {
index,
accountId,
planType,
subscriptionUntil,
exists: true,
};
}
export function listClaudeAccounts(): ClaudeAccountSummary[] {
const out: ClaudeAccountSummary[] = [];
const def = readClaudeAccount(0);
if (def) out.push(def);
const dir = path.join(os.homedir(), '.claude-accounts');
if (fs.existsSync(dir)) {
const entries = fs.readdirSync(dir);
const indices = entries
.map((e) => Number.parseInt(e, 10))
.filter((n) => Number.isFinite(n) && n >= 1)
.sort((a, b) => a - b);
for (const i of indices) {
const acc = readClaudeAccount(i);
if (acc) out.push(acc);
}
}
return out;
}
export function listCodexAccounts(): CodexAccountSummary[] {
const out: CodexAccountSummary[] = [];
const def = readCodexAccount(0);
if (def) out.push(def);
const dir = path.join(os.homedir(), '.codex-accounts');
if (fs.existsSync(dir)) {
const entries = fs.readdirSync(dir);
const indices = entries
.map((e) => Number.parseInt(e, 10))
.filter((n) => Number.isFinite(n) && n >= 1)
.sort((a, b) => a - b);
for (const i of indices) {
const acc = readCodexAccount(i);
if (acc) out.push(acc);
}
}
return out;
}
function pickEnvValue(content: string, key: string): string | undefined {
const re = new RegExp(`^${key}=(.*)$`, 'm');
const match = content.match(re);
if (!match) return undefined;
return match[1].trim().replace(/^['"]|['"]$/g, '');
}
function readEnvFile(): string {
const file = envFilePath();
if (!fs.existsSync(file)) return '';
return fs.readFileSync(file, 'utf-8');
}
function readEnvOrProcess(key: string): string | undefined {
const fromFile = pickEnvValue(readEnvFile(), key);
if (fromFile !== undefined) return fromFile;
const fromProc = process.env[key];
return fromProc;
}
export function getModelConfig(): ModelConfigSnapshot {
const out: Partial<ModelConfigSnapshot> = {};
for (const role of ROLE_KEYS) {
const model = readEnvOrProcess(`${role}_MODEL`) ?? '';
const effort = readEnvOrProcess(`${role}_EFFORT`) ?? '';
(out as Record<string, ModelRoleConfig>)[role.toLowerCase()] = {
model,
effort,
};
}
return out as ModelConfigSnapshot;
}
export interface ModelUpdateInput {
owner?: Partial<ModelRoleConfig>;
reviewer?: Partial<ModelRoleConfig>;
arbiter?: Partial<ModelRoleConfig>;
}
function setOrInsertEnvLine(
content: string,
key: string,
value: string,
): string {
const re = new RegExp(`^${key}=.*$`, 'm');
if (re.test(content)) {
return content.replace(re, `${key}=${value}`);
}
const trimmed = content.replace(/\s*$/, '');
return `${trimmed}\n${key}=${value}\n`;
}
export function updateModelConfig(
input: ModelUpdateInput,
): ModelConfigSnapshot {
const file = envFilePath();
let content = '';
if (fs.existsSync(file)) {
content = fs.readFileSync(file, 'utf-8');
}
for (const role of ROLE_KEYS) {
const update = (
input as Record<string, Partial<ModelRoleConfig> | undefined>
)[role.toLowerCase()];
if (!update) continue;
if (update.model !== undefined) {
content = setOrInsertEnvLine(content, `${role}_MODEL`, update.model);
}
if (update.effort !== undefined) {
content = setOrInsertEnvLine(content, `${role}_EFFORT`, update.effort);
}
}
const tempPath = `${file}.tmp`;
fs.writeFileSync(tempPath, content, { mode: 0o600 });
fs.renameSync(tempPath, file);
return getModelConfig();
}
export function removeAccountDirectory(
provider: 'claude' | 'codex',
index: number,
): void {
if (!Number.isFinite(index) || index < 1) {
throw new Error('cannot remove default account (index 0)');
}
const baseDir =
provider === 'claude'
? path.join(os.homedir(), '.claude-accounts', String(index))
: path.join(os.homedir(), '.codex-accounts', String(index));
if (!fs.existsSync(baseDir)) {
throw new Error(`account directory not found: ${baseDir}`);
}
fs.rmSync(baseDir, { recursive: true, force: true });
}
export function addClaudeAccountFromToken(token: string): {
index: number;
accountId: string | null;
} {
const trimmed = token.trim();
if (!trimmed) throw new Error('empty token');
const parts = trimmed.split('.');
if (parts.length < 2) {
throw new Error('invalid OAuth token: expected JWT format');
}
let payload: Record<string, unknown> = {};
try {
payload = JSON.parse(
Buffer.from(parts[1], 'base64').toString('utf-8'),
) as Record<string, unknown>;
} catch {
throw new Error('invalid OAuth token: payload not parseable');
}
const exp = typeof payload.exp === 'number' ? payload.exp * 1000 : 0;
const accountId = typeof payload.sub === 'string' ? payload.sub : null;
const baseDir = path.join(os.homedir(), '.claude-accounts');
if (!fs.existsSync(baseDir)) {
fs.mkdirSync(baseDir, { recursive: true, mode: 0o700 });
}
let nextIndex = 1;
while (
fs.existsSync(path.join(baseDir, String(nextIndex), '.credentials.json'))
) {
nextIndex += 1;
}
const dir = path.join(baseDir, String(nextIndex));
fs.mkdirSync(dir, { recursive: true, mode: 0o700 });
const creds = {
claudeAiOauth: {
accessToken: trimmed,
refreshToken: '',
expiresAt: exp,
scopes: ['user:profile', 'user:inference'],
},
};
const credsPath = path.join(dir, '.credentials.json');
const tempPath = `${credsPath}.tmp`;
fs.writeFileSync(tempPath, JSON.stringify(creds, null, 2), { mode: 0o600 });
fs.renameSync(tempPath, credsPath);
return { index: nextIndex, accountId };
}

View File

@@ -2,6 +2,7 @@ import { describe, expect, it } from 'vitest';
import type { UsageRow } from './dashboard-usage-rows.js';
import {
buildWebUsageRowsForSnapshot,
formatStatusHeader,
renderUsageTable,
summarizeWatcherTasks,
@@ -107,3 +108,94 @@ describe('renderUsageTable', () => {
expect(lines).toEqual(['_조회 불가_']);
});
});
describe('buildWebUsageRowsForSnapshot', () => {
it('keeps real Claude and Kimi rows ahead of Codex rows for web snapshots', () => {
const rows = buildWebUsageRowsForSnapshot({
serviceAgentType: 'claude-code',
claudeAccounts: [
{
index: 0,
masked: 'claude-1',
isActive: true,
isRateLimited: false,
usage: {
five_hour: {
utilization: 0.2,
resets_at: '2026-04-26T12:00:00.000Z',
},
seven_day: {
utilization: 0.4,
resets_at: '2026-04-27T12:00:00.000Z',
},
},
},
{
index: 1,
masked: 'claude-2',
isActive: false,
isRateLimited: false,
usage: {
five_hour: {
utilization: 0.3,
resets_at: '2026-04-26T12:00:00.000Z',
},
seven_day: {
utilization: 0.5,
resets_at: '2026-04-27T12:00:00.000Z',
},
},
},
],
kimiUsage: {
fiveHour: { pct: 18, resetTime: '2026-04-26T12:00:00.000Z' },
weekly: { pct: 29, resetTime: '2026-04-27T12:00:00.000Z' },
},
codexRows: [
{
name: 'Codex1',
h5pct: 25,
h5reset: '1h',
d7pct: 35,
d7reset: '2d',
},
],
});
const names = rows.map((row) => row.name);
expect(names[0]).toMatch(/^Claude1/);
expect(names[1]).toMatch(/^Claude2/);
expect(names[2]).toMatch(/^Kimi/);
expect(names[3]).toBe('Codex1');
});
it('does not invent Claude or Kimi rows for codex-only services', () => {
const rows = buildWebUsageRowsForSnapshot({
serviceAgentType: 'codex',
claudeAccounts: [
{
index: 0,
masked: 'claude-1',
isActive: true,
isRateLimited: false,
usage: null,
},
],
kimiUsage: {
fiveHour: { pct: 18, resetTime: '2026-04-26T12:00:00.000Z' },
weekly: { pct: 29, resetTime: '2026-04-27T12:00:00.000Z' },
},
codexRows: [
{
name: 'Codex1',
h5pct: 25,
h5reset: '1h',
d7pct: 35,
d7reset: '2d',
},
],
});
expect(rows.map((row) => row.name)).toEqual(['Codex1']);
});
});

View File

@@ -14,9 +14,14 @@ import {
STATUS_SHOW_ROOM_DETAILS,
STATUS_SHOW_ROOMS,
USAGE_DASHBOARD_ENABLED,
CODEX_WARMUP_CONFIG,
getMoaConfig,
} from './config.js';
import { fetchKimiUsage, buildKimiUsageRows } from './kimi-usage.js';
import {
fetchKimiUsage,
buildKimiUsageRows,
type KimiUsageData,
} from './kimi-usage.js';
import { getGlobalFailoverInfo } from './service-routing.js';
import {
fetchAllClaudeUsage,
@@ -28,6 +33,7 @@ import {
refreshActiveCodexUsage,
refreshAllCodexAccountUsage,
} from './codex-usage-collector.js';
import { runCodexWarmupCycle } from './codex-warmup.js';
import {
composeDashboardContent,
formatElapsed,
@@ -91,7 +97,7 @@ const RENDERER_USAGE_REFRESH_MS = 30_000;
let statusMessageId: string | null = null;
let cachedUsageContent = '';
let cachedClaudeAccounts: ClaudeAccountUsage[] = [];
let cachedKimiUsage: import('./kimi-usage.js').KimiUsageData | null = null;
let cachedKimiUsage: KimiUsageData | null = null;
let usageUpdateInProgress = false;
let channelMetaCache = new Map<string, ChannelMeta>();
let channelMetaLastRefresh = 0;
@@ -100,6 +106,8 @@ let dashboardUpdateLogged = false;
let cachedCodexUsageRows: UsageRow[] = [];
/** Codex service only: ISO timestamp of last successful usage fetch. */
let codexUsageFetchedAt: string | null = null;
/** Renderer service only: ISO timestamp of last successful Claude/Kimi usage render. */
let rendererUsageFetchedAt: string | null = null;
export interface WatcherTaskSummary {
active: number;
@@ -229,9 +237,47 @@ function formatRoomName(
return base;
}
export function buildWebUsageRowsForSnapshot(args: {
serviceAgentType: AgentType;
claudeAccounts: ClaudeAccountUsage[];
kimiUsage: KimiUsageData | null;
codexRows: UsageRow[];
}): UsageRow[] {
const rows: UsageRow[] = [];
if (args.serviceAgentType === 'claude-code') {
rows.push(...buildClaudeUsageRows(args.claudeAccounts));
rows.push(...buildKimiUsageRows(args.kimiUsage));
}
rows.push(...args.codexRows);
return rows;
}
function buildUsageSnapshotRows(opts: UnifiedDashboardOptions): {
rows: UsageRow[];
fetchedAt: string | null;
} {
const rows = buildWebUsageRowsForSnapshot({
serviceAgentType: opts.serviceAgentType,
claudeAccounts: cachedClaudeAccounts,
kimiUsage: cachedKimiUsage,
codexRows: cachedCodexUsageRows,
});
const fetchedAt =
[rendererUsageFetchedAt, codexUsageFetchedAt]
.filter((value): value is string => !!value)
.sort()
.at(-1) ?? null;
return { rows, fetchedAt };
}
function writeLocalStatusSnapshot(opts: UnifiedDashboardOptions): void {
const groups = opts.roomBindings();
const statuses = opts.queue.getStatuses(Object.keys(groups));
const usageSnapshot = buildUsageSnapshotRows(opts);
writeStatusSnapshot({
serviceId: opts.serviceId,
@@ -265,8 +311,10 @@ function writeLocalStatusSnapshot(opts: UnifiedDashboardOptions): void {
pendingMessages: boolean;
pendingTasks: number;
}>,
...(cachedCodexUsageRows.length > 0 && { usageRows: cachedCodexUsageRows }),
...(codexUsageFetchedAt && { usageRowsFetchedAt: codexUsageFetchedAt }),
...(usageSnapshot.rows.length > 0 && { usageRows: usageSnapshot.rows }),
...(usageSnapshot.fetchedAt && {
usageRowsFetchedAt: usageSnapshot.fetchedAt,
}),
});
}
@@ -645,6 +693,7 @@ async function refreshUsageCache(): Promise<void> {
usageUpdateInProgress = true;
try {
cachedUsageContent = await buildUsageContent();
rendererUsageFetchedAt = new Date().toISOString();
} catch (err) {
logger.warn({ err }, 'Failed to build usage content');
} finally {
@@ -735,18 +784,49 @@ export async function startUnifiedDashboard(
cachedCodexUsageRows = result.rows;
if (result.fetchedAt) codexUsageFetchedAt = result.fetchedAt;
};
void refreshAllCodexAccountUsage().then((r) => {
applyCodexRefresh(r);
return refreshActiveCodexUsage().then(applyCodexRefresh);
});
const isWarmupRuntimeBusy = () => {
const groups = opts.roomBindings();
return opts.queue
.getStatuses(Object.keys(groups))
.some((status) => status.status === 'processing');
};
let codexWarmupInFlight = false;
const runCodexWarmup = async () => {
if (!CODEX_WARMUP_CONFIG.enabled || codexWarmupInFlight) return;
codexWarmupInFlight = true;
try {
const result = await runCodexWarmupCycle(CODEX_WARMUP_CONFIG, {
shouldSkip: isWarmupRuntimeBusy,
});
if (result.status === 'warmed') {
applyCodexRefresh(await refreshAllCodexAccountUsage());
}
} catch (err) {
logger.warn({ err }, 'Codex warm-up cycle failed unexpectedly');
} finally {
codexWarmupInFlight = false;
}
};
void refreshAllCodexAccountUsage()
.then((r) => {
applyCodexRefresh(r);
return refreshActiveCodexUsage().then(applyCodexRefresh);
})
.then(() => runCodexWarmup());
setInterval(
() => void refreshActiveCodexUsage().then(applyCodexRefresh),
opts.usageUpdateInterval,
);
setInterval(
() => void refreshAllCodexAccountUsage().then(applyCodexRefresh),
() =>
void refreshAllCodexAccountUsage()
.then(applyCodexRefresh)
.then(() => runCodexWarmup()),
CODEX_FULL_SCAN_INTERVAL,
);
if (CODEX_WARMUP_CONFIG.enabled) {
setInterval(() => void runCodexWarmup(), CODEX_WARMUP_CONFIG.intervalMs);
}
logger.info(
{

View File

@@ -106,6 +106,66 @@ describe('verification helpers', () => {
});
});
it('verifies nested pnpm workspaces under a bun parent without tripping corepack project specs', async () => {
const parentDir = fs.mkdtempSync(
path.join(os.tmpdir(), 'ejclaw-verification-corepack-parent-'),
);
fs.writeFileSync(
path.join(parentDir, 'package.json'),
JSON.stringify({ packageManager: 'bun@1.3.11' }),
);
const repoDir = path.join(parentDir, 'owner');
fs.mkdirSync(repoDir, { recursive: true });
fs.writeFileSync(
path.join(repoDir, 'package.json'),
JSON.stringify({
name: 'nested-pnpm-workspace',
scripts: {
typecheck:
'node -e "process.stdout.write(process.env.COREPACK_ENABLE_PROJECT_SPEC || \'missing\')"',
},
}),
);
fs.writeFileSync(
path.join(repoDir, 'pnpm-lock.yaml'),
[
"lockfileVersion: '6.0'",
'settings:',
' autoInstallPeers: true',
' excludeLinksFromLockfile: false',
'importers:',
' .: {}',
'',
].join('\n'),
);
fs.mkdirSync(path.join(repoDir, 'node_modules', '.bin'), {
recursive: true,
});
fs.writeFileSync(
path.join(repoDir, 'node_modules', '.bin', 'placeholder'),
'',
);
expect(hasInstalledNodeModules(repoDir)).toBe(false);
const expectedSnapshotId = computeVerificationSnapshot(repoDir).snapshotId;
const result = await runVerificationRequest(
{
requestId: 'req-corepack-parent-pnpm',
profile: 'typecheck',
expectedSnapshotId,
},
{ repoDir },
);
expect(result.ok).toBe(true);
expect(result.exitCode).toBe(0);
expect(result.command).toBe('corepack pnpm run typecheck');
expect(result.stdout).toContain('0');
expect(result.snapshotId).toBe(expectedSnapshotId);
});
it('computes a stable snapshot over the readable workspace inputs', () => {
const repoDir = fs.mkdtempSync(path.join(os.tmpdir(), 'ejclaw-snapshot-'));
fs.mkdirSync(path.join(repoDir, 'src'), { recursive: true });

View File

@@ -5,9 +5,11 @@ import path from 'path';
import { TIMEZONE } from './config.js';
import {
buildWorkspaceCommandEnvironment,
buildWorkspaceScriptCommand,
ensureWorkspaceDependenciesInstalled,
hasInstalledNodeModules,
type WorkspacePackageManager,
} from './workspace-package-manager.js';
import { computeVerificationSnapshotId } from '../shared/verification-snapshot.js';
@@ -44,6 +46,7 @@ export interface VerificationSnapshot {
}
interface VerificationCommandSpec {
packageManager: WorkspacePackageManager;
file: string;
args: string[];
commandText: string;
@@ -111,6 +114,7 @@ export function buildVerificationCommand(
: 'lint';
const command = buildWorkspaceScriptCommand(repoDir, scriptName);
return {
packageManager: command.packageManager,
file: command.file,
args: command.args,
commandText: command.commandText,
@@ -334,7 +338,11 @@ export async function runVerificationRequest(
command.args,
{
cwd: repoDir,
env: directExecution.env,
env: buildWorkspaceCommandEnvironment(
repoDir,
command.packageManager,
directExecution.env,
),
},
);
const afterSnapshot = computeVerificationSnapshot(repoDir);

View File

@@ -0,0 +1,478 @@
import { describe, expect, it } from 'vitest';
import type { StatusSnapshot } from './status-dashboard.js';
import type { PairedTurnAttemptRecord, PairedTurnRecord } from './db.js';
import type {
NewMessage,
PairedTask,
PairedTurnOutput,
ScheduledTask,
} from './types.js';
import {
buildWebDashboardRoomActivity,
buildWebDashboardOverview,
sanitizeScheduledTask,
} from './web-dashboard-data.js';
function makeTask(overrides: Partial<ScheduledTask>): ScheduledTask {
return {
id: 'task-1',
group_folder: 'general',
chat_jid: 'dc:general',
agent_type: null,
status_message_id: null,
status_started_at: null,
prompt: 'secret long prompt that should not be exposed in full',
schedule_type: 'cron',
schedule_value: '* * * * *',
context_mode: 'group',
next_run: '2026-04-26T05:00:00.000Z',
last_run: null,
last_result: null,
status: 'active',
created_at: '2026-04-26T04:00:00.000Z',
...overrides,
};
}
function makePairedTask(overrides: Partial<PairedTask>): PairedTask {
return {
id: 'paired-1',
chat_jid: 'dc:general',
group_folder: 'general',
owner_service_id: 'codex-main',
reviewer_service_id: 'claude-reviewer',
owner_agent_type: 'codex',
reviewer_agent_type: 'claude-code',
arbiter_agent_type: 'codex',
title: 'Dashboard PR',
source_ref: null,
plan_notes: null,
review_requested_at: null,
round_trip_count: 0,
owner_failure_count: 0,
owner_step_done_streak: 0,
finalize_step_done_count: 0,
task_done_then_user_reopen_count: 0,
empty_step_done_streak: 0,
status: 'review_ready',
arbiter_verdict: null,
arbiter_requested_at: null,
completion_reason: null,
created_at: '2026-04-26T04:00:00.000Z',
updated_at: '2026-04-26T04:30:00.000Z',
...overrides,
};
}
describe('web dashboard data', () => {
it('builds overview counts from status snapshots and scheduled tasks', () => {
const snapshots: StatusSnapshot[] = [
{
serviceId: 'codex-main',
agentType: 'codex',
assistantName: 'Codex',
updatedAt: '2026-04-26T04:59:00.000Z',
entries: [
{
jid: 'dc:1',
name: '#general',
folder: 'general',
agentType: 'codex',
status: 'processing',
elapsedMs: 1200,
pendingMessages: true,
pendingTasks: 2,
},
{
jid: 'dc:2',
name: '#brain',
folder: 'brain',
agentType: 'claude-code',
status: 'inactive',
elapsedMs: null,
pendingMessages: false,
pendingTasks: 0,
},
],
usageRows: [
{
name: 'codex-a',
h5pct: 10,
h5reset: '1h',
d7pct: 20,
d7reset: '2d',
},
],
},
];
const overview = buildWebDashboardOverview({
now: '2026-04-26T05:00:00.000Z',
snapshots,
tasks: [
makeTask({
id: 'watch-1',
prompt: '[BACKGROUND CI WATCH] owner/repo#1',
status: 'active',
}),
makeTask({ id: 'cron-1', prompt: 'regular cleanup', status: 'paused' }),
],
});
expect(overview.rooms.total).toBe(2);
expect(overview.rooms.active).toBe(1);
expect(overview.rooms.waiting).toBe(0);
expect(overview.rooms.inactive).toBe(1);
expect(overview.tasks.total).toBe(2);
expect(overview.tasks.active).toBe(1);
expect(overview.tasks.paused).toBe(1);
expect(overview.tasks.watchers.active).toBe(1);
expect(overview.usage.rows).toHaveLength(1);
});
it('deduplicates full usage rows from renderer and codex snapshots', () => {
const snapshots: StatusSnapshot[] = [
{
serviceId: 'codex-main',
agentType: 'codex',
assistantName: 'Codex',
updatedAt: '2026-04-26T04:59:00.000Z',
entries: [],
usageRowsFetchedAt: '2026-04-26T04:59:00.000Z',
usageRows: [
{
name: 'Codex1',
h5pct: 20,
h5reset: '1h',
d7pct: 30,
d7reset: '2d',
},
{
name: 'Codex2',
h5pct: 15,
h5reset: '1h',
d7pct: 18,
d7reset: '2d',
},
],
},
{
serviceId: 'claude-main',
agentType: 'claude-code',
assistantName: 'Claude',
updatedAt: '2026-04-26T05:00:00.000Z',
entries: [],
usageRowsFetchedAt: '2026-04-26T05:00:00.000Z',
usageRows: [
{
name: 'Claude1 Max',
h5pct: 66,
h5reset: '2h',
d7pct: 40,
d7reset: '4d',
},
{
name: 'Kimi',
h5pct: 10,
h5reset: '3h',
d7pct: 12,
d7reset: '5d',
},
{
name: 'Codex1',
h5pct: 25,
h5reset: '55m',
d7pct: 35,
d7reset: '2d',
},
],
},
];
const overview = buildWebDashboardOverview({
now: '2026-04-26T05:01:00.000Z',
snapshots,
tasks: [],
});
expect(overview.usage.rows.map((row) => row.name)).toEqual([
'Claude1 Max',
'Kimi',
'Codex1',
'Codex2',
]);
expect(
overview.usage.rows.filter((row) => row.name === 'Codex1'),
).toHaveLength(1);
expect(overview.usage.fetchedAt).toBe('2026-04-26T05:00:00.000Z');
});
it('does not expose full scheduled task prompts through API payloads', () => {
const sanitized = sanitizeScheduledTask(
makeTask({
prompt: 'x'.repeat(220),
last_result: 'ok',
}),
);
expect(sanitized).not.toHaveProperty('prompt');
expect(sanitized.promptPreview.length).toBeLessThanOrEqual(123);
expect(sanitized.promptLength).toBe(220);
});
it('redacts common secret values from scheduled task prompt previews', () => {
const sanitized = sanitizeScheduledTask(
makeTask({
prompt:
'deploy with OPENAI_API_KEY=sk-abcdefghijklmnopqrstuvwxyz123456 and BOT_TOKEN=plain-secret-value',
}),
);
expect(sanitized.promptPreview).toContain('OPENAI_API_KEY=<redacted>');
expect(sanitized.promptPreview).toContain('BOT_TOKEN=<redacted>');
expect(sanitized.promptPreview).not.toContain(
'sk-abcdefghijklmnopqrstuvwxyz123456',
);
expect(sanitized.promptPreview).not.toContain('plain-secret-value');
});
it('builds redacted room activity from messages and paired turn output', () => {
const task = makePairedTask({
id: 'paired-room-1',
chat_jid: 'dc:ops',
status: 'in_review',
round_trip_count: 3,
updated_at: '2026-04-26T05:30:00.000Z',
});
const turn: PairedTurnRecord = {
turn_id: 'turn-1',
task_id: task.id,
task_updated_at: task.updated_at,
role: 'reviewer',
intent_kind: 'reviewer-turn',
state: 'queued',
executor_service_id: null,
executor_agent_type: null,
attempt_no: 0,
created_at: '2026-04-26T05:19:00.000Z',
updated_at: '2026-04-26T05:31:00.000Z',
completed_at: null,
last_error: null,
};
const attempt: PairedTurnAttemptRecord = {
attempt_id: 'turn-1:attempt:2',
parent_attempt_id: null,
parent_handoff_id: null,
continuation_handoff_id: null,
turn_id: 'turn-1',
task_id: task.id,
task_updated_at: task.updated_at,
role: 'reviewer',
intent_kind: 'reviewer-turn',
state: 'running',
executor_service_id: 'claude-reviewer',
executor_agent_type: 'claude-code',
active_run_id: 'run-reviewer-1',
attempt_no: 2,
created_at: '2026-04-26T05:20:00.000Z',
updated_at: '2026-04-26T05:31:00.000Z',
completed_at: null,
last_error: 'OPENAI_API_KEY=plain-secret-value',
};
const outputs: PairedTurnOutput[] = [
{
id: 1,
task_id: task.id,
turn_number: 1,
role: 'owner',
output_text: 'owner output',
verdict: 'step_done',
created_at: '2026-04-26T05:25:00.000Z',
},
{
id: 2,
task_id: task.id,
turn_number: 2,
role: 'reviewer',
output_text: 'reviewer output with BOT_TOKEN=plain-secret-value',
verdict: null,
created_at: '2026-04-26T05:30:00.000Z',
},
];
const messages: NewMessage[] = [
{
id: 'msg-1',
chat_jid: 'dc:ops',
sender: 'user-1',
sender_name: '눈쟁이',
content: 'latest message',
timestamp: '2026-04-26T05:29:00.000Z',
is_from_me: false,
is_bot_message: false,
message_source_kind: 'human',
},
];
const activity = buildWebDashboardRoomActivity({
serviceId: 'codex-main',
entry: {
jid: 'dc:ops',
name: '#ops',
folder: 'ops',
agentType: 'codex',
status: 'processing',
elapsedMs: 15_000,
pendingMessages: true,
pendingTasks: 1,
},
pairedTask: task,
turns: [turn],
attempts: [attempt],
outputs,
messages,
outputLimit: 1,
});
expect(activity.pairedTask).toMatchObject({
id: 'paired-room-1',
roundTripCount: 3,
currentTurn: {
role: 'reviewer',
state: 'running',
attemptNo: 2,
lastError: 'OPENAI_API_KEY=<redacted>',
},
outputs: [
{
turnNumber: 2,
role: 'reviewer',
outputText: 'reviewer output with BOT_TOKEN=<redacted>',
},
],
});
expect(activity.messages).toMatchObject([
{ senderName: '눈쟁이', content: 'latest message' },
]);
});
it('builds typed inbox items from pending rooms, paired tasks, and CI failures', () => {
const snapshots: StatusSnapshot[] = [
{
serviceId: 'codex-main',
agentType: 'codex',
assistantName: 'Codex',
updatedAt: '2026-04-26T05:00:00.000Z',
entries: [
{
jid: 'dc:ops',
name: '#ops',
folder: 'ops',
agentType: 'codex',
status: 'waiting',
elapsedMs: 1000,
pendingMessages: true,
pendingTasks: 0,
},
{
jid: 'dc:idle',
name: '#idle',
folder: 'idle',
agentType: 'codex',
status: 'inactive',
elapsedMs: null,
pendingMessages: false,
pendingTasks: 0,
},
],
},
];
const overview = buildWebDashboardOverview({
now: '2026-04-26T05:10:00.000Z',
snapshots,
pairedTasks: [
makePairedTask({
id: 'review-1',
status: 'review_ready',
review_requested_at: '2026-04-26T05:03:00.000Z',
}),
makePairedTask({
id: 'merge-1',
status: 'merge_ready',
title: 'Ready to merge',
updated_at: '2026-04-26T05:04:00.000Z',
}),
makePairedTask({
id: 'done-1',
status: 'completed',
}),
],
tasks: [
makeTask({
id: 'ci-1',
prompt:
'[BACKGROUND CI WATCH]\nWatch target:\nPR #21\n\nCheck instructions:\nwatch',
last_run: '2026-04-26T05:05:00.000Z',
last_result: 'Error: BOT_TOKEN=plain-secret-value failed',
status: 'paused',
}),
makeTask({
id: 'ci-2',
prompt:
'[BACKGROUND CI WATCH]\nWatch target:\nPR #22\n\nCheck instructions:\nwatch',
last_run: '2026-04-26T05:07:00.000Z',
last_result: 'Error: BOT_TOKEN=plain-secret-value failed',
status: 'active',
}),
makeTask({
id: 'cron-1',
prompt: 'regular cron',
last_run: '2026-04-26T05:06:00.000Z',
last_result: 'Error: non-watch task failed',
}),
],
});
expect(overview.inbox.map((item) => item.kind)).toEqual([
'ci-failure',
'approval',
'reviewer-request',
'pending-room',
]);
expect(overview.inbox).toContainEqual(
expect.objectContaining({
id: 'room:codex-main:dc:ops',
kind: 'pending-room',
severity: 'info',
roomJid: 'dc:ops',
serviceId: 'codex-main',
occurredAt: '2026-04-26T05:00:00.000Z',
createdAt: '2026-04-26T05:10:00.000Z',
}),
);
expect(overview.inbox).toContainEqual(
expect.objectContaining({
id: 'paired:review-1:review_ready',
kind: 'reviewer-request',
severity: 'warn',
serviceId: 'claude-reviewer',
taskStatus: 'review_ready',
occurredAt: '2026-04-26T05:03:00.000Z',
}),
);
const ciFailure = overview.inbox.find((item) => item.kind === 'ci-failure');
expect(ciFailure).toMatchObject({
id: 'ci:ci-2',
severity: 'error',
occurrences: 2,
source: 'scheduled-task',
taskId: 'ci-2',
lastOccurredAt: '2026-04-26T05:07:00.000Z',
});
expect(ciFailure?.summary).toContain('BOT_TOKEN=<redacted>');
expect(ciFailure?.summary).not.toContain('plain-secret-value');
expect(overview.inbox.some((item) => item.taskId === 'cron-1')).toBe(false);
expect(overview.inbox.some((item) => item.taskId === 'done-1')).toBe(false);
});
});

668
src/web-dashboard-data.ts Normal file
View File

@@ -0,0 +1,668 @@
import { isWatchCiTask } from './task-watch-status.js';
import type { PairedTurnAttemptRecord, PairedTurnRecord } from './db.js';
import type { StatusSnapshot, UsageRowSnapshot } from './status-dashboard.js';
import type {
NewMessage,
PairedTask,
PairedTurnOutput,
ScheduledTask,
} from './types.js';
export interface SanitizedScheduledTask {
id: string;
groupFolder: string;
chatJid: string;
agentType: ScheduledTask['agent_type'];
ciProvider: ScheduledTask['ci_provider'];
ciMetadata: ScheduledTask['ci_metadata'];
scheduleType: ScheduledTask['schedule_type'];
scheduleValue: string;
contextMode: ScheduledTask['context_mode'];
nextRun: string | null;
lastRun: string | null;
lastResult: string | null;
status: ScheduledTask['status'];
suspendedUntil: string | null;
createdAt: string;
promptPreview: string;
promptLength: number;
isWatcher: boolean;
}
export interface WebDashboardOverview {
generatedAt: string;
services: Array<{
serviceId: string;
assistantName: string;
agentType: StatusSnapshot['agentType'];
updatedAt: string;
totalRooms: number;
activeRooms: number;
}>;
rooms: {
total: number;
active: number;
waiting: number;
inactive: number;
};
tasks: {
total: number;
active: number;
paused: number;
completed: number;
watchers: {
active: number;
paused: number;
completed: number;
};
};
usage: {
rows: UsageRowSnapshot[];
fetchedAt: string | null;
};
inbox: InboxItem[];
}
export interface WebDashboardRoomMessage {
id: string;
sender: string;
senderName: string;
content: string;
timestamp: string;
isFromMe: boolean;
isBotMessage: boolean;
sourceKind: NonNullable<NewMessage['message_source_kind']>;
}
export interface WebDashboardRoomTurn {
turnId: string;
role: PairedTurnRecord['role'];
intentKind: PairedTurnRecord['intent_kind'];
state: PairedTurnRecord['state'];
attemptNo: number;
executorServiceId: string | null;
executorAgentType: PairedTurnRecord['executor_agent_type'];
activeRunId: string | null;
createdAt: string;
updatedAt: string;
completedAt: string | null;
lastError: string | null;
progressText: string | null;
progressUpdatedAt: string | null;
}
export interface WebDashboardRoomTurnOutput {
id: number;
turnNumber: number;
role: PairedTurnOutput['role'];
verdict: PairedTurnOutput['verdict'] | null;
createdAt: string;
outputText: string;
}
export interface WebDashboardRoomActivity {
serviceId: string;
jid: string;
name: string;
folder: string;
agentType: StatusSnapshot['agentType'];
status: StatusSnapshot['entries'][number]['status'];
elapsedMs: number | null;
pendingMessages: boolean;
pendingTasks: number;
messages: WebDashboardRoomMessage[];
pairedTask: {
id: string;
title: string | null;
status: PairedTask['status'];
roundTripCount: number;
updatedAt: string;
currentTurn: WebDashboardRoomTurn | null;
outputs: WebDashboardRoomTurnOutput[];
} | null;
}
export type InboxItemKind =
| 'pending-room'
| 'reviewer-request'
| 'approval'
| 'arbiter-request'
| 'ci-failure'
| 'mention';
export type InboxItemSeverity = 'info' | 'warn' | 'error';
export interface InboxItem {
id: string;
groupKey: string;
kind: InboxItemKind;
severity: InboxItemSeverity;
title: string;
summary: string;
occurredAt: string;
lastOccurredAt: string;
createdAt: string;
occurrences: number;
source: 'status-snapshot' | 'paired-task' | 'scheduled-task';
roomJid?: string;
roomName?: string;
groupFolder?: string;
serviceId?: string;
taskId?: string;
taskStatus?: string;
}
const SECRET_ASSIGNMENT_RE =
/\b([A-Z0-9_]*(?:TOKEN|SECRET|PASSWORD|API_KEY|AUTH|PRIVATE_KEY)[A-Z0-9_]*)\s*=\s*([^\s"'`]+)/gi;
const SECRET_VALUE_RE =
/\b(?:sk-[A-Za-z0-9_-]{12,}|gh[pousr]_[A-Za-z0-9_]{12,}|xox[baprs]-[A-Za-z0-9-]{12,}|eyJ[A-Za-z0-9_-]{20,}\.[A-Za-z0-9_-]{20,}\.[A-Za-z0-9_-]{10,})\b/g;
const ROOM_MESSAGE_PREVIEW_MAX_LENGTH = 900;
const ROOM_OUTPUT_PREVIEW_MAX_LENGTH = 1800;
function redactSensitiveText(value: string): string {
return value
.replace(SECRET_ASSIGNMENT_RE, '$1=<redacted>')
.replace(SECRET_VALUE_RE, '<redacted-token>');
}
function truncateText(value: string, maxLength = 120): string {
if (value.length <= maxLength) return value;
return `${value.slice(0, maxLength)}...`;
}
function buildPromptPreview(prompt: string): string {
return truncateText(redactSensitiveText(prompt).replace(/\s+/g, ' ').trim());
}
function buildInboxPreview(value: string): string {
return truncateText(redactSensitiveText(value).replace(/\s+/g, ' ').trim());
}
function buildRoomPreview(value: string, maxLength: number): string {
return truncateText(redactSensitiveText(value).trim(), maxLength);
}
function stableHash(value: string): string {
let hash = 0;
for (let index = 0; index < value.length; index += 1) {
hash = (hash * 31 + value.charCodeAt(index)) | 0;
}
return Math.abs(hash).toString(36);
}
function collectUsageRows(snapshots: StatusSnapshot[]): UsageRowSnapshot[] {
const rows: UsageRowSnapshot[] = [];
const seen = new Set<string>();
const sortedSnapshots = [...snapshots].sort((a, b) =>
(b.usageRowsFetchedAt ?? b.updatedAt).localeCompare(
a.usageRowsFetchedAt ?? a.updatedAt,
),
);
for (const snapshot of sortedSnapshots) {
for (const row of snapshot.usageRows ?? []) {
if (seen.has(row.name)) continue;
seen.add(row.name);
rows.push(row);
}
}
return rows;
}
function failedTaskResult(task: ScheduledTask): string | null {
if (!task.last_result) return null;
const normalized = task.last_result.toLowerCase();
if (
normalized.includes('fail') ||
normalized.includes('error') ||
normalized.includes('timeout') ||
normalized.includes('cancel') ||
normalized.includes('reject')
) {
return buildInboxPreview(task.last_result);
}
return null;
}
function pairedTaskInboxKind(
status: PairedTask['status'],
): InboxItemKind | null {
if (status === 'merge_ready') return 'approval';
if (status === 'review_ready' || status === 'in_review') {
return 'reviewer-request';
}
if (status === 'arbiter_requested' || status === 'in_arbitration') {
return 'arbiter-request';
}
return null;
}
function collectInboxItems(args: {
snapshots: StatusSnapshot[];
tasks: ScheduledTask[];
pairedTasks: PairedTask[];
createdAt: string;
}): InboxItem[] {
const items: InboxItem[] = [];
for (const snapshot of args.snapshots) {
for (const entry of snapshot.entries) {
if (!entry.pendingMessages && entry.pendingTasks === 0) continue;
const parts: string[] = [];
if (entry.pendingMessages) parts.push('pending messages');
if (entry.pendingTasks > 0) parts.push(`${entry.pendingTasks} tasks`);
items.push({
id: `room:${snapshot.serviceId}:${entry.jid}`,
groupKey: `room:${snapshot.serviceId}:${entry.jid}`,
kind: 'pending-room',
severity: entry.pendingTasks > 0 ? 'warn' : 'info',
title: entry.name || entry.folder || entry.jid,
summary: parts.join(' · '),
occurredAt: snapshot.updatedAt,
lastOccurredAt: snapshot.updatedAt,
createdAt: args.createdAt,
occurrences: 1,
source: 'status-snapshot',
roomJid: entry.jid,
roomName: entry.name,
groupFolder: entry.folder,
serviceId: snapshot.serviceId,
});
}
}
for (const task of args.pairedTasks) {
const kind = pairedTaskInboxKind(task.status);
if (!kind) continue;
items.push({
id: `paired:${task.id}:${task.status}`,
groupKey: `paired:${task.id}:${task.status}`,
kind,
severity: kind === 'arbiter-request' ? 'error' : 'warn',
title: task.title || task.group_folder,
summary: task.status,
occurredAt:
kind === 'arbiter-request'
? (task.arbiter_requested_at ?? task.updated_at)
: kind === 'reviewer-request'
? (task.review_requested_at ?? task.updated_at)
: task.updated_at,
lastOccurredAt:
kind === 'arbiter-request'
? (task.arbiter_requested_at ?? task.updated_at)
: kind === 'reviewer-request'
? (task.review_requested_at ?? task.updated_at)
: task.updated_at,
createdAt: args.createdAt,
occurrences: 1,
source: 'paired-task',
roomJid: task.chat_jid,
groupFolder: task.group_folder,
serviceId:
kind === 'reviewer-request'
? task.reviewer_service_id
: task.owner_service_id,
taskId: task.id,
taskStatus: task.status,
});
}
for (const task of args.tasks) {
if (!isWatchCiTask(task)) continue;
const result = failedTaskResult(task);
if (!result) continue;
items.push({
id: `ci:${task.id}`,
groupKey: `ci:${stableHash(result.toLowerCase())}`,
kind: 'ci-failure',
severity: task.status === 'paused' ? 'error' : 'warn',
title: 'CI watcher failed',
summary: result,
occurredAt: task.last_run ?? task.created_at,
lastOccurredAt: task.last_run ?? task.created_at,
createdAt: args.createdAt,
occurrences: 1,
source: 'scheduled-task',
roomJid: task.chat_jid,
groupFolder: task.group_folder,
serviceId: task.agent_type ?? undefined,
taskId: task.id,
taskStatus: task.status,
});
}
const severityRank: Record<InboxItemSeverity, number> = {
error: 0,
warn: 1,
info: 2,
};
const groupedItems = new Map<string, InboxItem>();
for (const item of items) {
const existing = groupedItems.get(item.groupKey);
if (!existing) {
groupedItems.set(item.groupKey, item);
continue;
}
const occurrences = existing.occurrences + item.occurrences;
const severity =
severityRank[item.severity] < severityRank[existing.severity]
? item.severity
: existing.severity;
const latest =
item.lastOccurredAt.localeCompare(existing.lastOccurredAt) > 0
? item
: existing;
groupedItems.set(item.groupKey, {
...latest,
severity,
occurrences,
lastOccurredAt:
item.lastOccurredAt.localeCompare(existing.lastOccurredAt) > 0
? item.lastOccurredAt
: existing.lastOccurredAt,
});
}
return [...groupedItems.values()]
.sort((a, b) => {
const severityDelta = severityRank[a.severity] - severityRank[b.severity];
if (severityDelta !== 0) return severityDelta;
return b.lastOccurredAt.localeCompare(a.lastOccurredAt);
})
.slice(0, 50);
}
export function sanitizeScheduledTask(
task: ScheduledTask,
): SanitizedScheduledTask {
return {
id: task.id,
groupFolder: task.group_folder,
chatJid: task.chat_jid,
agentType: task.agent_type,
ciProvider: task.ci_provider ?? null,
ciMetadata: task.ci_metadata ?? null,
scheduleType: task.schedule_type,
scheduleValue: task.schedule_value,
contextMode: task.context_mode,
nextRun: task.next_run,
lastRun: task.last_run,
lastResult: task.last_result,
status: task.status,
suspendedUntil: task.suspended_until ?? null,
createdAt: task.created_at,
promptPreview: buildPromptPreview(task.prompt),
promptLength: task.prompt.length,
isWatcher: isWatchCiTask(task),
};
}
function sanitizeRoomMessage(message: NewMessage): WebDashboardRoomMessage {
return {
id: message.id,
sender: message.sender,
senderName: message.sender_name || message.sender,
content: buildRoomPreview(
message.content ?? '',
ROOM_MESSAGE_PREVIEW_MAX_LENGTH,
),
timestamp: message.timestamp,
isFromMe: !!message.is_from_me,
isBotMessage: !!message.is_bot_message,
sourceKind: message.message_source_kind ?? 'human',
};
}
/**
* SSOT for live progress: scan the `messages` table for the most recent
* TASK_STATUS_MESSAGE_PREFIX-prefixed message that belongs to the current
* turn (sender role match + timestamp >= turn.created_at). The Discord
* bridge ingests every edit of the bot's live message back into messages,
* so this single source replaces the legacy paired_turns.progress_text
* column for derivation.
*/
const TASK_STATUS_PREFIX = '';
function turnRoleFromSenderName(name: string | null | undefined): string {
const v = (name ?? '').toLowerCase();
if (v.includes('오너') || v.includes('owner')) return 'owner';
if (v.includes('리뷰어') || v.includes('reviewer')) return 'reviewer';
if (v.includes('중재자') || v.includes('arbiter')) return 'arbiter';
return 'human';
}
function deriveLiveProgress(
turnRole: string,
turnCreatedAt: string,
messages: NewMessage[],
): { progressText: string | null; progressUpdatedAt: string | null } {
const startMs = new Date(turnCreatedAt).getTime();
const targetRole = turnRoleFromSenderName(turnRole);
for (let i = messages.length - 1; i >= 0; i--) {
const m = messages[i];
const content = m.content ?? '';
if (!content.startsWith(TASK_STATUS_PREFIX)) continue;
const ts = m.timestamp ? new Date(m.timestamp).getTime() : 0;
if (Number.isFinite(startMs) && ts < startMs) continue;
const role = turnRoleFromSenderName(m.sender_name);
if (role !== targetRole) continue;
const body = content.slice(TASK_STATUS_PREFIX.length);
const stripped = body.replace(/\n\n\d+[초smhMSH]?$/, '').trim();
if (!stripped) continue;
return { progressText: stripped, progressUpdatedAt: m.timestamp };
}
return { progressText: null, progressUpdatedAt: null };
}
function sanitizeRoomTurn(
turn: PairedTurnRecord,
attempt: PairedTurnAttemptRecord | null,
messages: NewMessage[] = [],
): WebDashboardRoomTurn {
const role = attempt?.role ?? turn.role;
const createdAt = attempt?.created_at ?? turn.created_at;
const completedAt = attempt?.completed_at ?? turn.completed_at;
// Read paired_turns.progress_text first (written by recordTurnProgress
// when the controller is in the loop). If that's empty AND the turn is
// active, fall back to scanning the messages table for the most recent
// TASK_STATUS-prefixed message — same content the Discord bridge ingests.
let progressText = turn.progress_text ?? null;
let progressUpdatedAt = turn.progress_updated_at ?? null;
if ((!progressText || !progressText.trim()) && completedAt === null) {
const live = deriveLiveProgress(role, createdAt, messages);
if (live.progressText) {
progressText = live.progressText;
progressUpdatedAt = live.progressUpdatedAt;
}
}
return {
turnId: turn.turn_id,
role,
intentKind: attempt?.intent_kind ?? turn.intent_kind,
state: attempt?.state ?? turn.state,
attemptNo: attempt?.attempt_no ?? turn.attempt_no,
executorServiceId: attempt?.executor_service_id ?? turn.executor_service_id,
executorAgentType: attempt?.executor_agent_type ?? turn.executor_agent_type,
activeRunId: attempt?.active_run_id ?? null,
createdAt,
updatedAt: attempt?.updated_at ?? turn.updated_at,
completedAt,
lastError:
(attempt?.last_error ?? turn.last_error)
? buildRoomPreview(
attempt?.last_error ?? turn.last_error ?? '',
ROOM_MESSAGE_PREVIEW_MAX_LENGTH,
)
: null,
progressText,
progressUpdatedAt,
};
}
function sanitizeRoomTurnOutput(
output: PairedTurnOutput,
): WebDashboardRoomTurnOutput {
return {
id: output.id,
turnNumber: output.turn_number,
role: output.role,
verdict: output.verdict ?? null,
createdAt: output.created_at,
outputText: buildRoomPreview(
output.output_text,
ROOM_OUTPUT_PREVIEW_MAX_LENGTH,
),
};
}
export function buildWebDashboardRoomActivity(args: {
serviceId: string;
entry: StatusSnapshot['entries'][number];
pairedTask: PairedTask | null;
turns: PairedTurnRecord[];
attempts: PairedTurnAttemptRecord[];
outputs: PairedTurnOutput[];
messages: NewMessage[];
outputLimit?: number;
}): WebDashboardRoomActivity {
const latestAttemptByTurnId = new Map<string, PairedTurnAttemptRecord>();
for (const attempt of args.attempts) {
const previous = latestAttemptByTurnId.get(attempt.turn_id);
if (!previous || attempt.attempt_no > previous.attempt_no) {
latestAttemptByTurnId.set(attempt.turn_id, attempt);
}
}
const currentTurn =
[...args.turns].sort((a, b) =>
b.updated_at.localeCompare(a.updated_at),
)[0] ?? null;
const currentAttempt = currentTurn
? (latestAttemptByTurnId.get(currentTurn.turn_id) ?? null)
: null;
const outputLimit = args.outputLimit ?? 4;
return {
serviceId: args.serviceId,
jid: args.entry.jid,
name: args.entry.name,
folder: args.entry.folder,
agentType: args.entry.agentType,
status: args.entry.status,
elapsedMs: args.entry.elapsedMs,
pendingMessages: args.entry.pendingMessages,
pendingTasks: args.entry.pendingTasks,
messages: args.messages.map(sanitizeRoomMessage),
pairedTask: args.pairedTask
? {
id: args.pairedTask.id,
title: args.pairedTask.title,
status: args.pairedTask.status,
roundTripCount: args.pairedTask.round_trip_count,
updatedAt: args.pairedTask.updated_at,
currentTurn: currentTurn
? sanitizeRoomTurn(currentTurn, currentAttempt, args.messages)
: null,
outputs: args.outputs.slice(-outputLimit).map(sanitizeRoomTurnOutput),
}
: null,
};
}
export function buildWebDashboardOverview(args: {
now?: string;
snapshots: StatusSnapshot[];
tasks: ScheduledTask[];
pairedTasks?: PairedTask[];
}): WebDashboardOverview {
const generatedAt = args.now ?? new Date().toISOString();
const rooms = {
total: 0,
active: 0,
waiting: 0,
inactive: 0,
};
const services = args.snapshots.map((snapshot) => {
let activeRooms = 0;
for (const entry of snapshot.entries) {
rooms.total += 1;
if (entry.status === 'processing') {
rooms.active += 1;
activeRooms += 1;
} else if (entry.status === 'waiting') {
rooms.waiting += 1;
activeRooms += 1;
} else {
rooms.inactive += 1;
}
}
return {
serviceId: snapshot.serviceId,
assistantName: snapshot.assistantName,
agentType: snapshot.agentType,
updatedAt: snapshot.updatedAt,
totalRooms: snapshot.entries.length,
activeRooms,
};
});
const tasks = {
total: args.tasks.length,
active: 0,
paused: 0,
completed: 0,
watchers: {
active: 0,
paused: 0,
completed: 0,
},
};
for (const task of args.tasks) {
if (task.status === 'active') tasks.active += 1;
if (task.status === 'paused') tasks.paused += 1;
if (task.status === 'completed') tasks.completed += 1;
if (isWatchCiTask(task)) {
if (task.status === 'active') tasks.watchers.active += 1;
if (task.status === 'paused') tasks.watchers.paused += 1;
if (task.status === 'completed') tasks.watchers.completed += 1;
}
}
const usageRows = collectUsageRows(args.snapshots);
const usageFetchedAt =
args.snapshots
.map((snapshot) => snapshot.usageRowsFetchedAt)
.filter((value): value is string => !!value)
.sort()
.at(-1) ?? null;
return {
generatedAt,
services,
rooms,
tasks,
usage: {
rows: usageRows,
fetchedAt: usageFetchedAt,
},
inbox: collectInboxItems({
snapshots: args.snapshots,
tasks: args.tasks,
pairedTasks: args.pairedTasks ?? [],
createdAt: generatedAt,
}),
};
}

File diff suppressed because it is too large Load Diff

1659
src/web-dashboard-server.ts Normal file

File diff suppressed because it is too large Load Diff

View File

@@ -13,6 +13,7 @@ vi.mock('child_process', () => ({
}));
import {
buildWorkspaceCommandEnvironment,
buildWorkspaceScriptCommand,
detectWorkspacePackageManager,
ensureWorkspaceDependenciesInstalled,
@@ -139,6 +140,13 @@ describe('workspace package manager helpers', () => {
packageManager: 'pnpm',
commandText: 'corepack pnpm install --frozen-lockfile',
});
expect(execFileSyncMock).toHaveBeenCalledWith(
'corepack',
['pnpm', 'install', '--frozen-lockfile'],
expect.objectContaining({
cwd: repoDir,
}),
);
expect(hasInstalledNodeModules(repoDir)).toBe(true);
execFileSyncMock.mockClear();
@@ -209,4 +217,59 @@ describe('workspace package manager helpers', () => {
expect(hasInstalledNodeModules(repoDir)).toBe(true);
expect(execFileSyncMock).not.toHaveBeenCalled();
});
it('disables corepack project specs only for lockfile-selected pnpm workspaces under a conflicting ancestor', () => {
const parentDir = path.join(tempRoot, 'parent');
fs.mkdirSync(parentDir, { recursive: true });
fs.writeFileSync(
path.join(parentDir, 'package.json'),
JSON.stringify({ packageManager: 'bun@1.3.11' }),
);
const repoDir = path.join(parentDir, 'child');
fs.mkdirSync(repoDir, { recursive: true });
fs.writeFileSync(
path.join(repoDir, 'package.json'),
JSON.stringify({ name: 'child', scripts: { typecheck: 'tsc --noEmit' } }),
);
fs.writeFileSync(
path.join(repoDir, 'pnpm-lock.yaml'),
'lockfileVersion: 9.0\n',
);
expect(
buildWorkspaceCommandEnvironment(repoDir, 'pnpm', { PATH: '/tmp/bin' }),
).toMatchObject({
PATH: '/tmp/bin',
COREPACK_ENABLE_PROJECT_SPEC: '0',
});
});
it('keeps an explicitly pinned pnpm workspace packageManager under a bun ancestor', () => {
const parentDir = path.join(tempRoot, 'parent');
fs.mkdirSync(parentDir, { recursive: true });
fs.writeFileSync(
path.join(parentDir, 'package.json'),
JSON.stringify({ packageManager: 'bun@1.3.11' }),
);
const repoDir = path.join(parentDir, 'child');
fs.mkdirSync(repoDir, { recursive: true });
fs.writeFileSync(
path.join(repoDir, 'package.json'),
JSON.stringify({
name: 'child',
packageManager: 'pnpm@10.11.0',
scripts: { typecheck: 'tsc --noEmit' },
}),
);
fs.writeFileSync(
path.join(repoDir, 'pnpm-lock.yaml'),
'lockfileVersion: 9.0\n',
);
expect(
buildWorkspaceCommandEnvironment(repoDir, 'pnpm', { PATH: '/tmp/bin' }),
).toEqual({ PATH: '/tmp/bin' });
});
});

View File

@@ -18,6 +18,7 @@ export interface WorkspaceDependencyInstallResult {
commandText?: string;
}
const COREPACK_PROJECT_SPEC_DISABLED = '0';
const INSTALL_STATE_FILENAME = '.ejclaw-install-state.json';
const NODE_MODULES_NOISE_ENTRIES = new Set([
'.cache',
@@ -117,6 +118,52 @@ function buildCorepackCommand(
};
}
function findNearestAncestorPackageManager(
repoDir: string,
): WorkspacePackageManager | null {
let currentDir = path.dirname(path.resolve(repoDir));
while (true) {
const packageManager = detectPackageManagerFromField(
readPackageJsonMetadata(currentDir)?.packageManager,
);
if (packageManager) {
return packageManager;
}
const parentDir = path.dirname(currentDir);
if (parentDir === currentDir) {
return null;
}
currentDir = parentDir;
}
}
export function buildWorkspaceCommandEnvironment(
repoDir: string,
packageManager: WorkspacePackageManager,
baseEnv: NodeJS.ProcessEnv = process.env,
): NodeJS.ProcessEnv {
if (packageManager !== 'pnpm') {
return baseEnv;
}
const localPackageManager = readPackageJsonMetadata(repoDir)?.packageManager;
if (typeof localPackageManager === 'string' && localPackageManager.trim()) {
return baseEnv;
}
const ancestorPackageManager = findNearestAncestorPackageManager(repoDir);
if (!ancestorPackageManager || ancestorPackageManager === packageManager) {
return baseEnv;
}
return {
...baseEnv,
COREPACK_ENABLE_PROJECT_SPEC: COREPACK_PROJECT_SPEC_DISABLED,
};
}
function computeInstallFingerprint(repoDir: string): string | null {
const packageJsonPath = path.join(repoDir, 'package.json');
if (!fs.existsSync(packageJsonPath)) {
@@ -359,6 +406,7 @@ export function ensureWorkspaceDependenciesInstalled(
try {
execFileSync(command.file, command.args, {
cwd: repoDir,
env: buildWorkspaceCommandEnvironment(repoDir, command.packageManager),
encoding: 'utf-8',
stdio: ['ignore', 'pipe', 'pipe'],
});
@@ -420,6 +468,7 @@ export function detectPnpmStorePath(workspaceDir: string): string | null {
try {
const storePath = execFileSync(file, args, {
cwd: workspaceDir,
env: buildWorkspaceCommandEnvironment(workspaceDir, 'pnpm'),
encoding: 'utf-8',
stdio: ['ignore', 'pipe', 'pipe'],
timeout: 5000,