Compare commits
19 Commits
a9095d95a8
...
e752524252
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
e752524252 | ||
|
|
04cfe97a14 | ||
|
|
d8abcf0621 | ||
|
|
3f73197e6c | ||
|
|
d57dd68fc0 | ||
|
|
65ef6e8830 | ||
|
|
8600213bcc | ||
|
|
9108174380 | ||
|
|
cf6d1f89fc | ||
|
|
0064654d8f | ||
|
|
e21308db1f | ||
|
|
77426fc15c | ||
|
|
8a71484934 | ||
|
|
3930c21625 | ||
|
|
44c54a759c | ||
|
|
6036430c60 | ||
|
|
142929ba39 | ||
|
|
58811b2700 | ||
|
|
6ab38ca461 |
@@ -43,6 +43,9 @@ GitHub `main` 베이스 위에 운영 중 검증된 변경을 현재 코드 구
|
||||
- 사용량 윈도우 정렬 프라이머: Codex/Claude 초기화 시점을 맞추기 위한 정렬 신호를 조건 없이 발사.
|
||||
- 리뷰어 무응답 방지: 리뷰어가 verdict 없이 연속 실패할 때 재시도 상한(기본 2회)으로 핑퐁 루프를 끊고 arbiter/사용자로 에스컬레이션.
|
||||
- 대시보드: 등록된 방 별칭을 Discord 채널명과 분리해 보존.
|
||||
- 상태 대시보드 갱신 정책: 기본 갱신을 벽시계 분 경계(:00초)에 정렬하고, 새 채팅 메시지·에이전트 활성 변화 시 즉시(1.5s 디바운스) 갱신. 상태 메시지 편집이 실패하면 15초 간격 2회 재시도 후에만 새 메시지로 재게시하며(봇 재시작 직후에는 즉시 재표시), 편집 실패로 인한 반복 재생성(알림 스팸)을 제거.
|
||||
- Claude 세션 토큰 write-back: 각 Claude 턴 종료 시 세션이 rotate한 OAuth 토큰을 표준 자격증명(data/claude)으로 수렴시켜, 본체 리프레시 루프와 세션 CLI가 같은 토큰 패밀리를 각자 갱신하다 발생하던 `invalid_grant`(리프레시 토큰 패밀리 폐기 → 재로그인 강제)를 방지. 세션 자격증명 스캔 경로도 실제 구조(`services/<id>/.claude`, `tasks/<id>` 하위)에 맞게 정정해 fan-out/stale 정리가 실제 세션 파일을 덮도록 함. Codex는 기존 write-back(`syncCodexSessionAuthBack`)을 사용.
|
||||
- 채널 등록 해제 도구(`scripts/deregister-room.ts`): 채널 등록 해제 시 paired 데이터·세션(reviewer·arbiter 포함)·워크스페이스·디스크 폴더(git worktree 포함)를 전부 정리하고 채팅 채널과 메시지만 보존. `--dry-run` 미리보기, main 방 보호, 다른 방과 공유하는 폴더 보호, 실행 전 DB 자동 백업 포함.
|
||||
|
||||
## Tribunal 시스템
|
||||
|
||||
|
||||
359
scripts/deregister-room.ts
Normal file
359
scripts/deregister-room.ts
Normal file
@@ -0,0 +1,359 @@
|
||||
#!/usr/bin/env bun
|
||||
/**
|
||||
* Deregister a chat room and purge all managed data, keeping ONLY the chat
|
||||
* channel record (chats) and its message history (messages).
|
||||
*
|
||||
* Usage:
|
||||
* bun scripts/deregister-room.ts [--dry-run] [--force] [--no-backup] <channelId|jid> [more...]
|
||||
*
|
||||
* What it removes for each target chat:
|
||||
* - room registration + role/skill overrides + channel owner lease
|
||||
* - paired tasks/turns/attempts/outputs/reservations/leases/projects/handoffs
|
||||
* - work items, scheduled tasks, task run logs
|
||||
* - sessions (by group folder) + router cursor entry
|
||||
* - on-disk group/workspace/session/ipc folders (git worktrees removed cleanly)
|
||||
*
|
||||
* What it KEEPS (never touched): the `chats` row and all `messages` rows.
|
||||
*
|
||||
* It does NOT restart the service. Restart ejclaw afterwards so the room is
|
||||
* dropped from the live in-memory bindings (the router loads bindings only at
|
||||
* startup), e.g. `systemctl --user restart ejclaw.service`.
|
||||
*
|
||||
* Safety:
|
||||
* - refuses to deregister a main room unless --force
|
||||
* - a group folder shared by another remaining room is NOT deleted on disk
|
||||
* (only that chat's DB rows are removed)
|
||||
* - --dry-run reports what would change without writing anything
|
||||
* - the DB is auto-backed up to /home/claude/ejclaw-db-backup-<ts>.db before
|
||||
* the purge unless --no-backup is passed
|
||||
*/
|
||||
import { execFileSync } from 'child_process';
|
||||
import fs from 'fs';
|
||||
import path from 'path';
|
||||
|
||||
import { Database } from 'bun:sqlite';
|
||||
|
||||
import { DATA_DIR, GROUPS_DIR, STORE_DIR } from '../src/config.js';
|
||||
|
||||
const args = process.argv.slice(2);
|
||||
const dryRun = args.includes('--dry-run');
|
||||
const force = args.includes('--force');
|
||||
const noBackup = args.includes('--no-backup');
|
||||
const ids = args.filter((a) => !a.startsWith('--'));
|
||||
|
||||
if (ids.length === 0) {
|
||||
console.error(
|
||||
'Usage: bun scripts/deregister-room.ts [--dry-run] [--force] [--no-backup] <channelId|jid> [more...]',
|
||||
);
|
||||
process.exit(2);
|
||||
}
|
||||
|
||||
function normalizeJid(id: string): string {
|
||||
if (/^(dc|tg|wa):/.test(id)) return id;
|
||||
if (/^\d+$/.test(id)) return `dc:${id}`;
|
||||
return id;
|
||||
}
|
||||
|
||||
const jids = [...new Set(ids.map(normalizeJid))];
|
||||
const repoRoot = path.resolve(GROUPS_DIR, '..');
|
||||
const db = new Database(path.join(STORE_DIR, 'messages.db'));
|
||||
|
||||
interface Target {
|
||||
jid: string;
|
||||
folders: string[];
|
||||
isMain: boolean;
|
||||
name: string | null;
|
||||
registered: boolean;
|
||||
}
|
||||
|
||||
// When a room was already unregistered (room_settings gone) the group folder is
|
||||
// no longer stored there, so recover it from the group_folder recorded on the
|
||||
// chat's leftover managed rows — otherwise sessions/disk/worktrees for an
|
||||
// already-unregistered room would be missed.
|
||||
function backtraceFolders(jid: string): string[] {
|
||||
const set = new Set<string>();
|
||||
for (const t of [
|
||||
'paired_tasks',
|
||||
'work_items',
|
||||
'scheduled_tasks',
|
||||
'service_handoffs',
|
||||
'paired_projects',
|
||||
]) {
|
||||
const rows = db
|
||||
.query(
|
||||
`SELECT DISTINCT group_folder FROM ${t} WHERE chat_jid=? AND group_folder IS NOT NULL`,
|
||||
)
|
||||
.all(jid) as Array<{ group_folder: string | null }>;
|
||||
for (const r of rows) if (r.group_folder) set.add(r.group_folder);
|
||||
}
|
||||
return [...set];
|
||||
}
|
||||
|
||||
const targets: Target[] = jids.map((jid) => {
|
||||
const r = db
|
||||
.query('SELECT folder, is_main, name FROM room_settings WHERE chat_jid=?')
|
||||
.get(jid) as
|
||||
| { folder: string | null; is_main: number | null; name: string | null }
|
||||
| undefined;
|
||||
const folders = r?.folder ? [r.folder] : backtraceFolders(jid);
|
||||
return {
|
||||
jid,
|
||||
folders,
|
||||
isMain: r?.is_main === 1,
|
||||
name: r?.name ?? null,
|
||||
registered: Boolean(r),
|
||||
};
|
||||
});
|
||||
|
||||
const mainTargets = targets.filter((t) => t.isMain);
|
||||
if (mainTargets.length && !force) {
|
||||
console.error(
|
||||
'Refusing to deregister main room(s):',
|
||||
mainTargets.map((t) => t.jid).join(', '),
|
||||
'\nPass --force only if you really intend to unregister the main control room.',
|
||||
);
|
||||
process.exit(3);
|
||||
}
|
||||
|
||||
const targetJidSet = new Set(jids);
|
||||
function folderShared(folder: string): boolean {
|
||||
const rows = db
|
||||
.query('SELECT chat_jid FROM room_settings WHERE folder=?')
|
||||
.all(folder) as Array<{ chat_jid: string }>;
|
||||
return rows.some((r) => !targetJidSet.has(r.chat_jid));
|
||||
}
|
||||
|
||||
const safeFolders = new Set<string>();
|
||||
const sharedFolders = new Set<string>();
|
||||
for (const t of targets) {
|
||||
for (const folder of t.folders) {
|
||||
if (folderShared(folder)) sharedFolders.add(folder);
|
||||
else safeFolders.add(folder);
|
||||
}
|
||||
}
|
||||
|
||||
console.log('=== targets ===');
|
||||
for (const t of targets) {
|
||||
console.log(
|
||||
` ${t.jid} | name=${t.name ?? '(unregistered)'} | folder=${t.folders.length ? t.folders.join(',') : '-'} | registered=${t.registered}${t.isMain ? ' | MAIN' : ''}`,
|
||||
);
|
||||
}
|
||||
if (sharedFolders.size) {
|
||||
console.log(
|
||||
' NOTE shared folders kept on disk (used by another room):',
|
||||
[...sharedFolders].join(', '),
|
||||
);
|
||||
}
|
||||
|
||||
const qJ = jids.map(() => '?').join(',');
|
||||
const taskIds = (
|
||||
db
|
||||
.query(`SELECT id FROM paired_tasks WHERE chat_jid IN (${qJ})`)
|
||||
.all(...jids) as Array<{
|
||||
id: string;
|
||||
}>
|
||||
).map((r) => r.id);
|
||||
const qT = taskIds.map(() => '?').join(',');
|
||||
// task_run_logs.task_id references scheduled_tasks.id (NOT paired_tasks.id), so
|
||||
// its rows must be pruned by the chat's scheduled task ids.
|
||||
const scheduledTaskIds = (
|
||||
db
|
||||
.query(`SELECT id FROM scheduled_tasks WHERE chat_jid IN (${qJ})`)
|
||||
.all(...jids) as Array<{ id: string }>
|
||||
).map((r) => r.id);
|
||||
const qS = scheduledTaskIds.map(() => '?').join(',');
|
||||
const safeFolderList = [...safeFolders];
|
||||
|
||||
const count = (t: string, where: string, a: unknown[]): number =>
|
||||
(
|
||||
db.query(`SELECT COUNT(*) n FROM ${t} WHERE ${where}`).get(...a) as {
|
||||
n: number;
|
||||
}
|
||||
).n;
|
||||
|
||||
// Report / gather counts.
|
||||
const plan: Array<{ table: string; n: number; run: () => void }> = [];
|
||||
const addJid = (t: string) =>
|
||||
plan.push({
|
||||
table: t,
|
||||
n: count(t, `chat_jid IN (${qJ})`, jids),
|
||||
run: () =>
|
||||
db.query(`DELETE FROM ${t} WHERE chat_jid IN (${qJ})`).run(...jids),
|
||||
});
|
||||
const addTask = (t: string) => {
|
||||
if (!taskIds.length) return;
|
||||
plan.push({
|
||||
table: t,
|
||||
n: count(t, `task_id IN (${qT})`, taskIds),
|
||||
run: () =>
|
||||
db.query(`DELETE FROM ${t} WHERE task_id IN (${qT})`).run(...taskIds),
|
||||
});
|
||||
};
|
||||
|
||||
addTask('paired_turn_attempts');
|
||||
addTask('paired_turn_outputs');
|
||||
addTask('paired_turns');
|
||||
// Prune scheduled-task run logs before the scheduled_tasks rows they belong to.
|
||||
if (scheduledTaskIds.length) {
|
||||
plan.push({
|
||||
table: 'task_run_logs',
|
||||
n: count('task_run_logs', `task_id IN (${qS})`, scheduledTaskIds),
|
||||
run: () =>
|
||||
db
|
||||
.query(`DELETE FROM task_run_logs WHERE task_id IN (${qS})`)
|
||||
.run(...scheduledTaskIds),
|
||||
});
|
||||
}
|
||||
for (const t of [
|
||||
'paired_task_execution_leases',
|
||||
'paired_turn_reservations',
|
||||
'paired_tasks',
|
||||
'paired_projects',
|
||||
'service_handoffs',
|
||||
'work_items',
|
||||
'scheduled_tasks',
|
||||
'channel_owner',
|
||||
'room_role_overrides',
|
||||
'room_skill_overrides',
|
||||
'room_settings',
|
||||
]) {
|
||||
addJid(t);
|
||||
}
|
||||
// sessions are keyed by group folder — only purge folders not shared with a
|
||||
// surviving room. Reviewer/arbiter sessions are stored under the colon-suffixed
|
||||
// group_folder (e.g. "<folder>:reviewer"), so include those variants too.
|
||||
const sessionFolderKeys = safeFolderList.flatMap((f) => [
|
||||
f,
|
||||
`${f}:reviewer`,
|
||||
`${f}:arbiter`,
|
||||
]);
|
||||
if (sessionFolderKeys.length) {
|
||||
const qSess = sessionFolderKeys.map(() => '?').join(',');
|
||||
plan.push({
|
||||
table: 'sessions',
|
||||
n: count('sessions', `group_folder IN (${qSess})`, sessionFolderKeys),
|
||||
run: () =>
|
||||
db
|
||||
.query(`DELETE FROM sessions WHERE group_folder IN (${qSess})`)
|
||||
.run(...sessionFolderKeys),
|
||||
});
|
||||
}
|
||||
|
||||
// router cursor prune count
|
||||
const ras = db
|
||||
.query('SELECT value FROM router_state WHERE key=?')
|
||||
.get('last_agent_seq') as { value: string } | undefined;
|
||||
let routerPrune = 0;
|
||||
if (ras) {
|
||||
const obj = JSON.parse(ras.value) as Record<string, unknown>;
|
||||
routerPrune = jids.filter((j) => j in obj).length;
|
||||
}
|
||||
|
||||
console.log('\n=== DB rows to delete ===');
|
||||
for (const p of plan) if (p.n) console.log(` ${p.table}: ${p.n}`);
|
||||
if (routerPrune) console.log(` router_state.last_agent_seq: ${routerPrune}`);
|
||||
|
||||
// disk targets
|
||||
const diskDirs: string[] = [];
|
||||
for (const folder of safeFolders) {
|
||||
// Tribunal rooms keep separate reviewer/arbiter runtime dirs suffixed with
|
||||
// the role (e.g. data/sessions/<folder>-reviewer). Include those variants so
|
||||
// deregistration leaves no leftovers.
|
||||
const folderVariants = [folder, `${folder}-reviewer`, `${folder}-arbiter`];
|
||||
for (const d of [
|
||||
path.join(GROUPS_DIR, folder),
|
||||
...folderVariants.flatMap((f) => [
|
||||
path.join(DATA_DIR, 'workspaces', f),
|
||||
path.join(DATA_DIR, 'sessions', f),
|
||||
path.join(DATA_DIR, 'ipc', f),
|
||||
]),
|
||||
]) {
|
||||
if (fs.existsSync(d)) diskDirs.push(d);
|
||||
}
|
||||
}
|
||||
console.log('\n=== disk dirs to delete ===');
|
||||
for (const d of diskDirs) console.log(` ${d}`);
|
||||
|
||||
if (dryRun) {
|
||||
console.log('\n[dry-run] no changes made.');
|
||||
process.exit(0);
|
||||
}
|
||||
|
||||
// --- execute ---
|
||||
// Auto-backup the DB before the irreversible purge (skip with --no-backup).
|
||||
if (!noBackup) {
|
||||
const dbFile = path.join(STORE_DIR, 'messages.db');
|
||||
const stamp = new Date()
|
||||
.toISOString()
|
||||
.replace(/[-:]/g, '')
|
||||
.replace(/\..+$/, '')
|
||||
.replace('T', '-');
|
||||
const backup = path.join('/home/claude', `ejclaw-db-backup-${stamp}.db`);
|
||||
fs.copyFileSync(dbFile, backup);
|
||||
console.log(`DB backed up to ${backup}`);
|
||||
}
|
||||
|
||||
db.exec('BEGIN');
|
||||
try {
|
||||
for (const p of plan) p.run();
|
||||
if (ras && routerPrune) {
|
||||
const obj = JSON.parse(ras.value) as Record<string, unknown>;
|
||||
for (const j of jids) delete obj[j];
|
||||
db.query('UPDATE router_state SET value=? WHERE key=?').run(
|
||||
JSON.stringify(obj),
|
||||
'last_agent_seq',
|
||||
);
|
||||
}
|
||||
db.exec('COMMIT');
|
||||
} catch (e) {
|
||||
db.exec('ROLLBACK');
|
||||
console.error('ROLLBACK due to error:', (e as Error).message);
|
||||
process.exit(1);
|
||||
}
|
||||
|
||||
// remove git worktrees that live under a safe workspace folder, then rm -rf.
|
||||
function listWorktrees(): string[] {
|
||||
try {
|
||||
const out = execFileSync('git', ['worktree', 'list', '--porcelain'], {
|
||||
cwd: repoRoot,
|
||||
encoding: 'utf-8',
|
||||
});
|
||||
return out
|
||||
.split('\n')
|
||||
.filter((l) => l.startsWith('worktree '))
|
||||
.map((l) => l.slice('worktree '.length).trim());
|
||||
} catch {
|
||||
return [];
|
||||
}
|
||||
}
|
||||
const worktrees = listWorktrees();
|
||||
for (const folder of safeFolders) {
|
||||
const wsDir = path.join(DATA_DIR, 'workspaces', folder);
|
||||
for (const wt of worktrees) {
|
||||
if (wt === wsDir || wt.startsWith(wsDir + path.sep)) {
|
||||
try {
|
||||
execFileSync('git', ['worktree', 'remove', '--force', wt], {
|
||||
cwd: repoRoot,
|
||||
stdio: 'inherit',
|
||||
});
|
||||
console.log(` git worktree removed: ${wt}`);
|
||||
} catch (e) {
|
||||
console.warn(
|
||||
` worktree remove failed (${wt}): ${(e as Error).message}`,
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
for (const d of diskDirs) {
|
||||
fs.rmSync(d, { recursive: true, force: true });
|
||||
console.log(` removed ${d}`);
|
||||
}
|
||||
try {
|
||||
execFileSync('git', ['worktree', 'prune'], { cwd: repoRoot });
|
||||
} catch {
|
||||
/* ignore */
|
||||
}
|
||||
|
||||
console.log('\nDone. Restart ejclaw to drop the room(s) from live bindings:');
|
||||
console.log(' systemctl --user restart ejclaw.service');
|
||||
@@ -9,6 +9,7 @@ import path from 'path';
|
||||
import { GROUPS_DIR } from '../src/config.js';
|
||||
import { assignRoom, initDatabase } from '../src/db.js';
|
||||
import { isValidGroupFolder } from '../src/group-folder.js';
|
||||
import { ensurePairedWorkspaceProvisioned } from '../src/paired-workspace-manager.js';
|
||||
import { logger } from '../src/logger.js';
|
||||
import { emitStatus } from './status.js';
|
||||
|
||||
@@ -109,6 +110,24 @@ export async function run(args: string[]): Promise<void> {
|
||||
});
|
||||
logger.info('Assigned room through canonical room service');
|
||||
|
||||
// Provision the paired workspace so a tribunal room's reviewer/arbiter work
|
||||
// immediately on registration. Without this the room has work_dir=null and
|
||||
// the paired flow never creates a task, so only the owner ever runs.
|
||||
const canonicalWorkDir = ensurePairedWorkspaceProvisioned({
|
||||
chatJid: parsed.jid,
|
||||
groupFolder: parsed.folder,
|
||||
});
|
||||
assignRoom(parsed.jid, {
|
||||
name: parsed.name,
|
||||
folder: parsed.folder,
|
||||
isMain: parsed.isMain,
|
||||
workDir: canonicalWorkDir,
|
||||
});
|
||||
logger.info(
|
||||
{ folder: parsed.folder, canonicalWorkDir },
|
||||
'Provisioned paired workspace for room so reviewer/arbiter run on registration',
|
||||
);
|
||||
|
||||
fs.mkdirSync(path.join(GROUPS_DIR, parsed.folder, 'logs'), {
|
||||
recursive: true,
|
||||
});
|
||||
|
||||
@@ -17,6 +17,7 @@ import {
|
||||
type PreparedCodexSessionAuth,
|
||||
} from './agent-runner-environment.js';
|
||||
import { syncCodexSessionAuthBack } from './codex-token-rotation.js';
|
||||
import { syncClaudeSessionAuthBack } from './token-refresh.js';
|
||||
import { runSpawnedAgentProcess } from './agent-runner-process.js';
|
||||
import { getStoredRoomSkillOverrides } from './db.js';
|
||||
export {
|
||||
@@ -119,6 +120,31 @@ function createCodexAuthSessionFinalizer(
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* After a Claude agent turn, adopt any token the session's Claude CLI refreshed
|
||||
* (rotated) back into the canonical credentials, so the main refresh loop never
|
||||
* ends up holding an already-rotated token (which Anthropic revokes as a family).
|
||||
* No-op for Codex sessions (sessionClaudeDir undefined).
|
||||
*/
|
||||
function createClaudeAuthSessionFinalizer(
|
||||
sessionClaudeDir: string | undefined,
|
||||
): () => void {
|
||||
let finalized = false;
|
||||
return () => {
|
||||
if (finalized) return;
|
||||
finalized = true;
|
||||
if (!sessionClaudeDir) return;
|
||||
try {
|
||||
syncClaudeSessionAuthBack(sessionClaudeDir);
|
||||
} catch (err) {
|
||||
logger.warn(
|
||||
{ err, sessionClaudeDir },
|
||||
'Failed to sync Claude session auth back to canonical credentials',
|
||||
);
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
export async function runAgentProcess(
|
||||
group: RegisteredGroup,
|
||||
input: AgentInput,
|
||||
@@ -199,6 +225,15 @@ export async function runAgentProcess(
|
||||
const finalizeCodexAuthSessionOnce = createCodexAuthSessionFinalizer(
|
||||
() => codexSessionAuth,
|
||||
);
|
||||
const finalizeClaudeAuthSessionOnce = createClaudeAuthSessionFinalizer(
|
||||
(group.agentType || 'claude-code') === 'codex'
|
||||
? undefined
|
||||
: env.CLAUDE_CONFIG_DIR,
|
||||
);
|
||||
const finalizeAuthSessionOnce = () => {
|
||||
finalizeCodexAuthSessionOnce();
|
||||
finalizeClaudeAuthSessionOnce();
|
||||
};
|
||||
|
||||
// Check if runner is built
|
||||
const distEntry = path.join(runnerDir, 'dist', 'index.js');
|
||||
@@ -255,14 +290,14 @@ export async function runAgentProcess(
|
||||
logsDir,
|
||||
startTime,
|
||||
onOutput,
|
||||
onTerminalStreamedOutputFlushed: finalizeCodexAuthSessionOnce,
|
||||
onTerminalStreamedOutputFlushed: finalizeAuthSessionOnce,
|
||||
})
|
||||
.then((output) => {
|
||||
finalizeCodexAuthSessionOnce();
|
||||
finalizeAuthSessionOnce();
|
||||
resolve(output);
|
||||
})
|
||||
.catch((err: unknown) => {
|
||||
finalizeCodexAuthSessionOnce();
|
||||
finalizeAuthSessionOnce();
|
||||
logger.error(
|
||||
{ err, processName, chatJid: input.chatJid, runId: input.runId },
|
||||
'Spawned agent process runner failed',
|
||||
|
||||
@@ -157,6 +157,48 @@ describe('codex-usage-collector fallback account usage', () => {
|
||||
]);
|
||||
});
|
||||
|
||||
it('handles a lone weekly window (secondary: null) without throwing and routes it to 7d', async () => {
|
||||
createDefaultCodexAuth(tempHome);
|
||||
const childProcess = await import('child_process');
|
||||
|
||||
vi.mocked(childProcess.spawn).mockImplementation(((
|
||||
_cmd: string,
|
||||
_args: readonly string[] | undefined,
|
||||
_opts?: { env?: Record<string, string> },
|
||||
) => {
|
||||
// Some plans (e.g. Plus) report only a weekly window in `primary` with
|
||||
// `secondary: null`. This previously threw on `secondary.usedPercent`.
|
||||
return createFakeChildProcess({
|
||||
codex: {
|
||||
limitName: 'Codex',
|
||||
primary: {
|
||||
usedPercent: 10,
|
||||
windowDurationMins: 10080, // 7 days
|
||||
resetsAt: new Date(Date.now() + 4 * 86_400_000).toISOString(),
|
||||
},
|
||||
secondary: null,
|
||||
},
|
||||
}) 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();
|
||||
|
||||
// Usage must be applied (not swallowed by a TypeError), and the lone weekly
|
||||
// window is shown as 7d while the missing 5h window renders as unknown.
|
||||
expect(result.fetchedAt).toEqual(expect.any(String));
|
||||
expect(result.rows).toEqual([
|
||||
expect.objectContaining({
|
||||
name: 'Codex',
|
||||
h5pct: -1,
|
||||
d7pct: 10,
|
||||
}),
|
||||
]);
|
||||
});
|
||||
|
||||
it('finds codex via ~/.hermes/node/bin when running under bun', async () => {
|
||||
createDefaultCodexAuth(tempHome);
|
||||
const childProcess = await import('child_process');
|
||||
|
||||
@@ -14,8 +14,16 @@ import { logger } from './logger.js';
|
||||
export interface CodexRateLimit {
|
||||
limitId?: string;
|
||||
limitName: string | null;
|
||||
primary: { usedPercent: number; resetsAt: string | number };
|
||||
secondary: { usedPercent: number; resetsAt: string | number };
|
||||
primary: {
|
||||
usedPercent: number;
|
||||
resetsAt: string | number;
|
||||
windowDurationMins?: number;
|
||||
};
|
||||
secondary: {
|
||||
usedPercent: number;
|
||||
resetsAt: string | number;
|
||||
windowDurationMins?: number;
|
||||
} | null;
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -291,8 +299,8 @@ export function applyCodexUsageToAccount(
|
||||
account: accountIndex + 1,
|
||||
buckets: usage.map((l) => ({
|
||||
id: l.limitId,
|
||||
h5: l.primary.usedPercent,
|
||||
d7: l.secondary.usedPercent,
|
||||
h5: l.primary?.usedPercent ?? null,
|
||||
d7: l.secondary?.usedPercent ?? null,
|
||||
})),
|
||||
},
|
||||
`Codex account #${accountIndex + 1}: ${usage.length} rate-limit bucket(s)`,
|
||||
@@ -312,13 +320,31 @@ export function applyCodexUsageToAccount(
|
||||
return;
|
||||
}
|
||||
|
||||
const pct = Math.round(effective.primary.usedPercent);
|
||||
const d7Pct = Math.round(effective.secondary.usedPercent);
|
||||
// Historical shape: primary = 5h window, secondary = 7d window. Some plans
|
||||
// (e.g. Plus) return only ONE window in `primary` (a weekly/7d window) with
|
||||
// `secondary: null`. Guard the null so we don't throw, and route a lone
|
||||
// weekly window into the 7d slot so usage still renders.
|
||||
// Store raw ISO timestamps (not pre-formatted strings) so the exhaustion
|
||||
// gate can compute "minutes until reset" later. The dashboard render path
|
||||
// formats these at display time via `formatResetRemaining`.
|
||||
const resetIso = toIsoMaybe(effective.primary.resetsAt);
|
||||
const resetD7Iso = toIsoMaybe(effective.secondary.resetsAt);
|
||||
let pct = Math.round(effective.primary.usedPercent);
|
||||
let resetIso = toIsoMaybe(effective.primary.resetsAt);
|
||||
let d7Pct = effective.secondary
|
||||
? Math.round(effective.secondary.usedPercent)
|
||||
: -1;
|
||||
let resetD7Iso = effective.secondary
|
||||
? toIsoMaybe(effective.secondary.resetsAt)
|
||||
: undefined;
|
||||
if (
|
||||
!effective.secondary &&
|
||||
(effective.primary.windowDurationMins ?? 0) > 1440
|
||||
) {
|
||||
// Lone window is weekly (> 1 day) — show it as 7d, leave 5h unknown.
|
||||
d7Pct = pct;
|
||||
resetD7Iso = resetIso;
|
||||
pct = -1;
|
||||
resetIso = undefined;
|
||||
}
|
||||
updateCodexAccountUsage(pct, resetIso, accountIndex, d7Pct, resetD7Iso);
|
||||
logger.info(
|
||||
{
|
||||
|
||||
@@ -259,7 +259,9 @@ export function loadConfig(): AppConfig {
|
||||
moa: buildMoaConfig(),
|
||||
status: {
|
||||
channelId: readText('STATUS_CHANNEL_ID') ?? '',
|
||||
updateInterval: 10000,
|
||||
// Base periodic refresh. Event-driven updates (new message, agent
|
||||
// activity change) refresh immediately between these ticks.
|
||||
updateInterval: 60000,
|
||||
usageUpdateInterval: 60000,
|
||||
showRooms: readBooleanUnlessFalse('STATUS_SHOW_ROOMS', true),
|
||||
showRoomDetails: readBooleanUnlessFalse('STATUS_SHOW_ROOM_DETAILS', true),
|
||||
|
||||
79
src/db/room-registration.test.ts
Normal file
79
src/db/room-registration.test.ts
Normal file
@@ -0,0 +1,79 @@
|
||||
import { describe, expect, it } from 'vitest';
|
||||
|
||||
import { openInitializedInMemoryDatabase } from './database-lifecycle.js';
|
||||
import {
|
||||
getStoredRoomRowsFromDatabase,
|
||||
getStoredRoomSettingsRowFromDatabase,
|
||||
insertStoredRoomSettings,
|
||||
} from './room-registration.js';
|
||||
|
||||
// Regression guard for the cgv-macro incident: a room_settings row that carried
|
||||
// an invalid mode_source ('room') was silently dropped from every binding
|
||||
// lookup, so the router ignored the channel, it vanished from the status list,
|
||||
// and re-registration wedged on the UNIQUE(chat_jid) constraint. The loader must
|
||||
// keep such a room routable by coercing the bad source to a valid one.
|
||||
describe('getStoredRoomSettingsRowFromDatabase mode_source resilience', () => {
|
||||
const jid = 'dc:invalid-source-room';
|
||||
const snapshot = {
|
||||
name: 'cgv-macro',
|
||||
folder: 'cgv-macro',
|
||||
triggerPattern: '',
|
||||
requiresTrigger: false,
|
||||
isMain: false,
|
||||
ownerAgentType: 'claude-code' as const,
|
||||
workDir: null,
|
||||
};
|
||||
|
||||
function seedWithSource(source: string) {
|
||||
const db = openInitializedInMemoryDatabase();
|
||||
insertStoredRoomSettings(db, jid, 'single', 'explicit', snapshot);
|
||||
db.prepare(
|
||||
'UPDATE room_settings SET mode_source = ? WHERE chat_jid = ?',
|
||||
).run(source, jid);
|
||||
return db;
|
||||
}
|
||||
|
||||
it('coerces an unrecognized mode_source to explicit instead of dropping the room', () => {
|
||||
const db = seedWithSource('room');
|
||||
|
||||
const stored = getStoredRoomSettingsRowFromDatabase(db, jid);
|
||||
expect(stored).toBeDefined();
|
||||
expect(stored?.roomMode).toBe('single');
|
||||
expect(stored?.modeSource).toBe('explicit');
|
||||
expect(stored?.folder).toBe('cgv-macro');
|
||||
|
||||
// And it stays visible in the full-room listing that feeds getAllRoomBindings.
|
||||
const listed = getStoredRoomRowsFromDatabase(db).map((r) => r.chatJid);
|
||||
expect(listed).toContain(jid);
|
||||
});
|
||||
|
||||
it('room_mode is guarded by a schema CHECK, but mode_source is not — hence the loader coercion', () => {
|
||||
const db = openInitializedInMemoryDatabase();
|
||||
insertStoredRoomSettings(db, jid, 'single', 'explicit', snapshot);
|
||||
|
||||
// room_mode cannot be corrupted at all: the column CHECK rejects the write.
|
||||
expect(() =>
|
||||
db
|
||||
.prepare('UPDATE room_settings SET room_mode = ? WHERE chat_jid = ?')
|
||||
.run('bogus', jid),
|
||||
).toThrow(/CHECK constraint/);
|
||||
|
||||
// mode_source has no such CHECK, so a bad value persists — which is exactly
|
||||
// the corruption the loader now tolerates instead of dropping the room.
|
||||
db.prepare(
|
||||
'UPDATE room_settings SET mode_source = ? WHERE chat_jid = ?',
|
||||
).run('room', jid);
|
||||
expect(getStoredRoomSettingsRowFromDatabase(db, jid)?.roomMode).toBe(
|
||||
'single',
|
||||
);
|
||||
});
|
||||
|
||||
it('preserves a valid mode_source unchanged', () => {
|
||||
const db = openInitializedInMemoryDatabase();
|
||||
insertStoredRoomSettings(db, jid, 'tribunal', 'inferred', snapshot);
|
||||
|
||||
expect(getStoredRoomSettingsRowFromDatabase(db, jid)?.modeSource).toBe(
|
||||
'inferred',
|
||||
);
|
||||
});
|
||||
});
|
||||
@@ -182,8 +182,23 @@ export function getStoredRoomSettingsRowFromDatabase(
|
||||
row?.room_mode === 'single' || row?.room_mode === 'tribunal'
|
||||
? row.room_mode
|
||||
: undefined;
|
||||
const source = normalizeRoomModeSource(row?.mode_source);
|
||||
if (!row || !roomMode || !source) return undefined;
|
||||
if (!row || !roomMode) return undefined;
|
||||
|
||||
// A row with a valid room_mode but an unrecognized mode_source must NOT be
|
||||
// silently dropped. Dropping it removes the room from every binding lookup,
|
||||
// so the router ignores the channel, it disappears from the status list, and
|
||||
// re-registration wedges on the UNIQUE(chat_jid) constraint (assignRoom's
|
||||
// "existing" probe also relies on this function). Coerce the source to
|
||||
// 'explicit' — which preserves the stored room_mode as-is — and warn so the
|
||||
// underlying corruption stays visible and self-heals on the next assignRoom.
|
||||
let source = normalizeRoomModeSource(row.mode_source);
|
||||
if (!source) {
|
||||
logger.warn(
|
||||
{ jid: chatJid, modeSource: row.mode_source ?? null },
|
||||
'room_settings row has invalid mode_source; coercing to explicit to keep room routable',
|
||||
);
|
||||
source = 'explicit';
|
||||
}
|
||||
|
||||
return {
|
||||
chatJid,
|
||||
|
||||
@@ -59,12 +59,28 @@ export class GroupQueue {
|
||||
return state;
|
||||
}
|
||||
|
||||
private onActivityChange: (() => void) | null = null;
|
||||
|
||||
setProcessMessagesFn(
|
||||
fn: (groupJid: string, context: GroupRunContext) => Promise<boolean>,
|
||||
): void {
|
||||
this.processMessagesFn = fn;
|
||||
}
|
||||
|
||||
/** Notified when a group starts or finishes a run (activeCount changes). */
|
||||
setOnActivityChange(fn: () => void): void {
|
||||
this.onActivityChange = fn;
|
||||
}
|
||||
|
||||
private notifyActivityChange(): void {
|
||||
if (!this.onActivityChange) return;
|
||||
try {
|
||||
this.onActivityChange();
|
||||
} catch (err) {
|
||||
logger.debug({ err }, 'onActivityChange callback threw');
|
||||
}
|
||||
}
|
||||
|
||||
/** Limit concurrency after restart to avoid API rate-limit storms. */
|
||||
enterRecoveryMode(): void {
|
||||
this.recoveryMode = true;
|
||||
@@ -528,6 +544,7 @@ export class GroupQueue {
|
||||
transitionRunPhase(state, groupJid, 'running_messages', { runId, reason });
|
||||
assertRunPhaseInvariants(state, groupJid);
|
||||
this.activeCount++;
|
||||
this.notifyActivityChange();
|
||||
|
||||
logger.info(
|
||||
{
|
||||
@@ -580,6 +597,7 @@ export class GroupQueue {
|
||||
resetRunState(state, groupJid);
|
||||
assertRunPhaseInvariants(state, groupJid);
|
||||
this.activeCount--;
|
||||
this.notifyActivityChange();
|
||||
this.drainGroup(groupJid);
|
||||
}
|
||||
}
|
||||
@@ -592,6 +610,7 @@ export class GroupQueue {
|
||||
assertRunPhaseInvariants(state, groupJid);
|
||||
this.activeCount++;
|
||||
this.activeTaskCount++;
|
||||
this.notifyActivityChange();
|
||||
|
||||
logger.info(
|
||||
{
|
||||
@@ -628,6 +647,7 @@ export class GroupQueue {
|
||||
assertRunPhaseInvariants(state, groupJid);
|
||||
this.activeCount--;
|
||||
this.activeTaskCount--;
|
||||
this.notifyActivityChange();
|
||||
this.drainGroup(groupJid);
|
||||
}
|
||||
}
|
||||
|
||||
13
src/index.ts
13
src/index.ts
@@ -52,7 +52,10 @@ import {
|
||||
} from './sender-allowlist.js';
|
||||
import { createMessageRuntime } from './message-runtime.js';
|
||||
import { nudgeSchedulerLoop, startSchedulerLoop } from './task-scheduler.js';
|
||||
import { startUnifiedDashboard } from './unified-dashboard.js';
|
||||
import {
|
||||
requestImmediateStatusUpdate,
|
||||
startUnifiedDashboard,
|
||||
} from './unified-dashboard.js';
|
||||
import { startUsagePrimer } from './usage-primer.js';
|
||||
import { startWebDashboardServer } from './web-dashboard-server.js';
|
||||
import { Channel, NewMessage, RegisteredGroup } from './types.js';
|
||||
@@ -395,6 +398,11 @@ async function main(): Promise<void> {
|
||||
}
|
||||
}
|
||||
storeMessage(msg);
|
||||
// A real chat message in a registered room: refresh the status dashboard
|
||||
// immediately instead of waiting for the next periodic tick.
|
||||
if (!msg.is_from_me && !msg.is_bot_message && roomBindings[chatJid]) {
|
||||
requestImmediateStatusUpdate();
|
||||
}
|
||||
},
|
||||
onChatMetadata: (
|
||||
chatJid: string,
|
||||
@@ -569,6 +577,9 @@ async function main(): Promise<void> {
|
||||
},
|
||||
purgeOnStart: true,
|
||||
});
|
||||
// Refresh the status dashboard immediately when an agent starts/finishes a run
|
||||
// (activity moves between rooms) rather than waiting for the periodic tick.
|
||||
queue.setOnActivityChange(requestImmediateStatusUpdate);
|
||||
webDashboardServer = startWebDashboardServer({
|
||||
...WEB_DASHBOARD,
|
||||
getRoomBindings: runtimeState.getRoomBindings,
|
||||
|
||||
@@ -285,6 +285,12 @@ export function createExecuteTurn(deps: CreateExecuteTurnDeps): ExecuteTurnFn {
|
||||
visiblePhase,
|
||||
};
|
||||
} finally {
|
||||
// Always tear down the controller's background timers. finish() clears
|
||||
// them on the normal path, but it lives in the try above — if runAgent
|
||||
// throws / is aborted / is killed, finish() is skipped and the 5s
|
||||
// progress ticker would otherwise keep editing the Discord message
|
||||
// forever (an orphaned "stuck at 0s" zombie). dispose() is idempotent.
|
||||
turnController.dispose();
|
||||
turnController.cancelPendingTypingDelay();
|
||||
logger.debug(
|
||||
{
|
||||
|
||||
@@ -162,6 +162,62 @@ describe('MessageTurnController outbound audit logging', () => {
|
||||
);
|
||||
});
|
||||
|
||||
it('dispose() stops the progress ticker so an aborted turn leaves no zombie edits', async () => {
|
||||
vi.useFakeTimers();
|
||||
try {
|
||||
const channel = makeChannel();
|
||||
const controller = new MessageTurnController({
|
||||
chatJid: 'dc:test-room',
|
||||
group: makeGroup(),
|
||||
runId: 'run-zombie-1',
|
||||
channel,
|
||||
idleTimeout: 1_000,
|
||||
failureFinalText: '실패',
|
||||
isClaudeCodeAgent: true,
|
||||
clearSession: vi.fn(),
|
||||
requestClose: vi.fn(),
|
||||
deliverFinalText: vi.fn().mockResolvedValue(true),
|
||||
deliveryRole: 'reviewer',
|
||||
deliveryServiceId: 'codex-review',
|
||||
pairedTurnIdentity: makeTurnIdentity(),
|
||||
});
|
||||
|
||||
await controller.start();
|
||||
// Subagent tool-activity creates the tracked progress message and starts
|
||||
// the 5s ticker.
|
||||
await controller.handleOutput({
|
||||
status: 'success',
|
||||
phase: 'tool-activity',
|
||||
agentId: 'sub-1',
|
||||
result: '작업 중',
|
||||
} as any);
|
||||
// Flush the async progress-message creation (.then → progressMessageId).
|
||||
await vi.advanceTimersByTimeAsync(0);
|
||||
|
||||
// Ticker is live: advancing past the 5s interval edits the message.
|
||||
await vi.advanceTimersByTimeAsync(5_000);
|
||||
expect(vi.mocked(channel.editMessage)).toHaveBeenCalled();
|
||||
|
||||
// Abnormal exit: finish() is NEVER called (agent threw / turn aborted).
|
||||
// The caller's finally must call dispose().
|
||||
controller.dispose();
|
||||
|
||||
const editsAfterDispose = vi.mocked(channel.editMessage!).mock.calls
|
||||
.length;
|
||||
await vi.advanceTimersByTimeAsync(30_000);
|
||||
|
||||
// No further edits — the orphaned "stuck at 0s" zombie ticker is gone.
|
||||
expect(vi.mocked(channel.editMessage!).mock.calls.length).toBe(
|
||||
editsAfterDispose,
|
||||
);
|
||||
|
||||
// Idempotent: calling dispose() again is a no-op and does not throw.
|
||||
expect(() => controller.dispose()).not.toThrow();
|
||||
} finally {
|
||||
vi.useRealTimers();
|
||||
}
|
||||
});
|
||||
|
||||
it('logs fallback progress audit when tracked message creation fails', async () => {
|
||||
const channel = makeChannel();
|
||||
const deliverFinalText = vi.fn().mockResolvedValue(true);
|
||||
|
||||
@@ -479,6 +479,26 @@ export class MessageTurnController {
|
||||
this.progressTicker = null;
|
||||
}
|
||||
|
||||
/**
|
||||
* Idempotent teardown of background timers (progress ticker + idle timer).
|
||||
*
|
||||
* `finish()` already tears these down on the normal completion path, but it
|
||||
* lives inside the caller's `try` block. When the agent run throws / is
|
||||
* aborted / is killed, control jumps straight to the caller's `finally` and
|
||||
* `finish()` is skipped — leaving the 5s progress ticker editing the Discord
|
||||
* message forever (an orphaned "stuck at 0s" zombie progress message).
|
||||
*
|
||||
* Callers MUST invoke this from a `finally` so timers are cleared on every
|
||||
* exit path. Safe to call more than once and safe to call after finish().
|
||||
*/
|
||||
dispose(): void {
|
||||
this.clearProgressTicker();
|
||||
if (this.idleTimer) {
|
||||
clearTimeout(this.idleTimer);
|
||||
this.idleTimer = null;
|
||||
}
|
||||
}
|
||||
|
||||
private resetProgressState(): void {
|
||||
this.clearProgressTicker();
|
||||
this.pendingProgressText = null;
|
||||
|
||||
@@ -123,6 +123,37 @@ describe('paired workspace manager', () => {
|
||||
fs.rmSync(tempRoot, { recursive: true, force: true });
|
||||
});
|
||||
|
||||
it('ensurePairedWorkspaceProvisioned inits a canonical repo with a commit and upserts the project', async () => {
|
||||
const { db, manager } = await loadModules();
|
||||
db._initTestDatabase();
|
||||
|
||||
const workDir = path.join(tempRoot, 'canon');
|
||||
const result = manager.ensurePairedWorkspaceProvisioned({
|
||||
chatJid: 'dc:test-room',
|
||||
groupFolder: 'test-group',
|
||||
workDir,
|
||||
});
|
||||
|
||||
expect(result).toBe(workDir);
|
||||
expect(fs.existsSync(path.join(workDir, '.git'))).toBe(true);
|
||||
// HEAD must resolve so owner-workspace provisioning can `git worktree add`.
|
||||
expect(runGit(['rev-parse', 'HEAD'], workDir)).toMatch(/^[0-9a-f]{40}$/);
|
||||
// A paired_projects row is created from the canonical work dir, without
|
||||
// which the tribunal flow throws "Paired project not found" and no
|
||||
// reviewer/arbiter ever runs.
|
||||
const project = db.getPairedProject('dc:test-room');
|
||||
expect(project?.canonical_work_dir).toBe(workDir);
|
||||
|
||||
// Idempotent — a second call reuses the repo and does not throw.
|
||||
expect(() =>
|
||||
manager.ensurePairedWorkspaceProvisioned({
|
||||
chatJid: 'dc:test-room',
|
||||
groupFolder: 'test-group',
|
||||
workDir,
|
||||
}),
|
||||
).not.toThrow();
|
||||
});
|
||||
|
||||
it('registers the owner workspace for reviewer execution when review is requested', async () => {
|
||||
const { db, manager } = await loadModules();
|
||||
db._initTestDatabase();
|
||||
|
||||
@@ -3,11 +3,12 @@ import crypto from 'crypto';
|
||||
import fs from 'fs';
|
||||
import path from 'path';
|
||||
|
||||
import { DATA_DIR } from './config.js';
|
||||
import { DATA_DIR, GROUPS_DIR } from './config.js';
|
||||
import {
|
||||
getPairedProject,
|
||||
getPairedTaskById,
|
||||
getPairedWorkspace,
|
||||
upsertPairedProject,
|
||||
upsertPairedWorkspace,
|
||||
} from './db.js';
|
||||
import { resolvePairedTaskWorkspacePath } from './group-folder.js';
|
||||
@@ -610,6 +611,62 @@ function getTaskAndProject(taskId: string): {
|
||||
return { task, canonicalWorkDir: project.canonical_work_dir };
|
||||
}
|
||||
|
||||
/**
|
||||
* Ensure a room's canonical paired workspace exists so the tribunal flow can
|
||||
* provision an owner worktree and run the reviewer/arbiter.
|
||||
*
|
||||
* Root cause this guards against: a room registered as tribunal with no
|
||||
* `work_dir` → `ensurePairedProject` returns null → no paired task is created →
|
||||
* reviewer/arbiter never run (the symptom seen in web-vstock and other rooms).
|
||||
*
|
||||
* This defaults the canonical work dir to `groups/<folder>`, guarantees it is a
|
||||
* standalone git repo with at least one commit (so `git worktree add` can
|
||||
* resolve HEAD during owner-workspace provisioning), and upserts the
|
||||
* `paired_projects` row. Idempotent — safe to call on every registration.
|
||||
*/
|
||||
export function ensurePairedWorkspaceProvisioned(args: {
|
||||
chatJid: string;
|
||||
groupFolder: string;
|
||||
workDir?: string | null;
|
||||
}): string {
|
||||
const canonicalWorkDir =
|
||||
args.workDir && args.workDir.trim().length > 0
|
||||
? args.workDir
|
||||
: path.join(GROUPS_DIR, args.groupFolder);
|
||||
|
||||
fs.mkdirSync(canonicalWorkDir, { recursive: true });
|
||||
|
||||
// Detect a standalone repo by its OWN `.git` only. A bare `git rev-parse`
|
||||
// would walk up to a parent repo (e.g. the EJClaw checkout that contains
|
||||
// groups/<folder>) and falsely report the dir as a repo — the exact bug that
|
||||
// produced stray EJClaw worktrees. So init a fresh repo when no local .git.
|
||||
if (!fs.existsSync(path.join(canonicalWorkDir, '.git'))) {
|
||||
runGit(['init', '-b', 'main'], canonicalWorkDir);
|
||||
}
|
||||
|
||||
if (!resolveCommit(canonicalWorkDir, 'HEAD')) {
|
||||
// Fresh repo with no commits — owner-workspace provisioning needs a
|
||||
// resolvable HEAD, so seed an empty initial commit with a local identity.
|
||||
runGit(['config', 'user.email', 'claude-bot@ejclaw'], canonicalWorkDir);
|
||||
runGit(['config', 'user.name', 'EJClaw'], canonicalWorkDir);
|
||||
runGit(
|
||||
['commit', '--allow-empty', '-m', 'chore: initialize paired workspace'],
|
||||
canonicalWorkDir,
|
||||
);
|
||||
}
|
||||
|
||||
const now = new Date().toISOString();
|
||||
upsertPairedProject({
|
||||
chat_jid: args.chatJid,
|
||||
group_folder: args.groupFolder,
|
||||
canonical_work_dir: canonicalWorkDir,
|
||||
created_at: now,
|
||||
updated_at: now,
|
||||
});
|
||||
|
||||
return canonicalWorkDir;
|
||||
}
|
||||
|
||||
function makeWorkspaceRecord(args: {
|
||||
taskId: string;
|
||||
role: PairedWorkspace['role'];
|
||||
|
||||
@@ -2,9 +2,23 @@ import { describe, expect, it } from 'vitest';
|
||||
|
||||
import {
|
||||
applyUpdatedTokensToEnvContent,
|
||||
collectSessionCredentialPaths,
|
||||
pickFreshestOAuth,
|
||||
shouldAdoptSessionOAuth,
|
||||
shouldStartTokenRefreshLoop,
|
||||
} from './token-refresh.js';
|
||||
|
||||
function creds(expiresAt: number, refreshToken = 'rt') {
|
||||
return {
|
||||
claudeAiOauth: {
|
||||
accessToken: `at-${expiresAt}`,
|
||||
refreshToken,
|
||||
expiresAt,
|
||||
scopes: [] as string[],
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
describe('shouldStartTokenRefreshLoop', () => {
|
||||
it('starts refresh for the Claude service', () => {
|
||||
expect(shouldStartTokenRefreshLoop('claude-code')).toBe(true);
|
||||
@@ -42,3 +56,110 @@ describe('shouldStartTokenRefreshLoop', () => {
|
||||
);
|
||||
});
|
||||
});
|
||||
|
||||
describe('pickFreshestOAuth', () => {
|
||||
it('adopts a session copy that expires later than the source', () => {
|
||||
const best = pickFreshestOAuth([
|
||||
{ label: 'source', creds: creds(1000) },
|
||||
{ label: 'sessionA', creds: creds(5000) },
|
||||
{ label: 'sessionB', creds: creds(3000) },
|
||||
]);
|
||||
expect(best?.label).toBe('sessionA');
|
||||
});
|
||||
|
||||
it('keeps the source on ties so we do not churn copies needlessly', () => {
|
||||
const best = pickFreshestOAuth([
|
||||
{ label: 'source', creds: creds(5000) },
|
||||
{ label: 'sessionA', creds: creds(5000) },
|
||||
]);
|
||||
expect(best?.label).toBe('source');
|
||||
});
|
||||
|
||||
it('ignores candidates without a refresh token or numeric expiry', () => {
|
||||
const best = pickFreshestOAuth([
|
||||
{ label: 'source', creds: creds(1000) },
|
||||
{ label: 'noRefresh', creds: creds(9000, '') },
|
||||
{ label: 'missing', creds: null },
|
||||
]);
|
||||
expect(best?.label).toBe('source');
|
||||
});
|
||||
|
||||
it('returns null when no candidate is usable', () => {
|
||||
expect(
|
||||
pickFreshestOAuth([
|
||||
{ label: 'source', creds: null },
|
||||
{ label: 'noRefresh', creds: creds(9000, '') },
|
||||
]),
|
||||
).toBeNull();
|
||||
});
|
||||
});
|
||||
|
||||
describe('shouldAdoptSessionOAuth (session→canonical write-back)', () => {
|
||||
const oauth = (expiresAt: number, extra: Record<string, unknown> = {}) => ({
|
||||
accessToken: `at-${expiresAt}`,
|
||||
refreshToken: `rt-${expiresAt}`,
|
||||
expiresAt,
|
||||
scopes: [] as string[],
|
||||
...extra,
|
||||
});
|
||||
|
||||
it('adopts a session token strictly newer than canonical', () => {
|
||||
expect(shouldAdoptSessionOAuth(oauth(2000), oauth(1000))).toBe(true);
|
||||
});
|
||||
|
||||
it('does not regress to an older or equal session token', () => {
|
||||
expect(shouldAdoptSessionOAuth(oauth(1000), oauth(2000))).toBe(false);
|
||||
expect(shouldAdoptSessionOAuth(oauth(2000), oauth(2000))).toBe(false);
|
||||
});
|
||||
|
||||
it('adopts when canonical is missing/invalid', () => {
|
||||
expect(shouldAdoptSessionOAuth(oauth(1000), null)).toBe(true);
|
||||
});
|
||||
|
||||
it('rejects an invalid session token', () => {
|
||||
expect(shouldAdoptSessionOAuth(null, oauth(1000))).toBe(false);
|
||||
expect(
|
||||
shouldAdoptSessionOAuth(
|
||||
{ accessToken: '', refreshToken: '', expiresAt: 5000, scopes: [] },
|
||||
oauth(1000),
|
||||
),
|
||||
).toBe(false);
|
||||
});
|
||||
|
||||
it('refuses to cross subscription types', () => {
|
||||
expect(
|
||||
shouldAdoptSessionOAuth(
|
||||
oauth(2000, { subscriptionType: 'pro' }),
|
||||
oauth(1000, { subscriptionType: 'max' }),
|
||||
),
|
||||
).toBe(false);
|
||||
});
|
||||
});
|
||||
|
||||
describe('collectSessionCredentialPaths (real session layout)', () => {
|
||||
it('builds service group + task credential paths, not <group>/.claude', () => {
|
||||
const tree: Record<string, string[]> = {
|
||||
'/s': ['iger', 'javis_bot'],
|
||||
'/s/iger/services': ['claude'],
|
||||
'/s/iger/services/claude/tasks': ['task-1'],
|
||||
'/s/javis_bot/services': ['claude'],
|
||||
'/s/javis_bot/services/claude/tasks': [],
|
||||
};
|
||||
const paths = collectSessionCredentialPaths('/s', (dir) => tree[dir] ?? []);
|
||||
expect(paths).toContain(
|
||||
'/s/iger/services/claude/.claude/.credentials.json',
|
||||
);
|
||||
expect(paths).toContain(
|
||||
'/s/iger/services/claude/tasks/task-1/.claude/.credentials.json',
|
||||
);
|
||||
expect(paths).toContain(
|
||||
'/s/javis_bot/services/claude/.claude/.credentials.json',
|
||||
);
|
||||
// Must NOT use the old broken <group>/.claude path that matched nothing.
|
||||
expect(paths).not.toContain('/s/iger/.claude/.credentials.json');
|
||||
});
|
||||
|
||||
it('returns nothing when there are no session folders', () => {
|
||||
expect(collectSessionCredentialPaths('/s', () => [])).toEqual([]);
|
||||
});
|
||||
});
|
||||
|
||||
@@ -12,7 +12,6 @@
|
||||
* the .env file so new tokens survive restarts.
|
||||
*/
|
||||
import fs from 'fs';
|
||||
import os from 'os';
|
||||
import { getErrorMessage, readJsonFile } from './utils.js';
|
||||
import path from 'path';
|
||||
|
||||
@@ -20,6 +19,10 @@ import { logger } from './logger.js';
|
||||
import { DATA_DIR } from './config.js';
|
||||
import { getAllTokens, updateTokenValue } from './token-rotation.js';
|
||||
import type { AgentType } from './types.js';
|
||||
import {
|
||||
getClaudeCredentialsPath,
|
||||
hasExplicitClaudeCredentialsPath,
|
||||
} from './claude-credentials-path.js';
|
||||
|
||||
const TOKEN_URL = 'https://api.anthropic.com/v1/oauth/token';
|
||||
const CLIENT_ID = '9d1c250a-e61b-44d9-88ed-5944d1962f5e';
|
||||
@@ -30,11 +33,85 @@ const DEFAULT_SCOPES = [
|
||||
'user:mcp_servers',
|
||||
];
|
||||
|
||||
// Check every 5 minutes, refresh if within 30 minutes of expiry
|
||||
// Check every 5 minutes.
|
||||
const CHECK_INTERVAL_MS = 5 * 60 * 1000;
|
||||
const REFRESH_BEFORE_EXPIRY_MS = 30 * 60 * 1000;
|
||||
// Refresh when within 60 minutes of expiry. This is intentionally EARLIER than
|
||||
// a spawned Claude Code agent's own internal refresh threshold: by refreshing
|
||||
// first and syncing the fresh token into every session dir, the app becomes the
|
||||
// consistent "winner" of the token-family rotation and agents rarely need to
|
||||
// rotate on their own. That shrinks the cross-process refresh race that was
|
||||
// revoking the whole token family (see loadFreshestCredentials below).
|
||||
const REFRESH_BEFORE_EXPIRY_MS = 60 * 60 * 1000;
|
||||
const REQUEST_TIMEOUT_MS = 15_000;
|
||||
|
||||
// After the refresh token itself has expired (or the endpoint keeps rejecting
|
||||
// us), retrying every CHECK_INTERVAL_MS forever is pointless — only a manual
|
||||
// browser re-login can recover. Keep trying for ~30 minutes, then give up and
|
||||
// flag the account as needing manual re-login. Incoming requests can then be
|
||||
// answered with a "please re-login" message instead of failing opaquely.
|
||||
const GIVE_UP_AFTER_MS = 30 * 60 * 1000;
|
||||
|
||||
/** User-facing message shown when an account needs a manual re-login. */
|
||||
export const RELOGIN_REQUIRED_MESSAGE =
|
||||
'로그인 정보를 찾을 수 없습니다. 재로그인 해주세요.';
|
||||
|
||||
interface AuthFailureState {
|
||||
firstFailureAt: number;
|
||||
gaveUp: boolean;
|
||||
}
|
||||
// Per-account continuous-refresh-failure tracking. Cleared as soon as a valid
|
||||
// token appears again (e.g. after a manual re-login rewrites the creds file).
|
||||
const authFailureByAccount = new Map<number, AuthFailureState>();
|
||||
|
||||
/**
|
||||
* True when automatic refresh for this account has been failing for longer than
|
||||
* GIVE_UP_AFTER_MS and we have stopped retrying. Request handlers use this to
|
||||
* reply with RELOGIN_REQUIRED_MESSAGE instead of running a doomed agent.
|
||||
*/
|
||||
export function isReloginRequired(accountIndex = 0): boolean {
|
||||
return authFailureByAccount.get(accountIndex)?.gaveUp === true;
|
||||
}
|
||||
|
||||
/** Clear failure tracking once a valid token is observed again. */
|
||||
function recordAuthSuccess(accountIndex: number): void {
|
||||
if (authFailureByAccount.has(accountIndex)) {
|
||||
authFailureByAccount.delete(accountIndex);
|
||||
logger.info(
|
||||
{ accountIndex },
|
||||
'Claude OAuth recovered — cleared re-login-required state',
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Record a hard refresh failure. Returns true once we have given up (been
|
||||
* failing for >= GIVE_UP_AFTER_MS), at which point the caller should stop
|
||||
* hitting the network.
|
||||
*/
|
||||
function recordAuthFailure(accountIndex: number): boolean {
|
||||
const now = Date.now();
|
||||
const state = authFailureByAccount.get(accountIndex);
|
||||
if (!state) {
|
||||
authFailureByAccount.set(accountIndex, {
|
||||
firstFailureAt: now,
|
||||
gaveUp: false,
|
||||
});
|
||||
return false;
|
||||
}
|
||||
if (!state.gaveUp && now - state.firstFailureAt >= GIVE_UP_AFTER_MS) {
|
||||
state.gaveUp = true;
|
||||
logger.error(
|
||||
{
|
||||
accountIndex,
|
||||
failingForMin: Math.round((now - state.firstFailureAt) / 60000),
|
||||
},
|
||||
'Claude OAuth refresh has failed for over 30 min — giving up automatic ' +
|
||||
'retries. Manual re-login required (run claude_relogin.sh).',
|
||||
);
|
||||
}
|
||||
return state.gaveUp;
|
||||
}
|
||||
|
||||
export function shouldStartTokenRefreshLoop(
|
||||
serviceAgentType: AgentType,
|
||||
): boolean {
|
||||
@@ -67,19 +144,25 @@ interface TokenResponse {
|
||||
* - Index 1+: ~/.claude-accounts/{index}/.credentials.json
|
||||
*/
|
||||
function getCredentialsPath(accountIndex: number): string {
|
||||
if (accountIndex === 0) {
|
||||
return path.join(os.homedir(), '.claude', '.credentials.json');
|
||||
const credsPath = getClaudeCredentialsPath(accountIndex, {
|
||||
allowHomeFallback:
|
||||
process.env.CLAUDE_TOKEN_REFRESH_USE_HOME_CREDENTIALS === 'true',
|
||||
});
|
||||
if (!credsPath) {
|
||||
throw new Error(
|
||||
`No Claude credentials path configured for account ${accountIndex}`,
|
||||
);
|
||||
}
|
||||
return path.join(
|
||||
os.homedir(),
|
||||
'.claude-accounts',
|
||||
String(accountIndex),
|
||||
'.credentials.json',
|
||||
);
|
||||
return credsPath;
|
||||
}
|
||||
|
||||
function readCredentials(accountIndex: number): CredentialsFile | null {
|
||||
const credsPath = getCredentialsPath(accountIndex);
|
||||
let credsPath: string;
|
||||
try {
|
||||
credsPath = getCredentialsPath(accountIndex);
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
if (!fs.existsSync(credsPath)) return null;
|
||||
const data = readJsonFile<CredentialsFile>(credsPath);
|
||||
if (!data) {
|
||||
@@ -102,19 +185,69 @@ function writeCredentials(accountIndex: number, creds: CredentialsFile): void {
|
||||
syncToSessionDirs(credsPath);
|
||||
}
|
||||
|
||||
function syncToSessionDirs(credsPath: string): void {
|
||||
/**
|
||||
* Pure builder for the on-disk session credential paths. Spawned agents run with
|
||||
* CLAUDE_CONFIG_DIR pointing at their session `.claude` dir and keep their own
|
||||
* copy of the credentials, which Claude Code may refresh (rotate) on its own.
|
||||
*
|
||||
* The real layout is:
|
||||
* <sessions>/<folder>/services/<serviceId>/.claude/.credentials.json (group session)
|
||||
* <sessions>/<folder>/services/<serviceId>/tasks/<taskId>/.claude/.credentials.json (task session)
|
||||
*
|
||||
* (An earlier version scanned <sessions>/<folder>/.claude/... which never
|
||||
* matched, so the fan-out and stale-copy scan silently missed every real
|
||||
* session file.) `listDirs` is injected so this is unit-testable.
|
||||
*/
|
||||
export function collectSessionCredentialPaths(
|
||||
sessionsDir: string,
|
||||
listDirs: (dir: string) => string[],
|
||||
): string[] {
|
||||
const CRED = ['.claude', '.credentials.json'] as const;
|
||||
const paths: string[] = [];
|
||||
for (const folder of listDirs(sessionsDir)) {
|
||||
const servicesDir = path.join(sessionsDir, folder, 'services');
|
||||
for (const serviceId of listDirs(servicesDir)) {
|
||||
const serviceDir = path.join(servicesDir, serviceId);
|
||||
paths.push(path.join(serviceDir, ...CRED));
|
||||
const tasksDir = path.join(serviceDir, 'tasks');
|
||||
for (const taskId of listDirs(tasksDir)) {
|
||||
paths.push(path.join(tasksDir, taskId, ...CRED));
|
||||
}
|
||||
}
|
||||
}
|
||||
return paths;
|
||||
}
|
||||
|
||||
function listSubdirectories(dir: string): string[] {
|
||||
try {
|
||||
return fs
|
||||
.readdirSync(dir, { withFileTypes: true })
|
||||
.filter((entry) => entry.isDirectory())
|
||||
.map((entry) => entry.name);
|
||||
} catch {
|
||||
return [];
|
||||
}
|
||||
}
|
||||
|
||||
function listSessionCredentialPaths(): string[] {
|
||||
const sessionsDir = path.join(DATA_DIR, 'sessions');
|
||||
try {
|
||||
if (!fs.existsSync(sessionsDir)) return;
|
||||
const groups = fs.readdirSync(sessionsDir);
|
||||
if (!fs.existsSync(sessionsDir)) return [];
|
||||
return collectSessionCredentialPaths(sessionsDir, listSubdirectories);
|
||||
} catch (err) {
|
||||
logger.warn(
|
||||
{ err: getErrorMessage(err) },
|
||||
'Failed to enumerate session credential dirs',
|
||||
);
|
||||
return [];
|
||||
}
|
||||
}
|
||||
|
||||
function syncToSessionDirs(credsPath: string): void {
|
||||
try {
|
||||
let synced = 0;
|
||||
for (const group of groups) {
|
||||
const dest = path.join(
|
||||
sessionsDir,
|
||||
group,
|
||||
'.claude',
|
||||
'.credentials.json',
|
||||
);
|
||||
for (const dest of listSessionCredentialPaths()) {
|
||||
if (dest === credsPath) continue;
|
||||
if (fs.existsSync(path.dirname(dest))) {
|
||||
fs.copyFileSync(credsPath, dest);
|
||||
synced++;
|
||||
@@ -134,6 +267,157 @@ function syncToSessionDirs(credsPath: string): void {
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Pure decision for the session→canonical write-back: adopt the session token
|
||||
* only when it is a valid credential that is strictly newer than canonical (the
|
||||
* session refreshed mid-turn) and does not cross subscription types. Never
|
||||
* regress canonical to an older/equal token.
|
||||
*/
|
||||
export function shouldAdoptSessionOAuth(
|
||||
session: OAuthCredentials | undefined | null,
|
||||
canonical: OAuthCredentials | undefined | null,
|
||||
): boolean {
|
||||
if (
|
||||
!session?.accessToken ||
|
||||
!session?.refreshToken ||
|
||||
typeof session.expiresAt !== 'number'
|
||||
) {
|
||||
return false;
|
||||
}
|
||||
if (
|
||||
canonical?.subscriptionType &&
|
||||
session.subscriptionType &&
|
||||
canonical.subscriptionType !== session.subscriptionType
|
||||
) {
|
||||
return false;
|
||||
}
|
||||
if (
|
||||
canonical &&
|
||||
typeof canonical.expiresAt === 'number' &&
|
||||
session.expiresAt <= canonical.expiresAt
|
||||
) {
|
||||
return false;
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
/**
|
||||
* Write-back (session → canonical). After a Claude agent turn, the child Claude
|
||||
* CLI running with CLAUDE_CONFIG_DIR=<sessionClaudeDir> may have refreshed
|
||||
* (rotated) the OAuth token in its own copy. If that copy is strictly newer than
|
||||
* the canonical credentials, adopt it so the canonical file always holds the
|
||||
* latest rotated token. Without this the main refresh loop can later refresh an
|
||||
* already-rotated token, which Anthropic rejects with invalid_grant and revokes
|
||||
* the whole token family (forcing a manual re-login). Mirrors the Codex
|
||||
* syncCodexSessionAuthBack write-back. Returns true if canonical was updated.
|
||||
*/
|
||||
export function syncClaudeSessionAuthBack(
|
||||
sessionClaudeDir: string,
|
||||
accountIndex = 0,
|
||||
): boolean {
|
||||
if (!sessionClaudeDir) return false;
|
||||
const sessionPath = path.join(sessionClaudeDir, '.credentials.json');
|
||||
let sessionCreds: CredentialsFile | null = null;
|
||||
try {
|
||||
if (!fs.existsSync(sessionPath)) return false;
|
||||
sessionCreds = JSON.parse(
|
||||
fs.readFileSync(sessionPath, 'utf-8'),
|
||||
) as CredentialsFile;
|
||||
} catch {
|
||||
return false;
|
||||
}
|
||||
const sOauth = sessionCreds?.claudeAiOauth;
|
||||
if (!sOauth) return false;
|
||||
const canonical = readCredentials(accountIndex);
|
||||
if (!shouldAdoptSessionOAuth(sOauth, canonical?.claudeAiOauth)) {
|
||||
return false;
|
||||
}
|
||||
|
||||
// writeCredentials also fans the adopted token out to every session dir, so
|
||||
// stale copies (old refresh tokens that could trigger family revocation) are
|
||||
// overwritten with the current one.
|
||||
writeCredentials(accountIndex, sessionCreds);
|
||||
recordAuthSuccess(accountIndex);
|
||||
logger.info(
|
||||
{
|
||||
accountIndex,
|
||||
newExpiryMin: Math.round((sOauth.expiresAt - Date.now()) / 60000),
|
||||
},
|
||||
'Adopted refreshed Claude session token back into canonical credentials',
|
||||
);
|
||||
return true;
|
||||
}
|
||||
|
||||
/**
|
||||
* Pure selector: from a list of candidate credential files, pick the one whose
|
||||
* access token expires latest (i.e. the most recently rotated, still-live
|
||||
* member of the OAuth token family). Candidates without a refresh token or a
|
||||
* numeric expiresAt are ignored. The first candidate should be the source file
|
||||
* so that ties resolve in its favour (only a strictly-newer copy wins).
|
||||
*/
|
||||
export function pickFreshestOAuth(
|
||||
candidates: Array<{ label: string; creds: CredentialsFile | null }>,
|
||||
): { label: string; creds: CredentialsFile } | null {
|
||||
let best: { label: string; creds: CredentialsFile } | null = null;
|
||||
for (const candidate of candidates) {
|
||||
const oauth = candidate.creds?.claudeAiOauth;
|
||||
if (!oauth?.refreshToken || typeof oauth.expiresAt !== 'number') continue;
|
||||
if (!best || oauth.expiresAt > best.creds.claudeAiOauth.expiresAt) {
|
||||
best = {
|
||||
label: candidate.label,
|
||||
creds: candidate.creds as CredentialsFile,
|
||||
};
|
||||
}
|
||||
}
|
||||
return best;
|
||||
}
|
||||
|
||||
/**
|
||||
* Only the primary account (index 0 / the service's CLAUDE_CREDENTIALS_PATH) is
|
||||
* mirrored into session dirs. Scanning them for other accounts would adopt the
|
||||
* wrong token family.
|
||||
*/
|
||||
function isPrimarySessionAccount(accountIndex: number): boolean {
|
||||
return accountIndex === 0;
|
||||
}
|
||||
|
||||
interface FreshestCredentials {
|
||||
creds: CredentialsFile;
|
||||
/** True when the freshest copy came from a session dir, not the source file. */
|
||||
adoptedFromSession: boolean;
|
||||
label: string;
|
||||
}
|
||||
|
||||
/**
|
||||
* Read the source credentials plus every session-dir copy and return the
|
||||
* freshest (latest-expiring) one. A spawned agent may have already rotated the
|
||||
* token family in its own session copy; adopting that live token instead of
|
||||
* re-POSTing our now-stale refresh token is what prevents reuse-detection from
|
||||
* revoking the entire family.
|
||||
*/
|
||||
function loadFreshestCredentials(
|
||||
accountIndex: number,
|
||||
): FreshestCredentials | null {
|
||||
const candidates: Array<{ label: string; creds: CredentialsFile | null }> = [
|
||||
{ label: 'source', creds: readCredentials(accountIndex) },
|
||||
];
|
||||
if (isPrimarySessionAccount(accountIndex)) {
|
||||
for (const sessionPath of listSessionCredentialPaths()) {
|
||||
candidates.push({
|
||||
label: sessionPath,
|
||||
creds: readJsonFile<CredentialsFile>(sessionPath),
|
||||
});
|
||||
}
|
||||
}
|
||||
const best = pickFreshestOAuth(candidates);
|
||||
if (!best) return null;
|
||||
return {
|
||||
creds: best.creds,
|
||||
adoptedFromSession: best.label !== 'source',
|
||||
label: best.label,
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Update CLAUDE_CODE_OAUTH_TOKENS in .env so refreshed tokens survive restarts.
|
||||
*/
|
||||
@@ -231,30 +515,87 @@ async function refreshToken(
|
||||
throw new Error('Token refresh failed');
|
||||
}
|
||||
|
||||
// Serialize all refreshes within this process so the 5-minute loop and an
|
||||
// on-demand forceRefreshToken() (triggered by a 401) can never rotate the same
|
||||
// token family concurrently — a self-inflicted reuse that revokes the family.
|
||||
let refreshMutex: Promise<unknown> = Promise.resolve();
|
||||
function serializeRefresh<T>(fn: () => Promise<T>): Promise<T> {
|
||||
const next = refreshMutex.then(fn, fn);
|
||||
// Keep the chain alive regardless of individual outcomes.
|
||||
refreshMutex = next.then(
|
||||
() => undefined,
|
||||
() => undefined,
|
||||
);
|
||||
return next;
|
||||
}
|
||||
|
||||
/**
|
||||
* Check and refresh a single account's credentials.
|
||||
* Returns the new access token if refreshed, null otherwise.
|
||||
* Returns the (new or adopted) access token if it changed, null otherwise.
|
||||
*/
|
||||
async function checkAndRefreshAccount(
|
||||
accountIndex: number,
|
||||
opts?: { force?: boolean },
|
||||
): Promise<string | null> {
|
||||
const creds = readCredentials(accountIndex);
|
||||
if (!creds?.claudeAiOauth) return null;
|
||||
return serializeRefresh(() => doCheckAndRefreshAccount(accountIndex, opts));
|
||||
}
|
||||
|
||||
const { expiresAt, refreshToken: rt } = creds.claudeAiOauth;
|
||||
async function doCheckAndRefreshAccount(
|
||||
accountIndex: number,
|
||||
opts?: { force?: boolean },
|
||||
): Promise<string | null> {
|
||||
const freshest = loadFreshestCredentials(accountIndex);
|
||||
if (!freshest?.creds.claudeAiOauth) return null;
|
||||
|
||||
const creds = freshest.creds;
|
||||
const oauth = creds.claudeAiOauth;
|
||||
const rt = oauth.refreshToken;
|
||||
if (!rt) {
|
||||
logger.debug({ accountIndex }, 'No refresh token in credentials, skipping');
|
||||
return null;
|
||||
}
|
||||
|
||||
const now = Date.now();
|
||||
const remaining = expiresAt - now;
|
||||
// A concurrent refresher (typically a spawned agent) may have already rotated
|
||||
// the family in its session copy. Converge the source file onto that newer
|
||||
// token so we never re-POST a superseded refresh token.
|
||||
if (freshest.adoptedFromSession) {
|
||||
writeCredentials(accountIndex, creds);
|
||||
logger.info(
|
||||
{
|
||||
accountIndex,
|
||||
remainingMin: Math.round((oauth.expiresAt - Date.now()) / 60000),
|
||||
},
|
||||
'Adopted newer Claude token from a concurrent refresher (skipped duplicate refresh)',
|
||||
);
|
||||
}
|
||||
|
||||
if (!opts?.force && remaining > REFRESH_BEFORE_EXPIRY_MS) {
|
||||
const now = Date.now();
|
||||
const remaining = oauth.expiresAt - now;
|
||||
|
||||
if (remaining > REFRESH_BEFORE_EXPIRY_MS) {
|
||||
// A valid token is present again (typically after a manual re-login rewrote
|
||||
// the credentials file) — drop any prior "re-login required" state.
|
||||
recordAuthSuccess(accountIndex);
|
||||
if (!opts?.force && !freshest.adoptedFromSession) {
|
||||
logger.debug(
|
||||
{ accountIndex, remainingMin: Math.round(remaining / 60000) },
|
||||
'Token still valid, no refresh needed',
|
||||
);
|
||||
}
|
||||
// Token is valid; only report a token when it actually changed (adopted) or
|
||||
// a caller forced the check and wants a usable token back to retry with.
|
||||
return freshest.adoptedFromSession || opts?.force
|
||||
? oauth.accessToken
|
||||
: null;
|
||||
}
|
||||
|
||||
// Already gave up on this account (refresh token dead). Stop hammering the
|
||||
// endpoint every 5 min — a manual re-login is required, which the
|
||||
// healthy-token branch above will detect and clear automatically.
|
||||
if (isReloginRequired(accountIndex)) {
|
||||
logger.debug(
|
||||
{ accountIndex, remainingMin: Math.round(remaining / 60000) },
|
||||
'Token still valid, no refresh needed',
|
||||
{ accountIndex },
|
||||
'Skipping refresh — re-login required until a manual re-login is performed',
|
||||
);
|
||||
return null;
|
||||
}
|
||||
@@ -271,17 +612,15 @@ async function checkAndRefreshAccount(
|
||||
);
|
||||
|
||||
try {
|
||||
const response = await refreshToken(
|
||||
rt,
|
||||
creds.claudeAiOauth.scopes || DEFAULT_SCOPES,
|
||||
);
|
||||
const response = await refreshToken(rt, oauth.scopes || DEFAULT_SCOPES);
|
||||
|
||||
creds.claudeAiOauth.accessToken = response.access_token;
|
||||
creds.claudeAiOauth.refreshToken = response.refresh_token || rt;
|
||||
creds.claudeAiOauth.expiresAt = now + response.expires_in * 1000;
|
||||
recordAuthSuccess(accountIndex);
|
||||
oauth.accessToken = response.access_token;
|
||||
oauth.refreshToken = response.refresh_token || rt;
|
||||
oauth.expiresAt = Date.now() + response.expires_in * 1000;
|
||||
|
||||
if (response.scope) {
|
||||
creds.claudeAiOauth.scopes = response.scope.split(' ');
|
||||
oauth.scopes = response.scope.split(' ');
|
||||
}
|
||||
|
||||
writeCredentials(accountIndex, creds);
|
||||
@@ -294,12 +633,40 @@ async function checkAndRefreshAccount(
|
||||
|
||||
return response.access_token;
|
||||
} catch (err) {
|
||||
// We may simply have lost the rotation race: another process refreshed the
|
||||
// family a moment ago, so our token is now "expired". Re-scan every copy —
|
||||
// if a valid token exists, adopt it instead of reporting a hard failure.
|
||||
const recovered = loadFreshestCredentials(accountIndex);
|
||||
const recoveredOauth = recovered?.creds.claudeAiOauth;
|
||||
if (
|
||||
recoveredOauth &&
|
||||
recoveredOauth.expiresAt - Date.now() > REFRESH_BEFORE_EXPIRY_MS
|
||||
) {
|
||||
if (recovered.adoptedFromSession) {
|
||||
writeCredentials(accountIndex, recovered.creds);
|
||||
}
|
||||
recordAuthSuccess(accountIndex);
|
||||
logger.warn(
|
||||
{
|
||||
accountIndex,
|
||||
remainingMin: Math.round(
|
||||
(recoveredOauth.expiresAt - Date.now()) / 60000,
|
||||
),
|
||||
},
|
||||
'Refresh failed but a concurrent refresher produced a valid token — adopted it',
|
||||
);
|
||||
return recoveredOauth.accessToken;
|
||||
}
|
||||
const gaveUp = recordAuthFailure(accountIndex);
|
||||
logger.error(
|
||||
{
|
||||
accountIndex,
|
||||
gaveUp,
|
||||
err: getErrorMessage(err),
|
||||
},
|
||||
'Failed to refresh Claude OAuth token — manual re-login may be required',
|
||||
gaveUp
|
||||
? 'Failed to refresh Claude OAuth token — giving up until manual re-login'
|
||||
: 'Failed to refresh Claude OAuth token — manual re-login may be required',
|
||||
);
|
||||
return null;
|
||||
}
|
||||
@@ -339,6 +706,23 @@ async function checkAndRefreshAll(): Promise<void> {
|
||||
let refreshInterval: ReturnType<typeof setInterval> | null = null;
|
||||
|
||||
export function startTokenRefreshLoop(): void {
|
||||
const refreshEnabled =
|
||||
process.env.CLAUDE_TOKEN_REFRESH_USE_CREDENTIALS === 'true' ||
|
||||
process.env.CLAUDE_TOKEN_REFRESH_USE_HOME_CREDENTIALS === 'true';
|
||||
if (!refreshEnabled) {
|
||||
logger.info(
|
||||
'Claude token auto-refresh disabled; EJClaw is using its dedicated OAuth token',
|
||||
);
|
||||
return;
|
||||
}
|
||||
if (
|
||||
process.env.CLAUDE_TOKEN_REFRESH_USE_HOME_CREDENTIALS !== 'true' &&
|
||||
!hasExplicitClaudeCredentialsPath()
|
||||
) {
|
||||
logger.info('Claude token auto-refresh disabled; no credentials path set');
|
||||
return;
|
||||
}
|
||||
|
||||
const allTokens = getAllTokens();
|
||||
|
||||
// Check if any credentials files exist (for any configured account)
|
||||
|
||||
@@ -3,7 +3,10 @@ import { describe, expect, it } from 'vitest';
|
||||
import type { UsageRow } from './dashboard-usage-rows.js';
|
||||
import {
|
||||
buildWebUsageRowsForSnapshot,
|
||||
createCoalescingTrigger,
|
||||
editStatusMessageWithRetry,
|
||||
formatStatusHeader,
|
||||
msUntilNextMinuteBoundary,
|
||||
getDashboardDuplicateCleanupIntervalMs,
|
||||
renderUsageTable,
|
||||
shouldPurgeDashboardChannelOnStart,
|
||||
@@ -228,3 +231,157 @@ describe('buildWebUsageRowsForSnapshot', () => {
|
||||
expect(rows.map((row) => row.name)).toEqual(['Codex1']);
|
||||
});
|
||||
});
|
||||
|
||||
describe('editStatusMessageWithRetry', () => {
|
||||
it('succeeds on the first attempt without sleeping', async () => {
|
||||
let calls = 0;
|
||||
let slept = 0;
|
||||
const ok = await editStatusMessageWithRetry({
|
||||
editOnce: async () => {
|
||||
calls++;
|
||||
},
|
||||
maxRetries: 2,
|
||||
retryDelayMs: 15_000,
|
||||
sleep: async () => {
|
||||
slept++;
|
||||
},
|
||||
});
|
||||
expect(ok).toBe(true);
|
||||
expect(calls).toBe(1);
|
||||
expect(slept).toBe(0);
|
||||
});
|
||||
|
||||
it('retries after a failure and succeeds on the second attempt', async () => {
|
||||
let calls = 0;
|
||||
const delays: number[] = [];
|
||||
const ok = await editStatusMessageWithRetry({
|
||||
editOnce: async () => {
|
||||
calls++;
|
||||
if (calls === 1) throw new Error('503');
|
||||
},
|
||||
maxRetries: 2,
|
||||
retryDelayMs: 15_000,
|
||||
sleep: async (ms) => {
|
||||
delays.push(ms);
|
||||
},
|
||||
});
|
||||
expect(ok).toBe(true);
|
||||
expect(calls).toBe(2);
|
||||
expect(delays).toEqual([15_000]); // one 15s wait between the two attempts
|
||||
});
|
||||
|
||||
it('gives up after maxRetries (3 total attempts) and reports each failure', async () => {
|
||||
let calls = 0;
|
||||
const attempts: Array<{ attempt: number; willRetry: boolean }> = [];
|
||||
let slept = 0;
|
||||
const ok = await editStatusMessageWithRetry({
|
||||
editOnce: async () => {
|
||||
calls++;
|
||||
throw new Error('503');
|
||||
},
|
||||
maxRetries: 2,
|
||||
retryDelayMs: 15_000,
|
||||
sleep: async () => {
|
||||
slept++;
|
||||
},
|
||||
onAttemptFailed: (attempt, willRetry) =>
|
||||
attempts.push({ attempt, willRetry }),
|
||||
});
|
||||
expect(ok).toBe(false);
|
||||
expect(calls).toBe(3); // initial + 2 retries
|
||||
expect(slept).toBe(2); // only between attempts, not after the last
|
||||
expect(attempts).toEqual([
|
||||
{ attempt: 1, willRetry: true },
|
||||
{ attempt: 2, willRetry: true },
|
||||
{ attempt: 3, willRetry: false },
|
||||
]);
|
||||
});
|
||||
|
||||
it('uses 15s spacing via the default sleep only when a retry is needed', async () => {
|
||||
// Sanity: with maxRetries 0, a single failure returns false and never sleeps.
|
||||
let slept = 0;
|
||||
const ok = await editStatusMessageWithRetry({
|
||||
editOnce: async () => {
|
||||
throw new Error('nope');
|
||||
},
|
||||
maxRetries: 0,
|
||||
retryDelayMs: 15_000,
|
||||
sleep: async () => {
|
||||
slept++;
|
||||
},
|
||||
});
|
||||
expect(ok).toBe(false);
|
||||
expect(slept).toBe(0);
|
||||
});
|
||||
|
||||
it('works with a channel method that relies on `this` when bound (regression)', async () => {
|
||||
class FakeChannel {
|
||||
client = { ok: true };
|
||||
edits = 0;
|
||||
async editMessage(): Promise<void> {
|
||||
// Mirrors the real editMessage: throws if `this` is lost.
|
||||
if (!this.client) throw new Error('this.client is undefined');
|
||||
this.edits++;
|
||||
}
|
||||
}
|
||||
const channel = new FakeChannel();
|
||||
|
||||
// A detached method reference loses `this` → every attempt fails.
|
||||
const detached = channel.editMessage;
|
||||
const detachedOk = await editStatusMessageWithRetry({
|
||||
editOnce: () => detached(),
|
||||
maxRetries: 0,
|
||||
retryDelayMs: 0,
|
||||
sleep: async () => {},
|
||||
});
|
||||
expect(detachedOk).toBe(false);
|
||||
expect(channel.edits).toBe(0);
|
||||
|
||||
// Binding to the channel (the fix) preserves `this` and succeeds.
|
||||
const bound = channel.editMessage.bind(channel);
|
||||
const boundOk = await editStatusMessageWithRetry({
|
||||
editOnce: () => bound(),
|
||||
maxRetries: 0,
|
||||
retryDelayMs: 0,
|
||||
sleep: async () => {},
|
||||
});
|
||||
expect(boundOk).toBe(true);
|
||||
expect(channel.edits).toBe(1);
|
||||
});
|
||||
});
|
||||
|
||||
describe('msUntilNextMinuteBoundary', () => {
|
||||
it('returns time to the next :00 boundary', () => {
|
||||
expect(msUntilNextMinuteBoundary(120_000)).toBe(60_000); // exactly on boundary
|
||||
expect(msUntilNextMinuteBoundary(120_000 + 15_000)).toBe(45_000); // 15s in
|
||||
expect(msUntilNextMinuteBoundary(120_000 + 59_000)).toBe(1_000); // 59s in
|
||||
expect(msUntilNextMinuteBoundary(120_000 + 1)).toBe(59_999); // 1ms in
|
||||
});
|
||||
});
|
||||
|
||||
describe('createCoalescingTrigger', () => {
|
||||
it('coalesces a burst into one run and re-arms after firing', () => {
|
||||
let runs = 0;
|
||||
let fire: (() => void) | null = null;
|
||||
const timer = {
|
||||
set: (cb: () => void) => {
|
||||
fire = cb;
|
||||
return 0 as unknown as ReturnType<typeof setTimeout>;
|
||||
},
|
||||
};
|
||||
const trigger = createCoalescingTrigger(() => runs++, 1500, timer);
|
||||
|
||||
trigger();
|
||||
trigger();
|
||||
trigger();
|
||||
expect(runs).toBe(0); // nothing runs until the debounce window elapses
|
||||
|
||||
fire!();
|
||||
expect(runs).toBe(1); // three calls collapsed into a single run
|
||||
|
||||
// A new request after the run schedules and fires again.
|
||||
trigger();
|
||||
fire!();
|
||||
expect(runs).toBe(2);
|
||||
});
|
||||
});
|
||||
|
||||
@@ -109,6 +109,30 @@ let usageUpdateInProgress = false;
|
||||
let channelMetaCache = new Map<string, ChannelMeta>();
|
||||
let channelMetaLastRefresh = 0;
|
||||
let dashboardUpdateLogged = false;
|
||||
/** Guards against overlapping status updates while a slow edit-retry runs. */
|
||||
let statusUpdateRunning = false;
|
||||
/** A refresh was requested while one was already running; run once more after. */
|
||||
let statusUpdatePending = false;
|
||||
/**
|
||||
* On a failed status-message edit, retry the edit this many extra times at
|
||||
* STATUS_EDIT_RETRY_DELAY_MS spacing before falling back to a fresh message.
|
||||
*/
|
||||
const STATUS_EDIT_MAX_RETRIES = 2;
|
||||
const STATUS_EDIT_RETRY_DELAY_MS = 15_000;
|
||||
/** Coalesce bursts of event-driven refresh requests into one update. */
|
||||
const IMMEDIATE_UPDATE_DEBOUNCE_MS = 1_500;
|
||||
/** Set by the renderer's startUnifiedDashboard so external events can nudge it. */
|
||||
let immediateUpdateTrigger: (() => void) | null = null;
|
||||
|
||||
/**
|
||||
* Request an out-of-band status refresh (e.g. a new chat message arrived or an
|
||||
* agent's activity changed) so the dashboard updates immediately instead of
|
||||
* waiting for the next periodic tick. No-op until the renderer dashboard starts;
|
||||
* bursts are debounced.
|
||||
*/
|
||||
export function requestImmediateStatusUpdate(): void {
|
||||
immediateUpdateTrigger?.();
|
||||
}
|
||||
/** Codex service only: cached usage rows written into the status snapshot. */
|
||||
let cachedCodexUsageRows: UsageRow[] = [];
|
||||
/** Codex service only: ISO timestamp of last successful usage fetch. */
|
||||
@@ -226,7 +250,7 @@ function getAgentDisplayName(
|
||||
return serviceId === 'codex-review' ? '코리뷰' : '코덱스';
|
||||
}
|
||||
|
||||
function formatRoomName(
|
||||
export function formatRoomName(
|
||||
jid: string,
|
||||
meta: ChannelMeta | undefined,
|
||||
fallbackName: string | undefined,
|
||||
@@ -238,7 +262,12 @@ function formatRoomName(
|
||||
(fallbackName && fallbackName !== jid ? fallbackName : undefined) ||
|
||||
jid;
|
||||
|
||||
if (jid.startsWith('dc:') && base !== jid && !base.startsWith('#')) {
|
||||
if (
|
||||
jid.startsWith('dc:') &&
|
||||
base !== jid &&
|
||||
!base.startsWith('#') &&
|
||||
!base.includes(' #')
|
||||
) {
|
||||
return `#${base}`;
|
||||
}
|
||||
return base;
|
||||
@@ -285,6 +314,9 @@ function writeLocalStatusSnapshot(opts: UnifiedDashboardOptions): void {
|
||||
const groups = opts.roomBindings();
|
||||
const statuses = opts.queue.getStatuses(Object.keys(groups));
|
||||
const usageSnapshot = buildUsageSnapshotRows(opts);
|
||||
const chatNameByJid = new Map(
|
||||
getAllChats().map((chat) => [chat.jid, chat.name]),
|
||||
);
|
||||
|
||||
writeStatusSnapshot({
|
||||
serviceId: opts.serviceId,
|
||||
@@ -298,6 +330,9 @@ function writeLocalStatusSnapshot(opts: UnifiedDashboardOptions): void {
|
||||
return {
|
||||
jid: status.jid,
|
||||
name: group.name,
|
||||
...(chatNameByJid.get(status.jid) && {
|
||||
chatName: chatNameByJid.get(status.jid),
|
||||
}),
|
||||
folder: group.folder,
|
||||
agentType: (group.agentType || opts.serviceAgentType) as
|
||||
| 'claude-code'
|
||||
@@ -311,6 +346,7 @@ function writeLocalStatusSnapshot(opts: UnifiedDashboardOptions): void {
|
||||
.filter(Boolean) as Array<{
|
||||
jid: string;
|
||||
name: string;
|
||||
chatName?: string;
|
||||
folder: string;
|
||||
agentType: 'claude-code' | 'codex';
|
||||
status: 'processing' | 'waiting' | 'inactive';
|
||||
@@ -342,6 +378,7 @@ function buildStatusContent(): string {
|
||||
pendingMessages: boolean;
|
||||
pendingTasks: number;
|
||||
name: string;
|
||||
chatName?: string;
|
||||
meta: ChannelMeta | undefined;
|
||||
}
|
||||
|
||||
@@ -359,6 +396,7 @@ function buildStatusContent(): string {
|
||||
pendingMessages: entry.pendingMessages,
|
||||
pendingTasks: entry.pendingTasks,
|
||||
name: entry.name,
|
||||
chatName: entry.chatName,
|
||||
meta: channelMetaCache.get(entry.jid),
|
||||
});
|
||||
byJid.set(entry.jid, existing);
|
||||
@@ -386,7 +424,8 @@ function buildStatusContent(): string {
|
||||
jid,
|
||||
meta,
|
||||
agents.find((agent) => agent.name && agent.name !== jid)?.name,
|
||||
chatNameByJid.get(jid),
|
||||
agents.find((agent) => agent.chatName && agent.chatName !== jid)
|
||||
?.chatName ?? chatNameByJid.get(jid),
|
||||
),
|
||||
meta,
|
||||
agents,
|
||||
@@ -750,6 +789,67 @@ async function refreshUsageCache(): Promise<void> {
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Attempt a status-message edit, retrying up to `maxRetries` extra times with
|
||||
* `retryDelayMs` spacing before giving up. Returns true if an edit succeeded,
|
||||
* false if every attempt failed (caller then reposts a fresh message).
|
||||
* `sleep` is injectable so tests can run without real delays.
|
||||
*/
|
||||
export async function editStatusMessageWithRetry(args: {
|
||||
editOnce: () => Promise<unknown>;
|
||||
maxRetries: number;
|
||||
retryDelayMs: number;
|
||||
onAttemptFailed?: (attempt: number, willRetry: boolean, err: unknown) => void;
|
||||
sleep?: (ms: number) => Promise<void>;
|
||||
}): Promise<boolean> {
|
||||
const sleep =
|
||||
args.sleep ?? ((ms: number) => new Promise((r) => setTimeout(r, ms)));
|
||||
for (let attempt = 0; attempt <= args.maxRetries; attempt++) {
|
||||
try {
|
||||
await args.editOnce();
|
||||
return true;
|
||||
} catch (err) {
|
||||
const willRetry = attempt < args.maxRetries;
|
||||
args.onAttemptFailed?.(attempt + 1, willRetry, err);
|
||||
if (willRetry) await sleep(args.retryDelayMs);
|
||||
}
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
/**
|
||||
* Returns a trigger that runs `run` at most once per `delayMs` window: the
|
||||
* first call schedules a run after `delayMs`, and further calls within that
|
||||
* window are folded into the same pending run. Timer fns are injectable for
|
||||
* tests.
|
||||
*/
|
||||
export function createCoalescingTrigger(
|
||||
run: () => void,
|
||||
delayMs: number,
|
||||
timer: {
|
||||
set: (cb: () => void, ms: number) => ReturnType<typeof setTimeout>;
|
||||
} = { set: (cb, ms) => setTimeout(cb, ms) },
|
||||
): () => void {
|
||||
let scheduled: ReturnType<typeof setTimeout> | null = null;
|
||||
return () => {
|
||||
if (scheduled) return;
|
||||
scheduled = timer.set(() => {
|
||||
scheduled = null;
|
||||
run();
|
||||
}, delayMs);
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Milliseconds from `nowMs` until the next wall-clock minute boundary (:00).
|
||||
* Minute boundaries align across timezones (offsets are whole minutes), so this
|
||||
* is timezone-agnostic. Exactly on a boundary returns a full minute.
|
||||
*/
|
||||
export function msUntilNextMinuteBoundary(nowMs: number): number {
|
||||
const rem = nowMs % 60_000;
|
||||
return rem === 0 ? 60_000 : 60_000 - rem;
|
||||
}
|
||||
|
||||
export async function startUnifiedDashboard(
|
||||
opts: UnifiedDashboardOptions,
|
||||
): Promise<void> {
|
||||
@@ -757,6 +857,10 @@ export async function startUnifiedDashboard(
|
||||
|
||||
const isRenderer = opts.serviceAgentType === 'claude-code';
|
||||
const statusJid = `dc:${opts.statusChannelId}`;
|
||||
// The first render after a (re)start should show the status message
|
||||
// immediately: edit the stored message if possible, but if that edit fails,
|
||||
// repost right away instead of waiting through the steady-state retry cycle.
|
||||
let firstStatusRender = true;
|
||||
if (isRenderer) {
|
||||
statusMessageId = readDashboardStatusMessageId(opts.statusChannelId);
|
||||
}
|
||||
@@ -780,21 +884,30 @@ export async function startUnifiedDashboard(
|
||||
const updateStatus = async () => {
|
||||
writeLocalStatusSnapshot(opts);
|
||||
if (!isRenderer) return;
|
||||
|
||||
const channel = findDiscordChannel(opts.channels);
|
||||
if (!channel) {
|
||||
logger.warn(
|
||||
{
|
||||
channelCount: opts.channels.length,
|
||||
names: opts.channels.map((c) => c.name),
|
||||
connected: opts.channels.map((c) => c.isConnected()),
|
||||
},
|
||||
'Dashboard: no connected Discord channel found',
|
||||
);
|
||||
// A failed edit now retries with delays, so a single updateStatus can run
|
||||
// for tens of seconds. Skip overlapping ticks (this is also why the base
|
||||
// periodic refresh does not fire mid-retry); remember the request so it runs
|
||||
// once the in-flight update finishes.
|
||||
if (statusUpdateRunning) {
|
||||
statusUpdatePending = true;
|
||||
return;
|
||||
}
|
||||
statusUpdateRunning = true;
|
||||
|
||||
try {
|
||||
const channel = findDiscordChannel(opts.channels);
|
||||
if (!channel) {
|
||||
logger.warn(
|
||||
{
|
||||
channelCount: opts.channels.length,
|
||||
names: opts.channels.map((c) => c.name),
|
||||
connected: opts.channels.map((c) => c.isConnected()),
|
||||
},
|
||||
'Dashboard: no connected Discord channel found',
|
||||
);
|
||||
return;
|
||||
}
|
||||
|
||||
await refreshChannelMeta(opts);
|
||||
const content = buildUnifiedDashboardContent();
|
||||
if (!content) {
|
||||
@@ -809,14 +922,38 @@ export async function startUnifiedDashboard(
|
||||
}
|
||||
|
||||
if (statusMessageId && channel.editMessage) {
|
||||
try {
|
||||
await channel.editMessage(statusJid, statusMessageId, content);
|
||||
// A transient Discord error (e.g. 503) on the periodic edit should not
|
||||
// immediately spawn a fresh status message. Retry the edit a couple of
|
||||
// times with a delay first; only give up (and repost) if all fail.
|
||||
const editId = statusMessageId;
|
||||
// Bind to the channel: a detached method reference loses `this` and the
|
||||
// edit throws ("this.client is undefined"), which would make every
|
||||
// update fail and repost a fresh (notifying) message.
|
||||
const editMessage = channel.editMessage.bind(channel);
|
||||
// First render after start: 0 retries → repost immediately if the edit
|
||||
// fails. Steady state: retry twice at 15s before reposting.
|
||||
const maxRetries = firstStatusRender ? 0 : STATUS_EDIT_MAX_RETRIES;
|
||||
const edited = await editStatusMessageWithRetry({
|
||||
editOnce: () => editMessage(statusJid, editId, content),
|
||||
maxRetries,
|
||||
retryDelayMs: STATUS_EDIT_RETRY_DELAY_MS,
|
||||
onAttemptFailed: (attempt, willRetry, err) =>
|
||||
logger.warn(
|
||||
{
|
||||
err,
|
||||
messageId: editId,
|
||||
attempt,
|
||||
maxAttempts: maxRetries + 1,
|
||||
willRetry,
|
||||
},
|
||||
willRetry
|
||||
? 'Dashboard status message edit failed; retrying in 15s'
|
||||
: 'Dashboard status message edit failed after retries; sending a fresh tracked message',
|
||||
),
|
||||
});
|
||||
if (edited) {
|
||||
writeDashboardStatusMessageId(opts.statusChannelId, statusMessageId);
|
||||
} catch (err) {
|
||||
logger.warn(
|
||||
{ err, messageId: statusMessageId },
|
||||
'Dashboard status message edit failed; sending a fresh tracked message',
|
||||
);
|
||||
} else {
|
||||
statusMessageId = null;
|
||||
}
|
||||
}
|
||||
@@ -828,6 +965,9 @@ export async function startUnifiedDashboard(
|
||||
writeDashboardStatusMessageId(opts.statusChannelId, id);
|
||||
}
|
||||
}
|
||||
// A render (edit or repost) happened this tick; subsequent ticks use the
|
||||
// steady-state retry policy.
|
||||
firstStatusRender = false;
|
||||
if (statusMessageId) {
|
||||
await cleanupDashboardDuplicateMessages(opts, statusMessageId);
|
||||
}
|
||||
@@ -841,10 +981,33 @@ export async function startUnifiedDashboard(
|
||||
} catch (err) {
|
||||
logger.warn({ err }, 'Dashboard update failed');
|
||||
statusMessageId = null;
|
||||
} finally {
|
||||
statusUpdateRunning = false;
|
||||
if (statusUpdatePending) {
|
||||
statusUpdatePending = false;
|
||||
immediateUpdateTrigger?.();
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
setInterval(updateStatus, opts.statusUpdateInterval);
|
||||
if (isRenderer) {
|
||||
// Event-driven refresh: external callers (new chat message, agent activity
|
||||
// change) call requestImmediateStatusUpdate(); bursts are coalesced.
|
||||
immediateUpdateTrigger = createCoalescingTrigger(
|
||||
() => void updateStatus(),
|
||||
IMMEDIATE_UPDATE_DEBOUNCE_MS,
|
||||
);
|
||||
}
|
||||
|
||||
// Base periodic refresh fires on each wall-clock minute boundary (:00) so the
|
||||
// minute-precision timestamp shown in the status message stays accurate.
|
||||
const scheduleMinuteBoundaryUpdate = () => {
|
||||
setTimeout(() => {
|
||||
void updateStatus();
|
||||
scheduleMinuteBoundaryUpdate();
|
||||
}, msUntilNextMinuteBoundary(Date.now()));
|
||||
};
|
||||
scheduleMinuteBoundaryUpdate();
|
||||
setInterval(() => {
|
||||
if (!isRenderer || !statusMessageId) return;
|
||||
void cleanupDashboardDuplicateMessages(opts, statusMessageId);
|
||||
|
||||
Reference in New Issue
Block a user