From 7e77f3794846c39b92b4711e2d893b04e8bcc733 Mon Sep 17 00:00:00 2001 From: Eyejoker Date: Mon, 16 Mar 2026 05:08:19 +0900 Subject: [PATCH] refactor: extract dashboard and message-runtime from index.ts - Extract dashboard code into src/dashboard.ts - Extract message runtime into src/message-runtime.ts - Improve group-queue with better concurrency handling - Update agent-runner with enhanced env/config passing - Simplify DB operations and cleanup unused code - Update README for Codex SDK architecture --- README.md | 51 +- setup/container.ts | 4 +- setup/register.ts | 25 +- src/agent-runner.ts | 79 ++- src/dashboard.ts | 520 ++++++++++++++++++ src/db.test.ts | 13 + src/db.ts | 68 +-- src/group-queue.ts | 146 +++++- src/index.ts | 987 ++--------------------------------- src/ipc.ts | 2 - src/message-runtime.ts | 528 +++++++++++++++++++ src/router.ts | 10 - src/session-commands.test.ts | 25 + src/session-commands.ts | 18 +- 14 files changed, 1392 insertions(+), 1084 deletions(-) create mode 100644 src/dashboard.ts create mode 100644 src/message-runtime.ts diff --git a/README.md b/README.md index 25df2a7..f39fdfe 100644 --- a/README.md +++ b/README.md @@ -15,29 +15,26 @@ Two AI agents running as parallel systemd services, communicating over Discord: - **Claude Code** — powered by Claude Agent SDK, trigger `@claude` -- **Codex** — powered by Codex app-server (JSON-RPC), trigger `@codex` +- **Codex** — powered by Codex SDK, trigger `@codex` -Each agent has its own store, data, and groups directories. Discord channels can be registered with either agent, or both (`both` agent type for shared channels). +Each agent has its own store, data, and groups directories. Discord channels are registered per agent service. ### Key Features - **Direct host processes** — no container overhead, agents run natively - **Bidirectional image support** — receive images as multimodal input, send as Discord attachments - **Skill sync** — single source of truth (`~/.claude/skills/`), auto-synced to all sessions -- **OAuth auto-refresh** — token lifecycle managed automatically for headless environments - **Priority queue** — per-group serialization, global concurrency limit, idle preemption -- **Auto-continue** — Codex text-only turns automatically retried to enforce task execution ## Architecture ``` Discord ──► SQLite ──► GroupQueue ──┬──► Claude Agent SDK (host process) - └──► Codex App-Server (JSON-RPC stdio) - ├── thread/start, thread/resume - ├── turn/start (streaming, multimodal) - ├── turn/steer (mid-turn injection) - ├── Auto-approval (bypass sandbox) - └── Auto-continue (text-only turn retry) + └──► Codex SDK (long-lived thread runner) + ├── thread start/resume + ├── multimodal input + ├── per-group model/effort config + └── follow-up messages via IPC ``` ### Directory Layout @@ -52,7 +49,6 @@ nanoclaw/ │ ├── router.ts # Outbound message formatting and routing │ ├── sender-allowlist.ts # Security: sender-based access control │ ├── session-commands.ts # Session commands (/compact) -│ ├── token-refresh.ts # OAuth auto-refresh + session directory sync │ ├── task-scheduler.ts # Scheduled tasks (cron/interval/once) │ ├── ipc.ts # IPC watcher and task processing │ ├── db.ts # SQLite operations @@ -62,7 +58,7 @@ nanoclaw/ │ └── discord.ts # Discord: mentions, images, typing, file attachments ├── runners/ │ ├── agent-runner/ # Claude Code runner (Agent SDK, multimodal input) -│ ├── codex-runner/ # Codex runner (app-server JSON-RPC, auto-continue) +│ ├── codex-runner/ # Codex runner (SDK thread wrapper) │ └── skills/ # Shared agent skills (browser, etc.) ├── store/ # Claude Code service DB ├── store-codex/ # Codex service DB @@ -75,17 +71,15 @@ nanoclaw/ └── logs/ # Service logs ``` -### Codex App-Server Integration +### Codex SDK Integration -The Codex runner (`runners/codex-runner/`) communicates with `codex app-server` via JSON-RPC over stdio: +The Codex runner (`runners/codex-runner/`) uses `@openai/codex-sdk` with one long-lived thread per group: -- **Session persistence**: Thread IDs stored in DB, sessions saved as JSONL on disk -- **Streaming**: `item/agentMessage/delta` notifications for real-time text -- **Mid-turn steering**: IPC messages injected via `turn/steer` during execution -- **Auto-approval**: `approvalPolicy: "never"` + `sandbox: "danger-full-access"` -- **Auto-continue**: Detects text-only turns (no tool execution) and automatically retries up to 5 times to nudge the agent into actually executing tasks -- **Multimodal input**: Image attachments converted to `localImage` input blocks in `turn/start` -- **Per-group config**: Model, effort, MCP servers configured per channel +- **Session persistence**: Thread IDs stored in DB and resumed per group +- **Follow-up messages**: Additional Discord messages are injected into the active runner via IPC +- **Auto-approval**: `approvalPolicy: "never"` + `sandboxMode: "danger-full-access"` +- **Multimodal input**: Image attachments converted to `local_image` SDK inputs +- **Per-group config**: Model and reasoning effort can be overridden per channel ### Image Handling @@ -102,15 +96,6 @@ Skills are managed from a single source of truth (`~/.claude/skills/` on the ser - Codex sessions: Same sources, synced to per-group `.codex/` directories - Skills auto-register as slash commands (`/name`) in Claude Code and `$name` in Codex -### OAuth Token Auto-Refresh - -`src/token-refresh.ts` handles Claude Code OAuth token lifecycle: - -- Checks every 5 minutes, refreshes 30 minutes before expiry -- Tries `platform.claude.com` then falls back to `api.anthropic.com` -- Syncs refreshed credentials to all per-group session directories -- Solves the known headless environment token expiry issue - ### GroupQueue `src/group-queue.ts` manages agent execution with: @@ -241,10 +226,10 @@ Channels are stored in each service's SQLite database (`registered_groups` table ```bash # Example: register a channel for Claude Code -sqlite3 store/nanoclaw.db "INSERT INTO registered_groups (jid, name, folder, agent_type, trigger_pattern) VALUES ('dc:CHANNEL_ID', 'my-channel', 'my-channel', 'claude-code', '@claude');" +sqlite3 store/messages.db "INSERT INTO registered_groups (jid, name, folder, trigger_pattern, added_at, agent_type) VALUES ('dc:CHANNEL_ID', 'my-channel', 'my-channel', '@claude', datetime('now'), 'claude-code');" # Example: register a channel for Codex -sqlite3 store-codex/nanoclaw.db "INSERT INTO registered_groups (jid, name, folder, agent_type, trigger_pattern) VALUES ('dc:CHANNEL_ID', 'my-channel', 'my-channel', 'codex', '@codex');" +sqlite3 store-codex/messages.db "INSERT INTO registered_groups (jid, name, folder, trigger_pattern, added_at, agent_type) VALUES ('dc:CHANNEL_ID', 'my-channel', 'my-channel', '@codex', datetime('now'), 'codex');" ``` Fields: @@ -254,7 +239,7 @@ Fields: | `jid` | `dc:` | | `name` | Display name | | `folder` | Group folder name (workspace directory) | -| `agent_type` | `claude-code`, `codex`, or `both` | +| `agent_type` | `claude-code` or `codex` | | `trigger_pattern` | Regex for activation (e.g., `@claude`) | | `work_dir` | Optional working directory override | | `container_config` | Optional JSON (e.g., `{"codexEffort":"high"}`) | diff --git a/setup/container.ts b/setup/container.ts index cdcb92d..4a405ea 100644 --- a/setup/container.ts +++ b/setup/container.ts @@ -29,14 +29,14 @@ export async function run(_args: string[]): Promise { // Verify runner entry points exist const agentRunner = path.join( projectRoot, - 'container', + 'runners', 'agent-runner', 'dist', 'index.js', ); const codexRunner = path.join( projectRoot, - 'container', + 'runners', 'codex-runner', 'dist', 'index.js', diff --git a/setup/register.ts b/setup/register.ts index 03ea7df..33e77cb 100644 --- a/setup/register.ts +++ b/setup/register.ts @@ -9,7 +9,7 @@ import path from 'path'; import Database from 'better-sqlite3'; -import { STORE_DIR } from '../src/config.js'; +import { SERVICE_AGENT_TYPE, STORE_DIR } from '../src/config.js'; import { isValidGroupFolder } from '../src/group-folder.js'; import { logger } from '../src/logger.js'; import { emitStatus } from './status.js'; @@ -113,15 +113,31 @@ export async function run(args: string[]): Promise { added_at TEXT NOT NULL, container_config TEXT, requires_trigger INTEGER DEFAULT 1, - is_main INTEGER DEFAULT 0 + is_main INTEGER DEFAULT 0, + agent_type TEXT DEFAULT 'claude-code', + work_dir TEXT )`); + try { + db.exec( + `ALTER TABLE registered_groups ADD COLUMN agent_type TEXT DEFAULT 'claude-code'`, + ); + } catch { + /* column already exists */ + } + + try { + db.exec(`ALTER TABLE registered_groups ADD COLUMN work_dir TEXT`); + } catch { + /* column already exists */ + } + const isMainInt = parsed.isMain ? 1 : 0; db.prepare( `INSERT OR REPLACE INTO registered_groups - (jid, name, folder, trigger_pattern, added_at, container_config, requires_trigger, is_main) - VALUES (?, ?, ?, ?, ?, NULL, ?, ?)`, + (jid, name, folder, trigger_pattern, added_at, container_config, requires_trigger, is_main, agent_type, work_dir) + VALUES (?, ?, ?, ?, ?, NULL, ?, ?, ?, NULL)`, ).run( parsed.jid, parsed.name, @@ -130,6 +146,7 @@ export async function run(args: string[]): Promise { timestamp, requiresTriggerInt, isMainInt, + SERVICE_AGENT_TYPE, ); db.close(); diff --git a/src/agent-runner.ts b/src/agent-runner.ts index c80b323..2518347 100644 --- a/src/agent-runner.ts +++ b/src/agent-runner.ts @@ -29,6 +29,7 @@ export interface AgentInput { sessionId?: string; groupFolder: string; chatJid: string; + runId?: string; isMain: boolean; isScheduledTask?: boolean; assistantName?: string; @@ -324,15 +325,19 @@ export async function runAgentProcess( group, input.isMain, ); + if (input.runId) { + env.NANOCLAW_RUN_ID = input.runId; + } const safeName = group.folder.replace(/[^a-zA-Z0-9-]/g, '-'); - const processName = `nanoclaw-${safeName}-${Date.now()}`; + const processSuffix = input.runId || `${Date.now()}`; + const processName = `nanoclaw-${safeName}-${processSuffix}`; // Check if runner is built const distEntry = path.join(runnerDir, 'dist', 'index.js'); if (!fs.existsSync(distEntry)) { logger.error( - { runnerDir }, + { runnerDir, chatJid: input.chatJid, runId: input.runId }, 'Runner not built. Run: cd runners/agent-runner && npm install && npm run build', ); return { @@ -345,6 +350,9 @@ export async function runAgentProcess( logger.info( { group: group.name, + chatJid: input.chatJid, + groupFolder: input.groupFolder, + runId: input.runId, processName, agentType: group.agentType || 'claude-code', isMain: input.isMain, @@ -386,7 +394,7 @@ export async function runAgentProcess( stdout += chunk.slice(0, remaining); stdoutTruncated = true; logger.warn( - { group: group.name, size: stdout.length }, + { group: group.name, chatJid: input.chatJid, runId: input.runId, size: stdout.length }, 'Agent stdout truncated due to size limit', ); } else { @@ -416,7 +424,7 @@ export async function runAgentProcess( outputChain = outputChain.then(() => onOutput(parsed)); } catch (err) { logger.warn( - { group: group.name, error: err }, + { group: group.name, chatJid: input.chatJid, runId: input.runId, error: err }, 'Failed to parse streamed output chunk', ); } @@ -432,7 +440,7 @@ export async function runAgentProcess( const killOnTimeout = () => { timedOut = true; logger.error( - { group: group.name, processName }, + { group: group.name, chatJid: input.chatJid, runId: input.runId, processName }, 'Agent timeout, sending SIGTERM', ); proc.kill('SIGTERM'); @@ -455,9 +463,15 @@ export async function runAgentProcess( for (const line of lines) { if (!line) continue; if (line.includes('Turn in progress')) { - logger.info({ group: group.name }, line.replace(/^\[.*?\]\s*/, '')); + logger.info( + { group: group.name, chatJid: input.chatJid, runId: input.runId }, + line.replace(/^\[.*?\]\s*/, ''), + ); } else { - logger.debug({ agent: group.folder }, line); + logger.debug( + { agent: group.folder, chatJid: input.chatJid, runId: input.runId }, + line, + ); } } // Stderr activity means agent is alive — reset timeout @@ -479,11 +493,13 @@ export async function runAgentProcess( if (timedOut) { const ts = new Date().toISOString().replace(/[:.]/g, '-'); fs.writeFileSync( - path.join(logsDir, `agent-${ts}.log`), + path.join(logsDir, `agent-${input.runId || 'adhoc'}-${ts}.log`), [ `=== Agent Run Log (TIMEOUT) ===`, `Timestamp: ${new Date().toISOString()}`, `Group: ${group.name}`, + `ChatJid: ${input.chatJid}`, + `RunId: ${input.runId || 'n/a'}`, `Process: ${processName}`, `Duration: ${duration}ms`, `Exit Code: ${code}`, @@ -493,7 +509,14 @@ export async function runAgentProcess( if (hadStreamingOutput) { logger.info( - { group: group.name, processName, duration, code }, + { + group: group.name, + chatJid: input.chatJid, + runId: input.runId, + processName, + duration, + code, + }, 'Agent timed out after output (idle cleanup)', ); outputChain.then(() => { @@ -511,7 +534,10 @@ export async function runAgentProcess( } const timestamp = new Date().toISOString().replace(/[:.]/g, '-'); - const logFile = path.join(logsDir, `agent-${timestamp}.log`); + const logFile = path.join( + logsDir, + `agent-${input.runId || 'adhoc'}-${timestamp}.log`, + ); const isVerbose = process.env.LOG_LEVEL === 'debug' || process.env.LOG_LEVEL === 'trace'; @@ -519,6 +545,9 @@ export async function runAgentProcess( `=== Agent Run Log ===`, `Timestamp: ${new Date().toISOString()}`, `Group: ${group.name}`, + `ChatJid: ${input.chatJid}`, + `GroupFolder: ${input.groupFolder}`, + `RunId: ${input.runId || 'n/a'}`, `IsMain: ${input.isMain}`, `AgentType: ${group.agentType || 'claude-code'}`, `Duration: ${duration}ms`, @@ -549,7 +578,14 @@ export async function runAgentProcess( if (code !== 0) { logger.error( - { group: group.name, code, duration, logFile }, + { + group: group.name, + chatJid: input.chatJid, + runId: input.runId, + code, + duration, + logFile, + }, 'Agent exited with error', ); resolve({ @@ -563,7 +599,13 @@ export async function runAgentProcess( if (onOutput) { outputChain.then(() => { logger.info( - { group: group.name, duration, newSessionId }, + { + group: group.name, + chatJid: input.chatJid, + runId: input.runId, + duration, + newSessionId, + }, 'Agent completed (streaming mode)', ); resolve({ status: 'success', result: null, newSessionId }); @@ -586,13 +628,19 @@ export async function runAgentProcess( } const output: AgentOutput = JSON.parse(jsonLine); logger.info( - { group: group.name, duration, status: output.status }, + { + group: group.name, + chatJid: input.chatJid, + runId: input.runId, + duration, + status: output.status, + }, 'Agent completed', ); resolve(output); } catch (err) { logger.error( - { group: group.name, error: err }, + { group: group.name, chatJid: input.chatJid, runId: input.runId, error: err }, 'Failed to parse agent output', ); resolve({ @@ -606,7 +654,7 @@ export async function runAgentProcess( proc.on('error', (err) => { clearTimeout(timeout); logger.error( - { group: group.name, processName, error: err }, + { group: group.name, chatJid: input.chatJid, runId: input.runId, processName, error: err }, 'Agent spawn error', ); resolve({ @@ -651,7 +699,6 @@ export function writeGroupsSnapshot( groupFolder: string, isMain: boolean, groups: AvailableGroup[], - registeredJids: Set, ): void { const groupIpcDir = resolveGroupIpcPath(groupFolder); fs.mkdirSync(groupIpcDir, { recursive: true }); diff --git a/src/dashboard.ts b/src/dashboard.ts new file mode 100644 index 0000000..68edc6f --- /dev/null +++ b/src/dashboard.ts @@ -0,0 +1,520 @@ +import { ChildProcess, execSync, spawn } from 'child_process'; +import fs from 'fs'; +import os from 'os'; +import path from 'path'; + +import { readEnvFile } from './env.js'; +import { GroupQueue, GroupStatus } from './group-queue.js'; +import { logger } from './logger.js'; +import { Channel, ChannelMeta, RegisteredGroup } from './types.js'; + +interface DashboardOptions { + assistantName: string; + channels: Channel[]; + queue: GroupQueue; + registeredGroups: () => Record; + statusChannelId: string; + statusUpdateInterval: number; + usageUpdateInterval: number; +} + +interface ClaudeUsageData { + five_hour?: { utilization: number; resets_at: string }; + seven_day?: { utilization: number; resets_at: string }; +} + +interface CodexRateLimit { + limitId?: string; + limitName: string | null; + primary: { usedPercent: number; resetsAt: string | number }; + secondary: { usedPercent: number; resetsAt: string | number }; +} + +const STATUS_ICONS: Record = { + processing: '🟡', + idle: '🟢', + waiting: '🔵', + inactive: '⚪', +}; + +const CHANNEL_META_REFRESH_MS = 300000; + +let statusMessageId: string | null = null; +let usageMessageId: string | null = null; +let usageUpdateInProgress = false; +let channelMetaCache = new Map(); +let channelMetaLastRefresh = 0; + +function findDiscordChannel(channels: Channel[]): Channel | undefined { + return channels.find((c) => c.name.startsWith('discord') && c.isConnected()); +} + +function formatElapsed(ms: number): string { + const s = Math.floor(ms / 1000); + if (s < 60) return `${s}s`; + if (s < 3600) { + const m = Math.floor(s / 60); + const rem = s % 60; + return `${m}m${rem.toString().padStart(2, '0')}s`; + } + const h = Math.floor(s / 3600); + const m = Math.floor((s % 3600) / 60); + if (h < 24) return `${h}h${m.toString().padStart(2, '0')}m`; + const d = Math.floor(h / 24); + const remH = h % 24; + return `${d}d${remH}h`; +} + +function formatResetKST(value: string | number): string { + try { + const date = + typeof value === 'number' ? new Date(value * 1000) : new Date(value); + return date.toLocaleString('ko-KR', { + timeZone: 'Asia/Seoul', + month: 'short', + day: 'numeric', + hour: '2-digit', + minute: '2-digit', + }); + } catch { + return String(value); + } +} + +async function refreshChannelMeta(opts: DashboardOptions): Promise { + const now = Date.now(); + if (now - channelMetaLastRefresh < CHANNEL_META_REFRESH_MS) return; + + const ch = opts.channels.find( + (c) => c.name.startsWith('discord') && c.isConnected() && c.getChannelMeta, + ); + if (!ch?.getChannelMeta) return; + + const jids = Object.keys(opts.registeredGroups()).filter((j) => + j.startsWith('dc:'), + ); + try { + channelMetaCache = await ch.getChannelMeta(jids); + channelMetaLastRefresh = now; + } catch (err) { + logger.debug({ err }, 'Failed to refresh channel metadata'); + } +} + +function getStatusLabel(status: GroupStatus): string { + if (status.status === 'processing') { + return `처리 중 (${formatElapsed(status.elapsedMs || 0)})`; + } + if (status.status === 'idle') return '대기 중'; + if (status.status === 'waiting') { + return status.pendingTasks > 0 + ? `큐 대기 (태스크 ${status.pendingTasks}개)` + : '큐 대기 (메시지)'; + } + return '비활성'; +} + +function buildStatusContent(opts: DashboardOptions): string { + const registeredGroups = opts.registeredGroups(); + const jids = Object.keys(registeredGroups); + const statuses = opts.queue.getStatuses(jids); + + const entries = statuses + .map((status) => ({ + status, + group: registeredGroups[status.jid], + meta: channelMetaCache.get(status.jid), + })) + .filter((entry) => entry.group); + + const categoryMap = new Map(); + for (const entry of entries) { + const category = entry.meta?.category || '기타'; + if (!categoryMap.has(category)) categoryMap.set(category, []); + categoryMap.get(category)!.push(entry); + } + + const sortedCategories = [...categoryMap.entries()].sort((a, b) => { + const posA = a[1][0]?.meta?.categoryPosition ?? 999; + const posB = b[1][0]?.meta?.categoryPosition ?? 999; + return posA - posB; + }); + + const sections: string[] = []; + let totalActive = 0; + let totalIdle = 0; + let total = 0; + + for (const [categoryName, categoryEntries] of sortedCategories) { + categoryEntries.sort( + (a, b) => (a.meta?.position ?? 999) - (b.meta?.position ?? 999), + ); + + const lines = categoryEntries.map((entry) => { + const icon = STATUS_ICONS[entry.status.status] || '⚪'; + const label = getStatusLabel(entry.status); + const name = entry.meta?.name ? `#${entry.meta.name}` : entry.group.name; + return ` ${icon} **${name}** — ${label}`; + }); + + if (channelMetaCache.size > 0 && categoryName !== '기타') { + sections.push(`📁 **${categoryName}**\n${lines.join('\n')}`); + } else { + sections.push(lines.join('\n')); + } + + totalActive += categoryEntries.filter( + (entry) => entry.status.status === 'processing', + ).length; + totalIdle += categoryEntries.filter( + (entry) => entry.status.status === 'idle', + ).length; + total += categoryEntries.length; + } + + const header = `**에이전트 상태** (${opts.assistantName}) — 활성 ${totalActive} | 대기 ${totalIdle} | 전체 ${total}`; + return `${header}\n\n${sections.join('\n\n')}\n\n_${new Date().toLocaleTimeString('ko-KR')}_`; +} + +async function fetchClaudeUsage(): Promise { + try { + const envToken = readEnvFile(['CLAUDE_CODE_OAUTH_TOKEN']); + let token = + process.env.CLAUDE_CODE_OAUTH_TOKEN || + envToken.CLAUDE_CODE_OAUTH_TOKEN || + ''; + if (!token) { + const configDir = + process.env.CLAUDE_CONFIG_DIR || path.join(os.homedir(), '.claude'); + const credsPath = path.join(configDir, '.credentials.json'); + if (!fs.existsSync(credsPath)) return null; + const creds = JSON.parse(fs.readFileSync(credsPath, 'utf-8')); + token = creds?.claudeAiOauth?.accessToken || ''; + } + if (!token) return null; + + const res = await fetch('https://api.anthropic.com/api/oauth/usage', { + headers: { + Authorization: `Bearer ${token}`, + 'anthropic-beta': 'oauth-2025-04-20', + }, + }); + if (!res.ok) return null; + return (await res.json()) as ClaudeUsageData; + } catch { + return null; + } +} + +async function fetchCodexUsage(): Promise { + const npmGlobalBin = path.join(os.homedir(), '.npm-global', 'bin', 'codex'); + const codexBin = fs.existsSync(npmGlobalBin) ? npmGlobalBin : 'codex'; + + return new Promise((resolve) => { + let done = false; + const finish = (value: CodexRateLimit[] | null) => { + if (done) return; + done = true; + clearTimeout(timer); + if (proc) { + try { + proc.kill(); + } catch { + /* ignore */ + } + } + resolve(value); + }; + + const timer = setTimeout(() => finish(null), 20000); + + let proc: ChildProcess | null = null; + try { + proc = spawn(codexBin, ['app-server'], { + stdio: ['pipe', 'pipe', 'pipe'], + env: { + ...(process.env as Record), + PATH: [ + path.dirname(process.execPath), + path.join(os.homedir(), '.npm-global', 'bin'), + process.env.PATH || '', + ].join(':'), + }, + }); + } catch { + resolve(null); + return; + } + + if (!proc.stdout || !proc.stdin) { + finish(null); + return; + } + + const stdout = proc.stdout; + const stdin = proc.stdin; + + proc.on('error', () => finish(null)); + proc.on('close', () => finish(null)); + + let buf = ''; + stdout.on('data', (chunk: Buffer) => { + buf += chunk.toString(); + const lines = buf.split('\n'); + buf = lines.pop() || ''; + for (const line of lines) { + if (!line.trim()) continue; + try { + const msg = JSON.parse(line); + if (msg.id === 1) { + stdin.write( + JSON.stringify({ + jsonrpc: '2.0', + id: 2, + method: 'account/rateLimits/read', + params: {}, + }) + '\n', + ); + } else if (msg.id === 2 && msg.result) { + const byId = msg.result.rateLimitsByLimitId; + if (byId && typeof byId === 'object') { + finish(Object.values(byId) as CodexRateLimit[]); + } else { + finish(null); + } + } + } catch { + /* non-JSON line */ + } + } + }); + + stdin.write( + JSON.stringify({ + jsonrpc: '2.0', + id: 1, + method: 'initialize', + params: { clientInfo: { name: 'usage-monitor', version: '1.0' } }, + }) + '\n', + ); + }); +} + +async function buildUsageContent(): Promise { + const lines: string[] = []; + const [claudeUsage, codexUsage] = await Promise.all([ + fetchClaudeUsage(), + fetchCodexUsage(), + ]); + + const bar = (pct: number) => { + const filled = Math.round(pct / 10); + return '█'.repeat(filled) + '░'.repeat(10 - filled); + }; + + lines.push('📊 *사용량*'); + + type UsageRow = { + name: string; + h5pct: number; + h5reset: string; + d7pct: number; + d7reset: string; + }; + const rows: UsageRow[] = []; + + if (claudeUsage) { + const h5 = claudeUsage.five_hour; + const d7 = claudeUsage.seven_day; + rows.push({ + name: 'Claude', + h5pct: h5 + ? h5.utilization > 1 + ? Math.round(h5.utilization) + : Math.round(h5.utilization * 100) + : -1, + h5reset: h5 ? formatResetKST(h5.resets_at) : '', + d7pct: d7 + ? d7.utilization > 1 + ? Math.round(d7.utilization) + : Math.round(d7.utilization * 100) + : -1, + d7reset: d7 ? formatResetKST(d7.resets_at) : '', + }); + } + + if (codexUsage && Array.isArray(codexUsage)) { + const relevant = codexUsage.filter( + (limit) => + limit.primary.usedPercent > 0 || limit.secondary.usedPercent > 0, + ); + const display = relevant.length > 0 ? relevant : codexUsage.slice(0, 1); + for (const limit of display) { + rows.push({ + name: 'Codex', + h5pct: Math.round(limit.primary.usedPercent), + h5reset: formatResetKST(limit.primary.resetsAt), + d7pct: Math.round(limit.secondary.usedPercent), + d7reset: formatResetKST(limit.secondary.resetsAt), + }); + } + } + + if (rows.length > 0) { + lines.push('```'); + lines.push(' 5-Hour 7-Day'); + for (const row of rows) { + const h5 = + row.h5pct >= 0 + ? `${bar(row.h5pct)} ${String(row.h5pct).padStart(3)}%` + : ' — '; + const d7 = + row.d7pct >= 0 + ? `${bar(row.d7pct)} ${String(row.d7pct).padStart(3)}%` + : ' — '; + lines.push(`${row.name.padEnd(8)}${h5} ${d7}`); + } + lines.push('```'); + } else { + lines.push('_조회 불가_'); + } + lines.push(''); + + lines.push('🖥️ *서버*'); + + const loadAvg = os.loadavg(); + const cpuCount = os.cpus().length; + const cpuPct = Math.round((loadAvg[1] / cpuCount) * 100); + + const totalMem = os.totalmem(); + const usedMem = totalMem - os.freemem(); + const memPct = Math.round((usedMem / totalMem) * 100); + const memUsedGB = (usedMem / 1073741824).toFixed(1); + const memTotalGB = (totalMem / 1073741824).toFixed(1); + + let diskPct = 0; + let diskUsedGB = '?'; + let diskTotalGB = '?'; + try { + const df = execSync('df -B1 / | tail -1', { + encoding: 'utf-8', + timeout: 5000, + }).trim(); + const parts = df.split(/\s+/); + const diskUsed = parseInt(parts[2], 10); + const diskTotal = parseInt(parts[1], 10); + diskPct = Math.round((diskUsed / diskTotal) * 100); + diskUsedGB = (diskUsed / 1073741824).toFixed(0); + diskTotalGB = (diskTotal / 1073741824).toFixed(0); + } catch { + /* ignore */ + } + + lines.push('```'); + lines.push(`${'CPU'.padEnd(8)}${bar(cpuPct)} ${String(cpuPct).padStart(3)}%`); + lines.push( + `${'Memory'.padEnd(8)}${bar(memPct)} ${String(memPct).padStart(3)}% ${memUsedGB}/${memTotalGB}GB`, + ); + lines.push( + `${'Disk'.padEnd(8)}${bar(diskPct)} ${String(diskPct).padStart(3)}% ${diskUsedGB}/${diskTotalGB}GB`, + ); + lines.push(`${'Uptime'.padEnd(8)}${formatElapsed(os.uptime() * 1000)}`); + lines.push('```'); + + return ( + lines.join('\n') + + `\n_${new Date().toLocaleTimeString('ko-KR', { + hour: '2-digit', + minute: '2-digit', + })}_` + ); +} + +export async function purgeDashboardChannel( + opts: Pick, +): Promise { + if (!opts.statusChannelId) return; + + const statusJid = `dc:${opts.statusChannelId}`; + const ch = opts.channels.find( + (channel) => + channel.name.startsWith('discord') && + channel.isConnected() && + channel.purgeChannel, + ); + if (ch?.purgeChannel) { + await ch.purgeChannel(statusJid); + } +} + +export async function startStatusDashboard( + opts: DashboardOptions, +): Promise { + if (!opts.statusChannelId) return; + + const statusJid = `dc:${opts.statusChannelId}`; + + const updateStatus = async () => { + const ch = findDiscordChannel(opts.channels); + if (!ch) return; + + try { + await refreshChannelMeta(opts); + const content = buildStatusContent(opts); + + if (statusMessageId && ch.editMessage) { + await ch.editMessage(statusJid, statusMessageId, content); + } else if (ch.sendAndTrack) { + const id = await ch.sendAndTrack(statusJid, content); + if (id) statusMessageId = id; + } + } catch (err) { + logger.debug({ err }, 'Status dashboard update failed'); + statusMessageId = null; + } + }; + + setInterval(updateStatus, opts.statusUpdateInterval); + await updateStatus(); + logger.info({ channelId: opts.statusChannelId }, 'Status dashboard started'); +} + +export async function startUsageDashboard( + opts: DashboardOptions, +): Promise { + if (!opts.statusChannelId) return; + if (process.env.USAGE_DASHBOARD !== 'true') return; + + const statusJid = `dc:${opts.statusChannelId}`; + + const updateUsage = async () => { + if (usageUpdateInProgress) return; + usageUpdateInProgress = true; + + const ch = findDiscordChannel(opts.channels); + if (!ch) { + usageUpdateInProgress = false; + return; + } + + try { + const content = await buildUsageContent(); + + if (usageMessageId && ch.editMessage) { + await ch.editMessage(statusJid, usageMessageId, content); + } else if (ch.sendAndTrack) { + const id = await ch.sendAndTrack(statusJid, content); + if (id) usageMessageId = id; + } + } catch (err) { + logger.debug({ err }, 'Usage dashboard update failed'); + usageMessageId = null; + } finally { + usageUpdateInProgress = false; + } + }; + + setInterval(updateUsage, opts.usageUpdateInterval); + await updateUsage(); + logger.info('Usage dashboard started'); +} diff --git a/src/db.test.ts b/src/db.test.ts index b40a3a7..0ded8c5 100644 --- a/src/db.test.ts +++ b/src/db.test.ts @@ -3,12 +3,15 @@ import { describe, it, expect, beforeEach } from 'vitest'; import { _initTestDatabase, createTask, + deleteSession, deleteTask, getAllChats, getAllRegisteredGroups, getMessagesSince, getNewMessages, + getSession, getTaskById, + setSession, setRegisteredGroup, storeChatMetadata, storeMessage, @@ -300,6 +303,16 @@ describe('getNewMessages', () => { }); }); +describe('session accessors', () => { + it('deletes only the current service session for a group', () => { + setSession('group-a', 'session-123'); + expect(getSession('group-a')).toBe('session-123'); + + deleteSession('group-a'); + expect(getSession('group-a')).toBeUndefined(); + }); +}); + // --- storeChatMetadata --- describe('storeChatMetadata', () => { diff --git a/src/db.ts b/src/db.ts index 79934fb..4ac6c63 100644 --- a/src/db.ts +++ b/src/db.ts @@ -249,20 +249,6 @@ export function storeChatMetadata( } } -/** - * Update chat name without changing timestamp for existing chats. - * New chats get the current time as their initial timestamp. - * Used during group metadata sync. - */ -export function updateChatName(chatJid: string, name: string): void { - db.prepare( - ` - INSERT INTO chats (jid, name, last_message_time) VALUES (?, ?, ?) - ON CONFLICT(jid) DO UPDATE SET name = excluded.name - `, - ).run(chatJid, name, new Date().toISOString()); -} - export interface ChatInfo { jid: string; name: string; @@ -286,27 +272,6 @@ export function getAllChats(): ChatInfo[] { .all() as ChatInfo[]; } -/** - * Get timestamp of last group metadata sync. - */ -export function getLastGroupSync(): string | null { - // Store sync time in a special chat entry - const row = db - .prepare(`SELECT last_message_time FROM chats WHERE jid = '__group_sync__'`) - .get() as { last_message_time: string } | undefined; - return row?.last_message_time || null; -} - -/** - * Record that group metadata was synced. - */ -export function setLastGroupSync(): void { - const now = new Date().toISOString(); - db.prepare( - `INSERT OR REPLACE INTO chats (jid, name, last_message_time) VALUES ('__group_sync__', '__group_sync__', ?)`, - ).run(now); -} - /** * Store a message with full content. * Only call this for registered groups where message history is needed. @@ -326,33 +291,6 @@ export function storeMessage(msg: NewMessage): void { ); } -/** - * Store a message directly. - */ -export function storeMessageDirect(msg: { - id: string; - chat_jid: string; - sender: string; - sender_name: string; - content: string; - timestamp: string; - is_from_me: boolean; - is_bot_message?: boolean; -}): void { - db.prepare( - `INSERT OR REPLACE INTO messages (id, chat_jid, sender, sender_name, content, timestamp, is_from_me, is_bot_message) VALUES (?, ?, ?, ?, ?, ?, ?, ?)`, - ).run( - msg.id, - msg.chat_jid, - msg.sender, - msg.sender_name, - msg.content, - msg.timestamp, - msg.is_from_me ? 1 : 0, - msg.is_bot_message ? 1 : 0, - ); -} - export function getNewMessages( jids: string[], lastTimestamp: string, @@ -606,6 +544,12 @@ export function setSession(groupFolder: string, sessionId: string): void { ).run(groupFolder, SERVICE_AGENT_TYPE, sessionId); } +export function deleteSession(groupFolder: string): void { + db.prepare( + 'DELETE FROM sessions WHERE group_folder = ? AND agent_type = ?', + ).run(groupFolder, SERVICE_AGENT_TYPE); +} + export function getAllSessions(): Record { const rows = db .prepare( diff --git a/src/group-queue.ts b/src/group-queue.ts index 10fa6c4..46c05e6 100644 --- a/src/group-queue.ts +++ b/src/group-queue.ts @@ -11,6 +11,11 @@ interface QueuedTask { fn: () => Promise; } +export interface GroupRunContext { + runId: string; + reason: 'messages' | 'drain'; +} + const MAX_RETRIES = 5; const BASE_RETRY_MS = 5000; @@ -19,6 +24,7 @@ interface GroupState { idleWaiting: boolean; isTaskProcess: boolean; runningTaskId: string | null; + currentRunId: string | null; pendingMessages: boolean; pendingTasks: QueuedTask[]; process: ChildProcess | null; @@ -40,8 +46,9 @@ export class GroupQueue { private groups = new Map(); private activeCount = 0; private waitingGroups: string[] = []; - private processMessagesFn: ((groupJid: string) => Promise) | null = - null; + private processMessagesFn: + | ((groupJid: string, context: GroupRunContext) => Promise) + | null = null; private shuttingDown = false; private getGroup(groupJid: string): GroupState { @@ -52,6 +59,7 @@ export class GroupQueue { idleWaiting: false, isTaskProcess: false, runningTaskId: null, + currentRunId: null, pendingMessages: false, pendingTasks: [], process: null, @@ -65,10 +73,16 @@ export class GroupQueue { return state; } - setProcessMessagesFn(fn: (groupJid: string) => Promise): void { + setProcessMessagesFn( + fn: (groupJid: string, context: GroupRunContext) => Promise, + ): void { this.processMessagesFn = fn; } + private createRunId(): string { + return `run-${Date.now().toString(36)}-${Math.random().toString(36).slice(2, 8)}`; + } + enqueueMessageCheck(groupJid: string, groupFolder?: string): void { if (this.shuttingDown) return; @@ -154,17 +168,38 @@ export class GroupQueue { state.process = proc; state.processName = processName; if (groupFolder) state.groupFolder = groupFolder; + logger.info( + { + groupJid, + runId: state.currentRunId, + processName, + groupFolder: state.groupFolder, + isTaskProcess: state.isTaskProcess, + }, + 'Registered active process for group', + ); } /** * Mark the agent process as idle-waiting (finished work, waiting for IPC input). * If tasks are pending, preempt the idle agent process immediately. */ - notifyIdle(groupJid: string): void { + notifyIdle(groupJid: string, runId?: string): void { const state = this.getGroup(groupJid); state.idleWaiting = true; + logger.info( + { + groupJid, + runId: runId ?? state.currentRunId, + pendingTasks: state.pendingTasks.length, + }, + 'Agent entered idle wait state', + ); if (state.pendingTasks.length > 0) { - this.closeStdin(groupJid); + this.closeStdin(groupJid, { + runId: runId ?? state.currentRunId ?? undefined, + reason: 'pending-task-preemption', + }); } } @@ -174,8 +209,19 @@ export class GroupQueue { */ sendMessage(groupJid: string, text: string): boolean { const state = this.getGroup(groupJid); - if (!state.active || !state.groupFolder || state.isTaskProcess) + if (!state.active || !state.groupFolder || state.isTaskProcess) { + logger.debug( + { + groupJid, + runId: state.currentRunId, + active: state.active, + groupFolder: state.groupFolder, + isTaskProcess: state.isTaskProcess, + }, + 'Cannot pipe follow-up message to active agent', + ); return false; + } state.idleWaiting = false; // Agent is about to receive work, no longer idle const inputDir = path.join(DATA_DIR, 'ipc', state.groupFolder, 'input'); @@ -186,8 +232,22 @@ export class GroupQueue { const tempPath = `${filepath}.tmp`; fs.writeFileSync(tempPath, JSON.stringify({ type: 'message', text })); fs.renameSync(tempPath, filepath); + logger.info( + { + groupJid, + runId: state.currentRunId, + groupFolder: state.groupFolder, + textLength: text.length, + filename, + }, + 'Queued follow-up message for active agent', + ); return true; - } catch { + } catch (err) { + logger.warn( + { groupJid, runId: state.currentRunId, groupFolder: state.groupFolder, err }, + 'Failed to queue follow-up message for active agent', + ); return false; } } @@ -195,7 +255,10 @@ export class GroupQueue { /** * Signal the active agent process to wind down by writing a close sentinel. */ - closeStdin(groupJid: string): void { + closeStdin( + groupJid: string, + metadata?: { runId?: string; reason?: string }, + ): void { const state = this.getGroup(groupJid); if (!state.active || !state.groupFolder) return; @@ -203,8 +266,26 @@ export class GroupQueue { try { fs.mkdirSync(inputDir, { recursive: true }); fs.writeFileSync(path.join(inputDir, '_close'), ''); - } catch { - // ignore + logger.info( + { + groupJid, + runId: metadata?.runId ?? state.currentRunId, + groupFolder: state.groupFolder, + reason: metadata?.reason ?? 'unspecified', + }, + 'Signaled active agent to close stdin', + ); + } catch (err) { + logger.warn( + { + groupJid, + runId: metadata?.runId ?? state.currentRunId, + groupFolder: state.groupFolder, + reason: metadata?.reason ?? 'unspecified', + err, + }, + 'Failed to signal active agent to close stdin', + ); } } @@ -213,36 +294,55 @@ export class GroupQueue { reason: 'messages' | 'drain', ): Promise { const state = this.getGroup(groupJid); + const runId = this.createRunId(); state.active = true; state.idleWaiting = false; state.isTaskProcess = false; + state.currentRunId = runId; state.pendingMessages = false; state.startedAt = Date.now(); this.activeCount++; - logger.debug( - { groupJid, reason, activeCount: this.activeCount }, - 'Starting agent process for group', + logger.info( + { groupJid, runId, reason, activeCount: this.activeCount }, + 'Starting group message run', ); + let outcome: 'success' | 'retry_scheduled' | 'error' = 'success'; try { if (this.processMessagesFn) { - const success = await this.processMessagesFn(groupJid); + const success = await this.processMessagesFn(groupJid, { runId, reason }); if (success) { state.retryCount = 0; } else { - this.scheduleRetry(groupJid, state); + outcome = 'retry_scheduled'; + this.scheduleRetry(groupJid, state, runId); } } } catch (err) { - logger.error({ groupJid, err }, 'Error processing messages for group'); - this.scheduleRetry(groupJid, state); + outcome = 'error'; + logger.error({ groupJid, runId, err }, 'Error processing messages for group'); + this.scheduleRetry(groupJid, state, runId); } finally { + const durationMs = state.startedAt ? Date.now() - state.startedAt : null; + logger.info( + { + groupJid, + runId, + reason, + outcome, + durationMs, + pendingMessages: state.pendingMessages, + pendingTasks: state.pendingTasks.length, + }, + 'Finished group message run', + ); state.active = false; state.startedAt = null; state.process = null; state.processName = null; state.groupFolder = null; + state.currentRunId = null; this.activeCount--; this.drainGroup(groupJid); } @@ -279,11 +379,15 @@ export class GroupQueue { } } - private scheduleRetry(groupJid: string, state: GroupState): void { + private scheduleRetry( + groupJid: string, + state: GroupState, + runId?: string, + ): void { state.retryCount++; if (state.retryCount > MAX_RETRIES) { logger.error( - { groupJid, retryCount: state.retryCount }, + { groupJid, runId, retryCount: state.retryCount }, 'Max retries exceeded, dropping messages (will retry on next incoming message)', ); state.retryCount = 0; @@ -292,7 +396,7 @@ export class GroupQueue { const delayMs = BASE_RETRY_MS * Math.pow(2, state.retryCount - 1); logger.info( - { groupJid, retryCount: state.retryCount, delayMs }, + { groupJid, runId, retryCount: state.retryCount, delayMs }, 'Scheduling retry with backoff', ); setTimeout(() => { @@ -412,7 +516,7 @@ export class GroupQueue { // via idle timeout or agent timeout. // This prevents reconnection restarts from killing working agents. const activeProcesses: string[] = []; - for (const [jid, state] of this.groups) { + for (const [, state] of this.groups) { if (state.process && !state.process.killed && state.processName) { activeProcesses.push(state.processName); } diff --git a/src/index.ts b/src/index.ts index 2848437..6824b73 100644 --- a/src/index.ts +++ b/src/index.ts @@ -1,6 +1,4 @@ -import { ChildProcess, execSync, spawn } from 'child_process'; import fs from 'fs'; -import os from 'os'; import path from 'path'; import { @@ -19,21 +17,16 @@ import { getChannelFactory, getRegisteredChannelNames, } from './channels/registry.js'; +import { AvailableGroup, writeGroupsSnapshot } from './agent-runner.js'; import { - AgentOutput, - runAgentProcess, - writeGroupsSnapshot, - writeTasksSnapshot, -} from './agent-runner.js'; + purgeDashboardChannel, + startStatusDashboard, + startUsageDashboard, +} from './dashboard.js'; import { - getAllChats, + deleteSession, getAllRegisteredGroups, getAllSessions, - getAllTasks, - getLastHumanMessageTimestamp, - getMessagesSince, - getNewMessages, - getRegisteredGroup, getRouterState, initDatabase, setRegisteredGroup, @@ -42,24 +35,21 @@ import { storeChatMetadata, storeMessage, } from './db.js'; -import { readEnvFile } from './env.js'; import { GroupQueue } from './group-queue.js'; import { resolveGroupFolderPath } from './group-folder.js'; import { startIpcWatcher } from './ipc.js'; -import { findChannel, formatMessages, formatOutbound } from './router.js'; +import { + createMessageRuntime, + getAvailableGroups as getAvailableGroupsForRegisteredGroups, +} from './message-runtime.js'; +import { findChannel, formatOutbound } from './router.js'; import { isSenderAllowed, - isTriggerAllowed, loadSenderAllowlist, shouldDropMessage, } from './sender-allowlist.js'; -import { - extractSessionCommand, - handleSessionCommand, - isSessionCommandAllowed, -} from './session-commands.js'; import { startSchedulerLoop } from './task-scheduler.js'; -import { Channel, ChannelMeta, NewMessage, RegisteredGroup } from './types.js'; +import { Channel, NewMessage, RegisteredGroup } from './types.js'; import { logger } from './logger.js'; // Re-export for backwards compatibility during refactor @@ -69,10 +59,34 @@ let lastTimestamp = ''; let sessions: Record = {}; let registeredGroups: Record = {}; let lastAgentTimestamp: Record = {}; -let messageLoopRunning = false; const channels: Channel[] = []; const queue = new GroupQueue(); +const messageRuntime = createMessageRuntime({ + assistantName: ASSISTANT_NAME, + idleTimeout: IDLE_TIMEOUT, + pollInterval: POLL_INTERVAL, + timezone: TIMEZONE, + triggerPattern: TRIGGER_PATTERN, + channels, + queue, + getRegisteredGroups: () => registeredGroups, + getSessions: () => sessions, + getLastTimestamp: () => lastTimestamp, + setLastTimestamp: (timestamp) => { + lastTimestamp = timestamp; + }, + getLastAgentTimestamps: () => lastAgentTimestamp, + saveState, + persistSession: (groupFolder, sessionId) => { + sessions[groupFolder] = sessionId; + setSession(groupFolder, sessionId); + }, + clearSession: (groupFolder) => { + delete sessions[groupFolder]; + deleteSession(groupFolder); + }, +}); function loadState(): void { lastTimestamp = getRouterState('last_timestamp') || ''; @@ -124,18 +138,8 @@ function registerGroup(jid: string, group: RegisteredGroup): void { * Get available groups list for the agent. * Returns groups ordered by most recent activity. */ -export function getAvailableGroups(): import('./agent-runner.js').AvailableGroup[] { - const chats = getAllChats(); - const registeredJids = new Set(Object.keys(registeredGroups)); - - return chats - .filter((c) => c.jid !== '__group_sync__' && c.is_group) - .map((c) => ({ - jid: c.jid, - name: c.name, - lastActivity: c.last_message_time, - isRegistered: registeredJids.has(c.jid), - })); +export function getAvailableGroups(): AvailableGroup[] { + return getAvailableGroupsForRegisteredGroups(registeredGroups); } /** @internal - exported for testing */ @@ -145,886 +149,6 @@ export function _setRegisteredGroups( registeredGroups = groups; } -/** - * Process all pending messages for a group. - * Called by the GroupQueue when it's this group's turn. - */ -async function processGroupMessages(chatJid: string): Promise { - const group = registeredGroups[chatJid]; - if (!group) return true; - - const channel = findChannel(channels, chatJid); - if (!channel) { - logger.warn({ chatJid }, 'No channel owns JID, skipping messages'); - return true; - } - - const isMainGroup = group.isMain === true; - - const sinceTimestamp = lastAgentTimestamp[chatJid] || ''; - const missedMessages = getMessagesSince( - chatJid, - sinceTimestamp, - ASSISTANT_NAME, - ); - - if (missedMessages.length === 0) return true; - - // --- Session command interception (before trigger check) --- - const cmdResult = await handleSessionCommand({ - missedMessages, - isMainGroup, - groupName: group.name, - triggerPattern: TRIGGER_PATTERN, - timezone: TIMEZONE, - deps: { - sendMessage: (text) => channel.sendMessage(chatJid, text), - setTyping: (typing) => - channel.setTyping?.(chatJid, typing) ?? Promise.resolve(), - runAgent: (prompt, onOutput) => - runAgent(group, prompt, chatJid, onOutput), - closeStdin: () => queue.closeStdin(chatJid), - advanceCursor: (ts) => { - lastAgentTimestamp[chatJid] = ts; - saveState(); - }, - formatMessages, - canSenderInteract: (msg) => { - const hasTrigger = TRIGGER_PATTERN.test(msg.content.trim()); - const reqTrigger = !isMainGroup && group.requiresTrigger !== false; - return ( - isMainGroup || - !reqTrigger || - (hasTrigger && - (msg.is_from_me || - isTriggerAllowed(chatJid, msg.sender, loadSenderAllowlist()))) - ); - }, - }, - }); - if (cmdResult.handled) return cmdResult.success; - // --- End session command interception --- - - // For non-main groups, check if trigger is required and present - if (!isMainGroup && group.requiresTrigger !== false) { - const allowlistCfg = loadSenderAllowlist(); - const hasTrigger = missedMessages.some( - (m) => - TRIGGER_PATTERN.test(m.content.trim()) && - (m.is_from_me || isTriggerAllowed(chatJid, m.sender, allowlistCfg)), - ); - if (!hasTrigger) { - return true; - } - } - - const prompt = formatMessages(missedMessages, TIMEZONE); - - // Advance cursor so the piping path in startMessageLoop won't re-fetch - // these messages. Save the old cursor so we can roll back on error. - const previousCursor = lastAgentTimestamp[chatJid] || ''; - lastAgentTimestamp[chatJid] = - missedMessages[missedMessages.length - 1].timestamp; - saveState(); - - logger.info( - { group: group.name, messageCount: missedMessages.length }, - 'Processing messages', - ); - - // Track idle timer for closing stdin when agent is idle - let idleTimer: ReturnType | null = null; - - const resetIdleTimer = () => { - if (idleTimer) clearTimeout(idleTimer); - idleTimer = setTimeout(() => { - logger.debug({ group: group.name }, 'Idle timeout, closing agent stdin'); - queue.closeStdin(chatJid); - }, IDLE_TIMEOUT); - }; - - let hadError = false; - let outputSentToUser = false; - - await channel.setTyping?.(chatJid, true); - - const output = await runAgent(group, prompt, chatJid, async (result) => { - if (result.result) { - const raw = - typeof result.result === 'string' - ? result.result - : JSON.stringify(result.result); - const text = raw.replace(/[\s\S]*?<\/internal>/g, '').trim(); - logger.info({ group: group.name }, `Agent output: ${raw.slice(0, 200)}`); - if (text) { - await channel.sendMessage(chatJid, text); - outputSentToUser = true; - } - } - - // Always clear typing and reset idle timer on any output (including null results) - await channel.setTyping?.(chatJid, false); - resetIdleTimer(); - - if (result.status === 'success') { - queue.notifyIdle(chatJid); - } - - if (result.status === 'error') { - hadError = true; - } - }); - - await channel.setTyping?.(chatJid, false); - - if (output === 'error') { - hadError = true; - } - - if (idleTimer) clearTimeout(idleTimer); - - if (hadError) { - if (outputSentToUser) { - logger.warn( - { group: group.name }, - 'Agent error after output was sent, skipping cursor rollback to prevent duplicates', - ); - return true; - } - lastAgentTimestamp[chatJid] = previousCursor; - saveState(); - logger.warn( - { group: group.name }, - 'Agent error, rolled back message cursor for retry', - ); - return false; - } - - return true; -} - -async function runAgent( - group: RegisteredGroup, - prompt: string, - chatJid: string, - onOutput?: (output: AgentOutput) => Promise, -): Promise<'success' | 'error'> { - const isMain = group.isMain === true; - const sessionId = sessions[group.folder]; - - // Update tasks snapshot for agent to read (filtered by group) - const tasks = getAllTasks(); - writeTasksSnapshot( - group.folder, - isMain, - tasks.map((t) => ({ - id: t.id, - groupFolder: t.group_folder, - prompt: t.prompt, - schedule_type: t.schedule_type, - schedule_value: t.schedule_value, - status: t.status, - next_run: t.next_run, - })), - ); - - // Update available groups snapshot (main group only can see all groups) - const availableGroups = getAvailableGroups(); - writeGroupsSnapshot( - group.folder, - isMain, - availableGroups, - new Set(Object.keys(registeredGroups)), - ); - - // Wrap onOutput to track session ID from streamed results - const wrappedOnOutput = onOutput - ? async (output: AgentOutput) => { - if (output.newSessionId) { - sessions[group.folder] = output.newSessionId; - setSession(group.folder, output.newSessionId); - } - await onOutput(output); - } - : undefined; - - try { - const output = await runAgentProcess( - group, - { - prompt, - sessionId, - groupFolder: group.folder, - chatJid, - isMain, - assistantName: ASSISTANT_NAME, - }, - (proc, processName) => - queue.registerProcess(chatJid, proc, processName, group.folder), - wrappedOnOutput, - ); - - if (output.newSessionId) { - sessions[group.folder] = output.newSessionId; - setSession(group.folder, output.newSessionId); - } - - if (output.status === 'error') { - logger.error( - { group: group.name, error: output.error }, - 'Agent process error', - ); - return 'error'; - } - - return 'success'; - } catch (err) { - logger.error({ group: group.name, err }, 'Agent error'); - return 'error'; - } -} - -// ── Status & Usage Dashboards ─────────────────────────────────── - -function formatElapsed(ms: number): string { - const s = Math.floor(ms / 1000); - if (s < 60) return `${s}s`; - if (s < 3600) { - const m = Math.floor(s / 60); - const rem = s % 60; - return `${m}m${rem.toString().padStart(2, '0')}s`; - } - const h = Math.floor(s / 3600); - const m = Math.floor((s % 3600) / 60); - if (h < 24) return `${h}h${m.toString().padStart(2, '0')}m`; - const d = Math.floor(h / 24); - const remH = h % 24; - return `${d}d${remH}h`; -} - -function formatBytes(bytes: number): string { - if (bytes >= 1073741824) return `${(bytes / 1073741824).toFixed(1)}GB`; - if (bytes >= 1048576) return `${(bytes / 1048576).toFixed(0)}MB`; - return `${(bytes / 1024).toFixed(0)}KB`; -} - -function usageEmoji(pct: number): string { - if (pct >= 80) return '🔴'; - if (pct >= 50) return '🟡'; - return '🟢'; -} - -function formatResetKST(value: string | number): string { - try { - // Handle unix timestamp (seconds) or ISO string - const date = - typeof value === 'number' ? new Date(value * 1000) : new Date(value); - return date.toLocaleString('ko-KR', { - timeZone: 'Asia/Seoul', - month: 'short', - day: 'numeric', - hour: '2-digit', - minute: '2-digit', - }); - } catch { - return String(value); - } -} - -const STATUS_ICONS: Record = { - processing: '🟡', - idle: '🟢', - waiting: '🔵', - inactive: '⚪', -}; - -let statusMessageId: string | null = null; -let usageMessageId: string | null = null; - -// Cache for Discord channel metadata (name, position, category) -let channelMetaCache = new Map(); -let channelMetaLastRefresh = 0; -const CHANNEL_META_REFRESH_MS = 300000; // 5 minutes - -async function refreshChannelMeta(): Promise { - const now = Date.now(); - if (now - channelMetaLastRefresh < CHANNEL_META_REFRESH_MS) return; - - const ch = channels.find( - (c) => c.name.startsWith('discord') && c.isConnected() && c.getChannelMeta, - ); - if (!ch?.getChannelMeta) return; - - const jids = Object.keys(registeredGroups).filter((j) => j.startsWith('dc:')); - try { - channelMetaCache = await ch.getChannelMeta(jids); - channelMetaLastRefresh = now; - } catch (err) { - logger.debug({ err }, 'Failed to refresh channel metadata'); - } -} - -function getStatusLabel(s: import('./group-queue.js').GroupStatus): string { - if (s.status === 'processing') - return `처리 중 (${formatElapsed(s.elapsedMs || 0)})`; - if (s.status === 'idle') return '대기 중'; - if (s.status === 'waiting') - return s.pendingTasks > 0 - ? `큐 대기 (태스크 ${s.pendingTasks}개)` - : '큐 대기 (메시지)'; - return '비활성'; -} - -function buildStatusContent(): string { - const jids = Object.keys(registeredGroups); - const statuses = queue.getStatuses(jids); - - const entries = statuses - .map((s) => ({ - status: s, - group: registeredGroups[s.jid], - meta: channelMetaCache.get(s.jid), - })) - .filter((e) => e.group); - - // Group by category - const categoryMap = new Map(); - for (const entry of entries) { - const cat = entry.meta?.category || '기타'; - if (!categoryMap.has(cat)) categoryMap.set(cat, []); - categoryMap.get(cat)!.push(entry); - } - - // Sort categories by position - const sortedCategories = [...categoryMap.entries()].sort((a, b) => { - const posA = a[1][0]?.meta?.categoryPosition ?? 999; - const posB = b[1][0]?.meta?.categoryPosition ?? 999; - return posA - posB; - }); - - const sections: string[] = []; - let totalActive = 0; - let totalIdle = 0; - let total = 0; - - for (const [catName, catEntries] of sortedCategories) { - catEntries.sort( - (a, b) => (a.meta?.position ?? 999) - (b.meta?.position ?? 999), - ); - - const lines = catEntries.map((e) => { - const icon = STATUS_ICONS[e.status.status] || '⚪'; - const label = getStatusLabel(e.status); - // Prefer actual Discord channel name over DB-stored name - const name = e.meta?.name ? `#${e.meta.name}` : e.group.name; - return ` ${icon} **${name}** — ${label}`; - }); - - if (channelMetaCache.size > 0 && catName !== '기타') { - sections.push(`📁 **${catName}**\n${lines.join('\n')}`); - } else { - sections.push(lines.join('\n')); - } - - totalActive += catEntries.filter( - (e) => e.status.status === 'processing', - ).length; - totalIdle += catEntries.filter((e) => e.status.status === 'idle').length; - total += catEntries.length; - } - - const header = `**에이전트 상태** (${ASSISTANT_NAME}) — 활성 ${totalActive} | 대기 ${totalIdle} | 전체 ${total}`; - return `${header}\n\n${sections.join('\n\n')}\n\n_${new Date().toLocaleTimeString('ko-KR')}_`; -} - -// ── API Usage Fetchers ────────────────────────────────────────── - -interface ClaudeUsageData { - five_hour?: { utilization: number; resets_at: string }; - seven_day?: { utilization: number; resets_at: string }; -} - -interface CodexRateLimit { - limitId?: string; - limitName: string | null; - primary: { usedPercent: number; resetsAt: string | number }; - secondary: { usedPercent: number; resetsAt: string | number }; -} - -async function fetchClaudeUsage(): Promise { - try { - const envToken = readEnvFile(['CLAUDE_CODE_OAUTH_TOKEN']); - let token = - process.env.CLAUDE_CODE_OAUTH_TOKEN || - envToken.CLAUDE_CODE_OAUTH_TOKEN || - ''; - if (!token) { - const configDir = - process.env.CLAUDE_CONFIG_DIR || path.join(os.homedir(), '.claude'); - const credsPath = path.join(configDir, '.credentials.json'); - if (!fs.existsSync(credsPath)) return null; - const creds = JSON.parse(fs.readFileSync(credsPath, 'utf-8')); - token = creds?.claudeAiOauth?.accessToken || ''; - } - if (!token) return null; - - const res = await fetch('https://api.anthropic.com/api/oauth/usage', { - headers: { - Authorization: `Bearer ${token}`, - 'anthropic-beta': 'oauth-2025-04-20', - }, - }); - if (!res.ok) return null; - return (await res.json()) as ClaudeUsageData; - } catch { - return null; - } -} - -async function fetchCodexUsage(): Promise { - // Find codex binary - const npmGlobalBin = path.join(os.homedir(), '.npm-global', 'bin', 'codex'); - const codexBin = fs.existsSync(npmGlobalBin) ? npmGlobalBin : 'codex'; - - return new Promise((resolve) => { - let done = false; - const finish = (val: CodexRateLimit[] | null) => { - if (done) return; - done = true; - clearTimeout(timer); - try { - proc.kill(); - } catch { - /* ignore */ - } - resolve(val); - }; - - const timer = setTimeout(() => finish(null), 20000); - - let proc: ChildProcess; - try { - proc = spawn(codexBin, ['app-server'], { - stdio: ['pipe', 'pipe', 'pipe'], - env: { - ...(process.env as Record), - PATH: [ - path.dirname(process.execPath), - path.join(os.homedir(), '.npm-global', 'bin'), - process.env.PATH || '', - ].join(':'), - }, - }); - } catch { - resolve(null); - return; - } - - proc.on('error', () => finish(null)); - proc.on('close', () => finish(null)); - - let buf = ''; - proc.stdout!.on('data', (chunk: Buffer) => { - buf += chunk.toString(); - const lines = buf.split('\n'); - buf = lines.pop() || ''; - for (const line of lines) { - if (!line.trim()) continue; - try { - const msg = JSON.parse(line); - if (msg.id === 1) { - // Initialize done, query rate limits - proc.stdin!.write( - JSON.stringify({ - jsonrpc: '2.0', - id: 2, - method: 'account/rateLimits/read', - params: {}, - }) + '\n', - ); - } else if (msg.id === 2 && msg.result) { - // Extract rate limits from rateLimitsByLimitId object - const byId = msg.result.rateLimitsByLimitId; - if (byId && typeof byId === 'object') { - finish(Object.values(byId) as CodexRateLimit[]); - } else { - finish(null); - } - } - } catch { - /* non-JSON line, skip */ - } - } - }); - - // Send initialize - proc.stdin!.write( - JSON.stringify({ - jsonrpc: '2.0', - id: 1, - method: 'initialize', - params: { clientInfo: { name: 'usage-monitor', version: '1.0' } }, - }) + '\n', - ); - }); -} - -// ── Usage Dashboard Builder ───────────────────────────────────── - -async function buildUsageContent(): Promise { - const lines: string[] = []; - - // Fetch API usage in parallel - const [claudeUsage, codexUsage] = await Promise.all([ - fetchClaudeUsage(), - fetchCodexUsage(), - ]); - - // Progress bar helper - const bar = (pct: number) => { - const filled = Math.round(pct / 10); - return '█'.repeat(filled) + '░'.repeat(10 - filled); - }; - - // Unified usage section with progress bars - lines.push('📊 *사용량*'); - - type UsageRow = { name: string; h5pct: number; h5reset: string; d7pct: number; d7reset: string }; - const rows: UsageRow[] = []; - - if (claudeUsage) { - const h5 = claudeUsage.five_hour; - const d7 = claudeUsage.seven_day; - rows.push({ - name: 'Claude', - h5pct: h5 ? (h5.utilization > 1 ? Math.round(h5.utilization) : Math.round(h5.utilization * 100)) : -1, - h5reset: h5 ? formatResetKST(h5.resets_at) : '', - d7pct: d7 ? (d7.utilization > 1 ? Math.round(d7.utilization) : Math.round(d7.utilization * 100)) : -1, - d7reset: d7 ? formatResetKST(d7.resets_at) : '', - }); - } - - if (codexUsage && Array.isArray(codexUsage)) { - const relevant = codexUsage.filter( - (l) => l.primary.usedPercent > 0 || l.secondary.usedPercent > 0, - ); - const display = relevant.length > 0 ? relevant : codexUsage.slice(0, 1); - for (const limit of display) { - rows.push({ - name: 'Codex', - h5pct: Math.round(limit.primary.usedPercent), - h5reset: formatResetKST(limit.primary.resetsAt), - d7pct: Math.round(limit.secondary.usedPercent), - d7reset: formatResetKST(limit.secondary.resetsAt), - }); - } - } - - if (rows.length > 0) { - lines.push('```'); - lines.push(' 5-Hour 7-Day'); - for (const r of rows) { - const h5 = r.h5pct >= 0 ? `${bar(r.h5pct)} ${String(r.h5pct).padStart(3)}%` : ' — '; - const d7 = r.d7pct >= 0 ? `${bar(r.d7pct)} ${String(r.d7pct).padStart(3)}%` : ' — '; - lines.push(`${r.name.padEnd(8)}${h5} ${d7}`); - } - lines.push('```'); - } else { - lines.push('_조회 불가_'); - } - lines.push(''); - - // System resources with progress bars - lines.push('🖥️ *서버*'); - - const loadAvg = os.loadavg(); - const cpuCount = os.cpus().length; - const cpuPct = Math.round((loadAvg[1] / cpuCount) * 100); // 5min avg - - const totalMem = os.totalmem(); - const usedMem = totalMem - os.freemem(); - const memPct = Math.round((usedMem / totalMem) * 100); - const memUsedGB = (usedMem / 1073741824).toFixed(1); - const memTotalGB = (totalMem / 1073741824).toFixed(1); - - let diskPct = 0; - let diskUsedGB = '?'; - let diskTotalGB = '?'; - try { - const df = execSync('df -B1 / | tail -1', { - encoding: 'utf-8', - timeout: 5000, - }).trim(); - const parts = df.split(/\s+/); - const diskUsed = parseInt(parts[2], 10); - const diskTotal = parseInt(parts[1], 10); - diskPct = Math.round((diskUsed / diskTotal) * 100); - diskUsedGB = (diskUsed / 1073741824).toFixed(0); - diskTotalGB = (diskTotal / 1073741824).toFixed(0); - } catch { - /* ignore */ - } - - lines.push('```'); - lines.push(`${'CPU'.padEnd(8)}${bar(cpuPct)} ${String(cpuPct).padStart(3)}%`); - lines.push(`${'Memory'.padEnd(8)}${bar(memPct)} ${String(memPct).padStart(3)}% ${memUsedGB}/${memTotalGB}GB`); - lines.push(`${'Disk'.padEnd(8)}${bar(diskPct)} ${String(diskPct).padStart(3)}% ${diskUsedGB}/${diskTotalGB}GB`); - lines.push(`${'Uptime'.padEnd(8)}${formatElapsed(os.uptime() * 1000)}`); - lines.push('```'); - - return ( - lines.join('\n') + - `\n_${new Date().toLocaleTimeString('ko-KR', { hour: '2-digit', minute: '2-digit' })}_` - ); -} - -// ── Dashboard Lifecycle ───────────────────────────────────────── - -async function startStatusDashboard(): Promise { - if (!STATUS_CHANNEL_ID) return; - - const statusJid = `dc:${STATUS_CHANNEL_ID}`; - - const findDiscordChannel = () => - channels.find((c) => c.name.startsWith('discord') && c.isConnected()); - - const updateStatus = async () => { - const ch = findDiscordChannel(); - if (!ch) return; - - try { - await refreshChannelMeta(); - const content = buildStatusContent(); - - if (statusMessageId && ch.editMessage) { - await ch.editMessage(statusJid, statusMessageId, content); - } else if (ch.sendAndTrack) { - const id = await ch.sendAndTrack(statusJid, content); - if (id) statusMessageId = id; - } - } catch (err) { - logger.debug({ err }, 'Status dashboard update failed'); - statusMessageId = null; - } - }; - - setInterval(updateStatus, STATUS_UPDATE_INTERVAL); - await updateStatus(); - logger.info({ channelId: STATUS_CHANNEL_ID }, 'Status dashboard started'); -} - -let usageUpdateInProgress = false; - -async function startUsageDashboard(): Promise { - if (!STATUS_CHANNEL_ID) return; - // Only one service should show usage (set USAGE_DASHBOARD=true on that service) - if (process.env.USAGE_DASHBOARD !== 'true') return; - - const statusJid = `dc:${STATUS_CHANNEL_ID}`; - - const findDiscordChannel = () => - channels.find((c) => c.name.startsWith('discord') && c.isConnected()); - - const updateUsage = async () => { - if (usageUpdateInProgress) return; - usageUpdateInProgress = true; - - const ch = findDiscordChannel(); - if (!ch) { - usageUpdateInProgress = false; - return; - } - - try { - const content = await buildUsageContent(); - - if (usageMessageId && ch.editMessage) { - await ch.editMessage(statusJid, usageMessageId, content); - } else if (ch.sendAndTrack) { - const id = await ch.sendAndTrack(statusJid, content); - if (id) usageMessageId = id; - } - } catch (err) { - logger.debug({ err }, 'Usage dashboard update failed'); - usageMessageId = null; - } finally { - usageUpdateInProgress = false; - } - }; - - setInterval(updateUsage, USAGE_UPDATE_INTERVAL); - await updateUsage(); - logger.info('Usage dashboard started'); -} - -async function startMessageLoop(): Promise { - if (messageLoopRunning) { - logger.debug('Message loop already running, skipping duplicate start'); - return; - } - messageLoopRunning = true; - - logger.info(`NanoClaw running (trigger: @${ASSISTANT_NAME})`); - - while (true) { - try { - const jids = Object.keys(registeredGroups); - const { messages, newTimestamp } = getNewMessages( - jids, - lastTimestamp, - ASSISTANT_NAME, - ); - - if (messages.length > 0) { - logger.info({ count: messages.length }, 'New messages'); - - // Advance the "seen" cursor for all messages immediately - lastTimestamp = newTimestamp; - saveState(); - - // Deduplicate by group - const messagesByGroup = new Map(); - for (const msg of messages) { - const existing = messagesByGroup.get(msg.chat_jid); - if (existing) { - existing.push(msg); - } else { - messagesByGroup.set(msg.chat_jid, [msg]); - } - } - - for (const [chatJid, groupMessages] of messagesByGroup) { - const group = registeredGroups[chatJid]; - if (!group) continue; - - const channel = findChannel(channels, chatJid); - if (!channel) { - logger.warn({ chatJid }, 'No channel owns JID, skipping messages'); - continue; - } - - const isMainGroup = group.isMain === true; - - // --- Bot-collaboration timeout --- - // If all new messages are from bots, only process if a human - // sent a message within the last 12 hours. - const BOT_COLLAB_TIMEOUT_MS = 12 * 60 * 60 * 1000; - const allFromBots = groupMessages.every( - (m) => m.is_from_me || !!m.is_bot_message, - ); - if (allFromBots) { - const lastHuman = getLastHumanMessageTimestamp(chatJid); - if ( - !lastHuman || - Date.now() - new Date(lastHuman).getTime() > BOT_COLLAB_TIMEOUT_MS - ) { - logger.info( - { chatJid, lastHuman }, - 'Bot-collaboration timeout: no human message within 12h, skipping', - ); - continue; - } - } - // --- End bot-collaboration timeout --- - - // --- Session command interception (message loop) --- - // Scan ALL messages in the batch for a session command. - const loopCmdMsg = groupMessages.find( - (m) => extractSessionCommand(m.content, TRIGGER_PATTERN) !== null, - ); - - if (loopCmdMsg) { - // Only close active agent if the sender is authorized — otherwise an - // untrusted user could kill in-flight work by sending /compact (DoS). - // closeStdin no-ops internally when no agent is active. - if ( - isSessionCommandAllowed( - isMainGroup, - loopCmdMsg.is_from_me === true, - ) - ) { - queue.closeStdin(chatJid); - } - // Enqueue so processGroupMessages handles auth + cursor advancement. - // Don't pipe via IPC — slash commands need a fresh agent with - // string prompt (not MessageStream) for SDK recognition. - queue.enqueueMessageCheck(chatJid); - continue; - } - // --- End session command interception --- - - const needsTrigger = !isMainGroup && group.requiresTrigger !== false; - - // For non-main groups, only act on trigger messages. - // Non-trigger messages accumulate in DB and get pulled as - // context when a trigger eventually arrives. - if (needsTrigger) { - const allowlistCfg = loadSenderAllowlist(); - const hasTrigger = groupMessages.some( - (m) => - TRIGGER_PATTERN.test(m.content.trim()) && - (m.is_from_me || - isTriggerAllowed(chatJid, m.sender, allowlistCfg)), - ); - if (!hasTrigger) continue; - } - - // Pull all messages since lastAgentTimestamp so non-trigger - // context that accumulated between triggers is included. - const allPending = getMessagesSince( - chatJid, - lastAgentTimestamp[chatJid] || '', - ASSISTANT_NAME, - ); - const messagesToSend = - allPending.length > 0 ? allPending : groupMessages; - const formatted = formatMessages(messagesToSend, TIMEZONE); - - if (queue.sendMessage(chatJid, formatted)) { - logger.debug( - { chatJid, count: messagesToSend.length }, - 'Piped messages to active agent', - ); - lastAgentTimestamp[chatJid] = - messagesToSend[messagesToSend.length - 1].timestamp; - saveState(); - // Show typing indicator while the agent processes the piped message - channel - .setTyping?.(chatJid, true) - ?.catch((err) => - logger.warn({ chatJid, err }, 'Failed to set typing indicator'), - ); - } else { - // No active agent — enqueue for a new one - queue.enqueueMessageCheck(chatJid, group.folder); - } - } - } - } catch (err) { - logger.error({ err }, 'Error in message loop'); - } - await new Promise((resolve) => setTimeout(resolve, POLL_INTERVAL)); - } -} - -/** - * Startup recovery: check for unprocessed messages in registered groups. - * Handles crash between advancing lastTimestamp and processing messages. - */ -function recoverPendingMessages(): void { - for (const [chatJid, group] of Object.entries(registeredGroups)) { - const sinceTimestamp = lastAgentTimestamp[chatJid] || ''; - const pending = getMessagesSince(chatJid, sinceTimestamp, ASSISTANT_NAME); - if (pending.length > 0) { - logger.info( - { group: group.name, pendingCount: pending.length }, - 'Recovery: found unprocessed messages', - ); - queue.enqueueMessageCheck(chatJid, group.folder); - } - } -} - async function main(): Promise { initDatabase(); logger.info('Database initialized'); @@ -1125,24 +249,23 @@ async function main(): Promise { ); }, getAvailableGroups, - writeGroupsSnapshot: (gf, im, ag, rj) => - writeGroupsSnapshot(gf, im, ag, rj), + writeGroupsSnapshot: (gf, im, ag) => writeGroupsSnapshot(gf, im, ag), }); - queue.setProcessMessagesFn(processGroupMessages); - recoverPendingMessages(); - // Purge old messages in status channel before creating fresh dashboards - if (STATUS_CHANNEL_ID) { - const statusJid = `dc:${STATUS_CHANNEL_ID}`; - const ch = channels.find( - (c) => c.name.startsWith('discord') && c.isConnected() && c.purgeChannel, - ); - if (ch?.purgeChannel) { - await ch.purgeChannel(statusJid); - } - } - await startStatusDashboard(); - await startUsageDashboard(); - startMessageLoop().catch((err) => { + queue.setProcessMessagesFn(messageRuntime.processGroupMessages); + messageRuntime.recoverPendingMessages(); + const dashboardOpts = { + assistantName: ASSISTANT_NAME, + channels, + queue, + registeredGroups: () => registeredGroups, + statusChannelId: STATUS_CHANNEL_ID, + statusUpdateInterval: STATUS_UPDATE_INTERVAL, + usageUpdateInterval: USAGE_UPDATE_INTERVAL, + }; + await purgeDashboardChannel(dashboardOpts); + await startStatusDashboard(dashboardOpts); + await startUsageDashboard(dashboardOpts); + messageRuntime.startMessageLoop().catch((err) => { logger.fatal({ err }, 'Message loop crashed unexpectedly'); process.exit(1); }); diff --git a/src/ipc.ts b/src/ipc.ts index eaa1b1d..8f04592 100644 --- a/src/ipc.ts +++ b/src/ipc.ts @@ -20,7 +20,6 @@ export interface IpcDeps { groupFolder: string, isMain: boolean, availableGroups: AvailableGroup[], - registeredJids: Set, ) => void; } @@ -405,7 +404,6 @@ export async function processTaskIpc( sourceGroup, true, availableGroups, - new Set(Object.keys(registeredGroups)), ); } else { logger.warn( diff --git a/src/message-runtime.ts b/src/message-runtime.ts new file mode 100644 index 0000000..1834c20 --- /dev/null +++ b/src/message-runtime.ts @@ -0,0 +1,528 @@ +import { + AgentOutput, + AvailableGroup, + runAgentProcess, + writeGroupsSnapshot, + writeTasksSnapshot, +} from './agent-runner.js'; +import { + getAllChats, + getAllTasks, + getLastHumanMessageTimestamp, + getMessagesSince, + getNewMessages, +} from './db.js'; +import { GroupQueue, GroupRunContext } from './group-queue.js'; +import { findChannel, formatMessages } from './router.js'; +import { + isTriggerAllowed, + loadSenderAllowlist, +} from './sender-allowlist.js'; +import { + extractSessionCommand, + handleSessionCommand, + isSessionCommandAllowed, +} from './session-commands.js'; +import { Channel, NewMessage, RegisteredGroup } from './types.js'; +import { logger } from './logger.js'; + +export interface MessageRuntimeDeps { + assistantName: string; + idleTimeout: number; + pollInterval: number; + timezone: string; + triggerPattern: RegExp; + channels: Channel[]; + queue: GroupQueue; + getRegisteredGroups: () => Record; + getSessions: () => Record; + getLastTimestamp: () => string; + setLastTimestamp: (timestamp: string) => void; + getLastAgentTimestamps: () => Record; + saveState: () => void; + persistSession: (groupFolder: string, sessionId: string) => void; + clearSession: (groupFolder: string) => void; +} + +export function getAvailableGroups( + registeredGroups: Record, +): AvailableGroup[] { + const chats = getAllChats(); + const registeredJids = new Set(Object.keys(registeredGroups)); + + return chats + .filter((chat) => chat.jid !== '__group_sync__' && chat.is_group) + .map((chat) => ({ + jid: chat.jid, + name: chat.name, + lastActivity: chat.last_message_time, + isRegistered: registeredJids.has(chat.jid), + })); +} + +export function createMessageRuntime(deps: MessageRuntimeDeps): { + processGroupMessages: ( + chatJid: string, + context: GroupRunContext, + ) => Promise; + recoverPendingMessages: () => void; + startMessageLoop: () => Promise; +} { + let messageLoopRunning = false; + + const getCurrentAvailableGroups = (): AvailableGroup[] => + getAvailableGroups(deps.getRegisteredGroups()); + + const runAgent = async ( + group: RegisteredGroup, + prompt: string, + chatJid: string, + runId: string, + onOutput?: (output: AgentOutput) => Promise, + ): Promise<'success' | 'error'> => { + const isMain = group.isMain === true; + const sessions = deps.getSessions(); + const sessionId = sessions[group.folder]; + + const tasks = getAllTasks(); + writeTasksSnapshot( + group.folder, + isMain, + tasks.map((task) => ({ + id: task.id, + groupFolder: task.group_folder, + prompt: task.prompt, + schedule_type: task.schedule_type, + schedule_value: task.schedule_value, + status: task.status, + next_run: task.next_run, + })), + ); + + writeGroupsSnapshot(group.folder, isMain, getCurrentAvailableGroups()); + + const wrappedOnOutput = onOutput + ? async (output: AgentOutput) => { + if (output.newSessionId) { + deps.persistSession(group.folder, output.newSessionId); + } + await onOutput(output); + } + : undefined; + + try { + const output = await runAgentProcess( + group, + { + prompt, + sessionId, + groupFolder: group.folder, + chatJid, + runId, + isMain, + assistantName: deps.assistantName, + }, + (proc, processName) => + deps.queue.registerProcess(chatJid, proc, processName, group.folder), + wrappedOnOutput, + ); + + if (output.newSessionId) { + deps.persistSession(group.folder, output.newSessionId); + } + + if (output.status === 'error') { + logger.error( + { group: group.name, chatJid, runId, error: output.error }, + 'Agent process error', + ); + return 'error'; + } + + return 'success'; + } catch (err) { + logger.error({ group: group.name, chatJid, runId, err }, 'Agent error'); + return 'error'; + } + }; + + const processGroupMessages = async ( + chatJid: string, + context: GroupRunContext, + ): Promise => { + const { runId, reason } = context; + const registeredGroups = deps.getRegisteredGroups(); + const group = registeredGroups[chatJid]; + if (!group) { + logger.warn({ chatJid, runId, reason }, 'Registered group missing for queued run'); + return true; + } + + const channel = findChannel(deps.channels, chatJid); + if (!channel) { + logger.warn({ chatJid, runId, reason }, 'No channel owns JID, skipping messages'); + return true; + } + + const isMainGroup = group.isMain === true; + const lastAgentTimestamps = deps.getLastAgentTimestamps(); + const sinceTimestamp = lastAgentTimestamps[chatJid] || ''; + const missedMessages = getMessagesSince( + chatJid, + sinceTimestamp, + deps.assistantName, + ); + + if (missedMessages.length === 0) { + logger.info( + { chatJid, group: group.name, groupFolder: group.folder, runId, reason }, + 'No pending messages for queued run', + ); + return true; + } + + logger.info( + { + chatJid, + group: group.name, + groupFolder: group.folder, + runId, + reason, + messageCount: missedMessages.length, + sinceTimestamp, + }, + 'Loaded pending messages for queued run', + ); + + const cmdResult = await handleSessionCommand({ + missedMessages, + isMainGroup, + groupName: group.name, + runId, + triggerPattern: deps.triggerPattern, + timezone: deps.timezone, + deps: { + sendMessage: (text) => channel.sendMessage(chatJid, text), + setTyping: (typing) => + channel.setTyping?.(chatJid, typing) ?? Promise.resolve(), + runAgent: (prompt, onOutput) => + runAgent(group, prompt, chatJid, runId, onOutput), + closeStdin: () => + deps.queue.closeStdin(chatJid, { + runId, + reason: 'session-command', + }), + clearSession: () => deps.clearSession(group.folder), + advanceCursor: (timestamp) => { + lastAgentTimestamps[chatJid] = timestamp; + deps.saveState(); + }, + formatMessages, + canSenderInteract: (msg) => { + const hasTrigger = deps.triggerPattern.test(msg.content.trim()); + const requiresTrigger = + !isMainGroup && group.requiresTrigger !== false; + return ( + isMainGroup || + !requiresTrigger || + (hasTrigger && + (msg.is_from_me || + isTriggerAllowed( + chatJid, + msg.sender, + loadSenderAllowlist(), + ))) + ); + }, + }, + }); + if (cmdResult.handled) return cmdResult.success; + + if (!isMainGroup && group.requiresTrigger !== false) { + const allowlistCfg = loadSenderAllowlist(); + const hasTrigger = missedMessages.some( + (msg) => + deps.triggerPattern.test(msg.content.trim()) && + (msg.is_from_me || + isTriggerAllowed(chatJid, msg.sender, allowlistCfg)), + ); + if (!hasTrigger) { + logger.info( + { chatJid, group: group.name, groupFolder: group.folder, runId }, + 'Skipping queued run because no allowed trigger was found', + ); + return true; + } + } + + const prompt = formatMessages(missedMessages, deps.timezone); + const previousCursor = lastAgentTimestamps[chatJid] || ''; + lastAgentTimestamps[chatJid] = + missedMessages[missedMessages.length - 1].timestamp; + deps.saveState(); + + logger.info( + { + chatJid, + group: group.name, + groupFolder: group.folder, + runId, + messageCount: missedMessages.length, + }, + 'Dispatching queued messages to agent', + ); + + let idleTimer: ReturnType | null = null; + + const resetIdleTimer = () => { + if (idleTimer) clearTimeout(idleTimer); + idleTimer = setTimeout(() => { + logger.info( + { chatJid, group: group.name, groupFolder: group.folder, runId }, + 'Idle timeout reached, closing active agent stdin', + ); + deps.queue.closeStdin(chatJid, { + runId, + reason: 'idle-timeout', + }); + }, deps.idleTimeout); + }; + + let hadError = false; + let outputSentToUser = false; + + await channel.setTyping?.(chatJid, true); + + const output = await runAgent(group, prompt, chatJid, runId, async (result) => { + if (result.result) { + const raw = + typeof result.result === 'string' + ? result.result + : JSON.stringify(result.result); + const text = raw.replace(/[\s\S]*?<\/internal>/g, '').trim(); + logger.info( + { + chatJid, + group: group.name, + groupFolder: group.folder, + runId, + resultStatus: result.status, + }, + `Agent output: ${raw.slice(0, 200)}`, + ); + if (text) { + await channel.sendMessage(chatJid, text); + outputSentToUser = true; + } + } + + await channel.setTyping?.(chatJid, false); + resetIdleTimer(); + + if (result.status === 'success') { + deps.queue.notifyIdle(chatJid, runId); + } + + if (result.status === 'error') { + hadError = true; + } + }); + + await channel.setTyping?.(chatJid, false); + + if (output === 'error') { + hadError = true; + } + + if (idleTimer) clearTimeout(idleTimer); + + if (hadError) { + if (outputSentToUser) { + logger.warn( + { chatJid, group: group.name, groupFolder: group.folder, runId }, + 'Agent error after output was sent, skipping cursor rollback to prevent duplicates', + ); + return true; + } + lastAgentTimestamps[chatJid] = previousCursor; + deps.saveState(); + logger.warn( + { chatJid, group: group.name, groupFolder: group.folder, runId }, + 'Agent error, rolled back message cursor for retry', + ); + return false; + } + + logger.info( + { + chatJid, + group: group.name, + groupFolder: group.folder, + runId, + outputSentToUser, + }, + 'Queued run completed successfully', + ); + + return true; + }; + + const startMessageLoop = async (): Promise => { + if (messageLoopRunning) { + logger.debug('Message loop already running, skipping duplicate start'); + return; + } + messageLoopRunning = true; + + logger.info(`NanoClaw running (trigger: @${deps.assistantName})`); + + while (true) { + try { + const registeredGroups = deps.getRegisteredGroups(); + const jids = Object.keys(registeredGroups); + const { messages, newTimestamp } = getNewMessages( + jids, + deps.getLastTimestamp(), + deps.assistantName, + ); + + if (messages.length > 0) { + logger.info({ count: messages.length }, 'New messages'); + + deps.setLastTimestamp(newTimestamp); + deps.saveState(); + + const messagesByGroup = new Map(); + for (const msg of messages) { + const existing = messagesByGroup.get(msg.chat_jid); + if (existing) { + existing.push(msg); + } else { + messagesByGroup.set(msg.chat_jid, [msg]); + } + } + + for (const [chatJid, groupMessages] of messagesByGroup) { + const group = registeredGroups[chatJid]; + if (!group) continue; + + const channel = findChannel(deps.channels, chatJid); + if (!channel) { + logger.warn({ chatJid }, 'No channel owns JID, skipping messages'); + continue; + } + + const isMainGroup = group.isMain === true; + const allFromBots = groupMessages.every( + (msg) => msg.is_from_me || !!msg.is_bot_message, + ); + if (allFromBots) { + const lastHuman = getLastHumanMessageTimestamp(chatJid); + if ( + !lastHuman || + Date.now() - new Date(lastHuman).getTime() > + 12 * 60 * 60 * 1000 + ) { + logger.info( + { chatJid, lastHuman }, + 'Bot-collaboration timeout: no human message within 12h, skipping', + ); + continue; + } + } + + const loopCmdMsg = groupMessages.find( + (msg) => + extractSessionCommand(msg.content, deps.triggerPattern) !== null, + ); + + if (loopCmdMsg) { + if ( + isSessionCommandAllowed( + isMainGroup, + loopCmdMsg.is_from_me === true, + ) + ) { + deps.queue.closeStdin(chatJid); + } + deps.queue.enqueueMessageCheck(chatJid); + continue; + } + + const needsTrigger = + !isMainGroup && group.requiresTrigger !== false; + if (needsTrigger) { + const allowlistCfg = loadSenderAllowlist(); + const hasTrigger = groupMessages.some( + (msg) => + deps.triggerPattern.test(msg.content.trim()) && + (msg.is_from_me || + isTriggerAllowed(chatJid, msg.sender, allowlistCfg)), + ); + if (!hasTrigger) continue; + } + + const lastAgentTimestamps = deps.getLastAgentTimestamps(); + const allPending = getMessagesSince( + chatJid, + lastAgentTimestamps[chatJid] || '', + deps.assistantName, + ); + const messagesToSend = + allPending.length > 0 ? allPending : groupMessages; + const formatted = formatMessages(messagesToSend, deps.timezone); + + if (deps.queue.sendMessage(chatJid, formatted)) { + logger.debug( + { chatJid, count: messagesToSend.length }, + 'Piped messages to active agent', + ); + lastAgentTimestamps[chatJid] = + messagesToSend[messagesToSend.length - 1].timestamp; + deps.saveState(); + channel + .setTyping?.(chatJid, true) + ?.catch((err) => + logger.warn( + { chatJid, err }, + 'Failed to set typing indicator', + ), + ); + } else { + deps.queue.enqueueMessageCheck(chatJid, group.folder); + } + } + } + } catch (err) { + logger.error({ err }, 'Error in message loop'); + } + await new Promise((resolve) => setTimeout(resolve, deps.pollInterval)); + } + }; + + const recoverPendingMessages = (): void => { + const registeredGroups = deps.getRegisteredGroups(); + const lastAgentTimestamps = deps.getLastAgentTimestamps(); + for (const [chatJid, group] of Object.entries(registeredGroups)) { + const sinceTimestamp = lastAgentTimestamps[chatJid] || ''; + const pending = getMessagesSince( + chatJid, + sinceTimestamp, + deps.assistantName, + ); + if (pending.length > 0) { + logger.info( + { group: group.name, pendingCount: pending.length }, + 'Recovery: found unprocessed messages', + ); + deps.queue.enqueueMessageCheck(chatJid, group.folder); + } + } + }; + + return { + processGroupMessages, + recoverPendingMessages, + startMessageLoop, + }; +} diff --git a/src/router.ts b/src/router.ts index c14ca89..84ab214 100644 --- a/src/router.ts +++ b/src/router.ts @@ -34,16 +34,6 @@ export function formatOutbound(rawText: string): string { return text; } -export function routeOutbound( - channels: Channel[], - jid: string, - text: string, -): Promise { - const channel = channels.find((c) => c.ownsJid(jid) && c.isConnected()); - if (!channel) throw new Error(`No channel for JID: ${jid}`); - return channel.sendMessage(jid, text); -} - export function findChannel( channels: Channel[], jid: string, diff --git a/src/session-commands.test.ts b/src/session-commands.test.ts index 4a3cbfe..66e4c06 100644 --- a/src/session-commands.test.ts +++ b/src/session-commands.test.ts @@ -18,6 +18,10 @@ describe('extractSessionCommand', () => { expect(extractSessionCommand('@Andy /compact', trigger)).toBe('/compact'); }); + it('detects bare /clear', () => { + expect(extractSessionCommand('/clear', trigger)).toBe('/clear'); + }); + it('rejects /compact with extra text', () => { expect(extractSessionCommand('/compact now please', trigger)).toBeNull(); }); @@ -82,6 +86,7 @@ function makeDeps( setTyping: vi.fn().mockResolvedValue(undefined), runAgent: vi.fn().mockResolvedValue('success'), closeStdin: vi.fn(), + clearSession: vi.fn(), advanceCursor: vi.fn(), formatMessages: vi.fn().mockReturnValue(''), canSenderInteract: vi.fn().mockReturnValue(true), @@ -123,6 +128,26 @@ describe('handleSessionCommand', () => { expect(deps.advanceCursor).toHaveBeenCalledWith('100'); }); + it('handles authorized /clear without invoking the agent', async () => { + const deps = makeDeps(); + const result = await handleSessionCommand({ + missedMessages: [makeMsg('/clear')], + isMainGroup: true, + groupName: 'test', + triggerPattern: trigger, + timezone: 'UTC', + deps, + }); + expect(result).toEqual({ handled: true, success: true }); + expect(deps.closeStdin).toHaveBeenCalledTimes(1); + expect(deps.clearSession).toHaveBeenCalledTimes(1); + expect(deps.advanceCursor).toHaveBeenCalledWith('100'); + expect(deps.runAgent).not.toHaveBeenCalled(); + expect(deps.sendMessage).toHaveBeenCalledWith( + 'Current session cleared. The next message will start a new conversation.', + ); + }); + it('sends denial to interactable sender in non-main group', async () => { const deps = makeDeps(); const result = await handleSessionCommand({ diff --git a/src/session-commands.ts b/src/session-commands.ts index 1a3b2fb..20fdf53 100644 --- a/src/session-commands.ts +++ b/src/session-commands.ts @@ -12,6 +12,7 @@ export function extractSessionCommand( let text = content.trim(); text = text.replace(triggerPattern, '').trim(); if (text === '/compact') return '/compact'; + if (text === '/clear') return '/clear'; return null; } @@ -41,6 +42,7 @@ export interface SessionCommandDeps { onOutput: (result: AgentResult) => Promise, ) => Promise<'success' | 'error'>; closeStdin: () => void; + clearSession: () => void; advanceCursor: (timestamp: string) => void; formatMessages: (msgs: NewMessage[], timezone: string) => string; /** Whether the denied sender would normally be allowed to interact (for denial messages). */ @@ -63,6 +65,7 @@ export async function handleSessionCommand(opts: { missedMessages: NewMessage[]; isMainGroup: boolean; groupName: string; + runId?: string; triggerPattern: RegExp; timezone: string; deps: SessionCommandDeps; @@ -71,6 +74,7 @@ export async function handleSessionCommand(opts: { missedMessages, isMainGroup, groupName, + runId, triggerPattern, timezone, deps, @@ -98,7 +102,17 @@ export async function handleSessionCommand(opts: { } // AUTHORIZED: process pre-compact messages first, then run the command - logger.info({ group: groupName, command }, 'Session command'); + logger.info({ group: groupName, runId, command }, 'Session command'); + + if (command === '/clear') { + deps.closeStdin(); + deps.clearSession(); + deps.advanceCursor(cmdMsg.timestamp); + await deps.sendMessage( + 'Current session cleared. The next message will start a new conversation.', + ); + return { handled: true, success: true }; + } const cmdIndex = missedMessages.indexOf(cmdMsg); const preCompactMsgs = missedMessages.slice(0, cmdIndex); @@ -125,7 +139,7 @@ export async function handleSessionCommand(opts: { if (preResult === 'error' || hadPreError) { logger.warn( - { group: groupName }, + { group: groupName, runId }, 'Pre-compact processing failed, aborting session command', ); await deps.sendMessage(