diff --git a/prompts/claude-platform.md b/prompts/claude-platform.md index 8a3f180..6aaef48 100644 --- a/prompts/claude-platform.md +++ b/prompts/claude-platform.md @@ -1,6 +1,6 @@ # Claude Platform Rules -You are Andy, a personal assistant. +You are 클코, a personal assistant powered by Claude Code. ## Communication diff --git a/runners/agent-runner/src/index.ts b/runners/agent-runner/src/index.ts index 1e1f941..7ebcf6a 100644 --- a/runners/agent-runner/src/index.ts +++ b/runners/agent-runner/src/index.ts @@ -511,6 +511,18 @@ async function runQuery( NANOCLAW_IS_MAIN: containerInput.isMain ? '1' : '0', }, }, + ...(process.env.MEMENTO_MCP_SSE_URL + ? { + 'memento-mcp': { + command: process.env.MEMENTO_MCP_REMOTE_PATH || 'mcp-remote', + args: [ + process.env.MEMENTO_MCP_SSE_URL, + '--header', + `Authorization:Bearer ${process.env.MEMENTO_ACCESS_KEY || ''}`, + ], + }, + } + : {}), }, hooks: { PreCompact: [{ hooks: [createPreCompactHook(containerInput.assistantName)] }], @@ -539,7 +551,22 @@ async function runQuery( if (message.type === 'result') { resultCount++; const textResult = 'result' in message ? (message as { result?: string }).result : null; + const isError = message.subtype?.startsWith('error'); log(`Result #${resultCount}: subtype=${message.subtype}${textResult ? ` text=${textResult.slice(0, 200)}` : ''}`); + if (isError) { + // Log full error details for debugging + const msg = message as Record; + const errorDetail = JSON.stringify({ + subtype: message.subtype, + result: textResult?.slice(0, 500), + errors: msg.errors, + stop_reason: msg.stop_reason, + duration_ms: msg.duration_ms, + duration_api_ms: msg.duration_api_ms, + session_id: msg.session_id, + }); + log(`Error result detail: ${errorDetail}`); + } writeOutput({ status: 'success', result: textResult || null, @@ -747,7 +774,11 @@ async function main(): Promise { } } catch (err) { const errorMessage = err instanceof Error ? err.message : String(err); + const errorStack = err instanceof Error ? err.stack : undefined; + const errorCause = err instanceof Error && err.cause ? String(err.cause) : undefined; log(`Agent error: ${errorMessage}`); + if (errorStack) log(`Stack: ${errorStack}`); + if (errorCause) log(`Cause: ${errorCause}`); writeOutput({ status: 'error', result: null, diff --git a/runners/agent-runner/src/ipc-mcp-stdio.ts b/runners/agent-runner/src/ipc-mcp-stdio.ts index 3af73aa..72ae2b2 100644 --- a/runners/agent-runner/src/ipc-mcp-stdio.ts +++ b/runners/agent-runner/src/ipc-mcp-stdio.ts @@ -10,6 +10,11 @@ import { z } from 'zod'; import fs from 'fs'; import path from 'path'; import { CronExpressionParser } from 'cron-parser'; +import { + buildCiWatchPrompt, + DEFAULT_WATCH_CI_CONTEXT_MODE, + normalizeWatchCiIntervalSeconds, +} from './watch-ci.js'; const IPC_DIR = process.env.NANOCLAW_IPC_DIR || '/workspace/ipc'; const MESSAGES_DIR = path.join(IPC_DIR, 'messages'); @@ -152,6 +157,90 @@ SCHEDULE VALUE FORMAT (all times are LOCAL timezone): }, ); +server.tool( + 'watch_ci', + 'Schedule a background CI watcher that checks until a run or check reaches a terminal state, then sends one message and cancels itself.', + { + target: z + .string() + .describe( + 'What to watch, for example "PR #123 checks" or "GitHub Actions run 987654321".', + ), + check_instructions: z + .string() + .describe( + 'Exact steps or commands to check status and what details matter when it finishes.', + ), + poll_interval_seconds: z + .number() + .int() + .min(30) + .max(3600) + .default(60) + .describe('How often to poll in seconds. Default 60, minimum 30.'), + context_mode: z + .enum(['group', 'isolated']) + .default(DEFAULT_WATCH_CI_CONTEXT_MODE) + .describe( + 'group=runs with chat history and memory, isolated=fresh session (include all context in check_instructions). Default: isolated.', + ), + target_group_jid: z + .string() + .optional() + .describe( + '(Main group only) JID of the group to schedule the watcher for. Defaults to the current group.', + ), + }, + async (args) => { + let pollSeconds: number; + try { + pollSeconds = normalizeWatchCiIntervalSeconds(args.poll_interval_seconds); + } catch (error) { + return { + content: [ + { + type: 'text' as const, + text: error instanceof Error ? error.message : String(error), + }, + ], + isError: true, + }; + } + + const targetJid = + isMain && args.target_group_jid ? args.target_group_jid : chatJid; + const taskId = `task-${Date.now()}-${Math.random().toString(36).slice(2, 8)}`; + const prompt = buildCiWatchPrompt({ + taskId, + target: args.target, + checkInstructions: args.check_instructions, + }); + + const data = { + type: 'schedule_task', + taskId, + prompt, + schedule_type: 'interval' as const, + schedule_value: String(pollSeconds * 1000), + context_mode: args.context_mode || DEFAULT_WATCH_CI_CONTEXT_MODE, + targetJid, + createdBy: groupFolder, + timestamp: new Date().toISOString(), + }; + + writeIpcFile(TASKS_DIR, data); + + return { + content: [ + { + type: 'text' as const, + text: `CI watcher scheduled: ${taskId} (${pollSeconds}s)`, + }, + ], + }; + }, +); + server.tool( 'list_tasks', "List all scheduled tasks. From main: shows all tasks. From other groups: shows only that group's tasks.", diff --git a/runners/agent-runner/src/watch-ci.ts b/runners/agent-runner/src/watch-ci.ts new file mode 100644 index 0000000..b0ce26b --- /dev/null +++ b/runners/agent-runner/src/watch-ci.ts @@ -0,0 +1,72 @@ +export const DEFAULT_WATCH_CI_INTERVAL_SECONDS = 60; +export const MIN_WATCH_CI_INTERVAL_SECONDS = 30; +export const MAX_WATCH_CI_INTERVAL_SECONDS = 3600; +export const DEFAULT_WATCH_CI_CONTEXT_MODE = 'isolated'; + +export interface BuildCiWatchPromptArgs { + taskId: string; + target: string; + checkInstructions: string; +} + +export function normalizeWatchCiIntervalSeconds(seconds?: number): number { + if (seconds === undefined) { + return DEFAULT_WATCH_CI_INTERVAL_SECONDS; + } + + if (!Number.isInteger(seconds)) { + throw new Error('poll_interval_seconds must be an integer.'); + } + + if ( + seconds < MIN_WATCH_CI_INTERVAL_SECONDS || + seconds > MAX_WATCH_CI_INTERVAL_SECONDS + ) { + throw new Error( + `poll_interval_seconds must be between ${MIN_WATCH_CI_INTERVAL_SECONDS} and ${MAX_WATCH_CI_INTERVAL_SECONDS}.`, + ); + } + + return seconds; +} + +export function buildCiWatchPrompt({ + taskId, + target, + checkInstructions, +}: BuildCiWatchPromptArgs): string { + return ` +[BACKGROUND CI WATCH] + +You are running as an EJClaw background CI watcher. + +Watch target: +${target} + +Task ID: +${taskId} + +Check instructions: +${checkInstructions} + +Rules: +- Use the watch target and check instructions in this prompt as the source of truth for what to inspect. +- On each run, check whether the target is still queued, pending, running, in progress, or otherwise non-terminal. +- If it is still not finished, send no visible message and end this run quietly. +- If it reached a terminal state such as success, failure, cancelled, timed out, neutral, skipped, or action required: + 1. Send exactly one concise completion message with \`send_message\`. + 2. Format it as a short multiline summary when possible, not one long paragraph. + 3. Preferred shape: + - First line: \`CI 완료: \` + - Second line: \`판정: \` + - Then 2-4 flat bullet points with only the most important metrics, errors, or comparisons. + - Optional final line: \`다음: \` if a concrete follow-up is needed. + 4. Adapt the content to the specific CI. Do not invent fixed fields when they do not fit. + 5. Avoid tables unless they are clearly the shortest readable format. + 6. Keep the message compact and easy for other agents to parse. + 7. Call \`cancel_task\` with task_id "${taskId}" so this watcher stops itself. +- If you hit a transient problem such as a rate limit, network issue, or temporary auth failure, send no visible message and leave the task active for the next retry. +- Prefer no normal final response. Use \`send_message\` for the completion message, and keep any non-user-facing notes inside \`\` tags if needed. +- Do not claim continued monitoring after you cancel the task. +`.trim(); +} diff --git a/runners/agent-runner/test/watch-ci.test.ts b/runners/agent-runner/test/watch-ci.test.ts new file mode 100644 index 0000000..f838526 --- /dev/null +++ b/runners/agent-runner/test/watch-ci.test.ts @@ -0,0 +1,49 @@ +import { describe, expect, it } from 'vitest'; + +import { + buildCiWatchPrompt, + DEFAULT_WATCH_CI_CONTEXT_MODE, + normalizeWatchCiIntervalSeconds, +} from '../src/watch-ci.js'; + +describe('watch-ci helpers', () => { + it('builds a self-cancelling CI watch prompt', () => { + const prompt = buildCiWatchPrompt({ + taskId: 'task-123', + target: 'PR #42 checks', + checkInstructions: + 'Use gh pr checks 42 and summarize only terminal results.', + }); + + expect(prompt).toContain('PR #42 checks'); + expect(prompt).toContain('task-123'); + expect(prompt).toContain('cancel_task'); + expect(prompt).toContain('send_message'); + expect(prompt).toContain('gh pr checks 42'); + expect(prompt).toContain('CI 완료: '); + expect(prompt).toContain('판정: '); + expect(prompt).toContain( + 'Use the watch target and check instructions in this prompt as the source of truth', + ); + }); + + it('defaults CI watchers to isolated context', () => { + expect(DEFAULT_WATCH_CI_CONTEXT_MODE).toBe('isolated'); + }); + + it('normalizes valid poll intervals', () => { + expect(normalizeWatchCiIntervalSeconds()).toBe(60); + expect(normalizeWatchCiIntervalSeconds(30)).toBe(30); + expect(normalizeWatchCiIntervalSeconds(600)).toBe(600); + }); + + it('rejects invalid poll intervals', () => { + expect(() => normalizeWatchCiIntervalSeconds(29)).toThrow( + /between 30 and 3600/i, + ); + expect(() => normalizeWatchCiIntervalSeconds(3601)).toThrow( + /between 30 and 3600/i, + ); + expect(() => normalizeWatchCiIntervalSeconds(30.5)).toThrow(/integer/i); + }); +}); diff --git a/runners/codex-runner/package-lock.json b/runners/codex-runner/package-lock.json index 4f62306..8dab381 100644 --- a/runners/codex-runner/package-lock.json +++ b/runners/codex-runner/package-lock.json @@ -8,7 +8,7 @@ "name": "nanoclaw-codex-runner", "version": "1.0.0", "dependencies": { - "@openai/codex-sdk": "^0.115.0" + "@openai/codex": "^0.115.0" }, "devDependencies": { "@types/node": "^22.10.7", @@ -103,18 +103,6 @@ "node": ">=16" } }, - "node_modules/@openai/codex-sdk": { - "version": "0.115.0", - "resolved": "https://registry.npmjs.org/@openai/codex-sdk/-/codex-sdk-0.115.0.tgz", - "integrity": "sha512-BPoPhim0uUm3rzugY7JFaFJ+rPG/wMRhvKNFii//Dp3kTb+gFy3jrip7ijXqhswsnqXM3nwTiv7kYHt1+TMUPg==", - "license": "Apache-2.0", - "dependencies": { - "@openai/codex": "0.115.0" - }, - "engines": { - "node": ">=18" - } - }, "node_modules/@openai/codex-win32-arm64": { "name": "@openai/codex", "version": "0.115.0-win32-arm64", diff --git a/runners/codex-runner/package.json b/runners/codex-runner/package.json index a4c2d1d..2fe0579 100644 --- a/runners/codex-runner/package.json +++ b/runners/codex-runner/package.json @@ -2,14 +2,14 @@ "name": "nanoclaw-codex-runner", "version": "1.0.0", "type": "module", - "description": "Container-side Codex CLI runner for NanoClaw", + "description": "Codex app-server runner for NanoClaw", "main": "dist/index.js", "scripts": { "build": "tsc", "start": "node dist/index.js" }, "dependencies": { - "@openai/codex-sdk": "^0.115.0" + "@openai/codex": "^0.115.0" }, "devDependencies": { "@types/node": "^22.10.7", diff --git a/runners/codex-runner/src/app-server-client.ts b/runners/codex-runner/src/app-server-client.ts index 757210f..a3d44da 100644 --- a/runners/codex-runner/src/app-server-client.ts +++ b/runners/codex-runner/src/app-server-client.ts @@ -34,6 +34,7 @@ export interface CodexAppServerTurnOptions { cwd: string; model?: string; effort?: string; + onProgress?: (message: string) => void; } export interface CodexAppServerTurnResult { @@ -69,6 +70,7 @@ interface PendingRequest { interface ActiveTurn { threadId: string; state: AppServerTurnState; + onProgress?: (message: string) => void; resolve: (value: CodexAppServerTurnResult) => void; reject: (reason?: unknown) => void; } @@ -215,6 +217,7 @@ export class CodexAppServerClient { this.activeTurn = { threadId, state: createInitialAppServerTurnState(), + onProgress: options.onProgress, resolve, reject, }; @@ -366,6 +369,20 @@ export class CodexAppServerClient { private handleNotification(message: JsonRpcNotification): void { if (!this.activeTurn) return; + if (message.method === 'item/completed') { + const item = + (message.params?.item as Record | undefined) || + undefined; + if ( + item?.type === 'agentMessage' && + item.phase !== 'final_answer' && + typeof item.text === 'string' && + item.text.trim().length > 0 + ) { + this.activeTurn.onProgress?.(item.text); + } + } + this.activeTurn.state = reduceAppServerTurnState( this.activeTurn.state, message as AppServerTurnEvent, diff --git a/runners/codex-runner/src/index.ts b/runners/codex-runner/src/index.ts index 6dc6a2e..4395f1c 100644 --- a/runners/codex-runner/src/index.ts +++ b/runners/codex-runner/src/index.ts @@ -1,8 +1,7 @@ /** * NanoClaw Codex Runner * - * Default runtime is Codex app-server, with SDK fallback available via - * CODEX_RUNTIME=sdk or automatic fallback when app-server startup fails. + * App-server only runtime. * * Input protocol: * Stdin: Full ContainerInput JSON (read until EOF) @@ -13,7 +12,6 @@ * Each result is wrapped in OUTPUT_START_MARKER / OUTPUT_END_MARKER pairs. */ -import { Codex, type Thread, type ThreadOptions, type UserInput } from '@openai/codex-sdk'; import fs from 'fs'; import path from 'path'; @@ -38,6 +36,7 @@ interface ContainerInput { interface ContainerOutput { status: 'success' | 'error'; result: string | null; + phase?: 'progress' | 'final'; newSessionId?: string; error?: string; } @@ -51,7 +50,6 @@ const IPC_INPUT_DIR = path.join(IPC_DIR, 'input'); const IPC_INPUT_CLOSE_SENTINEL = path.join(IPC_INPUT_DIR, '_close'); const IPC_POLL_MS = 500; const MAX_TURNS = 100; -const CODEX_RUNTIME = (process.env.CODEX_RUNTIME || 'app-server').toLowerCase(); const OUTPUT_START_MARKER = '---NANOCLAW_OUTPUT_START---'; const OUTPUT_END_MARKER = '---NANOCLAW_OUTPUT_END---'; @@ -168,25 +166,6 @@ function extractImagePaths(text: string): { cleanText: string; imagePaths: strin }; } -function parseSdkInput(text: string): string | UserInput[] { - const { cleanText, imagePaths } = extractImagePaths(text); - if (imagePaths.length === 0) return text; - - const input: UserInput[] = []; - if (cleanText) { - input.push({ type: 'text', text: cleanText }); - } - for (const imgPath of imagePaths) { - if (fs.existsSync(imgPath)) { - input.push({ type: 'local_image', path: imgPath }); - log(`Adding image input: ${imgPath}`); - } else { - log(`Image not found, skipping: ${imgPath}`); - } - } - return input.length > 0 ? input : text; -} - function parseAppServerInput(text: string): AppServerInputItem[] { const { cleanText, imagePaths } = extractImagePaths(text); const input: AppServerInputItem[] = []; @@ -211,66 +190,43 @@ function parseAppServerInput(text: string): AppServerInputItem[] { return input; } -function getThreadOptions(): ThreadOptions { - const threadOptions: ThreadOptions = { - workingDirectory: EFFECTIVE_CWD, - approvalPolicy: 'never', - sandboxMode: 'danger-full-access', - networkAccessEnabled: true, - webSearchMode: 'live', - }; - if (CODEX_MODEL) threadOptions.model = CODEX_MODEL; - if (CODEX_EFFORT) { - threadOptions.modelReasoningEffort = - CODEX_EFFORT as ThreadOptions['modelReasoningEffort']; - } - return threadOptions; -} +function formatProgressElapsed(ms: number): string { + const elapsedSeconds = Math.floor(ms / 10_000) * 10; + const hours = Math.floor(elapsedSeconds / 3600); + const minutes = Math.floor((elapsedSeconds % 3600) / 60); + const seconds = elapsedSeconds % 60; + const parts: string[] = []; -async function executeSdkTurn( - thread: Thread, - input: string | UserInput[], -): Promise<{ result: string; error?: string }> { - const ac = new AbortController(); + if (hours > 0) parts.push(`${hours}시간`); + if (minutes > 0) parts.push(`${minutes}분`); + parts.push(`${seconds}초`); - let turnSeconds = 0; - const sentinel = setInterval(() => { - if (consumeCloseSentinel()) { - log('Close sentinel detected during SDK turn, aborting'); - ac.abort(); - return; - } - turnSeconds += 5; - if (turnSeconds % 60 === 0) { - log(`Turn in progress... (${Math.round(turnSeconds / 60)}min)`); - } - }, 5000); - - try { - const turn = await thread.run(input, { signal: ac.signal }); - return { result: turn.finalResponse }; - } catch (err) { - if (ac.signal.aborted) { - return { result: '' }; - } - return { - result: '', - error: err instanceof Error ? err.message : String(err), - }; - } finally { - clearInterval(sentinel); - } + return parts.join(' '); } async function executeAppServerTurn( client: CodexAppServerClient, threadId: string, prompt: string, -): Promise<{ result: string; error?: string }> { +): Promise<{ result: string | null; error?: string }> { + let lastProgressMessage: string | null = null; const activeTurn = await client.startTurn(threadId, parseAppServerInput(prompt), { cwd: EFFECTIVE_CWD, model: CODEX_MODEL || undefined, effort: CODEX_EFFORT || undefined, + onProgress: (message) => { + const trimmed = message.trim(); + if (!trimmed || trimmed === lastProgressMessage) { + return; + } + lastProgressMessage = trimmed; + writeOutput({ + status: 'success', + phase: 'progress', + result: trimmed, + newSessionId: threadId, + }); + }, }); let elapsedMs = 0; @@ -308,7 +264,7 @@ async function executeAppServerTurn( elapsedMs += IPC_POLL_MS; if (elapsedMs > 0 && elapsedMs % 60000 === 0) { - log(`Turn in progress... (${Math.round(elapsedMs / 60000)}min)`); + log(`Turn in progress... (${formatProgressElapsed(elapsedMs)})`); } setTimeout(() => void pollDuringTurn(), IPC_POLL_MS); }; @@ -318,13 +274,13 @@ async function executeAppServerTurn( try { const { state, result } = await activeTurn.wait(); if (state.status === 'completed') { - return { result: result || '' }; + return { result }; } if (state.status === 'interrupted' && consumeCloseSentinel()) { - return { result: result || '' }; + return { result }; } return { - result: result || '', + result, error: state.errorMessage || `Codex turn finished with status ${state.status}`, }; } finally { @@ -332,87 +288,6 @@ async function executeAppServerTurn( } } -async function runSdkSession( - containerInput: ContainerInput, - prompt: string, -): Promise { - const threadOptions = getThreadOptions(); - const codex = new Codex(); - - let thread: Thread; - if (containerInput.sessionId) { - thread = codex.resumeThread(containerInput.sessionId, threadOptions); - log(`Thread resuming (session: ${containerInput.sessionId})`); - } else { - thread = codex.startThread(threadOptions); - log('Thread started (new session)'); - } - - let turnCount = 0; - while (true) { - turnCount++; - if (turnCount > MAX_TURNS) { - log(`Turn limit reached (${MAX_TURNS}), exiting`); - writeOutput({ - status: 'success', - result: '[세션 턴 제한 도달. 새 메시지로 다시 시작됩니다.]', - newSessionId: thread.id || undefined, - }); - break; - } - - const input = parseSdkInput(prompt); - log(`Starting SDK turn ${turnCount}/${MAX_TURNS}...`); - - let { result, error } = await executeSdkTurn(thread, input); - - if (error && turnCount === 1 && containerInput.sessionId) { - log(`Resume may have failed, retrying with new thread: ${error}`); - thread = codex.startThread(threadOptions); - ({ result, error } = await executeSdkTurn(thread, input)); - } - - if (consumeCloseSentinel()) { - if (result) { - writeOutput({ - status: 'success', - result, - newSessionId: thread.id || undefined, - }); - } - log('Close sentinel detected, exiting SDK runtime'); - break; - } - - if (error) { - log(`SDK turn error: ${error}`); - writeOutput({ - status: 'error', - result: result || null, - newSessionId: thread.id || undefined, - error, - }); - } else { - writeOutput({ - status: 'success', - result: result || null, - newSessionId: thread.id || undefined, - }); - } - - log('SDK turn done, waiting for next IPC message...'); - - const nextMessage = await waitForIpcMessage(); - if (nextMessage === null) { - log('Close sentinel received, exiting SDK runtime'); - break; - } - - log(`Got new SDK message (${nextMessage.length} chars)`); - prompt = nextMessage; - } -} - async function runAppServerCompact( client: CodexAppServerClient, threadId: string | undefined, @@ -529,6 +404,7 @@ async function runAppServerSession( writeOutput({ status: 'success', result: result || null, + ...(result ? { phase: 'final' as const } : {}), newSessionId: threadId, }); } @@ -549,10 +425,6 @@ async function runAppServerSession( } } -function shouldUseAppServer(): boolean { - return CODEX_RUNTIME !== 'sdk'; -} - // ── Main ────────────────────────────────────────────────────────── async function main(): Promise { @@ -595,28 +467,9 @@ async function main(): Promise { prompt += '\n' + pending.join('\n'); } - const preferAppServer = shouldUseAppServer(); try { - if (preferAppServer) { - try { - log(`Runtime selected: app-server (${CODEX_RUNTIME})`); - await runAppServerSession(containerInput, prompt); - return; - } catch (err) { - if (CODEX_RUNTIME === 'app-server') { - log( - `App-server runtime failed, falling back to SDK: ${ - err instanceof Error ? err.message : String(err) - }`, - ); - } else { - throw err; - } - } - } - - log('Runtime selected: sdk'); - await runSdkSession(containerInput, prompt); + log('Runtime selected: app-server'); + await runAppServerSession(containerInput, prompt); } catch (err) { const errorMessage = err instanceof Error ? err.message : String(err); log(`Runner error: ${errorMessage}`); diff --git a/scripts/migrate-unify.cjs b/scripts/migrate-unify.cjs new file mode 100644 index 0000000..99420b0 --- /dev/null +++ b/scripts/migrate-unify.cjs @@ -0,0 +1,220 @@ +/** + * Data Unification Migration Script + * Merges codex data into the primary (claude) directories. + * + * Run AFTER stopping both services. + */ +const Database = require('/home/clone-ej/nanoclaw/node_modules/better-sqlite3'); +const fs = require('fs'); +const path = require('path'); + +const BASE = '/home/clone-ej/nanoclaw'; +const CLAUDE_DB = path.join(BASE, 'store/messages.db'); +const CODEX_DB = path.join(BASE, 'store-codex/messages.db'); + +console.log('=== NanoClaw Data Unification ===\n'); + +// 1. Open both DBs +const primary = new Database(CLAUDE_DB); +const codex = new Database(CODEX_DB); + +primary.pragma('journal_mode = WAL'); +primary.pragma('busy_timeout = 5000'); + +// 2. Merge registered_groups (codex → primary, skip JID conflicts) +console.log('--- Merging registered_groups ---'); +const codexGroups = codex.prepare('SELECT * FROM registered_groups').all(); +const insertGroup = primary.prepare(` + INSERT OR IGNORE INTO registered_groups + (jid, name, folder, trigger_pattern, added_at, container_config, requires_trigger, is_main, agent_type, work_dir) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) +`); + +let groupsAdded = 0; +let groupsSkipped = 0; +for (const g of codexGroups) { + // Check if JID already exists with different agent_type + const existing = primary.prepare('SELECT agent_type FROM registered_groups WHERE jid = ?').get(g.jid); + if (existing) { + // Same JID, different bots — both need to be registered + // Since JID is PK, we can't have duplicates. The group with claude-code type stays, + // and the codex one is already differentiated by agent_type in the same row. + // But wait — in the unified DB, the same JID can only appear once. + // For shared channels (both bots respond), we keep the claude registration + // and the codex service will also load it IF we adjust the filter. + // Actually, the codex service filters by agent_type='codex', so it won't see claude-code groups. + // We need BOTH registrations for shared channels. + // Solution: codex groups with different folder names get inserted. + // Same JID + same folder = skip (duplicate) + // Same JID + different folder = need a unique folder for codex + if (existing.agent_type !== g.agent_type) { + // Same JID but different agent types — need both + // Change JID to make it unique: append agent type suffix + const codexJid = g.jid + ':codex'; + const existsCodex = primary.prepare('SELECT 1 FROM registered_groups WHERE jid = ?').get(codexJid); + if (!existsCodex) { + insertGroup.run(codexJid, g.name, g.folder, g.trigger_pattern, g.added_at, + g.container_config, g.requires_trigger, g.is_main, g.agent_type, g.work_dir); + groupsAdded++; + console.log(` + ${g.folder} (codex, jid=${codexJid})`); + } else { + groupsSkipped++; + } + } else { + groupsSkipped++; + } + } else { + insertGroup.run(g.jid, g.name, g.folder, g.trigger_pattern, g.added_at, + g.container_config, g.requires_trigger, g.is_main, g.agent_type, g.work_dir); + groupsAdded++; + console.log(` + ${g.folder} (${g.agent_type})`); + } +} +console.log(` Added: ${groupsAdded}, Skipped: ${groupsSkipped}\n`); + +// 3. Merge sessions (codex → primary, agent_type='codex') +console.log('--- Merging sessions ---'); +const codexSessions = codex.prepare('SELECT * FROM sessions').all(); +const insertSession = primary.prepare( + 'INSERT OR IGNORE INTO sessions (group_folder, agent_type, session_id) VALUES (?, ?, ?)' +); +let sessionsAdded = 0; +for (const s of codexSessions) { + insertSession.run(s.group_folder, s.agent_type, s.session_id); + sessionsAdded++; + console.log(` + ${s.group_folder} [${s.agent_type}]`); +} +console.log(` Added: ${sessionsAdded}\n`); + +// 4. Merge router_state (codex → primary, already prefixed) +console.log('--- Merging router_state ---'); +const codexState = codex.prepare('SELECT * FROM router_state').all(); +const insertState = primary.prepare( + 'INSERT OR IGNORE INTO router_state (key, value) VALUES (?, ?)' +); +for (const s of codexState) { + insertState.run(s.key, s.value); + console.log(` + ${s.key}`); +} +console.log(''); + +// 5. Merge messages (codex → primary, skip duplicates by PK) +console.log('--- Merging messages ---'); +const codexMsgs = codex.prepare('SELECT * FROM messages').all(); +const insertMsg = primary.prepare(` + INSERT OR IGNORE INTO messages + (id, chat_jid, sender, sender_name, content, timestamp, is_from_me, is_bot_message) + VALUES (?, ?, ?, ?, ?, ?, ?, ?) +`); +let msgsAdded = 0; +let msgsSkipped = 0; +const batchInsert = primary.transaction((msgs) => { + for (const m of msgs) { + const result = insertMsg.run(m.id, m.chat_jid, m.sender, m.sender_name, + m.content, m.timestamp, m.is_from_me, m.is_bot_message); + if (result.changes > 0) msgsAdded++; + else msgsSkipped++; + } +}); +batchInsert(codexMsgs); +console.log(` Added: ${msgsAdded}, Skipped (duplicates): ${msgsSkipped}\n`); + +// 6. Merge chats (codex → primary, UPSERT with newer timestamp) +console.log('--- Merging chats ---'); +const codexChats = codex.prepare('SELECT * FROM chats').all(); +const upsertChat = primary.prepare(` + INSERT INTO chats (jid, name, last_message_time, channel, is_group) VALUES (?, ?, ?, ?, ?) + ON CONFLICT(jid) DO UPDATE SET + name = COALESCE(excluded.name, name), + last_message_time = MAX(last_message_time, excluded.last_message_time), + channel = COALESCE(excluded.channel, channel), + is_group = COALESCE(excluded.is_group, is_group) +`); +let chatsUpserted = 0; +for (const c of codexChats) { + upsertChat.run(c.jid, c.name, c.last_message_time, c.channel, c.is_group); + chatsUpserted++; +} +console.log(` Upserted: ${chatsUpserted}\n`); + +// 7. Merge scheduled_tasks (codex → primary, skip duplicates) +const codexTasks = codex.prepare('SELECT * FROM scheduled_tasks').all(); +if (codexTasks.length > 0) { + console.log('--- Merging scheduled_tasks ---'); + const insertTask = primary.prepare(` + INSERT OR IGNORE INTO scheduled_tasks + (id, group_folder, chat_jid, prompt, schedule_type, schedule_value, next_run, last_run, last_result, status, created_at, context_mode) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + `); + for (const t of codexTasks) { + insertTask.run(t.id, t.group_folder, t.chat_jid, t.prompt, t.schedule_type, + t.schedule_value, t.next_run, t.last_run, t.last_result, t.status, t.created_at, t.context_mode); + console.log(` + ${t.id}`); + } + console.log(''); +} + +// Close DBs +primary.close(); +codex.close(); + +// 8. Merge groups directories +console.log('--- Merging groups-codex/ → groups/ ---'); +const codexGroupsDir = path.join(BASE, 'groups-codex'); +const claudeGroupsDir = path.join(BASE, 'groups'); +if (fs.existsSync(codexGroupsDir)) { + const entries = fs.readdirSync(codexGroupsDir); + for (const entry of entries) { + const src = path.join(codexGroupsDir, entry); + const dst = path.join(claudeGroupsDir, entry); + if (fs.existsSync(dst)) { + // Merge: copy files that don't exist in dst + if (fs.statSync(src).isDirectory()) { + const files = fs.readdirSync(src); + for (const f of files) { + const srcFile = path.join(src, f); + const dstFile = path.join(dst, f); + if (!fs.existsSync(dstFile)) { + fs.cpSync(srcFile, dstFile, { recursive: true }); + console.log(` cp ${entry}/${f}`); + } + } + } + } else { + fs.cpSync(src, dst, { recursive: true }); + console.log(` + ${entry}/`); + } + } +} +console.log(''); + +// 9. Merge data-codex/sessions/ → data/sessions/ +console.log('--- Merging data-codex/sessions/ → data/sessions/ ---'); +const codexDataSessions = path.join(BASE, 'data-codex/sessions'); +const claudeDataSessions = path.join(BASE, 'data/sessions'); +if (fs.existsSync(codexDataSessions)) { + const entries = fs.readdirSync(codexDataSessions); + for (const entry of entries) { + const src = path.join(codexDataSessions, entry); + const dst = path.join(claudeDataSessions, entry); + if (fs.existsSync(dst)) { + // Both exist — merge .codex/ subdirectory + const codexSubdir = path.join(src, '.codex'); + const dstCodexSubdir = path.join(dst, '.codex'); + if (fs.existsSync(codexSubdir) && !fs.existsSync(dstCodexSubdir)) { + fs.cpSync(codexSubdir, dstCodexSubdir, { recursive: true }); + console.log(` cp ${entry}/.codex/`); + } + } else { + fs.cpSync(src, dst, { recursive: true }); + console.log(` + ${entry}/`); + } + } +} +console.log(''); + +console.log('=== Migration complete ==='); +console.log('Next steps:'); +console.log('1. Update .env.codex to remove NANOCLAW_STORE_DIR, NANOCLAW_DATA_DIR, NANOCLAW_GROUPS_DIR'); +console.log('2. Restart both services'); +console.log('3. Verify, then rename old dirs to .bak'); diff --git a/src/agent-runner.test.ts b/src/agent-runner.test.ts index 1c7e59e..8df241d 100644 --- a/src/agent-runner.test.ts +++ b/src/agent-runner.test.ts @@ -210,4 +210,46 @@ describe('agent-runner timeout behavior', () => { expect(result.status).toBe('success'); expect(result.newSessionId).toBe('session-456'); }); + + it('preserves streamed progress phase metadata', async () => { + const onOutput = vi.fn(async () => {}); + const resultPromise = runAgentProcess( + testGroup, + testInput, + () => {}, + onOutput, + ); + + emitOutputMarker(fakeProc, { + status: 'success', + result: '생각 중...', + phase: 'progress', + newSessionId: 'session-progress', + }); + emitOutputMarker(fakeProc, { + status: 'success', + result: '최종 답변', + phase: 'final', + newSessionId: 'session-progress', + }); + + await vi.advanceTimersByTimeAsync(10); + fakeProc.emit('close', 0); + await vi.advanceTimersByTimeAsync(10); + + const result = await resultPromise; + expect(result.status).toBe('success'); + expect(onOutput).toHaveBeenCalledWith( + expect.objectContaining({ + result: '생각 중...', + phase: 'progress', + }), + ); + expect(onOutput).toHaveBeenCalledWith( + expect.objectContaining({ + result: '최종 답변', + phase: 'final', + }), + ); + }); }); diff --git a/src/agent-runner.ts b/src/agent-runner.ts index 3ba7227..7c06dfe 100644 --- a/src/agent-runner.ts +++ b/src/agent-runner.ts @@ -44,6 +44,7 @@ export interface AgentInput { export interface AgentOutput { status: 'success' | 'error'; result: string | null; + phase?: 'progress' | 'final'; newSessionId?: string; error?: string; } @@ -164,10 +165,17 @@ function prepareGroupEnvironment( 'CODEX_OPENAI_API_KEY', 'CODEX_MODEL', 'CODEX_EFFORT', + 'MEMENTO_MCP_SSE_URL', + 'MEMENTO_ACCESS_KEY', + 'MEMENTO_MCP_REMOTE_PATH', ]); // Build a clean env without Claude Code nesting detection variables const cleanEnv = { ...(process.env as Record) }; + // Merge .env file values (readEnvFile only reads file, doesn't set process.env) + for (const [k, v] of Object.entries(envVars)) { + if (v && !cleanEnv[k]) cleanEnv[k] = v; + } delete cleanEnv.CLAUDECODE; delete cleanEnv.CLAUDE_CODE_ENTRYPOINT; @@ -290,8 +298,9 @@ function prepareGroupEnvironment( let toml = fs.existsSync(configTomlPath) ? fs.readFileSync(configTomlPath, 'utf-8') : ''; - // Remove existing nanoclaw MCP section if present (to refresh env vars) + // Remove existing nanoclaw/memento MCP sections if present (to refresh env vars) toml = toml.replace(/\n?\[mcp_servers\.nanoclaw\][\s\S]*?(?=\n\[|$)/, ''); + toml = toml.replace(/\n?\[mcp_servers\.memento-mcp\][\s\S]*?(?=\n\[|$)/, ''); const mcpSection = ` [mcp_servers.nanoclaw] command = "node" @@ -303,7 +312,24 @@ NANOCLAW_CHAT_JID = ${JSON.stringify(group.folder)} NANOCLAW_GROUP_FOLDER = ${JSON.stringify(group.folder)} NANOCLAW_IS_MAIN = ${JSON.stringify(isMain ? '1' : '0')} `; - toml = toml.trimEnd() + '\n' + mcpSection; + // Inject memento-mcp if MEMENTO_MCP_SSE_URL is set + const mementoSseUrl = + envVars.MEMENTO_MCP_SSE_URL || process.env.MEMENTO_MCP_SSE_URL; + const mementoAccessKey = + envVars.MEMENTO_ACCESS_KEY || process.env.MEMENTO_ACCESS_KEY || ''; + const mementoRemotePath = + envVars.MEMENTO_MCP_REMOTE_PATH || + process.env.MEMENTO_MCP_REMOTE_PATH || + 'mcp-remote'; + const mementoSection = mementoSseUrl + ? ` +[mcp_servers.memento-mcp] +command = ${JSON.stringify(mementoRemotePath)} +args = [${JSON.stringify(mementoSseUrl)}, "--header", ${JSON.stringify(`Authorization:Bearer ${mementoAccessKey}`)}] +` + : ''; + + toml = toml.trimEnd() + '\n' + mcpSection + mementoSection; fs.writeFileSync(configTomlPath, toml); } @@ -461,6 +487,18 @@ export async function runAgentProcess( } hadStreamingOutput = true; resetTimeout(); + if (parsed.status === 'error') { + logger.warn( + { + group: group.name, + chatJid: input.chatJid, + runId: input.runId, + error: parsed.error, + newSessionId: parsed.newSessionId, + }, + 'Streamed agent error output', + ); + } outputChain = outputChain.then(() => onOutput(parsed)); } catch (err) { logger.warn( diff --git a/src/bot-message-filter.test.ts b/src/bot-message-filter.test.ts new file mode 100644 index 0000000..4b69143 --- /dev/null +++ b/src/bot-message-filter.test.ts @@ -0,0 +1,141 @@ +import { describe, expect, it } from 'vitest'; + +import { filterProcessableMessages } from './bot-message-filter.js'; +import { NewMessage } from './types.js'; + +function makeMsg(overrides: Partial = {}): NewMessage { + return { + id: '1', + chat_jid: 'dc:1', + sender: 'user-1', + sender_name: 'User', + content: 'hello', + timestamp: '2026-03-20T00:00:00.000Z', + is_from_me: false, + is_bot_message: false, + ...overrides, + }; +} + +const OWN_BOT_ID = 'my-bot-123'; + +describe('filterProcessableMessages', () => { + it('filters bot-authored messages in normal rooms', () => { + const result = filterProcessableMessages( + [ + makeMsg({ id: 'human-1', content: 'human' }), + makeMsg({ + id: 'bot-1', + sender: 'bot-1', + sender_name: 'Bot', + content: 'status report', + is_bot_message: true, + }), + ], + false, + ); + + expect(result).toHaveLength(1); + expect(result[0].id).toBe('human-1'); + }); + + it('keeps other bot messages in paired rooms', () => { + const isOwn = (m: NewMessage) => + m.is_bot_message === true && m.sender === OWN_BOT_ID; + const result = filterProcessableMessages( + [ + makeMsg({ id: 'human-1', content: 'human' }), + makeMsg({ + id: 'bot-1', + sender: 'other-bot-456', + sender_name: 'OtherBot', + content: 'status report', + is_bot_message: true, + }), + ], + true, + isOwn, + ); + + expect(result).toHaveLength(2); + }); + + it('filters own bot messages in paired rooms', () => { + const isOwn = (m: NewMessage) => + m.is_bot_message === true && m.sender === OWN_BOT_ID; + const result = filterProcessableMessages( + [ + makeMsg({ id: 'human-1', content: 'human' }), + makeMsg({ + id: 'own-1', + sender: OWN_BOT_ID, + sender_name: 'MyBot', + content: 'my own output', + is_bot_message: true, + }), + makeMsg({ + id: 'other-1', + sender: 'other-bot-456', + sender_name: 'OtherBot', + content: 'partner response', + is_bot_message: true, + }), + ], + true, + isOwn, + ); + + expect(result).toHaveLength(2); + expect(result[0].id).toBe('human-1'); + expect(result[1].id).toBe('other-1'); + }); + + it('keeps all bot messages in paired rooms without isOwnMessage', () => { + const result = filterProcessableMessages( + [ + makeMsg({ id: 'human-1', content: 'human' }), + makeMsg({ + id: 'bot-1', + sender: 'bot-1', + sender_name: 'Bot', + content: 'status report', + is_bot_message: true, + }), + ], + true, + ); + + expect(result).toHaveLength(2); + }); + + it('filters session command control bot messages in paired rooms', () => { + const isOwn = (m: NewMessage) => + m.is_bot_message === true && m.sender === OWN_BOT_ID; + const result = filterProcessableMessages( + [ + makeMsg({ id: 'human-1', content: 'human' }), + makeMsg({ + id: 'control-1', + sender: 'other-bot-456', + sender_name: 'OtherBot', + content: + 'Current session cleared. The next message will start a new conversation.', + is_bot_message: true, + }), + makeMsg({ + id: 'other-1', + sender: 'other-bot-456', + sender_name: 'OtherBot', + content: 'partner response', + is_bot_message: true, + }), + ], + true, + isOwn, + ); + + expect(result).toHaveLength(2); + expect(result[0].id).toBe('human-1'); + expect(result[1].id).toBe('other-1'); + }); +}); diff --git a/src/bot-message-filter.ts b/src/bot-message-filter.ts new file mode 100644 index 0000000..216f77b --- /dev/null +++ b/src/bot-message-filter.ts @@ -0,0 +1,31 @@ +import { isSessionCommandControlMessage } from './session-commands.js'; +import { NewMessage } from './types.js'; + +/** + * Filter messages before processing. + * - Normal rooms: drop all bot messages. + * - Paired rooms (allowBotMessages=true): keep other bot's messages, + * but drop messages authored by this service's own bot (via isOwnMessage). + */ +export function filterProcessableMessages( + messages: NewMessage[], + allowBotMessages: boolean, + isOwnMessage?: (msg: NewMessage) => boolean, +): NewMessage[] { + const withoutControlMessages = messages.filter( + (message) => + !( + message.is_bot_message && + isSessionCommandControlMessage(message.content) + ), + ); + + if (allowBotMessages) { + // In paired rooms, allow other bot messages but filter own bot's output + if (isOwnMessage) { + return withoutControlMessages.filter((m) => !isOwnMessage(m)); + } + return withoutControlMessages; + } + return withoutControlMessages.filter((message) => !message.is_bot_message); +} diff --git a/src/channels/discord.test.ts b/src/channels/discord.test.ts index 5db7420..3687772 100644 --- a/src/channels/discord.test.ts +++ b/src/channels/discord.test.ts @@ -26,6 +26,12 @@ vi.mock('../logger.js', () => ({ }, })); +const isPairedRoomJidMock = vi.hoisted(() => vi.fn(() => false)); + +vi.mock('../db.js', () => ({ + isPairedRoomJid: isPairedRoomJidMock, +})); + // --- discord.js mock --- type Handler = (...args: any[]) => any; @@ -196,6 +202,7 @@ async function triggerMessage(message: any) { describe('DiscordChannel', () => { beforeEach(() => { vi.clearAllMocks(); + isPairedRoomJidMock.mockReturnValue(false); }); afterEach(() => { @@ -301,18 +308,54 @@ describe('DiscordChannel', () => { expect(opts.onMessage).not.toHaveBeenCalled(); }); - it('delivers bot messages with is_bot_message flag', async () => { + it('ignores its own bot messages', async () => { const opts = createTestOpts(); const channel = new DiscordChannel('test-token', opts); await channel.connect(); - const msg = createMessage({ isBot: true, content: 'I am a bot' }); + const msg = createMessage({ + authorId: '999888777', + isBot: true, + content: 'I am the connected bot', + }); + await triggerMessage(msg); + + expect(opts.onMessage).not.toHaveBeenCalled(); + }); + + it('ignores other bot messages in normal rooms', async () => { + const opts = createTestOpts(); + const channel = new DiscordChannel('test-token', opts); + await channel.connect(); + + const msg = createMessage({ + authorId: '111222333', + isBot: true, + content: 'I am another bot', + }); + await triggerMessage(msg); + + expect(opts.onMessage).not.toHaveBeenCalled(); + }); + + it('delivers other bot messages in paired rooms', async () => { + isPairedRoomJidMock.mockReturnValue(true); + + const opts = createTestOpts(); + const channel = new DiscordChannel('test-token', opts); + await channel.connect(); + + const msg = createMessage({ + authorId: '111222333', + isBot: true, + content: 'I am another bot', + }); await triggerMessage(msg); expect(opts.onMessage).toHaveBeenCalledWith( - expect.any(String), + 'dc:1234567890123456', expect.objectContaining({ - content: 'I am a bot', + content: 'I am another bot', is_bot_message: true, }), ); diff --git a/src/channels/discord.ts b/src/channels/discord.ts index f683366..1ff94ce 100644 --- a/src/channels/discord.ts +++ b/src/channels/discord.ts @@ -17,6 +17,7 @@ import { DATA_DIR, TRIGGER_PATTERN, } from '../config.js'; +import { isPairedRoomJid } from '../db.js'; import { readEnvFile } from '../env.js'; import { logger } from '../logger.js'; @@ -158,6 +159,7 @@ import { AgentType, Channel, ChannelMeta, + NewMessage, OnChatMetadata, OnInboundMessage, RegisteredGroup, @@ -202,11 +204,12 @@ export class DiscordChannel implements Channel { }); this.client.on(Events.MessageCreate, async (message: Message) => { - // Ignore own messages only - if (message.author.id === this.client?.user?.id) return; - const channelId = message.channelId; const chatJid = `dc:${channelId}`; + const isOwnBotMessage = message.author.id === this.client?.user?.id; + if (isOwnBotMessage) return; + if (message.author.bot && !isPairedRoomJid(chatJid)) return; + let content = message.content; const timestamp = message.createdAt.toISOString(); const senderName = @@ -481,6 +484,10 @@ export class DiscordChannel implements Channel { return this.client !== null && this.client.isReady(); } + isOwnMessage(msg: NewMessage): boolean { + return !!msg.is_bot_message && msg.sender === this.client?.user?.id; + } + ownsJid(jid: string): boolean { if (!jid.startsWith('dc:')) return false; if (!this.agentTypeFilter) return true; diff --git a/src/claude-usage.ts b/src/claude-usage.ts new file mode 100644 index 0000000..3ed83c1 --- /dev/null +++ b/src/claude-usage.ts @@ -0,0 +1,183 @@ +import { spawn } from 'child_process'; + +import { logger } from './logger.js'; + +export interface ClaudeUsageData { + five_hour?: { utilization: number; resets_at: string }; + seven_day?: { utilization: number; resets_at: string }; +} + +const CLAUDE_EXPECT_TIMEOUT_MS = 25000; +const ANSI_RE = /\u001b(?:[@-Z\\-_]|\[[0-?]*[ -/]*[@-~])/g; + +const EXPECT_PROGRAM = ` +set timeout 20 +log_user 1 +match_max 200000 +set binary $env(CLAUDE_BINARY) +spawn -noecho -- $binary --setting-sources user --allowed-tools "" +expect { + -re "Do you trust the files in this folder\\\\?" { send "y\\r"; exp_continue } + -re "Quick safety check:" { send "\\r"; exp_continue } + -re "Yes, I trust this folder" { send "\\r"; exp_continue } + -re "Ready to code here\\\\?" { send "\\r"; exp_continue } + -re "Press Enter to continue" { send "\\r"; exp_continue } + timeout {} +} +send "/usage\\r" +set deadline [expr {[clock seconds] + 20}] +while {[clock seconds] < $deadline} { + expect { + -re "Do you trust the files in this folder\\\\?" { send "y\\r"; exp_continue } + -re "Quick safety check:" { send "\\r"; exp_continue } + -re "Yes, I trust this folder" { send "\\r"; exp_continue } + -re "Ready to code here\\\\?" { send "\\r"; exp_continue } + -re "Press Enter to continue" { send "\\r"; exp_continue } + -re "Current session" { after 2000; exit 0 } + -re "Failed to load usage data" { after 500; exit 2 } + eof { exit 3 } + timeout { send "\\r" } + } +} +exit 4 +`; + +function normalizeLines(rawText: string): string[] { + return rawText + .replace(ANSI_RE, '') + .replace(/\r/g, '\n') + .split('\n') + .map((line) => line.replace(/\s+/g, ' ').trim()) + .filter(Boolean); +} + +function parsePercent(windowText: string): number | null { + const match = windowText.match(/(\d{1,3})%\s*(used|left)\b/i); + if (!match) return null; + const value = parseInt(match[1], 10); + if (Number.isNaN(value)) return null; + return match[2].toLowerCase() === 'left' ? 100 - value : value; +} + +function parseWindow( + lines: string[], + labels: string[], +): { utilization: number; resets_at: string } | null { + const normalizedLabels = labels.map((label) => label.toLowerCase()); + for (let i = 0; i < lines.length; i++) { + const line = lines[i].toLowerCase(); + if (!normalizedLabels.some((label) => line.includes(label))) continue; + + const windowLines = lines.slice(i, i + 6); + const windowText = windowLines.join('\n'); + const utilization = parsePercent(windowText); + if (utilization === null) continue; + + const resetLine = windowLines.find((candidate) => + candidate.toLowerCase().startsWith('resets'), + ); + + return { + utilization, + resets_at: resetLine || '', + }; + } + + return null; +} + +export function parseClaudeUsagePanel(rawText: string): ClaudeUsageData | null { + const lines = normalizeLines(rawText); + if (lines.length === 0) return null; + if ( + lines.some((line) => + line.toLowerCase().includes('failed to load usage data'), + ) + ) { + return null; + } + + const fiveHour = parseWindow(lines, ['Current session']); + if (!fiveHour) return null; + + const sevenDay = + parseWindow(lines, ['Current week (all models)']) || + parseWindow(lines, [ + 'Current week (Sonnet only)', + 'Current week (Sonnet)', + ]) || + parseWindow(lines, ['Current week (Opus)']); + + return { + five_hour: fiveHour, + ...(sevenDay ? { seven_day: sevenDay } : {}), + }; +} + +export async function fetchClaudeUsageViaCli( + binary = 'claude', +): Promise { + return new Promise((resolve) => { + let output = ''; + let finished = false; + + const finish = (value: ClaudeUsageData | null) => { + if (finished) return; + finished = true; + clearTimeout(timer); + resolve(value); + }; + + let proc: ReturnType | null = null; + try { + proc = spawn('expect', ['-c', EXPECT_PROGRAM], { + stdio: ['ignore', 'pipe', 'pipe'], + env: { + ...(process.env as Record), + CLAUDE_BINARY: binary, + }, + }); + } catch (err) { + logger.debug({ err }, 'Claude CLI PTY probe unavailable'); + finish(null); + return; + } + + const timer = setTimeout(() => { + try { + proc?.kill('SIGTERM'); + } catch { + /* ignore */ + } + finish(null); + }, CLAUDE_EXPECT_TIMEOUT_MS); + + if (!proc.stdout || !proc.stderr) { + finish(null); + return; + } + + proc.stdout.setEncoding('utf8'); + proc.stderr.setEncoding('utf8'); + proc.stdout.on('data', (chunk: string) => { + output += chunk; + }); + proc.stderr.on('data', (chunk: string) => { + output += chunk; + }); + proc.on('error', (err) => { + logger.debug({ err }, 'Claude CLI PTY probe failed to start'); + finish(null); + }); + proc.on('close', () => { + const parsed = parseClaudeUsagePanel(output); + if (!parsed && output.trim()) { + logger.debug( + { tail: output.slice(-400) }, + 'Claude CLI PTY probe produced unparsable output', + ); + } + finish(parsed); + }); + }); +} diff --git a/src/config.ts b/src/config.ts index f031f32..0fed572 100644 --- a/src/config.ts +++ b/src/config.ts @@ -10,6 +10,7 @@ const envConfig = readEnvFile([ 'SERVICE_ID', 'SERVICE_AGENT_TYPE', 'SESSION_COMMAND_ALLOWED_SENDERS', + 'SESSION_COMMAND_USER_IDS', 'USAGE_DASHBOARD', ]); @@ -89,12 +90,15 @@ export const USAGE_DASHBOARD_ENABLED = export const TIMEZONE = process.env.TZ || Intl.DateTimeFormat().resolvedOptions().timeZone; +const rawSessionCommandAllowedSenders = + process.env.SESSION_COMMAND_ALLOWED_SENDERS || + process.env.SESSION_COMMAND_USER_IDS || + envConfig.SESSION_COMMAND_ALLOWED_SENDERS || + envConfig.SESSION_COMMAND_USER_IDS || + ''; + const SESSION_COMMAND_ALLOWED_SENDERS = new Set( - ( - process.env.SESSION_COMMAND_ALLOWED_SENDERS || - envConfig.SESSION_COMMAND_ALLOWED_SENDERS || - '' - ) + rawSessionCommandAllowedSenders .split(',') .map((value) => value.trim()) .filter(Boolean), diff --git a/src/db.test.ts b/src/db.test.ts index 58ed3a6..ad36427 100644 --- a/src/db.test.ts +++ b/src/db.test.ts @@ -201,6 +201,7 @@ describe('getMessagesSince', () => { ); const botMsgs = msgs.filter((m) => m.content === 'bot reply'); expect(botMsgs).toHaveLength(1); + expect(botMsgs[0].is_bot_message).toBe(true); }); it('returns all messages including bot when sinceTimestamp is empty', () => { @@ -483,4 +484,29 @@ describe('registered group isMain', () => { expect(group).toBeDefined(); expect(group.isMain).toBeUndefined(); }); + + it('filters duplicate jid registrations by agent type', () => { + setRegisteredGroup('dc:shared', { + name: 'Shared Room Claude', + folder: 'shared-room', + trigger: '@Andy', + added_at: '2024-01-01T00:00:00.000Z', + agentType: 'claude-code', + }); + setRegisteredGroup('dc:shared', { + name: 'Shared Room Codex', + folder: 'shared-room', + trigger: '@Andy', + added_at: '2024-01-01T00:00:00.000Z', + agentType: 'codex', + }); + + const claudeGroups = getAllRegisteredGroups('claude-code'); + const codexGroups = getAllRegisteredGroups('codex'); + + expect(claudeGroups['dc:shared']?.agentType).toBe('claude-code'); + expect(claudeGroups['dc:shared']?.name).toBe('Shared Room Claude'); + expect(codexGroups['dc:shared']?.agentType).toBe('codex'); + expect(codexGroups['dc:shared']?.name).toBe('Shared Room Codex'); + }); }); diff --git a/src/db.ts b/src/db.ts index cb75f67..34e3a4c 100644 --- a/src/db.ts +++ b/src/db.ts @@ -378,6 +378,16 @@ export function storeMessage(msg: NewMessage): void { ); } +function normalizeMessageRow( + row: NewMessage & { is_from_me?: boolean | number; is_bot_message?: boolean | number }, +): NewMessage { + return { + ...row, + is_from_me: !!row.is_from_me, + is_bot_message: !!row.is_bot_message, + }; +} + export function getNewMessages( jids: string[], lastTimestamp: string, @@ -404,14 +414,16 @@ export function getNewMessages( const rows = db .prepare(sql) - .all(lastTimestamp, ...jids, `${botPrefix}:%`, limit) as NewMessage[]; + .all(lastTimestamp, ...jids, `${botPrefix}:%`, limit) as Array< + NewMessage & { is_from_me?: boolean | number; is_bot_message?: boolean | number } + >; let newTimestamp = lastTimestamp; for (const row of rows) { if (row.timestamp > newTimestamp) newTimestamp = row.timestamp; } - return { messages: rows, newTimestamp }; + return { messages: rows.map(normalizeMessageRow), newTimestamp }; } export function getMessagesSince( @@ -434,9 +446,12 @@ export function getMessagesSince( LIMIT ? ) ORDER BY timestamp `; - return db + const rows = db .prepare(sql) - .all(chatJid, sinceTimestamp, `${botPrefix}:%`, limit) as NewMessage[]; + .all(chatJid, sinceTimestamp, `${botPrefix}:%`, limit) as Array< + NewMessage & { is_from_me?: boolean | number; is_bot_message?: boolean | number } + >; + return rows.map(normalizeMessageRow); } export function getLastHumanMessageTimestamp(chatJid: string): string | null { @@ -657,10 +672,17 @@ export function getAllSessions(): Record { export function getRegisteredGroup( jid: string, + agentType?: string, ): (RegisteredGroup & { jid: string }) | undefined { - const row = db - .prepare('SELECT * FROM registered_groups WHERE jid = ?') - .get(jid) as + const row = ( + agentType + ? db + .prepare( + 'SELECT * FROM registered_groups WHERE jid = ? AND agent_type = ?', + ) + .get(jid, agentType) + : db.prepare('SELECT * FROM registered_groups WHERE jid = ?').get(jid) + ) as | { jid: string; name: string; @@ -718,6 +740,13 @@ export function setRegisteredGroup(jid: string, group: RegisteredGroup): void { ); } +export function updateRegisteredGroupName(jid: string, name: string): void { + db.prepare('UPDATE registered_groups SET name = ? WHERE jid = ?').run( + name, + jid, + ); +} + export function getAllRegisteredGroups( agentTypeFilter?: string, ): Record { diff --git a/src/index.ts b/src/index.ts index fc385c3..f8ff672 100644 --- a/src/index.ts +++ b/src/index.ts @@ -7,6 +7,7 @@ import { ASSISTANT_NAME, IDLE_TIMEOUT, POLL_INTERVAL, + SERVICE_AGENT_TYPE, isSessionCommandSenderAllowed, STATUS_CHANNEL_ID, STATUS_UPDATE_INTERVAL, @@ -36,14 +37,18 @@ import { getRegisteredGroup, getRouterState, initDatabase, + isPairedRoomJid, setRegisteredGroup, setRouterState, + updateRegisteredGroupName, deleteSession, setSession, storeChatMetadata, storeMessage, } from './db.js'; +import { filterProcessableMessages } from './bot-message-filter.js'; import { GroupQueue } from './group-queue.js'; +import { readEnvFile } from './env.js'; import { resolveGroupFolderPath } from './group-folder.js'; import { startIpcWatcher } from './ipc.js'; import { findChannel, formatMessages, formatOutbound } from './router.js'; @@ -58,6 +63,11 @@ import { handleSessionCommand, isSessionCommandAllowed, } from './session-commands.js'; +import { shouldResetSessionOnAgentFailure } from './session-recovery.js'; +import { + readStatusSnapshots, + writeStatusSnapshot, +} from './status-dashboard.js'; import { startSchedulerLoop } from './task-scheduler.js'; import { Channel, ChannelMeta, NewMessage, RegisteredGroup } from './types.js'; import { logger } from './logger.js'; @@ -74,6 +84,20 @@ let messageLoopRunning = false; const channels: Channel[] = []; const queue = new GroupQueue(); +function advanceLastAgentCursor(chatJid: string, timestamp: string): void { + lastAgentTimestamp[chatJid] = timestamp; + saveState(); +} + +function getProcessableMessages( + chatJid: string, + messages: NewMessage[], + channel?: import('./types.js').Channel, +): NewMessage[] { + const isOwn = channel?.isOwnMessage?.bind(channel); + return filterProcessableMessages(messages, isPairedRoomJid(chatJid), isOwn); +} + function loadState(): void { lastTimestamp = getRouterState('last_timestamp') || ''; const agentTs = getRouterState('last_agent_timestamp'); @@ -84,9 +108,11 @@ function loadState(): void { lastAgentTimestamp = {}; } sessions = getAllSessions(); - registeredGroups = getAllRegisteredGroups(); + // Load only this service's registrations. The DB can hold both + // claude-code and codex rows for the same Discord JID. + registeredGroups = getAllRegisteredGroups(SERVICE_AGENT_TYPE); logger.info( - { groupCount: Object.keys(registeredGroups).length }, + { groupCount: Object.keys(registeredGroups).length, agentType: SERVICE_AGENT_TYPE }, 'State loaded', ); } @@ -167,13 +193,20 @@ async function processGroupMessages(chatJid: string): Promise { const isMainGroup = group.isMain === true; const sinceTimestamp = lastAgentTimestamp[chatJid] || ''; - const missedMessages = getMessagesSince( + const rawMissedMessages = getMessagesSince( chatJid, sinceTimestamp, ASSISTANT_NAME, ); + const missedMessages = getProcessableMessages(chatJid, rawMissedMessages, channel); - if (missedMessages.length === 0) return true; + if (missedMessages.length === 0) { + const lastIgnored = rawMissedMessages[rawMissedMessages.length - 1]; + if (lastIgnored) { + advanceLastAgentCursor(chatJid, lastIgnored.timestamp); + } + return true; + } // --- Session command interception (before trigger check) --- const cmdResult = await handleSessionCommand({ @@ -191,8 +224,7 @@ async function processGroupMessages(chatJid: string): Promise { closeStdin: () => queue.closeStdin(chatJid), clearSession: () => clearSession(group.folder), advanceCursor: (ts) => { - lastAgentTimestamp[chatJid] = ts; - saveState(); + advanceLastAgentCursor(chatJid, ts); }, formatMessages, isAdminSender: (msg) => isSessionCommandSenderAllowed(msg.sender), @@ -230,9 +262,7 @@ async function processGroupMessages(chatJid: string): Promise { // 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(); + advanceLastAgentCursor(chatJid, missedMessages[missedMessages.length - 1].timestamp); logger.info( { group: group.name, messageCount: missedMessages.length }, @@ -251,29 +281,150 @@ async function processGroupMessages(chatJid: string): Promise { }; let hadError = false; - let outputSentToUser = false; + let latestProgressText: string | null = null; + let latestProgressRendered: string | null = null; + let progressMessageId: string | null = null; + let progressStartedAt: number | null = null; + let progressTicker: ReturnType | null = null; + let finalOutputSentToUser = false; + let progressOutputSentToUser = false; + let poisonedSessionDetected = false; + const isClaudeCodeAgent = (group.agentType || 'claude-code') === 'claude-code'; + + const clearProgressTicker = () => { + if (progressTicker) { + clearInterval(progressTicker); + progressTicker = null; + } + }; + + const renderProgressMessage = (text: string) => { + const elapsedSeconds = + progressStartedAt === null + ? 0 + : Math.floor((Date.now() - progressStartedAt) / 10_000) * 10; + const hours = Math.floor(elapsedSeconds / 3600); + const minutes = Math.floor((elapsedSeconds % 3600) / 60); + const seconds = elapsedSeconds % 60; + const elapsedParts: string[] = []; + + if (hours > 0) elapsedParts.push(`${hours}시간`); + if (minutes > 0) elapsedParts.push(`${minutes}분`); + elapsedParts.push(`${seconds}초`); + + return `${text}\n\n${elapsedParts.join(' ')}`; + }; + + const syncTrackedProgressMessage = async () => { + if (!progressMessageId || !channel.editMessage || !latestProgressText) { + return; + } + + const rendered = renderProgressMessage(latestProgressText); + if (rendered === latestProgressRendered) { + return; + } + + try { + await channel.editMessage(chatJid, progressMessageId, rendered); + latestProgressRendered = rendered; + } catch { + clearProgressTicker(); + progressMessageId = null; + } + }; + + const ensureProgressTicker = () => { + if (!progressMessageId || !channel.editMessage || progressTicker) { + return; + } + + progressTicker = setInterval(() => { + void syncTrackedProgressMessage(); + }, 10_000); + }; + + const finalizeProgressMessage = async () => { + await syncTrackedProgressMessage(); + clearProgressTicker(); + }; + + const sendProgressMessage = async (text: string) => { + if (!text || text === latestProgressText) { + return; + } + + latestProgressText = text; + if (progressStartedAt === null) { + progressStartedAt = Date.now(); + } + const rendered = renderProgressMessage(text); + + if (progressMessageId && channel.editMessage) { + await syncTrackedProgressMessage(); + progressOutputSentToUser = true; + return; + } + + if (!channel.sendAndTrack) { + return; + } + + progressMessageId = await channel.sendAndTrack(chatJid, rendered); + if (progressMessageId) { + latestProgressRendered = rendered; + ensureProgressTicker(); + progressOutputSentToUser = true; + } + }; await channel.setTyping?.(chatJid, true); const output = await runAgent(group, prompt, chatJid, async (result) => { + if ( + isClaudeCodeAgent && + shouldResetSessionOnAgentFailure(result) && + !poisonedSessionDetected + ) { + poisonedSessionDetected = true; + hadError = true; + clearSession(group.folder); + queue.closeStdin(chatJid); + logger.warn( + { group: group.name, chatJid }, + 'Detected poisoned Claude session from streamed output, forcing close', + ); + } + if (result.result) { const raw = typeof result.result === 'string' ? result.result : JSON.stringify(result.result); - const text = raw.replace(/[\s\S]*?<\/internal>/g, '').trim(); + const text = formatOutbound(raw); logger.info({ group: group.name }, `Agent output: ${raw.slice(0, 200)}`); - if (text) { - await channel.sendMessage(chatJid, text); - outputSentToUser = true; + if (result.phase === 'progress') { + if (text) { + await sendProgressMessage(text); + } + return; } + if (text) { + await finalizeProgressMessage(); + await channel.sendMessage(chatJid, text); + finalOutputSentToUser = true; + } + } else { + await finalizeProgressMessage(); } // Always clear typing and reset idle timer on any output (including null results) await channel.setTyping?.(chatJid, false); - resetIdleTimer(); + if (!poisonedSessionDetected) { + resetIdleTimer(); + } - if (result.status === 'success') { + if (result.status === 'success' && !poisonedSessionDetected) { queue.notifyIdle(chatJid); } @@ -288,10 +439,12 @@ async function processGroupMessages(chatJid: string): Promise { hadError = true; } + clearProgressTicker(); + if (idleTimer) clearTimeout(idleTimer); if (hadError) { - if (outputSentToUser) { + if (finalOutputSentToUser || progressOutputSentToUser) { logger.warn( { group: group.name }, 'Agent error after output was sent, skipping cursor rollback to prevent duplicates', @@ -446,12 +599,16 @@ const STATUS_ICONS: Record = { }; let statusMessageId: string | null = null; +// eslint-disable-next-line @typescript-eslint/no-unused-vars -- kept for compat let usageMessageId: string | null = null; +let cachedUsageContent: string = ''; +let cachedClaudeUsageData: ClaudeUsageData | 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 +const STATUS_SNAPSHOT_MAX_AGE_MS = 60000; async function refreshChannelMeta(): Promise { const now = Date.now(); @@ -462,16 +619,40 @@ async function refreshChannelMeta(): Promise { ); if (!ch?.getChannelMeta) return; - const jids = Object.keys(registeredGroups).filter((j) => j.startsWith('dc:')); + // Include jids from both local registeredGroups and other service snapshots + const localJids = Object.keys(registeredGroups).filter((j) => j.startsWith('dc:')); + const snapshotJids = readStatusSnapshots(STATUS_SNAPSHOT_MAX_AGE_MS) + .flatMap((s) => s.entries.map((e) => e.jid)) + .filter((j) => j.startsWith('dc:')); + const jids = [...new Set([...localJids, ...snapshotJids])]; try { channelMetaCache = await ch.getChannelMeta(jids); channelMetaLastRefresh = now; + + // Auto-sync DB group names with Discord channel names + for (const [jid, meta] of channelMetaCache) { + if (!meta.name) continue; + const group = registeredGroups[jid]; + if (group && group.name !== meta.name) { + logger.info( + { jid, oldName: group.name, newName: meta.name }, + 'Syncing group name to Discord channel name', + ); + group.name = meta.name; + updateRegisteredGroupName(jid, meta.name); + } + } } catch (err) { logger.debug({ err }, 'Failed to refresh channel metadata'); } } -function getStatusLabel(s: import('./group-queue.js').GroupStatus): string { +function getStatusLabel(s: { + status: 'processing' | 'idle' | 'waiting' | 'inactive'; + elapsedMs: number | null; + pendingMessages: boolean; + pendingTasks: number; +}): string { if (s.status === 'processing') return `처리 중 (${formatElapsed(s.elapsedMs || 0)})`; if (s.status === 'idle') return '대기 중'; @@ -482,27 +663,133 @@ function getStatusLabel(s: import('./group-queue.js').GroupStatus): string { return '비활성'; } -function buildStatusContent(): string { +function getAgentDisplayName(agentType: 'claude-code' | 'codex'): string { + return agentType === 'codex' ? '코덱스' : '클코'; +} + +function formatRoomName( + jid: string, + meta: ChannelMeta | undefined, + fallbackName: string | undefined, + chatName: string | undefined, +): string { + const base = + meta?.name || + (chatName && chatName !== jid ? chatName : undefined) || + (fallbackName && fallbackName !== jid ? fallbackName : undefined) || + jid; + + if (jid.startsWith('dc:') && base !== jid && !base.startsWith('#')) { + return `#${base}`; + } + return base; +} + +function writeLocalStatusSnapshot(): void { 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); + writeStatusSnapshot({ + agentType: SERVICE_AGENT_TYPE, + assistantName: ASSISTANT_NAME, + updatedAt: new Date().toISOString(), + entries: statuses + .map((status) => { + const group = registeredGroups[status.jid]; + if (!group) return null; + return { + jid: status.jid, + name: group.name, + folder: group.folder, + agentType: (group.agentType || SERVICE_AGENT_TYPE) as + | 'claude-code' + | 'codex', + status: status.status, + elapsedMs: status.elapsedMs, + pendingMessages: status.pendingMessages, + pendingTasks: status.pendingTasks, + }; + }) + .filter( + ( + entry, + ): entry is { + jid: string; + name: string; + folder: string; + agentType: 'claude-code' | 'codex'; + status: 'processing' | 'idle' | 'waiting' | 'inactive'; + elapsedMs: number | null; + pendingMessages: boolean; + pendingTasks: number; + } => Boolean(entry), + ), + }); +} - // 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); +function buildStatusContent(): string { + const snapshots = readStatusSnapshots(STATUS_SNAPSHOT_MAX_AGE_MS); + const chatNameByJid = new Map(getAllChats().map((chat) => [chat.jid, chat.name])); + + // Collect all entries keyed by jid, with agent type info + interface RoomEntry { + agentType: 'claude-code' | 'codex'; + status: 'processing' | 'idle' | 'waiting' | 'inactive'; + elapsedMs: number | null; + pendingMessages: boolean; + pendingTasks: number; + name: string; + meta: ChannelMeta | undefined; + } + const byJid = new Map(); + + for (const snapshot of snapshots) { + const agentType = snapshot.agentType as 'claude-code' | 'codex'; + for (const entry of snapshot.entries) { + const arr = byJid.get(entry.jid) || []; + arr.push({ + agentType, + status: entry.status, + elapsedMs: entry.elapsedMs, + pendingMessages: entry.pendingMessages, + pendingTasks: entry.pendingTasks, + name: entry.name, + meta: channelMetaCache.get(entry.jid), + }); + byJid.set(entry.jid, arr); + } + } + + // Group by category, then render rooms + interface RoomInfo { + jid: string; + name: string; + meta: ChannelMeta | undefined; + agents: RoomEntry[]; + } + const categoryMap = new Map(); + let totalActive = 0; + let totalRooms = 0; + + for (const [jid, agents] of byJid) { + const meta = agents[0]?.meta; + const cat = meta?.category || '기타'; + if (!categoryMap.has(cat)) categoryMap.set(cat, []); + categoryMap.get(cat)!.push({ + jid, + name: formatRoomName( + jid, + meta, + agents.find((agent) => agent.name && agent.name !== jid)?.name, + chatNameByJid.get(jid), + ), + meta, + agents, + }); + totalRooms++; + if (agents.some((a) => a.status === 'processing')) totalActive++; } - // 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; @@ -510,38 +797,36 @@ function buildStatusContent(): string { }); const sections: string[] = []; - let totalActive = 0; - let totalIdle = 0; - let total = 0; - - for (const [catName, catEntries] of sortedCategories) { - catEntries.sort( + for (const [catName, rooms] of sortedCategories) { + rooms.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}`; - }); + const roomLines: string[] = []; + for (const room of rooms) { + // Sort agents: claude-code first, codex second + room.agents.sort((a, b) => + a.agentType === b.agentType ? 0 : a.agentType === 'claude-code' ? -1 : 1, + ); - if (channelMetaCache.size > 0 && catName !== '기타') { - sections.push(`📁 **${catName}**\n${lines.join('\n')}`); - } else { - sections.push(lines.join('\n')); + const agentParts = room.agents.map((a) => { + const icon = STATUS_ICONS[a.status] || '⚪'; + const label = getStatusLabel(a); + const tag = getAgentDisplayName(a.agentType); + return `${tag} ${icon} ${label}`; + }); + roomLines.push(` **${room.name}** — ${agentParts.join(' | ')}`); } - totalActive += catEntries.filter( - (e) => e.status.status === 'processing', - ).length; - totalIdle += catEntries.filter((e) => e.status.status === 'idle').length; - total += catEntries.length; + if (channelMetaCache.size > 0 && catName !== '기타') { + sections.push(`📁 **${catName}**\n${roomLines.join('\n')}`); + } else { + sections.push(roomLines.join('\n')); + } } - const header = `**에이전트 상태** (${ASSISTANT_NAME}) — 활성 ${totalActive} | 대기 ${totalIdle} | 전체 ${total}`; - return `${header}\n\n${sections.join('\n\n')}\n\n_${new Date().toLocaleTimeString('ko-KR')}_`; + const header = `**📊 에이전트 상태** — 활성 ${totalActive} / ${totalRooms}`; + return `${header}\n\n${sections.join('\n\n')}`; } // ── API Usage Fetchers ────────────────────────────────────────── @@ -558,15 +843,29 @@ interface CodexRateLimit { secondary: { usedPercent: number; resetsAt: string | number }; } -async function fetchClaudeUsage(): Promise { - try { - 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; +let usageApiBackoffUntil = 0; - const creds = JSON.parse(fs.readFileSync(credsPath, 'utf-8')); - const token = creds?.claudeAiOauth?.accessToken; +async function fetchClaudeUsage(): Promise { + // Skip if in backoff period (after 429) + if (Date.now() < usageApiBackoffUntil) { + logger.debug('Skipping usage API call (backoff active)'); + return null; + } + + 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', { @@ -575,9 +874,26 @@ async function fetchClaudeUsage(): Promise { 'anthropic-beta': 'oauth-2025-04-20', }, }); - if (!res.ok) return null; + if (!res.ok) { + const body = await res.text().catch(() => ''); + if (res.status === 429) { + // Back off for 10 minutes on rate limit + usageApiBackoffUntil = Date.now() + 600_000; + logger.warn( + { status: 429, retryAfter: res.headers.get('retry-after'), body: body.slice(0, 200) }, + 'Usage API rate limited (429), backing off 10min', + ); + } else { + logger.warn( + { status: res.status, body: body.slice(0, 200) }, + 'Usage API returned non-OK status', + ); + } + return null; + } return (await res.json()) as ClaudeUsageData; - } catch { + } catch (err) { + logger.debug({ err }, 'Usage API fetch failed'); return null; } } @@ -674,76 +990,117 @@ async function fetchCodexUsage(): Promise { async function buildUsageContent(): Promise { const lines: string[] = []; + const activeSnapshots = readStatusSnapshots(STATUS_SNAPSHOT_MAX_AGE_MS); + const hasActiveClaudeWork = activeSnapshots.some( + (snapshot) => + snapshot.agentType === 'claude-code' && + snapshot.entries.some( + (entry) => + entry.status === 'processing' || entry.status === 'waiting', + ), + ); - // Fetch API usage in parallel - const [claudeUsage, codexUsage] = await Promise.all([ - fetchClaudeUsage(), + const [liveClaudeUsage, codexUsage] = await Promise.all([ + hasActiveClaudeWork ? Promise.resolve(null) : fetchClaudeUsage(), fetchCodexUsage(), ]); + const claudeUsage = liveClaudeUsage || cachedClaudeUsageData; + const claudeUsageIsCached = !liveClaudeUsage && !!cachedClaudeUsageData; + if (liveClaudeUsage) { + cachedClaudeUsageData = liveClaudeUsage; + } + + 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[] = []; - // Claude Code usage if (claudeUsage) { - lines.push('☁️ *Claude Code*'); - if (claudeUsage.five_hour) { - // utilization may be fraction (0-1) or percentage (0-100) - const raw = claudeUsage.five_hour.utilization; - const pct = raw > 1 ? Math.round(raw) : Math.round(raw * 100); - lines.push( - `• 5시간: ${usageEmoji(pct)} ${pct}% (리셋: ${formatResetKST(claudeUsage.five_hour.resets_at)})`, - ); - } - if (claudeUsage.seven_day) { - const raw = claudeUsage.seven_day.utilization; - const pct = raw > 1 ? Math.round(raw) : Math.round(raw * 100); - lines.push( - `• 7일: ${usageEmoji(pct)} ${pct}% (리셋: ${formatResetKST(claudeUsage.seven_day.resets_at)})`, - ); - } - lines.push(''); + const h5 = claudeUsage.five_hour; + const d7 = claudeUsage.seven_day; + rows.push({ + name: claudeUsageIsCached ? 'Claude*' : '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) : '', + }); } - // Codex usage if (codexUsage && Array.isArray(codexUsage)) { - lines.push('🤖 *Codex CLI*'); - for (const limit of codexUsage) { - const p = Math.round(limit.primary.usedPercent); - const s = Math.round(limit.secondary.usedPercent); - const name = limit.limitName || limit.limitId || 'Codex'; - lines.push( - `• ${name} 5시간: ${usageEmoji(p)} ${p}% (리셋: ${formatResetKST(limit.primary.resetsAt)})`, - ); - lines.push( - `• ${name} 7일: ${usageEmoji(s)} ${s}% (리셋: ${formatResetKST(limit.secondary.resetsAt)})`, - ); + 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), + }); } - lines.push(''); } - if (!claudeUsage && !codexUsage) { - lines.push('_API 사용량 조회 불가_'); - lines.push(''); + 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('_조회 불가_'); } + if (claudeUsageIsCached) { + lines.push('_* Claude 사용량은 작업 중일 때는 캐시값 유지_'); + } + lines.push(''); - // System resources lines.push('🖥️ *서버*'); const loadAvg = os.loadavg(); const cpuCount = os.cpus().length; - const cpuPct1 = Math.round((loadAvg[0] / cpuCount) * 100); - const cpuPct5 = Math.round((loadAvg[1] / cpuCount) * 100); - const cpuPct15 = Math.round((loadAvg[2] / cpuCount) * 100); - lines.push( - `• CPU: ${usageEmoji(cpuPct1)} ${cpuPct1}% (1m) | ${cpuPct5}% (5m) | ${cpuPct15}% (15m)`, - ); + const cpuPct = Math.round((loadAvg[1] / cpuCount) * 100); const totalMem = os.totalmem(); const usedMem = totalMem - os.freemem(); const memPct = Math.round((usedMem / totalMem) * 100); - lines.push( - `• 메모리: ${usageEmoji(memPct)} ${memPct}% (${(usedMem / 1073741824).toFixed(1)}GB / ${(totalMem / 1073741824).toFixed(1)}GB)`, - ); + const memUsedGB = (usedMem / 1073741824).toFixed(1); + const memTotalGB = (totalMem / 1073741824).toFixed(1); - let diskLine = '• 디스크: 확인 불가'; + let diskPct = 0; + let diskUsedGB = '?'; + let diskTotalGB = '?'; try { const df = execSync('df -B1 / | tail -1', { encoding: 'utf-8', @@ -752,38 +1109,81 @@ async function buildUsageContent(): Promise { const parts = df.split(/\s+/); const diskUsed = parseInt(parts[2], 10); const diskTotal = parseInt(parts[1], 10); - const diskPct = Math.round((diskUsed / diskTotal) * 100); - diskLine = `• 디스크: ${usageEmoji(diskPct)} ${diskPct}% (${(diskUsed / 1073741824).toFixed(1)}GB / ${(diskTotal / 1073741824).toFixed(1)}GB)`; + diskPct = Math.round((diskUsed / diskTotal) * 100); + diskUsedGB = (diskUsed / 1073741824).toFixed(0); + diskTotalGB = (diskTotal / 1073741824).toFixed(0); } catch { /* ignore */ } - lines.push(diskLine); - lines.push(`• 업타임: ${formatElapsed(os.uptime() * 1000)}`); - - return ( - lines.join('\n') + - `\n\n_${new Date().toLocaleTimeString('ko-KR', { hour: '2-digit', minute: '2-digit' })}_` + 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'); } -// ── Dashboard Lifecycle ───────────────────────────────────────── +// ── Unified Dashboard Lifecycle ────────────────────────────────── + +let usageUpdateInProgress = false; + +async function refreshUsageCache(): Promise { + if (usageUpdateInProgress) return; + usageUpdateInProgress = true; + try { + cachedUsageContent = await buildUsageContent(); + } catch { + /* keep previous cache */ + } finally { + usageUpdateInProgress = false; + } +} + +function buildUnifiedDashboard(): string { + const status = buildStatusContent(); + const parts = [status]; + + if (cachedUsageContent) { + parts.push(cachedUsageContent); + } + + parts.push( + `_${new Date().toLocaleTimeString('ko-KR', { hour: '2-digit', minute: '2-digit' })}_`, + ); + return parts.join('\n\n'); +} async function startStatusDashboard(): Promise { if (!STATUS_CHANNEL_ID) return; + const isRenderer = SERVICE_AGENT_TYPE === 'claude-code'; const statusJid = `dc:${STATUS_CHANNEL_ID}`; const findDiscordChannel = () => channels.find((c) => c.name.startsWith('discord') && c.isConnected()); + // Initial usage fetch + if (isRenderer) { + await refreshUsageCache(); + } + const updateStatus = async () => { + writeLocalStatusSnapshot(); + if (!isRenderer) return; + const ch = findDiscordChannel(); if (!ch) return; try { await refreshChannelMeta(); - const content = buildStatusContent(); + const content = buildUnifiedDashboard(); if (statusMessageId && ch.editMessage) { await ch.editMessage(statusJid, statusMessageId, content); @@ -792,58 +1192,29 @@ async function startStatusDashboard(): Promise { if (id) statusMessageId = id; } } catch (err) { - logger.debug({ err }, 'Status dashboard update failed'); + logger.debug({ err }, 'Dashboard update failed'); statusMessageId = null; } }; + // Status updates every 10s setInterval(updateStatus, STATUS_UPDATE_INTERVAL); await updateStatus(); - logger.info({ channelId: STATUS_CHANNEL_ID }, 'Status dashboard started'); + + // Usage cache refreshes every 5min (only on renderer) + if (isRenderer) { + setInterval(refreshUsageCache, USAGE_UPDATE_INTERVAL); + } + + logger.info( + { channelId: STATUS_CHANNEL_ID, isRenderer, agentType: SERVICE_AGENT_TYPE }, + isRenderer ? 'Unified dashboard started' : 'Status snapshot updater started', + ); } -let usageUpdateInProgress = false; - +// Legacy compat — now handled inside startStatusDashboard 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'); + // Usage is now integrated into the unified dashboard } async function startMessageLoop(): Promise { @@ -893,12 +1264,25 @@ async function startMessageLoop(): Promise { } const isMainGroup = group.isMain === true; + const processableGroupMessages = getProcessableMessages( + chatJid, + groupMessages, + channel, + ); + + if (processableGroupMessages.length === 0) { + const lastIgnored = groupMessages[groupMessages.length - 1]; + if (lastIgnored) { + advanceLastAgentCursor(chatJid, lastIgnored.timestamp); + } + continue; + } // --- 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( + const allFromBots = processableGroupMessages.every( (m) => m.is_from_me || !!m.is_bot_message, ); if (allFromBots) { @@ -918,7 +1302,7 @@ async function startMessageLoop(): Promise { // --- Session command interception (message loop) --- // Scan ALL messages in the batch for a session command. - const loopCmdMsg = groupMessages.find( + const loopCmdMsg = processableGroupMessages.find( (m) => extractSessionCommand(m.content, TRIGGER_PATTERN) !== null, ); @@ -950,7 +1334,7 @@ async function startMessageLoop(): Promise { // context when a trigger eventually arrives. if (needsTrigger) { const allowlistCfg = loadSenderAllowlist(); - const hasTrigger = groupMessages.some( + const hasTrigger = processableGroupMessages.some( (m) => TRIGGER_PATTERN.test(m.content.trim()) && (m.is_from_me || @@ -966,8 +1350,11 @@ async function startMessageLoop(): Promise { lastAgentTimestamp[chatJid] || '', ASSISTANT_NAME, ); + const processablePending = getProcessableMessages(chatJid, allPending, channel); const messagesToSend = - allPending.length > 0 ? allPending : groupMessages; + processablePending.length > 0 + ? processablePending + : processableGroupMessages; const formatted = formatMessages(messagesToSend, TIMEZONE); if (queue.sendMessage(chatJid, formatted)) { @@ -975,9 +1362,10 @@ async function startMessageLoop(): Promise { { chatJid, count: messagesToSend.length }, 'Piped messages to active agent', ); - lastAgentTimestamp[chatJid] = - messagesToSend[messagesToSend.length - 1].timestamp; - saveState(); + advanceLastAgentCursor( + chatJid, + messagesToSend[messagesToSend.length - 1].timestamp, + ); // Show typing indicator while the agent processes the piped message channel .setTyping?.(chatJid, true) @@ -1004,13 +1392,20 @@ async function startMessageLoop(): Promise { function recoverPendingMessages(): void { for (const [chatJid, group] of Object.entries(registeredGroups)) { const sinceTimestamp = lastAgentTimestamp[chatJid] || ''; - const pending = getMessagesSince(chatJid, sinceTimestamp, ASSISTANT_NAME); + const rawPending = getMessagesSince(chatJid, sinceTimestamp, ASSISTANT_NAME); + const recoveryChannel = findChannel(channels, chatJid); + const pending = getProcessableMessages(chatJid, rawPending, recoveryChannel ?? undefined); if (pending.length > 0) { logger.info( { group: group.name, pendingCount: pending.length }, 'Recovery: found unprocessed messages', ); queue.enqueueMessageCheck(chatJid, group.folder); + } else if (rawPending.length > 0) { + advanceLastAgentCursor( + chatJid, + rawPending[rawPending.length - 1].timestamp, + ); } } } @@ -1121,7 +1516,7 @@ async function main(): Promise { queue.setProcessMessagesFn(processGroupMessages); recoverPendingMessages(); // Purge old messages in status channel before creating fresh dashboards - if (STATUS_CHANNEL_ID) { + if (STATUS_CHANNEL_ID && SERVICE_AGENT_TYPE === 'claude-code') { const statusJid = `dc:${STATUS_CHANNEL_ID}`; const ch = channels.find( (c) => c.name.startsWith('discord') && c.isConnected() && c.purgeChannel, diff --git a/src/session-recovery.test.ts b/src/session-recovery.test.ts new file mode 100644 index 0000000..0723595 --- /dev/null +++ b/src/session-recovery.test.ts @@ -0,0 +1,34 @@ +import { describe, expect, it } from 'vitest'; + +import { shouldResetSessionOnAgentFailure } from './session-recovery.js'; + +describe('shouldResetSessionOnAgentFailure', () => { + it('matches many-image dimension limit errors', () => { + expect( + shouldResetSessionOnAgentFailure({ + result: + 'An image in the conversation exceeds the dimension limit for many-image requests (2000px). Start a new session with fewer images.', + error: undefined, + }), + ).toBe(true); + }); + + it('matches the error field too', () => { + expect( + shouldResetSessionOnAgentFailure({ + result: null, + error: + 'fatal: An image in the conversation exceeds the dimension limit for many-image requests (2000px). Start a new session with fewer images.', + }), + ).toBe(true); + }); + + it('does not match unrelated agent failures', () => { + expect( + shouldResetSessionOnAgentFailure({ + result: null, + error: 'Claude Code process exited with code 1', + }), + ).toBe(false); + }); +}); diff --git a/src/session-recovery.ts b/src/session-recovery.ts new file mode 100644 index 0000000..b1b5f15 --- /dev/null +++ b/src/session-recovery.ts @@ -0,0 +1,26 @@ +import type { AgentOutput } from './agent-runner.js'; + +const SESSION_RESET_PATTERNS = [ + /An image in the conversation exceeds the dimension limit for many-image requests \(2000px\)\./i, + /Start a new session with fewer images\./i, +]; + +function toText(value: string | object | null | undefined): string[] { + if (!value) return []; + if (typeof value === 'string') return [value]; + + try { + return [JSON.stringify(value)]; + } catch { + return []; + } +} + +export function shouldResetSessionOnAgentFailure( + output: Pick, +): boolean { + const texts = [...toText(output.result), ...toText(output.error)]; + return texts.some((text) => + SESSION_RESET_PATTERNS.some((pattern) => pattern.test(text)), + ); +} diff --git a/src/status-dashboard.ts b/src/status-dashboard.ts new file mode 100644 index 0000000..3ca4b91 --- /dev/null +++ b/src/status-dashboard.ts @@ -0,0 +1,65 @@ +import fs from 'fs'; +import path from 'path'; + +import { CACHE_DIR } from './config.js'; +import type { GroupStatus } from './group-queue.js'; +import type { AgentType } from './types.js'; + +export interface StatusSnapshotEntry { + jid: string; + name: string; + folder: string; + agentType: AgentType; + status: GroupStatus['status']; + elapsedMs: number | null; + pendingMessages: boolean; + pendingTasks: number; +} + +export interface StatusSnapshot { + agentType: AgentType; + assistantName: string; + updatedAt: string; + entries: StatusSnapshotEntry[]; +} + +const STATUS_SNAPSHOT_DIR = path.join(CACHE_DIR, 'status-dashboard'); + +export function writeStatusSnapshot(snapshot: StatusSnapshot): void { + fs.mkdirSync(STATUS_SNAPSHOT_DIR, { recursive: true }); + const targetPath = path.join( + STATUS_SNAPSHOT_DIR, + `${snapshot.agentType}.json`, + ); + const tempPath = `${targetPath}.tmp`; + fs.writeFileSync(tempPath, JSON.stringify(snapshot, null, 2)); + fs.renameSync(tempPath, targetPath); +} + +export function readStatusSnapshots(maxAgeMs: number): StatusSnapshot[] { + if (!fs.existsSync(STATUS_SNAPSHOT_DIR)) return []; + + const now = Date.now(); + const snapshots: StatusSnapshot[] = []; + + for (const entry of fs.readdirSync(STATUS_SNAPSHOT_DIR)) { + if (!entry.endsWith('.json')) continue; + const snapshotPath = path.join(STATUS_SNAPSHOT_DIR, entry); + + try { + const raw = fs.readFileSync(snapshotPath, 'utf-8'); + const parsed = JSON.parse(raw) as StatusSnapshot; + if (!parsed.updatedAt || !parsed.agentType || !Array.isArray(parsed.entries)) + continue; + + const ageMs = now - new Date(parsed.updatedAt).getTime(); + if (Number.isNaN(ageMs) || ageMs > maxAgeMs) continue; + + snapshots.push(parsed); + } catch { + continue; + } + } + + return snapshots; +}