From 779d57c392b293829d41cdea29aeb26be47d9420 Mon Sep 17 00:00:00 2001 From: Eyejoker Date: Thu, 19 Mar 2026 02:57:14 +0900 Subject: [PATCH] Stabilize paired-room orchestration and Codex app-server flow --- prompts/claude-paired-room.md | 20 + prompts/claude-platform.md | 46 ++ prompts/codex-paired-room.md | 18 + prompts/codex-platform.md | 27 + runners/agent-runner/src/index.ts | 11 - runners/codex-runner/package.json | 2 +- runners/codex-runner/pnpm-lock.yaml | 119 +++++ runners/codex-runner/src/app-server-client.ts | 459 ++++++++++++++++ runners/codex-runner/src/app-server-state.ts | 174 ++++++ runners/codex-runner/src/index.ts | 505 ++++++++++++++---- .../test/app-server-state.test.ts | 83 +++ setup/register.test.ts | 10 +- setup/register.ts | 10 +- src/agent-runner.test.ts | 38 ++ src/agent-runner.ts | 49 +- src/channels/discord.test.ts | 44 ++ src/channels/discord.ts | 12 + src/db.test.ts | 48 +- src/db.ts | 139 +++-- src/group-queue.test.ts | 23 + src/group-queue.ts | 34 +- src/message-runtime.test.ts | 461 ++++++++++++++++ src/message-runtime.ts | 137 ++++- src/platform-prompts.test.ts | 58 ++ src/platform-prompts.ts | 60 +++ src/session-commands.test.ts | 23 + src/session-commands.ts | 21 +- src/session-recovery.test.ts | 34 ++ src/session-recovery.ts | 26 + src/task-scheduler.ts | 3 + src/types.ts | 2 + vitest.config.ts | 6 +- 32 files changed, 2517 insertions(+), 185 deletions(-) create mode 100644 prompts/claude-paired-room.md create mode 100644 prompts/claude-platform.md create mode 100644 prompts/codex-paired-room.md create mode 100644 prompts/codex-platform.md create mode 100644 runners/codex-runner/pnpm-lock.yaml create mode 100644 runners/codex-runner/src/app-server-client.ts create mode 100644 runners/codex-runner/src/app-server-state.ts create mode 100644 runners/codex-runner/test/app-server-state.test.ts create mode 100644 src/message-runtime.test.ts create mode 100644 src/platform-prompts.test.ts create mode 100644 src/platform-prompts.ts create mode 100644 src/session-recovery.test.ts create mode 100644 src/session-recovery.ts diff --git a/prompts/claude-paired-room.md b/prompts/claude-paired-room.md new file mode 100644 index 0000000..fc002d2 --- /dev/null +++ b/prompts/claude-paired-room.md @@ -0,0 +1,20 @@ +# Claude Paired Room Rules + +This room has both Claude and Codex. +Both of you can read the same room conversation and respond in the same thread. + +Your default role is review, test planning, verification, and risk checking. + +Discussion and design debate are shared responsibilities. You can challenge Codex, refine its approach, and propose alternatives when they are stronger. + +Keep coordination with Codex public by default. Use `` only for content that truly needs to stay hidden from the room. + +When Codex is already implementing, prefer: +- clarifying requirements +- surfacing edge cases and regressions +- proposing focused tests +- reviewing results and calling out risks + +Let Codex take the lead on implementation in most cases. + +You can still implement when the user explicitly asks you to, when Codex is blocked, or when a small targeted patch is the fastest way to verify a point. diff --git a/prompts/claude-platform.md b/prompts/claude-platform.md new file mode 100644 index 0000000..9ce497f --- /dev/null +++ b/prompts/claude-platform.md @@ -0,0 +1,46 @@ +# Claude Platform Rules + +You are Andy, a personal assistant. + +## Communication + +Your output is sent directly to the user or Discord group. + +You also have `mcp__nanoclaw__send_message`, which sends a message immediately while you are still working. Use it when you want to acknowledge a request before starting longer work. + +### Internal thoughts + +Use `` only for genuinely hidden content. + +If part of your output is internal reasoning rather than something for the user, wrap it in `` tags: + +```text +Compiled all three reports, ready to summarize. + +Here are the key findings from the research... +``` + +Text inside `` tags is logged but not sent to the user. + +Prefer public replies for coordination, status updates, review comments, and anything Codex or the user should react to. + +### Sub-agents and teammates + +When working as a sub-agent or teammate, only use `send_message` if the main agent explicitly asked you to. + +## Memory + +The group folder may contain a `conversations/` directory with searchable history from earlier sessions. Use it when you need prior context. + +When you learn something important: +- Create files for structured data when that is genuinely useful +- Split files larger than 500 lines into smaller folders or documents +- Keep an index if you start building a larger memory structure + +## Message formatting + +Do not use markdown headings in chat replies. Keep messages clean and readable for Discord. + +- Use concise paragraphs or simple lists +- Use fenced code blocks when showing code +- Prefer plain links over markdown link syntax diff --git a/prompts/codex-paired-room.md b/prompts/codex-paired-room.md new file mode 100644 index 0000000..99cdfb8 --- /dev/null +++ b/prompts/codex-paired-room.md @@ -0,0 +1,18 @@ +# Codex Paired Room Rules + +This room has both Claude and Codex. +Both of you can read the same room conversation and respond in the same thread. + +Your default role is implementation, debugging, command execution, and concrete code changes. + +Take the lead on implementation in this room unless the user explicitly redirects the work. + +Discussion and design debate are shared responsibilities. Engage with Claude critically and evaluate its feedback on the merits. + +Treat Claude's feedback as input to inspect, test, and reason through, not as something to accept automatically. + +When Claude is already reviewing or testing, prefer: +- making the code change +- running commands and checks +- narrowing the bug or failure +- reporting concrete results back to the room diff --git a/prompts/codex-platform.md b/prompts/codex-platform.md new file mode 100644 index 0000000..1cb9c86 --- /dev/null +++ b/prompts/codex-platform.md @@ -0,0 +1,27 @@ +# Codex Platform Rules + +You are 코덱스, a participant in a Discord chat. + +## Core rules + +- Respond directly to messages. Do not provide reply suggestions or draft responses for someone else to send. +- Respond in Korean. +- When coding, debugging, or file work is needed, do it directly. + +## Communication + +Your output is sent directly to the Discord group. + +- Keep answers concise unless more detail is genuinely needed +- Give conclusions and concrete next steps, not hidden reasoning +- Use code blocks for commands or code when helpful +- Do not claim you will keep watching, monitor later, report back later, or continue tracking unless you actually scheduled a NanoClaw task with `schedule_task` +- If no task was scheduled, do not imply that background tracking is active. If future follow-up is needed, tell the user to ping you again or explicitly ask for scheduling +- When you do schedule background follow-up, mention that it was scheduled. Include the task ID only when it is useful for later reference + +## Working style + +- Prefer reading the current workspace before making assumptions +- Modify only what is needed for the task +- Verify changes when you can instead of claiming they should work +- For CI/status/watch requests that require future follow-up, prefer `schedule_task` over leaving the chat session idle diff --git a/runners/agent-runner/src/index.ts b/runners/agent-runner/src/index.ts index 6600a8b..b962a85 100644 --- a/runners/agent-runner/src/index.ts +++ b/runners/agent-runner/src/index.ts @@ -62,7 +62,6 @@ interface SDKUserMessage { // Paths configurable via env vars (defaults to container paths for backwards compat) const GROUP_DIR = process.env.NANOCLAW_GROUP_DIR || '/workspace/group'; const IPC_DIR = process.env.NANOCLAW_IPC_DIR || '/workspace/ipc'; -const GLOBAL_DIR = process.env.NANOCLAW_GLOBAL_DIR || '/workspace/global'; const EXTRA_BASE = process.env.NANOCLAW_EXTRA_DIR || '/workspace/extra'; // Optional: override cwd (agent works in this directory instead of GROUP_DIR) const WORK_DIR = process.env.NANOCLAW_WORK_DIR || ''; @@ -450,13 +449,6 @@ async function runQuery( let messageCount = 0; let resultCount = 0; - // Load global CLAUDE.md as additional system context (shared across all groups) - const globalClaudeMdPath = path.join(GLOBAL_DIR, 'CLAUDE.md'); - let globalClaudeMd: string | undefined; - if (!containerInput.isMain && fs.existsSync(globalClaudeMdPath)) { - globalClaudeMd = fs.readFileSync(globalClaudeMdPath, 'utf-8'); - } - // Discover additional directories const extraDirs: string[] = []; @@ -507,9 +499,6 @@ async function runQuery( additionalDirectories: extraDirs.length > 0 ? extraDirs : undefined, resume: sessionId, resumeSessionAt: resumeAt, - systemPrompt: globalClaudeMd - ? { type: 'preset' as const, preset: 'claude_code' as const, append: globalClaudeMd } - : undefined, allowedTools: [ 'Bash', 'Read', 'Write', 'Edit', 'Glob', 'Grep', diff --git a/runners/codex-runner/package.json b/runners/codex-runner/package.json index 51cd5c2..a4c2d1d 100644 --- a/runners/codex-runner/package.json +++ b/runners/codex-runner/package.json @@ -9,7 +9,7 @@ "start": "node dist/index.js" }, "dependencies": { - "@openai/codex-sdk": "^0.114.0" + "@openai/codex-sdk": "^0.115.0" }, "devDependencies": { "@types/node": "^22.10.7", diff --git a/runners/codex-runner/pnpm-lock.yaml b/runners/codex-runner/pnpm-lock.yaml new file mode 100644 index 0000000..e397a43 --- /dev/null +++ b/runners/codex-runner/pnpm-lock.yaml @@ -0,0 +1,119 @@ +lockfileVersion: '9.0' + +settings: + autoInstallPeers: true + excludeLinksFromLockfile: false + +importers: + + .: + dependencies: + '@openai/codex-sdk': + specifier: ^0.115.0 + version: 0.115.0 + devDependencies: + '@types/node': + specifier: ^22.10.7 + version: 22.19.15 + typescript: + specifier: ^5.7.3 + version: 5.9.3 + +packages: + + '@openai/codex-sdk@0.115.0': + resolution: {integrity: sha512-BPoPhim0uUm3rzugY7JFaFJ+rPG/wMRhvKNFii//Dp3kTb+gFy3jrip7ijXqhswsnqXM3nwTiv7kYHt1+TMUPg==} + engines: {node: '>=18'} + + '@openai/codex@0.115.0': + resolution: {integrity: sha512-uu689DHUzvuPcb39hJ+7fqy++TAvY32w9VggDpcz3HS0Sx0WadWoAPPcMK547P2T6AqfMsAtA8kspkR3tqErOg==} + engines: {node: '>=16'} + hasBin: true + + '@openai/codex@0.115.0-darwin-arm64': + resolution: {integrity: sha512-wTWV+YlDTL0y0mL+FMmbzhilm+dPkbIZTe/lH3LwBzcEMhgSGLsPdZLfAiMND4wFE21HS7H0pjMogNQEMLQmqg==} + engines: {node: '>=16'} + cpu: [arm64] + os: [darwin] + + '@openai/codex@0.115.0-darwin-x64': + resolution: {integrity: sha512-EPzgymU4CFp83qjv29wXFwhWib3zC8g6SLTJGh2OpcJiOYyLY0bO53FB+QchL1jC9gm1uLyD5j5F3ddI0AbjIg==} + engines: {node: '>=16'} + cpu: [x64] + os: [darwin] + + '@openai/codex@0.115.0-linux-arm64': + resolution: {integrity: sha512-40+SCyI+LvVx/iX30qH7dTQzWt9MZxDaK2E6YRB4hoYej5UALrhkFNzweHa5uvqvhmqGZCuagZOYB8285eGCgw==} + engines: {node: '>=16'} + cpu: [arm64] + os: [linux] + + '@openai/codex@0.115.0-linux-x64': + resolution: {integrity: sha512-+6eRd2p4KMrhQvuF7XpjYIdCA2Ok2LbhOq+ywZdLpHbmwQ/Yynd0gJ/Q90xCeo2vloCwl9WsbsiVVSahG5FyWg==} + engines: {node: '>=16'} + cpu: [x64] + os: [linux] + + '@openai/codex@0.115.0-win32-arm64': + resolution: {integrity: sha512-yaYhQ0kPL9Kithmrr1vh5A4c+gqAMhMZeA/htTtkpWW8RQkmwgcYYpX/Ky2cEzu0w51pvxQQy63LgHftc+pg1Q==} + engines: {node: '>=16'} + cpu: [arm64] + os: [win32] + + '@openai/codex@0.115.0-win32-x64': + resolution: {integrity: sha512-wtYf2HJCB+p+Uj7o1W08fRgZMzKCVGRQ4YdSfLRNfF4wwPqX5XsK9blsv7brukn5J9lclYxagiR6qGvbfHbD7w==} + engines: {node: '>=16'} + cpu: [x64] + os: [win32] + + '@types/node@22.19.15': + resolution: {integrity: sha512-F0R/h2+dsy5wJAUe3tAU6oqa2qbWY5TpNfL/RGmo1y38hiyO1w3x2jPtt76wmuaJI4DQnOBu21cNXQ2STIUUWg==} + + typescript@5.9.3: + resolution: {integrity: sha512-jl1vZzPDinLr9eUt3J/t7V6FgNEw9QjvBPdysz9KfQDD41fQrC2Y4vKQdiaUpFT4bXlb1RHhLpp8wtm6M5TgSw==} + engines: {node: '>=14.17'} + hasBin: true + + undici-types@6.21.0: + resolution: {integrity: sha512-iwDZqg0QAGrg9Rav5H4n0M64c3mkR59cJ6wQp+7C4nI0gsmExaedaYLNO44eT4AtBBwjbTiGPMlt2Md0T9H9JQ==} + +snapshots: + + '@openai/codex-sdk@0.115.0': + dependencies: + '@openai/codex': 0.115.0 + + '@openai/codex@0.115.0': + optionalDependencies: + '@openai/codex-darwin-arm64': '@openai/codex@0.115.0-darwin-arm64' + '@openai/codex-darwin-x64': '@openai/codex@0.115.0-darwin-x64' + '@openai/codex-linux-arm64': '@openai/codex@0.115.0-linux-arm64' + '@openai/codex-linux-x64': '@openai/codex@0.115.0-linux-x64' + '@openai/codex-win32-arm64': '@openai/codex@0.115.0-win32-arm64' + '@openai/codex-win32-x64': '@openai/codex@0.115.0-win32-x64' + + '@openai/codex@0.115.0-darwin-arm64': + optional: true + + '@openai/codex@0.115.0-darwin-x64': + optional: true + + '@openai/codex@0.115.0-linux-arm64': + optional: true + + '@openai/codex@0.115.0-linux-x64': + optional: true + + '@openai/codex@0.115.0-win32-arm64': + optional: true + + '@openai/codex@0.115.0-win32-x64': + optional: true + + '@types/node@22.19.15': + dependencies: + undici-types: 6.21.0 + + typescript@5.9.3: {} + + undici-types@6.21.0: {} diff --git a/runners/codex-runner/src/app-server-client.ts b/runners/codex-runner/src/app-server-client.ts new file mode 100644 index 0000000..11693c9 --- /dev/null +++ b/runners/codex-runner/src/app-server-client.ts @@ -0,0 +1,459 @@ +import { spawn, type ChildProcessWithoutNullStreams } from 'child_process'; +import { createRequire } from 'module'; +import path from 'path'; + +import { + createInitialAppServerTurnState, + getAppServerTurnResult, + isAppServerTurnFinished, + reduceAppServerTurnState, + type AppServerTurnEvent, + type AppServerTurnState, +} from './app-server-state.js'; + +export interface AppServerInputItemText { + type: 'text'; + text: string; +} + +export interface AppServerInputItemLocalImage { + type: 'localImage'; + path: string; +} + +export type AppServerInputItem = + | AppServerInputItemText + | AppServerInputItemLocalImage; + +export interface CodexAppServerThreadOptions { + cwd: string; + model?: string; +} + +export interface CodexAppServerTurnOptions { + cwd: string; + model?: string; + effort?: string; + onProgress?: (message: string) => void; +} + +export interface CodexAppServerTurnResult { + state: AppServerTurnState; + result: string | null; +} + +interface JsonRpcResponse { + id: number; + result?: unknown; + error?: { + code?: number; + message?: string; + data?: unknown; + }; +} + +interface JsonRpcNotification { + method: string; + params?: Record; +} + +interface JsonRpcServerRequest extends JsonRpcNotification { + id: number; +} + +interface PendingRequest { + method: string; + resolve: (value: unknown) => void; + reject: (reason?: unknown) => void; +} + +interface ActiveTurn { + threadId: string; + state: AppServerTurnState; + onProgress?: (message: string) => void; + resolve: (value: CodexAppServerTurnResult) => void; + reject: (reason?: unknown) => void; +} + +export interface CodexAppServerClientOptions { + cwd: string; + env?: NodeJS.ProcessEnv; + log: (message: string) => void; +} + +export class CodexAppServerClient { + private readonly cwd: string; + private readonly env: NodeJS.ProcessEnv; + private readonly log: (message: string) => void; + private readonly pending = new Map(); + private readonly require = createRequire(import.meta.url); + private nextId = 1; + private stdoutBuffer = ''; + private activeTurn: ActiveTurn | null = null; + private proc: ChildProcessWithoutNullStreams | null = null; + + constructor(options: CodexAppServerClientOptions) { + this.cwd = options.cwd; + this.env = options.env || process.env; + this.log = options.log; + } + + async start(): Promise { + if (this.proc) return; + + const codexPackagePath = this.require.resolve('@openai/codex/package.json'); + const codexBin = path.join(path.dirname(codexPackagePath), 'bin', 'codex.js'); + + this.proc = spawn(process.execPath, [codexBin, 'app-server'], { + cwd: this.cwd, + env: this.env, + stdio: ['pipe', 'pipe', 'pipe'], + }); + + this.proc.stdout.setEncoding('utf8'); + this.proc.stdout.on('data', (chunk: string) => { + this.stdoutBuffer += chunk; + const lines = this.stdoutBuffer.split('\n'); + this.stdoutBuffer = lines.pop() || ''; + for (const line of lines) { + const trimmed = line.trim(); + if (!trimmed) continue; + this.handleStdoutLine(trimmed); + } + }); + + this.proc.stderr.setEncoding('utf8'); + this.proc.stderr.on('data', (chunk: string) => { + for (const line of chunk.split('\n')) { + const trimmed = line.trim(); + if (!trimmed) continue; + this.log(`[app-server] ${trimmed}`); + } + }); + + this.proc.on('close', (code) => { + const error = new Error( + `Codex app-server exited with code ${code ?? 'unknown'}`, + ); + this.rejectAll(error); + }); + + this.proc.on('error', (error) => { + this.rejectAll(error); + }); + + await this.request('initialize', { + clientInfo: { + name: 'nanoclaw_codex_runner', + title: 'NanoClaw Codex Runner', + version: '1.0.0', + }, + capabilities: { + experimentalApi: true, + optOutNotificationMethods: [ + 'item/agentMessage/delta', + 'item/plan/delta', + 'item/reasoning/textDelta', + 'item/reasoning/summaryTextDelta', + 'item/reasoning/summaryPartAdded', + ], + }, + }); + this.notify('initialized', {}); + } + + async close(): Promise { + if (!this.proc) return; + const proc = this.proc; + this.proc = null; + try { + proc.kill('SIGTERM'); + } catch { + /* ignore */ + } + } + + async startOrResumeThread( + sessionId: string | undefined, + options: CodexAppServerThreadOptions, + ): Promise { + const params = { + cwd: options.cwd, + model: options.model, + approvalPolicy: 'never', + sandbox: 'danger-full-access', + serviceName: 'nanoclaw', + }; + + const result = sessionId + ? await this.request('thread/resume', { + threadId: sessionId, + ...params, + }) + : await this.request('thread/start', params); + + const thread = (result as { thread?: { id?: string } }).thread; + if (!thread?.id) { + throw new Error('Codex app-server did not return a thread id.'); + } + return thread.id; + } + + async startTurn( + threadId: string, + input: AppServerInputItem[], + options: CodexAppServerTurnOptions, + ): Promise<{ + turnId: string; + steer: (nextInput: AppServerInputItem[]) => Promise; + interrupt: () => Promise; + wait: () => Promise; + }> { + if (this.activeTurn) { + throw new Error('A Codex app-server turn is already active.'); + } + + const turnPromise = new Promise((resolve, reject) => { + this.activeTurn = { + threadId, + state: createInitialAppServerTurnState(), + onProgress: options.onProgress, + resolve, + reject, + }; + }); + + let turnId = ''; + try { + const response = (await this.request('turn/start', { + threadId, + input, + cwd: options.cwd, + approvalPolicy: 'never', + sandboxPolicy: { + type: 'dangerFullAccess', + networkAccess: true, + }, + model: options.model, + effort: options.effort, + summary: 'concise', + })) as { turn?: { id?: string; status?: string } }; + + turnId = response.turn?.id || ''; + if (!turnId) { + throw new Error('Codex app-server did not return a turn id.'); + } + + const activeTurn = this.activeTurn as ActiveTurn | null; + if (activeTurn !== null) { + activeTurn.state = reduceAppServerTurnState(activeTurn.state, { + method: 'turn/started', + params: { + turn: { + id: turnId, + status: response.turn?.status || 'inProgress', + }, + }, + }); + } + } catch (error) { + this.activeTurn = null; + throw error; + } + + return { + turnId, + steer: async (nextInput) => { + await this.request('turn/steer', { + threadId, + input: nextInput, + expectedTurnId: turnId, + }); + }, + interrupt: async () => { + await this.request('turn/interrupt', { + threadId, + turnId, + }); + }, + wait: async () => turnPromise, + }; + } + + async startCompaction(threadId: string): Promise { + if (this.activeTurn) { + throw new Error('A Codex app-server turn is already active.'); + } + + const turnPromise = new Promise((resolve, reject) => { + this.activeTurn = { + threadId, + state: createInitialAppServerTurnState(), + resolve, + reject, + }; + }); + + try { + await this.request('thread/compact/start', { threadId }); + } catch (error) { + this.activeTurn = null; + throw error; + } + + return turnPromise; + } + + private handleStdoutLine(line: string): void { + let message: JsonRpcResponse | JsonRpcNotification | JsonRpcServerRequest; + try { + message = JSON.parse(line); + } catch { + this.log(`[app-server] non-JSON stdout: ${line}`); + return; + } + + if ( + typeof (message as JsonRpcResponse).id === 'number' && + ('result' in message || 'error' in message) && + !('method' in message) + ) { + this.handleResponse(message as JsonRpcResponse); + return; + } + + if ( + typeof (message as JsonRpcServerRequest).id === 'number' && + typeof (message as JsonRpcServerRequest).method === 'string' + ) { + this.handleServerRequest(message as JsonRpcServerRequest); + return; + } + + if (typeof (message as JsonRpcNotification).method === 'string') { + this.handleNotification(message as JsonRpcNotification); + } + } + + private handleResponse(message: JsonRpcResponse): void { + const pending = this.pending.get(message.id); + if (!pending) return; + this.pending.delete(message.id); + + if (message.error) { + pending.reject( + new Error( + message.error.message || + `${pending.method} failed with JSON-RPC error ${message.error.code ?? 'unknown'}`, + ), + ); + return; + } + + pending.resolve(message.result); + } + + private handleServerRequest(message: JsonRpcServerRequest): void { + if (message.method.endsWith('/requestApproval')) { + this.respond(message.id, 'acceptForSession'); + return; + } + + this.respondError( + message.id, + -32601, + `NanoClaw does not handle server request ${message.method}`, + ); + } + + 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, + ); + + if (!isAppServerTurnFinished(this.activeTurn.state)) { + return; + } + + const activeTurn = this.activeTurn; + this.activeTurn = null; + activeTurn.resolve({ + state: activeTurn.state, + result: getAppServerTurnResult(activeTurn.state), + }); + } + + private rejectAll(error: unknown): void { + for (const pending of this.pending.values()) { + pending.reject(error); + } + this.pending.clear(); + + if (this.activeTurn) { + const activeTurn = this.activeTurn; + this.activeTurn = null; + activeTurn.reject(error); + } + } + + private request(method: string, params?: unknown): Promise { + const id = this.nextId++; + const payload = { + jsonrpc: '2.0', + id, + method, + params, + }; + + return new Promise((resolve, reject) => { + this.pending.set(id, { method, resolve, reject }); + this.write(payload); + }); + } + + private notify(method: string, params?: unknown): void { + this.write({ + jsonrpc: '2.0', + method, + params, + }); + } + + private respond(id: number, result: unknown): void { + this.write({ + jsonrpc: '2.0', + id, + result, + }); + } + + private respondError(id: number, code: number, message: string): void { + this.write({ + jsonrpc: '2.0', + id, + error: { code, message }, + }); + } + + private write(message: Record): void { + if (!this.proc?.stdin.writable) { + throw new Error('Codex app-server stdin is not writable.'); + } + this.proc.stdin.write(JSON.stringify(message) + '\n'); + } +} diff --git a/runners/codex-runner/src/app-server-state.ts b/runners/codex-runner/src/app-server-state.ts new file mode 100644 index 0000000..97c0b86 --- /dev/null +++ b/runners/codex-runner/src/app-server-state.ts @@ -0,0 +1,174 @@ +export interface AppServerAgentMessageItem { + type: 'agentMessage'; + text?: string | null; + phase?: 'commentary' | 'final_answer' | string | null; +} + +export interface AppServerContextCompactionItem { + type: 'contextCompaction'; + id?: string; +} + +export type AppServerItem = + | AppServerAgentMessageItem + | AppServerContextCompactionItem + | { + type: string; + [key: string]: unknown; + }; + +export interface AppServerTurnState { + turnId?: string; + status: 'pending' | 'inProgress' | 'completed' | 'failed' | 'interrupted'; + finalAnswer: string | null; + latestAgentMessage: string | null; + errorMessage: string | null; + compactionCompleted: boolean; +} + +export type AppServerTurnEvent = + | { + method: 'turn/started'; + params?: { turn?: { id?: string | null; status?: string | null } }; + } + | { + method: 'turn/completed'; + params?: { + turn?: { + id?: string | null; + status?: string | null; + error?: + | { message?: string | null } + | string + | null; + }; + }; + } + | { + method: 'item/completed'; + params?: { item?: AppServerItem | null }; + } + | { + method: 'error'; + params?: { + error?: { + message?: string | null; + codexErrorInfo?: { + httpStatusCode?: number | null; + type?: string | null; + } | null; + } | null; + }; + }; + +export function createInitialAppServerTurnState(): AppServerTurnState { + return { + status: 'pending', + finalAnswer: null, + latestAgentMessage: null, + errorMessage: null, + compactionCompleted: false, + }; +} + +export function reduceAppServerTurnState( + state: AppServerTurnState, + event: AppServerTurnEvent, +): AppServerTurnState { + if (event.method === 'turn/started') { + return { + ...state, + turnId: event.params?.turn?.id || state.turnId, + status: 'inProgress', + }; + } + + if (event.method === 'item/completed') { + const item = event.params?.item; + if (!item) return state; + + if (item.type === 'agentMessage') { + const text = + typeof item.text === 'string' && item.text.trim().length > 0 + ? item.text + : null; + if (!text) return state; + + if (item.phase === 'final_answer') { + return { + ...state, + finalAnswer: text, + latestAgentMessage: text, + }; + } + + return { + ...state, + latestAgentMessage: text, + }; + } + + if (item.type === 'contextCompaction') { + return { + ...state, + compactionCompleted: true, + }; + } + + return state; + } + + if (event.method === 'error') { + const error = event.params?.error; + const message = + typeof error?.message === 'string' && error.message.trim().length > 0 + ? error.message.trim() + : 'Codex app-server turn failed.'; + const httpStatusCode = error?.codexErrorInfo?.httpStatusCode; + + return { + ...state, + errorMessage: + typeof httpStatusCode === 'number' + ? `${message} (HTTP ${httpStatusCode})` + : message, + }; + } + + if (event.method === 'turn/completed') { + const turn = event.params?.turn; + const status = + turn?.status === 'completed' || + turn?.status === 'failed' || + turn?.status === 'interrupted' + ? turn.status + : 'completed'; + const turnError = + typeof turn?.error === 'string' + ? turn.error + : turn?.error?.message || null; + + return { + ...state, + turnId: turn?.id || state.turnId, + status, + errorMessage: turnError || state.errorMessage, + }; + } + + return state; +} + +export function isAppServerTurnFinished(state: AppServerTurnState): boolean { + return ( + state.status === 'completed' || + state.status === 'failed' || + state.status === 'interrupted' + ); +} + +export function getAppServerTurnResult( + state: AppServerTurnState, +): string | null { + return state.finalAnswer; +} diff --git a/runners/codex-runner/src/index.ts b/runners/codex-runner/src/index.ts index 710e2a7..d161e6b 100644 --- a/runners/codex-runner/src/index.ts +++ b/runners/codex-runner/src/index.ts @@ -1,9 +1,8 @@ /** - * NanoClaw Codex Runner (SDK mode) + * NanoClaw Codex Runner * - * Uses @openai/codex-sdk which wraps `codex exec`. This ensures complete - * task execution per turn — the agent finishes all work before responding, - * unlike app-server mode which can end turns prematurely. + * Default runtime is Codex app-server, with SDK fallback available via + * CODEX_RUNTIME=sdk or automatic fallback when app-server startup fails. * * Input protocol: * Stdin: Full ContainerInput JSON (read until EOF) @@ -14,15 +13,20 @@ * Each result is wrapped in OUTPUT_START_MARKER / OUTPUT_END_MARKER pairs. */ -import { Codex, type Thread, type UserInput, type ThreadOptions } from '@openai/codex-sdk'; +import { Codex, type Thread, type ThreadOptions, type UserInput } from '@openai/codex-sdk'; import fs from 'fs'; import path from 'path'; +import { + CodexAppServerClient, + type AppServerInputItem, +} from './app-server-client.js'; + // ── Types ────────────────────────────────────────────────────────── interface ContainerInput { prompt: string; - sessionId?: string; // threadId from previous session + sessionId?: string; groupFolder: string; chatJid: string; isMain: boolean; @@ -34,6 +38,7 @@ interface ContainerInput { interface ContainerOutput { status: 'success' | 'error'; result: string | null; + phase?: 'progress' | 'final'; newSessionId?: string; error?: string; } @@ -47,6 +52,7 @@ 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---'; @@ -55,6 +61,8 @@ const EFFECTIVE_CWD = WORK_DIR || GROUP_DIR; const CODEX_MODEL = process.env.CODEX_MODEL || ''; const CODEX_EFFORT = process.env.CODEX_EFFORT || ''; +let closeRequested = false; + // ── Helpers ──────────────────────────────────────────────────────── function writeOutput(output: ContainerOutput): void { @@ -71,25 +79,33 @@ async function readStdin(): Promise { return new Promise((resolve, reject) => { let data = ''; process.stdin.setEncoding('utf8'); - process.stdin.on('data', (chunk: string) => { data += chunk; }); + process.stdin.on('data', (chunk: string) => { + data += chunk; + }); process.stdin.on('end', () => resolve(data)); process.stdin.on('error', reject); }); } -function shouldClose(): boolean { - if (fs.existsSync(IPC_INPUT_CLOSE_SENTINEL)) { - try { fs.unlinkSync(IPC_INPUT_CLOSE_SENTINEL); } catch { /* ignore */ } - return true; +function consumeCloseSentinel(): boolean { + if (closeRequested) return true; + if (!fs.existsSync(IPC_INPUT_CLOSE_SENTINEL)) return false; + + try { + fs.unlinkSync(IPC_INPUT_CLOSE_SENTINEL); + } catch { + /* ignore */ } - return false; + closeRequested = true; + return true; } function drainIpcInput(): string[] { try { fs.mkdirSync(IPC_INPUT_DIR, { recursive: true }); - const files = fs.readdirSync(IPC_INPUT_DIR) - .filter(f => f.endsWith('.json')) + const files = fs + .readdirSync(IPC_INPUT_DIR) + .filter((file) => file.endsWith('.json')) .sort(); const messages: string[] = []; @@ -102,8 +118,16 @@ function drainIpcInput(): string[] { messages.push(data.text); } } catch (err) { - log(`Failed to process input file ${file}: ${err instanceof Error ? err.message : String(err)}`); - try { fs.unlinkSync(filePath); } catch { /* ignore */ } + log( + `Failed to process input file ${file}: ${ + err instanceof Error ? err.message : String(err) + }`, + ); + try { + fs.unlinkSync(filePath); + } catch { + /* ignore */ + } } } return messages; @@ -116,7 +140,7 @@ function drainIpcInput(): string[] { function waitForIpcMessage(): Promise { return new Promise((resolve) => { const poll = () => { - if (shouldClose()) { + if (consumeCloseSentinel()) { resolve(null); return; } @@ -131,23 +155,25 @@ function waitForIpcMessage(): Promise { }); } -// ── Input Parsing ───────────────────────────────────────────────── - -/** - * Parse [Image: /path] patterns from text and build SDK input. - */ -function parseInput(text: string): string | UserInput[] { +function extractImagePaths(text: string): { cleanText: string; imagePaths: string[] } { const imagePattern = /\[Image:\s*(\/[^\]]+)\]/g; const imagePaths: string[] = []; - let match; + let match: RegExpExecArray | null; while ((match = imagePattern.exec(text)) !== null) { imagePaths.push(match[1].trim()); } + return { + cleanText: text.replace(imagePattern, '').trim(), + imagePaths, + }; +} + +function parseSdkInput(text: string): string | UserInput[] { + const { cleanText, imagePaths } = extractImagePaths(text); if (imagePaths.length === 0) return text; const input: UserInput[] = []; - const cleanText = text.replace(imagePattern, '').trim(); if (cleanText) { input.push({ type: 'text', text: cleanText }); } @@ -162,23 +188,56 @@ function parseInput(text: string): string | UserInput[] { return input.length > 0 ? input : text; } -// ── Turn Execution ──────────────────────────────────────────────── +function parseAppServerInput(text: string): AppServerInputItem[] { + const { cleanText, imagePaths } = extractImagePaths(text); + const input: AppServerInputItem[] = []; -/** - * Execute a single turn using the SDK. The SDK wraps `codex exec`, - * ensuring the agent completes all work before returning. - */ -async function executeTurn( + if (cleanText) { + input.push({ type: 'text', text: cleanText }); + } + + for (const imgPath of imagePaths) { + if (fs.existsSync(imgPath)) { + input.push({ type: 'localImage', path: imgPath }); + log(`Adding image input: ${imgPath}`); + } else { + log(`Image not found, skipping: ${imgPath}`); + } + } + + if (input.length === 0) { + input.push({ type: 'text', text }); + } + + 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; +} + +async function executeSdkTurn( thread: Thread, input: string | UserInput[], ): Promise<{ result: string; error?: string }> { const ac = new AbortController(); - // Poll close sentinel + heartbeat during turn let turnSeconds = 0; const sentinel = setInterval(() => { - if (shouldClose()) { - log('Close sentinel detected during turn, aborting'); + if (consumeCloseSentinel()) { + log('Close sentinel detected during SDK turn, aborting'); ac.abort(); return; } @@ -204,55 +263,97 @@ async function executeTurn( } } -// ── Main ────────────────────────────────────────────────────────── +async function executeAppServerTurn( + client: CodexAppServerClient, + threadId: string, + prompt: 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, + }); + }, + }); -async function main(): Promise { - let containerInput: ContainerInput; + let elapsedMs = 0; + let polling = true; + const pollDuringTurn = async () => { + if (!polling) return; + + if (consumeCloseSentinel()) { + log('Close sentinel detected during app-server turn, interrupting'); + polling = false; + try { + await activeTurn.interrupt(); + } catch (err) { + log( + `Failed to interrupt active turn: ${ + err instanceof Error ? err.message : String(err) + }`, + ); + } + return; + } + + const messages = drainIpcInput(); + if (messages.length > 0) { + const merged = messages.join('\n'); + log(`Steering active turn with ${messages.length} queued message(s)`); + try { + await activeTurn.steer(parseAppServerInput(merged)); + } catch (err) { + log( + `turn/steer failed: ${err instanceof Error ? err.message : String(err)}`, + ); + } + } + + elapsedMs += IPC_POLL_MS; + if (elapsedMs > 0 && elapsedMs % 60000 === 0) { + log(`Turn in progress... (${Math.round(elapsedMs / 60000)}min)`); + } + setTimeout(() => void pollDuringTurn(), IPC_POLL_MS); + }; + + setTimeout(() => void pollDuringTurn(), IPC_POLL_MS); try { - const stdinData = await readStdin(); - containerInput = JSON.parse(stdinData); - try { fs.unlinkSync('/tmp/input.json'); } catch { /* may not exist */ } - log(`Received input for group: ${containerInput.groupFolder}`); - } catch (err) { - writeOutput({ - status: 'error', - result: null, - error: `Failed to parse input: ${err instanceof Error ? err.message : String(err)}`, - }); - process.exit(1); + const { state, result } = await activeTurn.wait(); + if (state.status === 'completed') { + return { result }; + } + if (state.status === 'interrupted' && consumeCloseSentinel()) { + return { result }; + } + return { + result, + error: state.errorMessage || `Codex turn finished with status ${state.status}`, + }; + } finally { + polling = false; } +} - fs.mkdirSync(IPC_INPUT_DIR, { recursive: true }); - try { fs.unlinkSync(IPC_INPUT_CLOSE_SENTINEL); } catch { /* ignore */ } - - // Build initial prompt - let prompt = containerInput.prompt; - if (containerInput.isScheduledTask) { - prompt = `[SCHEDULED TASK]\n\n${prompt}`; - } - const pending = drainIpcInput(); - if (pending.length > 0) { - prompt += '\n' + pending.join('\n'); - } - - // Build thread options - 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']; - } - - // Create SDK instance (inherits env from parent — CODEX_HOME, OPENAI_API_KEY, etc.) +async function runSdkSession( + containerInput: ContainerInput, + prompt: string, +): Promise { + const threadOptions = getThreadOptions(); const codex = new Codex(); - // Start or resume thread (resume may fail on first run, fallback to new thread) let thread: Thread; if (containerInput.sessionId) { thread = codex.resumeThread(containerInput.sessionId, threadOptions); @@ -263,9 +364,147 @@ async function main(): Promise { } 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, +): Promise { + if (!threadId) { + writeOutput({ + status: 'success', + result: '현재 활성 Codex 세션이 없어 compact를 건너뜁니다.', + }); + return; + } + + const { state } = await client.startCompaction(threadId); + if (state.status === 'failed') { + writeOutput({ + status: 'error', + result: null, + newSessionId: threadId, + error: state.errorMessage || 'Conversation compaction failed.', + }); + return; + } + + writeOutput({ + status: 'success', + result: state.compactionCompleted + ? 'Conversation compacted.' + : 'Compaction requested but contextCompaction was not observed.', + newSessionId: threadId, + }); +} + +async function runAppServerSession( + containerInput: ContainerInput, + prompt: string, +): Promise { + const client = new CodexAppServerClient({ + cwd: EFFECTIVE_CWD, + env: process.env, + log, + }); + + await client.start(); + + let threadId: string | undefined; try { - // Main turn loop + try { + threadId = await client.startOrResumeThread(containerInput.sessionId, { + cwd: EFFECTIVE_CWD, + model: CODEX_MODEL || undefined, + }); + log( + containerInput.sessionId + ? `App-server thread resumed (${threadId})` + : `App-server thread started (${threadId})`, + ); + } catch (err) { + if (!containerInput.sessionId) throw err; + log( + `App-server resume failed, retrying with new thread: ${ + err instanceof Error ? err.message : String(err) + }`, + ); + threadId = await client.startOrResumeThread(undefined, { + cwd: EFFECTIVE_CWD, + model: CODEX_MODEL || undefined, + }); + log(`App-server thread restarted (${threadId})`); + } + + const trimmedPrompt = prompt.trim(); + if (trimmedPrompt === '/compact') { + await runAppServerCompact(client, threadId); + return; + } + + let turnCount = 0; while (true) { turnCount++; if (turnCount > MAX_TURNS) { @@ -273,59 +512,127 @@ async function main(): Promise { writeOutput({ status: 'success', result: '[세션 턴 제한 도달. 새 메시지로 다시 시작됩니다.]', - newSessionId: thread.id || undefined, + newSessionId: threadId, }); break; } - const input = parseInput(prompt); - log(`Starting turn ${turnCount}/${MAX_TURNS}...`); + log(`Starting app-server turn ${turnCount}/${MAX_TURNS}...`); + const { result, error } = await executeAppServerTurn(client, threadId, prompt); - let { result, error } = await executeTurn(thread, input); - - // Fallback: if resume failed on first turn, retry with a new thread - if (error && turnCount === 1 && containerInput.sessionId) { - log(`Resume may have failed, retrying with new thread: ${error}`); - thread = codex.startThread(threadOptions); - ({ result, error } = await executeTurn(thread, input)); - } - - // Check close sentinel - if (shouldClose()) { + if (consumeCloseSentinel()) { if (result) { - writeOutput({ status: 'success', result, newSessionId: thread.id || undefined }); + writeOutput({ + status: 'success', + result, + newSessionId: threadId, + }); } - log('Close sentinel detected, exiting'); + log('Close sentinel detected, exiting app-server runtime'); break; } if (error) { - log(`Turn error: ${error}`); + log(`App-server turn error: ${error}`); writeOutput({ status: 'error', result: result || null, - newSessionId: thread.id || undefined, + newSessionId: threadId, error, }); } else { writeOutput({ status: 'success', result: result || null, - newSessionId: thread.id || undefined, + ...(result ? { phase: 'final' as const } : {}), + newSessionId: threadId, }); } - log('Turn done, waiting for next IPC message...'); + log('App-server turn done, waiting for next IPC message...'); const nextMessage = await waitForIpcMessage(); if (nextMessage === null) { - log('Close sentinel received, exiting'); + log('Close sentinel received, exiting app-server runtime'); break; } - log(`Got new message (${nextMessage.length} chars)`); + log(`Got new app-server message (${nextMessage.length} chars)`); prompt = nextMessage; } + } finally { + await client.close(); + } +} + +function shouldUseAppServer(): boolean { + return CODEX_RUNTIME !== 'sdk'; +} + +// ── Main ────────────────────────────────────────────────────────── + +async function main(): Promise { + let containerInput: ContainerInput; + + try { + const stdinData = await readStdin(); + containerInput = JSON.parse(stdinData); + try { + fs.unlinkSync('/tmp/input.json'); + } catch { + /* may not exist */ + } + log(`Received input for group: ${containerInput.groupFolder}`); + } catch (err) { + writeOutput({ + status: 'error', + result: null, + error: `Failed to parse input: ${ + err instanceof Error ? err.message : String(err) + }`, + }); + process.exit(1); + } + + fs.mkdirSync(IPC_INPUT_DIR, { recursive: true }); + closeRequested = false; + try { + fs.unlinkSync(IPC_INPUT_CLOSE_SENTINEL); + } catch { + /* ignore */ + } + + let prompt = containerInput.prompt; + if (containerInput.isScheduledTask) { + prompt = `[SCHEDULED TASK]\n\n${prompt}`; + } + const pending = drainIpcInput(); + if (pending.length > 0) { + 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); } catch (err) { const errorMessage = err instanceof Error ? err.message : String(err); log(`Runner error: ${errorMessage}`); diff --git a/runners/codex-runner/test/app-server-state.test.ts b/runners/codex-runner/test/app-server-state.test.ts new file mode 100644 index 0000000..87462c7 --- /dev/null +++ b/runners/codex-runner/test/app-server-state.test.ts @@ -0,0 +1,83 @@ +import { describe, expect, it } from 'vitest'; + +import { + createInitialAppServerTurnState, + getAppServerTurnResult, + reduceAppServerTurnState, +} from '../src/app-server-state.js'; + +describe('codex app-server turn state', () => { + it('prefers final_answer over commentary when building final output', () => { + let state = createInitialAppServerTurnState(); + + state = reduceAppServerTurnState(state, { + method: 'item/completed', + params: { + item: { + type: 'agentMessage', + phase: 'commentary', + text: '중간 설명입니다.', + }, + }, + }); + + state = reduceAppServerTurnState(state, { + method: 'item/completed', + params: { + item: { + type: 'agentMessage', + phase: 'final_answer', + text: '최종 답변입니다.', + }, + }, + }); + + expect(getAppServerTurnResult(state)).toBe('최종 답변입니다.'); + }); + + it('does not treat commentary-only messages as a final output', () => { + const state = reduceAppServerTurnState(createInitialAppServerTurnState(), { + method: 'item/completed', + params: { + item: { + type: 'agentMessage', + phase: 'commentary', + text: '작업 중 상태입니다.', + }, + }, + }); + + expect(state.latestAgentMessage).toBe('작업 중 상태입니다.'); + expect(getAppServerTurnResult(state)).toBeNull(); + }); + + it('records compaction completion from contextCompaction items', () => { + const state = reduceAppServerTurnState(createInitialAppServerTurnState(), { + method: 'item/completed', + params: { + item: { + type: 'contextCompaction', + id: 'ctx_1', + }, + }, + }); + + expect(state.compactionCompleted).toBe(true); + }); + + it('captures upstream HTTP status from app-server error events', () => { + const state = reduceAppServerTurnState(createInitialAppServerTurnState(), { + method: 'error', + params: { + error: { + message: 'Too many requests', + codexErrorInfo: { + httpStatusCode: 429, + }, + }, + }, + }); + + expect(state.errorMessage).toBe('Too many requests (HTTP 429)'); + }); +}); diff --git a/setup/register.test.ts b/setup/register.test.ts index d47d95c..d4dcda7 100644 --- a/setup/register.test.ts +++ b/setup/register.test.ts @@ -12,14 +12,18 @@ import Database from 'better-sqlite3'; function createTestDb(): Database.Database { const db = new Database(':memory:'); db.exec(`CREATE TABLE IF NOT EXISTS registered_groups ( - jid TEXT PRIMARY KEY, + jid TEXT NOT NULL, name TEXT NOT NULL, - folder TEXT NOT NULL UNIQUE, + folder TEXT NOT NULL, trigger_pattern TEXT NOT NULL, 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 NOT NULL DEFAULT 'claude-code', + work_dir TEXT, + PRIMARY KEY (jid, agent_type), + UNIQUE (folder, agent_type) )`); return db; } diff --git a/setup/register.ts b/setup/register.ts index 33e77cb..7f5516d 100644 --- a/setup/register.ts +++ b/setup/register.ts @@ -106,16 +106,18 @@ export async function run(args: string[]): Promise { const db = new Database(dbPath); // Ensure schema exists db.exec(`CREATE TABLE IF NOT EXISTS registered_groups ( - jid TEXT PRIMARY KEY, + jid TEXT NOT NULL, name TEXT NOT NULL, - folder TEXT NOT NULL UNIQUE, + folder TEXT NOT NULL, trigger_pattern TEXT NOT NULL, added_at TEXT NOT NULL, container_config TEXT, requires_trigger INTEGER DEFAULT 1, is_main INTEGER DEFAULT 0, - agent_type TEXT DEFAULT 'claude-code', - work_dir TEXT + agent_type TEXT NOT NULL DEFAULT 'claude-code', + work_dir TEXT, + PRIMARY KEY (jid, agent_type), + UNIQUE (folder, agent_type) )`); try { diff --git a/src/agent-runner.test.ts b/src/agent-runner.test.ts index ebb1de2..dd14196 100644 --- a/src/agent-runner.test.ts +++ b/src/agent-runner.test.ts @@ -1,6 +1,7 @@ import { describe, it, expect, beforeEach, vi, afterEach } from 'vitest'; import { EventEmitter } from 'events'; import { PassThrough } from 'stream'; +import fs from 'fs'; // Sentinel markers must match agent-runner.ts const OUTPUT_START_MARKER = '---NANOCLAW_OUTPUT_START---'; @@ -212,4 +213,41 @@ describe('agent-runner timeout behavior', () => { expect(result.status).toBe('success'); expect(result.newSessionId).toBe('session-456'); }); + + it('materializes repo-managed Claude rules into the session CLAUDE.md', async () => { + vi.mocked(fs.existsSync).mockImplementation((p: fs.PathLike) => { + const path = String(p); + return ( + path.includes('dist/index.js') || + path.endsWith('/prompts/claude-platform.md') || + path.endsWith('/global/CLAUDE.md') + ); + }); + vi.mocked(fs.readFileSync).mockImplementation((p: fs.PathOrFileDescriptor) => { + const path = String(p); + if (path.endsWith('/prompts/claude-platform.md')) { + return 'Platform Claude Rules'; + } + if (path.endsWith('/global/CLAUDE.md')) { + return 'Global Claude Memory'; + } + return ''; + }); + + const resultPromise = runAgentProcess( + testGroup, + testInput, + () => {}, + undefined, + ); + + fakeProc.emit('close', 0); + await vi.advanceTimersByTimeAsync(10); + await resultPromise; + + expect(fs.writeFileSync).toHaveBeenCalledWith( + '/tmp/nanoclaw-test-data/sessions/test-group/.claude/CLAUDE.md', + 'Platform Claude Rules\n\n---\n\nGlobal Claude Memory\n', + ); + }); }); diff --git a/src/agent-runner.ts b/src/agent-runner.ts index e4a1cba..163f78e 100644 --- a/src/agent-runner.ts +++ b/src/agent-runner.ts @@ -18,6 +18,11 @@ import { import { resolveGroupFolderPath, resolveGroupIpcPath } from './group-folder.js'; import { logger } from './logger.js'; import { readEnvFile } from './env.js'; +import { isPairedRoomJid } from './db.js'; +import { + readPairedRoomPrompt, + readPlatformPrompt, +} from './platform-prompts.js'; import { RegisteredGroup } from './types.js'; // Sentinel markers for robust output parsing (must match agent-runner) @@ -39,6 +44,7 @@ export interface AgentInput { export interface AgentOutput { status: 'success' | 'error'; result: string | null; + phase?: 'progress' | 'final'; newSessionId?: string; error?: string; } @@ -50,6 +56,7 @@ export interface AgentOutput { function prepareGroupEnvironment( group: RegisteredGroup, isMain: boolean, + chatJid: string, ): { env: Record; groupDir: string; runnerDir: string } { const projectRoot = process.cwd(); const groupDir = resolveGroupFolderPath(group.folder); @@ -115,6 +122,31 @@ function prepareGroupEnvironment( // Global memory directory (for non-main groups) const globalDir = path.join(GROUPS_DIR, 'global'); + const globalClaudeMdPath = path.join(globalDir, 'CLAUDE.md'); + const isPairedRoom = isPairedRoomJid(chatJid); + + const claudePlatformPrompt = readPlatformPrompt('claude-code', projectRoot); + const claudePairedRoomPrompt = isPairedRoom + ? readPairedRoomPrompt('claude-code', projectRoot) + : undefined; + const globalClaudeMemory = + !isMain && fs.existsSync(globalClaudeMdPath) + ? fs.readFileSync(globalClaudeMdPath, 'utf-8').trim() + : undefined; + const sessionClaudeMd = [ + claudePlatformPrompt, + claudePairedRoomPrompt, + globalClaudeMemory, + ] + .filter((value): value is string => Boolean(value)) + .join('\n\n---\n\n') + .trim(); + const sessionClaudeMdPath = path.join(groupSessionsDir, 'CLAUDE.md'); + if (sessionClaudeMd) { + fs.writeFileSync(sessionClaudeMdPath, sessionClaudeMd + '\n'); + } else if (fs.existsSync(sessionClaudeMdPath)) { + fs.unlinkSync(sessionClaudeMdPath); + } // Additional mount directories (validated) const extraDirs: string[] = []; @@ -215,13 +247,27 @@ function prepareGroupEnvironment( const authSrc = path.join(hostCodexDir, 'auth.json'); const authDst = path.join(sessionCodexDir, 'auth.json'); if (fs.existsSync(authSrc)) fs.copyFileSync(authSrc, authDst); - for (const file of ['config.toml', 'config.json', 'AGENTS.md']) { + for (const file of ['config.toml', 'config.json']) { const src = path.join(hostCodexDir, file); const dst = path.join(sessionCodexDir, file); if (fs.existsSync(src)) { fs.copyFileSync(src, dst); } } + const sessionAgentsPath = path.join(sessionCodexDir, 'AGENTS.md'); + const codexPlatformPrompt = readPlatformPrompt('codex', projectRoot); + const codexPairedRoomPrompt = isPairedRoom + ? readPairedRoomPrompt('codex', projectRoot) + : undefined; + const sessionAgents = [codexPlatformPrompt, codexPairedRoomPrompt] + .filter((value): value is string => Boolean(value)) + .join('\n\n---\n\n') + .trim(); + if (sessionAgents) { + fs.writeFileSync(sessionAgentsPath, sessionAgents + '\n'); + } else if (fs.existsSync(sessionAgentsPath)) { + fs.unlinkSync(sessionAgentsPath); + } // Sync skills into Codex session dir // SSOT: ~/.claude/skills/ (shared with Claude Code) + runners/skills/ const codexSkillSources = [ @@ -324,6 +370,7 @@ export async function runAgentProcess( const { env, groupDir, runnerDir } = prepareGroupEnvironment( group, input.isMain, + input.chatJid, ); if (input.runId) { env.NANOCLAW_RUN_ID = input.runId; diff --git a/src/channels/discord.test.ts b/src/channels/discord.test.ts index 5db7420..06334ea 100644 --- a/src/channels/discord.test.ts +++ b/src/channels/discord.test.ts @@ -313,11 +313,55 @@ describe('DiscordChannel', () => { expect.any(String), expect.objectContaining({ content: 'I am a bot', + is_from_me: false, is_bot_message: true, }), ); }); + it('ignores self-authored bot messages', async () => { + const opts = createTestOpts(); + const channel = new DiscordChannel('test-token', opts); + await channel.connect(); + + const msg = createMessage({ + authorId: '999888777', + isBot: true, + content: 'I am this bot', + }); + await triggerMessage(msg); + + expect(opts.onMessage).not.toHaveBeenCalled(); + }); + + it('identifies stored messages authored by the current bot', async () => { + const opts = createTestOpts(); + const channel = new DiscordChannel('test-token', opts); + await channel.connect(); + + expect( + channel.isOwnMessage?.({ + id: 'msg_bot', + chat_jid: 'dc:1234567890123456', + sender: '999888777', + sender_name: 'Andy', + content: 'bot message', + timestamp: '2024-01-01T00:00:00.000Z', + is_bot_message: true, + }), + ).toBe(true); + expect( + channel.isOwnMessage?.({ + id: 'msg_user', + chat_jid: 'dc:1234567890123456', + sender: '55512345', + sender_name: 'Alice', + content: 'user message', + timestamp: '2024-01-01T00:00:00.000Z', + }), + ).toBe(false); + }); + it('uses member displayName when available (server nickname)', async () => { const opts = createTestOpts(); const channel = new DiscordChannel('test-token', opts); diff --git a/src/channels/discord.ts b/src/channels/discord.ts index f683366..a436629 100644 --- a/src/channels/discord.ts +++ b/src/channels/discord.ts @@ -19,6 +19,7 @@ import { } from '../config.js'; import { readEnvFile } from '../env.js'; import { logger } from '../logger.js'; +import { formatOutbound } from '../router.js'; const ATTACHMENTS_DIR = path.join(DATA_DIR, 'attachments'); const TRANSCRIPTION_CACHE_DIR = path.join(CACHE_DIR, 'transcriptions'); @@ -158,6 +159,7 @@ import { AgentType, Channel, ChannelMeta, + NewMessage, OnChatMetadata, OnInboundMessage, RegisteredGroup, @@ -446,6 +448,7 @@ export class DiscordChannel implements Channel { for (const [name, id] of Object.entries(mentionMap)) { cleaned = cleaned.replace(new RegExp(`@${name}`, 'g'), `<@${id}>`); } + cleaned = formatOutbound(cleaned); // Discord has a 2000 character limit per message — split if needed const MAX_LENGTH = 2000; @@ -454,6 +457,11 @@ export class DiscordChannel implements Channel { name: path.basename(f), })); + if (!cleaned && files.length === 0) { + logger.debug({ jid }, 'Skipping empty Discord outbound message'); + return; + } + if (cleaned.length <= MAX_LENGTH) { await textChannel.send({ content: cleaned || undefined, @@ -490,6 +498,10 @@ export class DiscordChannel implements Channel { return groupType === this.agentTypeFilter; } + isOwnMessage(msg: NewMessage): boolean { + return !!this.client?.user?.id && msg.sender === this.client.user.id; + } + async disconnect(): Promise { if (this.client) { this.client.destroy(); diff --git a/src/db.test.ts b/src/db.test.ts index 0ded8c5..56c7ab4 100644 --- a/src/db.test.ts +++ b/src/db.test.ts @@ -7,8 +7,10 @@ import { deleteTask, getAllChats, getAllRegisteredGroups, + getRegisteredAgentTypesForJid, getMessagesSince, getNewMessages, + isPairedRoomJid, getSession, getTaskById, setSession, @@ -190,13 +192,12 @@ describe('getMessagesSince', () => { '2024-01-01T00:00:02.000Z', 'Andy', ); - // Should exclude m1, m2 (before/at timestamp); m3 (bot) is now included expect(msgs).toHaveLength(2); expect(msgs[0].content).toBe('bot reply'); expect(msgs[1].content).toBe('third'); }); - it('includes bot messages with is_bot_message flag', () => { + it('includes bot messages from other senders', () => { const msgs = getMessagesSince( 'group@g.us', '2024-01-01T00:00:00.000Z', @@ -206,9 +207,8 @@ describe('getMessagesSince', () => { expect(botMsgs).toHaveLength(1); }); - it('returns all messages including bot when sinceTimestamp is empty', () => { + it('returns all messages when sinceTimestamp is empty', () => { const msgs = getMessagesSince('group@g.us', '', 'Andy'); - // 3 user messages + 1 bot message expect(msgs).toHaveLength(4); }); @@ -279,7 +279,6 @@ describe('getNewMessages', () => { '2024-01-01T00:00:00.000Z', 'Andy', ); - // Includes bot message, returns all 4 messages expect(messages).toHaveLength(4); expect(newTimestamp).toBe('2024-01-01T00:00:04.000Z'); }); @@ -290,7 +289,6 @@ describe('getNewMessages', () => { '2024-01-01T00:00:02.000Z', 'Andy', ); - // bot reply (00:00:03) + g1 msg2 (00:00:04) after ts expect(messages).toHaveLength(2); expect(messages[0].content).toBe('bot reply'); expect(messages[1].content).toBe('g1 msg2'); @@ -497,3 +495,41 @@ describe('registered group isMain', () => { expect(group.isMain).toBeUndefined(); }); }); + +describe('paired room registration', () => { + it('detects when both Claude and Codex are registered on the same jid', () => { + setRegisteredGroup('dc:123', { + name: 'Paired Room Claude', + folder: 'paired-claude', + trigger: '@Andy', + added_at: '2024-01-01T00:00:00.000Z', + agentType: 'claude-code', + }); + setRegisteredGroup('dc:123', { + name: 'Paired Room Codex', + folder: 'paired-codex', + trigger: '@Codex', + added_at: '2024-01-01T00:00:00.000Z', + agentType: 'codex', + }); + + expect(getRegisteredAgentTypesForJid('dc:123').sort()).toEqual([ + 'claude-code', + 'codex', + ]); + expect(isPairedRoomJid('dc:123')).toBe(true); + }); + + it('does not mark solo rooms as paired', () => { + setRegisteredGroup('dc:solo', { + name: 'Solo Claude Room', + folder: 'solo-claude', + trigger: '@Andy', + added_at: '2024-01-01T00:00:00.000Z', + agentType: 'claude-code', + }); + + expect(getRegisteredAgentTypesForJid('dc:solo')).toEqual(['claude-code']); + expect(isPairedRoomJid('dc:solo')).toBe(false); + }); +}); diff --git a/src/db.ts b/src/db.ts index 4ac6c63..b705b1b 100644 --- a/src/db.ts +++ b/src/db.ts @@ -13,6 +13,7 @@ import { isValidGroupFolder } from './group-folder.js'; import { logger } from './logger.js'; import { NewMessage, + AgentType, RegisteredGroup, ScheduledTask, TaskRunLog, @@ -82,13 +83,18 @@ function createSchema(database: Database.Database): void { PRIMARY KEY (group_folder, agent_type) ); CREATE TABLE IF NOT EXISTS registered_groups ( - jid TEXT PRIMARY KEY, + jid TEXT NOT NULL, name TEXT NOT NULL, - folder TEXT NOT NULL UNIQUE, + folder TEXT NOT NULL, trigger_pattern TEXT NOT NULL, added_at TEXT NOT NULL, container_config TEXT, - requires_trigger INTEGER DEFAULT 1 + requires_trigger INTEGER DEFAULT 1, + is_main INTEGER DEFAULT 0, + agent_type TEXT NOT NULL DEFAULT 'claude-code', + work_dir TEXT, + PRIMARY KEY (jid, agent_type), + UNIQUE (folder, agent_type) ); `); @@ -114,33 +120,80 @@ function createSchema(database: Database.Database): void { /* column already exists */ } - // Add is_main column if it doesn't exist (migration for existing DBs) - try { - database.exec( - `ALTER TABLE registered_groups ADD COLUMN is_main INTEGER DEFAULT 0`, + // Migrate registered_groups to composite keys so Claude/Codex can share a jid/folder. + const registeredGroupsSql = ( + database + .prepare( + `SELECT sql FROM sqlite_master WHERE type = 'table' AND name = 'registered_groups'`, + ) + .get() as { sql?: string } | undefined + )?.sql; + if ( + registeredGroupsSql && + !registeredGroupsSql.includes('PRIMARY KEY (jid, agent_type)') + ) { + const registeredGroupCols = database + .prepare('PRAGMA table_info(registered_groups)') + .all() as Array<{ name: string }>; + const hasIsMain = registeredGroupCols.some((col) => col.name === 'is_main'); + const hasAgentType = registeredGroupCols.some( + (col) => col.name === 'agent_type', ); + const hasWorkDir = registeredGroupCols.some((col) => col.name === 'work_dir'); + + database.exec(` + CREATE TABLE registered_groups_new ( + jid TEXT NOT NULL, + name TEXT NOT NULL, + folder TEXT NOT NULL, + trigger_pattern TEXT NOT NULL, + added_at TEXT NOT NULL, + container_config TEXT, + requires_trigger INTEGER DEFAULT 1, + is_main INTEGER DEFAULT 0, + agent_type TEXT NOT NULL DEFAULT 'claude-code', + work_dir TEXT, + PRIMARY KEY (jid, agent_type), + UNIQUE (folder, agent_type) + ); + `); + + database.exec(` + INSERT INTO registered_groups_new ( + jid, + name, + folder, + trigger_pattern, + added_at, + container_config, + requires_trigger, + is_main, + agent_type, + work_dir + ) + SELECT + jid, + name, + folder, + trigger_pattern, + added_at, + container_config, + requires_trigger, + ${hasIsMain ? 'COALESCE(is_main, 0)' : "CASE WHEN folder = 'main' THEN 1 ELSE 0 END"}, + ${hasAgentType ? "COALESCE(agent_type, 'claude-code')" : "'claude-code'"}, + ${hasWorkDir ? 'work_dir' : 'NULL'} + FROM registered_groups; + `); + + database.exec(` + DROP TABLE registered_groups; + ALTER TABLE registered_groups_new RENAME TO registered_groups; + `); + } else { // Backfill: existing rows with folder = 'main' are the main group database.exec( - `UPDATE registered_groups SET is_main = 1 WHERE folder = 'main'`, + `UPDATE registered_groups SET is_main = 1 WHERE folder = 'main' AND COALESCE(is_main, 0) = 0`, ); - } catch { - /* column already exists */ - } - - // Add agent_type column if it doesn't exist (migration for Codex support) - try { - database.exec( - `ALTER TABLE registered_groups ADD COLUMN agent_type TEXT DEFAULT 'claude-code'`, - ); - } catch { - /* column already exists */ - } - - // Add work_dir column if it doesn't exist (migration for per-group working directory) - try { - database.exec(`ALTER TABLE registered_groups ADD COLUMN work_dir TEXT`); - } catch { - /* column already exists */ } // Migrate sessions table to composite PK (group_folder, agent_type) @@ -300,9 +353,9 @@ export function getNewMessages( if (jids.length === 0) return { messages: [], newTimestamp: lastTimestamp }; const placeholders = jids.map(() => '?').join(','); - // Filter bot messages using both the is_bot_message flag AND the content - // prefix as a backstop for messages written before the migration ran. - // Subquery takes the N most recent, outer query re-sorts chronologically. + // Filter legacy prefixed outbound messages as a backstop for rows written + // before explicit bot flags existed. Self-message filtering is channel-specific + // and happens in message-runtime so cross-bot collaboration still works. const sql = ` SELECT * FROM ( SELECT id, chat_jid, sender, sender_name, content, timestamp, is_from_me, is_bot_message @@ -333,9 +386,9 @@ export function getMessagesSince( botPrefix: string, limit: number = 200, ): NewMessage[] { - // Filter bot messages using both the is_bot_message flag AND the content - // prefix as a backstop for messages written before the migration ran. - // Subquery takes the N most recent, outer query re-sorts chronologically. + // Filter legacy prefixed outbound messages as a backstop for rows written + // before explicit bot flags existed. Self-message filtering is channel-specific + // and happens in message-runtime so cross-bot collaboration still works. const sql = ` SELECT * FROM ( SELECT id, chat_jid, sender, sender_name, content, timestamp, is_from_me, is_bot_message @@ -681,6 +734,28 @@ export function getAllRegisteredGroups( return result; } +export function getRegisteredAgentTypesForJid(jid: string): AgentType[] { + if (!db) return []; + + const rows = db + .prepare('SELECT agent_type FROM registered_groups WHERE jid = ?') + .all(jid) as Array<{ agent_type: string | null }>; + + const types = new Set(); + for (const row of rows) { + const agentType = row.agent_type as AgentType | null; + if (agentType === 'claude-code' || agentType === 'codex') { + types.add(agentType); + } + } + return [...types]; +} + +export function isPairedRoomJid(jid: string): boolean { + const types = getRegisteredAgentTypesForJid(jid); + return types.includes('claude-code') && types.includes('codex'); +} + // --- JSON migration --- function migrateJsonState(): void { diff --git a/src/group-queue.test.ts b/src/group-queue.test.ts index abbe94d..9d01c6c 100644 --- a/src/group-queue.test.ts +++ b/src/group-queue.test.ts @@ -166,6 +166,29 @@ describe('GroupQueue', () => { expect(callCount).toBe(3); }); + it('does not bypass retry backoff when new messages arrive', async () => { + let callCount = 0; + + const processMessages = vi.fn(async () => { + callCount++; + return false; + }); + + queue.setProcessMessagesFn(processMessages); + queue.enqueueMessageCheck('group1@g.us'); + + await vi.advanceTimersByTimeAsync(10); + expect(callCount).toBe(1); + + queue.enqueueMessageCheck('group1@g.us'); + await vi.advanceTimersByTimeAsync(1000); + expect(callCount).toBe(1); + + await vi.advanceTimersByTimeAsync(4000); + await vi.advanceTimersByTimeAsync(10); + expect(callCount).toBe(2); + }); + // --- Shutdown prevents new enqueues --- it('prevents new enqueues after shutdown', async () => { diff --git a/src/group-queue.ts b/src/group-queue.ts index 17e013d..3ac6ed8 100644 --- a/src/group-queue.ts +++ b/src/group-queue.ts @@ -32,6 +32,8 @@ interface GroupState { processName: string | null; groupFolder: string | null; retryCount: number; + retryTimer: ReturnType | null; + retryScheduledAt: number | null; startedAt: number | null; } @@ -68,6 +70,8 @@ export class GroupQueue { processName: null, groupFolder: null, retryCount: 0, + retryTimer: null, + retryScheduledAt: null, startedAt: null, }; this.groups.set(groupJid, state); @@ -101,6 +105,22 @@ export class GroupQueue { return; } + if ( + state.retryScheduledAt !== null && + Date.now() < state.retryScheduledAt + ) { + state.pendingMessages = true; + logger.debug( + { + groupJid, + retryCount: state.retryCount, + retryScheduledAt: state.retryScheduledAt, + }, + 'Retry backoff active, message queued until retry window opens', + ); + return; + } + if (this.activeCount >= MAX_CONCURRENT_AGENTS) { state.pendingMessages = true; if (!this.waitingGroups.includes(groupJid)) { @@ -335,6 +355,7 @@ export class GroupQueue { }); if (success) { state.retryCount = 0; + state.retryScheduledAt = null; } else { outcome = 'retry_scheduled'; this.scheduleRetry(groupJid, state, runId); @@ -415,6 +436,11 @@ export class GroupQueue { ): void { state.retryCount++; if (state.retryCount > MAX_RETRIES) { + if (state.retryTimer) { + clearTimeout(state.retryTimer); + state.retryTimer = null; + } + state.retryScheduledAt = null; logger.error( { groupJid, runId, retryCount: state.retryCount }, 'Max retries exceeded, dropping messages (will retry on next incoming message)', @@ -424,11 +450,17 @@ export class GroupQueue { } const delayMs = BASE_RETRY_MS * Math.pow(2, state.retryCount - 1); + state.retryScheduledAt = Date.now() + delayMs; logger.info( { groupJid, runId, retryCount: state.retryCount, delayMs }, 'Scheduling retry with backoff', ); - setTimeout(() => { + if (state.retryTimer) { + clearTimeout(state.retryTimer); + } + state.retryTimer = setTimeout(() => { + state.retryTimer = null; + state.retryScheduledAt = null; if (!this.shuttingDown) { this.enqueueMessageCheck(groupJid); } diff --git a/src/message-runtime.test.ts b/src/message-runtime.test.ts new file mode 100644 index 0000000..146b550 --- /dev/null +++ b/src/message-runtime.test.ts @@ -0,0 +1,461 @@ +import { beforeEach, describe, expect, it, vi } from 'vitest'; + +vi.mock('./agent-runner.js', () => ({ + runAgentProcess: vi.fn(), + writeGroupsSnapshot: vi.fn(), + writeTasksSnapshot: vi.fn(), +})); + +vi.mock('./config.js', () => ({ + isSessionCommandSenderAllowed: vi.fn(() => false), +})); + +vi.mock('./db.js', () => ({ + getAllChats: vi.fn(() => []), + getAllTasks: vi.fn(() => []), + getLastHumanMessageTimestamp: vi.fn(() => null), + getMessagesSince: vi.fn(), + getNewMessages: vi.fn(() => ({ messages: [], newTimestamp: '' })), +})); + +vi.mock('./logger.js', () => ({ + logger: { + debug: vi.fn(), + info: vi.fn(), + warn: vi.fn(), + error: vi.fn(), + }, +})); + +vi.mock('./sender-allowlist.js', () => ({ + isTriggerAllowed: vi.fn(() => true), + loadSenderAllowlist: vi.fn(() => ({})), +})); + +vi.mock('./session-commands.js', () => ({ + extractSessionCommand: vi.fn(() => null), + handleSessionCommand: vi.fn(async () => ({ handled: false })), + isSessionCommandAllowed: vi.fn(() => true), + isSessionCommandControlMessage: vi.fn(() => false), +})); + +import * as agentRunner from './agent-runner.js'; +import * as db from './db.js'; +import { createMessageRuntime } from './message-runtime.js'; +import type { Channel, RegisteredGroup } from './types.js'; + +function makeGroup(agentType: 'claude-code' | 'codex'): RegisteredGroup { + return { + name: 'Test Group', + folder: `test-${agentType}`, + trigger: '@Andy', + added_at: new Date().toISOString(), + requiresTrigger: false, + agentType, + }; +} + +function makeChannel(chatJid: string): Channel { + return { + name: 'discord', + connect: vi.fn().mockResolvedValue(undefined), + sendMessage: vi.fn().mockResolvedValue(undefined), + isConnected: vi.fn(() => true), + ownsJid: vi.fn((jid: string) => jid === chatJid), + disconnect: vi.fn().mockResolvedValue(undefined), + setTyping: vi.fn().mockResolvedValue(undefined), + }; +} + +describe('createMessageRuntime', () => { + beforeEach(() => { + vi.clearAllMocks(); + }); + + it('clears Claude sessions and closes stdin immediately on poisoned output', async () => { + const chatJid = 'group@test'; + const group = makeGroup('claude-code'); + const channel = makeChannel(chatJid); + const closeStdin = vi.fn(); + const notifyIdle = vi.fn(); + const persistSession = vi.fn(); + const clearSession = vi.fn(); + const saveState = vi.fn(); + const lastAgentTimestamps: Record = {}; + + vi.mocked(db.getMessagesSince).mockReturnValue([ + { + id: 'msg-1', + chat_jid: chatJid, + sender: 'user@test', + sender_name: 'User', + content: 'hello', + timestamp: '2026-03-18T09:00:00.000Z', + }, + ]); + + vi.mocked(agentRunner.runAgentProcess).mockImplementation( + async (_group, _input, _onProcess, onOutput) => { + await onOutput?.({ + status: 'success', + result: + 'An image in the conversation exceeds the dimension limit for many-image requests (2000px). Start a new session with fewer images.', + newSessionId: 'session-123', + }); + return { + status: 'success', + result: null, + newSessionId: 'session-123', + }; + }, + ); + + const runtime = createMessageRuntime({ + assistantName: 'Andy', + idleTimeout: 1_000, + pollInterval: 1_000, + timezone: 'UTC', + triggerPattern: /^@Andy\b/i, + channels: [channel], + queue: { + registerProcess: vi.fn(), + closeStdin, + notifyIdle, + } as any, + getRegisteredGroups: () => ({ [chatJid]: group }), + getSessions: () => ({}), + getLastTimestamp: () => '', + setLastTimestamp: vi.fn(), + getLastAgentTimestamps: () => lastAgentTimestamps, + saveState, + persistSession, + clearSession, + }); + + const result = await runtime.processGroupMessages(chatJid, { + runId: 'run-1', + reason: 'messages', + }); + + expect(result).toBe(true); + expect(persistSession).toHaveBeenCalledWith(group.folder, 'session-123'); + expect(clearSession).toHaveBeenCalledWith(group.folder); + expect(closeStdin).toHaveBeenCalledWith(chatJid, { + runId: 'run-1', + reason: 'poisoned-session', + }); + expect(notifyIdle).not.toHaveBeenCalled(); + expect(channel.sendMessage).toHaveBeenCalledWith( + chatJid, + 'An image in the conversation exceeds the dimension limit for many-image requests (2000px). Start a new session with fewer images.', + ); + expect(lastAgentTimestamps[chatJid]).toBe('2026-03-18T09:00:00.000Z'); + expect(saveState).toHaveBeenCalled(); + }); + + it('does not apply the poisoned-session handling to Codex groups', async () => { + const chatJid = 'group@test'; + const group = makeGroup('codex'); + const channel = makeChannel(chatJid); + const closeStdin = vi.fn(); + const notifyIdle = vi.fn(); + const clearSession = vi.fn(); + + vi.mocked(db.getMessagesSince).mockReturnValue([ + { + id: 'msg-1', + chat_jid: chatJid, + sender: 'user@test', + sender_name: 'User', + content: 'hello', + timestamp: '2026-03-18T09:00:00.000Z', + }, + ]); + + vi.mocked(agentRunner.runAgentProcess).mockImplementation( + async (_group, _input, _onProcess, onOutput) => { + await onOutput?.({ + status: 'success', + result: + 'An image in the conversation exceeds the dimension limit for many-image requests (2000px). Start a new session with fewer images.', + newSessionId: 'session-456', + }); + return { + status: 'success', + result: null, + newSessionId: 'session-456', + }; + }, + ); + + const runtime = createMessageRuntime({ + assistantName: 'Andy', + idleTimeout: 1_000, + pollInterval: 1_000, + timezone: 'UTC', + triggerPattern: /^@Andy\b/i, + channels: [channel], + queue: { + registerProcess: vi.fn(), + closeStdin, + notifyIdle, + } as any, + getRegisteredGroups: () => ({ [chatJid]: group }), + getSessions: () => ({}), + getLastTimestamp: () => '', + setLastTimestamp: vi.fn(), + getLastAgentTimestamps: () => ({}), + saveState: vi.fn(), + persistSession: vi.fn(), + clearSession, + }); + + const result = await runtime.processGroupMessages(chatJid, { + runId: 'run-2', + reason: 'messages', + }); + + expect(result).toBe(true); + expect(clearSession).not.toHaveBeenCalled(); + expect(closeStdin).not.toHaveBeenCalled(); + expect(notifyIdle).toHaveBeenCalledWith(chatJid, 'run-2'); + }); + + it('streams Codex progress as a normal room message and only marks idle after the turn settles', async () => { + const chatJid = 'group@test'; + const group = makeGroup('codex'); + const channel = makeChannel(chatJid); + const notifyIdle = vi.fn(); + const persistSession = vi.fn(); + + vi.mocked(db.getMessagesSince).mockReturnValue([ + { + id: 'msg-1', + chat_jid: chatJid, + sender: 'user@test', + sender_name: 'User', + content: 'hello', + timestamp: '2026-03-19T00:00:00.000Z', + }, + ]); + + vi.mocked(agentRunner.runAgentProcess).mockImplementation( + async (_group, _input, _onProcess, onOutput) => { + await onOutput?.({ + status: 'success', + phase: 'progress', + result: 'CI 상태 확인 중입니다.', + newSessionId: 'session-progress', + }); + expect(notifyIdle).not.toHaveBeenCalled(); + await onOutput?.({ + status: 'success', + result: null, + newSessionId: 'session-progress', + }); + return { + status: 'success', + result: null, + newSessionId: 'session-progress', + }; + }, + ); + + const runtime = createMessageRuntime({ + assistantName: 'Andy', + idleTimeout: 1_000, + pollInterval: 1_000, + timezone: 'UTC', + triggerPattern: /^@Andy\b/i, + channels: [channel], + queue: { + registerProcess: vi.fn(), + closeStdin: vi.fn(), + notifyIdle, + } as any, + getRegisteredGroups: () => ({ [chatJid]: group }), + getSessions: () => ({}), + getLastTimestamp: () => '', + setLastTimestamp: vi.fn(), + getLastAgentTimestamps: () => ({}), + saveState: vi.fn(), + persistSession, + clearSession: vi.fn(), + }); + + const result = await runtime.processGroupMessages(chatJid, { + runId: 'run-progress', + reason: 'messages', + }); + + expect(result).toBe(true); + expect(channel.sendMessage).toHaveBeenCalledWith( + chatJid, + 'CI 상태 확인 중입니다.', + ); + expect(notifyIdle).toHaveBeenCalledTimes(1); + expect(notifyIdle).toHaveBeenCalledWith(chatJid, 'run-progress'); + expect(persistSession).toHaveBeenCalledWith( + group.folder, + 'session-progress', + ); + }); + + it('keeps progress separate from the final Codex answer', async () => { + const chatJid = 'group@test'; + const group = makeGroup('codex'); + const channel = makeChannel(chatJid); + const notifyIdle = vi.fn(); + + vi.mocked(db.getMessagesSince).mockReturnValue([ + { + id: 'msg-1', + chat_jid: chatJid, + sender: 'user@test', + sender_name: 'User', + content: 'hello', + timestamp: '2026-03-19T00:00:00.000Z', + }, + ]); + + vi.mocked(agentRunner.runAgentProcess).mockImplementation( + async (_group, _input, _onProcess, onOutput) => { + await onOutput?.({ + status: 'success', + phase: 'progress', + result: '테스트를 돌리는 중입니다.', + newSessionId: 'session-final', + }); + await onOutput?.({ + status: 'success', + phase: 'final', + result: '테스트가 끝났습니다.', + newSessionId: 'session-final', + }); + return { + status: 'success', + result: null, + newSessionId: 'session-final', + }; + }, + ); + + const runtime = createMessageRuntime({ + assistantName: 'Andy', + idleTimeout: 1_000, + pollInterval: 1_000, + timezone: 'UTC', + triggerPattern: /^@Andy\b/i, + channels: [channel], + queue: { + registerProcess: vi.fn(), + closeStdin: vi.fn(), + notifyIdle, + } as any, + getRegisteredGroups: () => ({ [chatJid]: group }), + getSessions: () => ({}), + getLastTimestamp: () => '', + setLastTimestamp: vi.fn(), + getLastAgentTimestamps: () => ({}), + saveState: vi.fn(), + persistSession: vi.fn(), + clearSession: vi.fn(), + }); + + const result = await runtime.processGroupMessages(chatJid, { + runId: 'run-final', + reason: 'messages', + }); + + expect(result).toBe(true); + expect(channel.sendMessage).toHaveBeenNthCalledWith( + 1, + chatJid, + '테스트를 돌리는 중입니다.', + ); + expect(channel.sendMessage).toHaveBeenNthCalledWith( + 2, + chatJid, + '테스트가 끝났습니다.', + ); + expect(notifyIdle).toHaveBeenCalledTimes(1); + expect(notifyIdle).toHaveBeenCalledWith(chatJid, 'run-final'); + }); + + it('does not roll back when a streamed progress message was already posted before an error', async () => { + const chatJid = 'group@test'; + const group = makeGroup('codex'); + const channel = makeChannel(chatJid); + const saveState = vi.fn(); + const lastAgentTimestamps: Record = {}; + + vi.mocked(db.getMessagesSince).mockReturnValue([ + { + id: 'msg-1', + chat_jid: chatJid, + sender: 'user@test', + sender_name: 'User', + content: 'hello', + timestamp: '2026-03-19T00:00:00.000Z', + }, + ]); + + vi.mocked(agentRunner.runAgentProcess).mockImplementation( + async (_group, _input, _onProcess, onOutput) => { + await onOutput?.({ + status: 'success', + phase: 'progress', + result: '중간 진행상황입니다.', + newSessionId: 'session-error', + }); + await onOutput?.({ + status: 'error', + result: null, + newSessionId: 'session-error', + error: 'temporary failure', + }); + return { + status: 'error', + result: null, + newSessionId: 'session-error', + error: 'temporary failure', + }; + }, + ); + + const runtime = createMessageRuntime({ + assistantName: 'Andy', + idleTimeout: 1_000, + pollInterval: 1_000, + timezone: 'UTC', + triggerPattern: /^@Andy\b/i, + channels: [channel], + queue: { + registerProcess: vi.fn(), + closeStdin: vi.fn(), + notifyIdle: vi.fn(), + } as any, + getRegisteredGroups: () => ({ [chatJid]: group }), + getSessions: () => ({}), + getLastTimestamp: () => '', + setLastTimestamp: vi.fn(), + getLastAgentTimestamps: () => lastAgentTimestamps, + saveState, + persistSession: vi.fn(), + clearSession: vi.fn(), + }); + + const result = await runtime.processGroupMessages(chatJid, { + runId: 'run-progress-error', + reason: 'messages', + }); + + expect(result).toBe(true); + expect(channel.sendMessage).toHaveBeenCalledWith( + chatJid, + '중간 진행상황입니다.', + ); + expect(lastAgentTimestamps[chatJid]).toBe('2026-03-19T00:00:00.000Z'); + expect(saveState).toHaveBeenCalled(); + }); +}); diff --git a/src/message-runtime.ts b/src/message-runtime.ts index 2eafdf4..56f88b5 100644 --- a/src/message-runtime.ts +++ b/src/message-runtime.ts @@ -14,13 +14,15 @@ import { } from './db.js'; import { isSessionCommandSenderAllowed } from './config.js'; import { GroupQueue, GroupRunContext } from './group-queue.js'; -import { findChannel, formatMessages } from './router.js'; +import { findChannel, formatMessages, formatOutbound } from './router.js'; import { isTriggerAllowed, loadSenderAllowlist } from './sender-allowlist.js'; import { extractSessionCommand, handleSessionCommand, isSessionCommandAllowed, + isSessionCommandControlMessage, } from './session-commands.js'; +import { shouldResetSessionOnAgentFailure } from './session-recovery.js'; import { Channel, NewMessage, RegisteredGroup } from './types.js'; import { logger } from './logger.js'; @@ -68,6 +70,18 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { } { let messageLoopRunning = false; + const filterOwnChannelMessages = (messages: NewMessage[]): NewMessage[] => + messages.filter((msg) => { + const channel = findChannel(deps.channels, msg.chat_jid); + if (channel?.isOwnMessage?.(msg)) { + return false; + } + if (msg.is_bot_message && isSessionCommandControlMessage(msg.content)) { + return false; + } + return true; + }); + const getCurrentAvailableGroups = (): AvailableGroup[] => getAvailableGroups(deps.getRegisteredGroups()); @@ -79,6 +93,7 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { onOutput?: (output: AgentOutput) => Promise, ): Promise<'success' | 'error'> => { const isMain = group.isMain === true; + const isClaudeCodeAgent = (group.agentType || 'claude-code') === 'claude-code'; const sessions = deps.getSessions(); const sessionId = sessions[group.folder]; @@ -99,11 +114,19 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { writeGroupsSnapshot(group.folder, isMain, getCurrentAvailableGroups()); + let resetSessionRequested = false; + const wrappedOnOutput = onOutput ? async (output: AgentOutput) => { if (output.newSessionId) { deps.persistSession(group.folder, output.newSessionId); } + if ( + isClaudeCodeAgent && + shouldResetSessionOnAgentFailure(output) + ) { + resetSessionRequested = true; + } await onOutput(output); } : undefined; @@ -129,6 +152,18 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { deps.persistSession(group.folder, output.newSessionId); } + if ( + isClaudeCodeAgent && + (resetSessionRequested || + shouldResetSessionOnAgentFailure(output)) + ) { + deps.clearSession(group.folder); + logger.warn( + { group: group.name, chatJid, runId }, + 'Cleared poisoned agent session after unrecoverable error', + ); + } + if (output.status === 'error') { logger.error( { group: group.name, chatJid, runId, error: output.error }, @@ -171,10 +206,12 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { const isMainGroup = group.isMain === true; const lastAgentTimestamps = deps.getLastAgentTimestamps(); const sinceTimestamp = lastAgentTimestamps[chatJid] || ''; - const missedMessages = getMessagesSince( - chatJid, - sinceTimestamp, - deps.assistantName, + const missedMessages = filterOwnChannelMessages( + getMessagesSince( + chatJid, + sinceTimestamp, + deps.assistantName, + ), ); if (missedMessages.length === 0) { @@ -280,6 +317,7 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { ); let idleTimer: ReturnType | null = null; + let latestProgressText: string | null = null; const resetIdleTimer = () => { if (idleTimer) clearTimeout(idleTimer); @@ -296,7 +334,20 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { }; let hadError = false; - let outputSentToUser = false; + let finalOutputSentToUser = false; + let progressOutputSentToUser = false; + let poisonedSessionDetected = false; + const isClaudeCodeAgent = (group.agentType || 'claude-code') === 'claude-code'; + + const sendProgressMessage = async (text: string) => { + if (!text || text === latestProgressText) { + return; + } + + latestProgressText = text; + await channel.sendMessage(chatJid, text); + progressOutputSentToUser = true; + }; await channel.setTyping?.(chatJid, true); @@ -306,14 +357,30 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { chatJid, runId, async (result) => { + if ( + isClaudeCodeAgent && + shouldResetSessionOnAgentFailure(result) && + !poisonedSessionDetected + ) { + poisonedSessionDetected = true; + hadError = true; + deps.clearSession(group.folder); + deps.queue.closeStdin(chatJid, { + runId, + reason: 'poisoned-session', + }); + logger.warn( + { chatJid, group: group.name, groupFolder: group.folder, runId }, + '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( { chatJid, @@ -324,16 +391,26 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { }, `Agent output: ${raw.slice(0, 200)}`, ); + if (result.phase === 'progress') { + if (text) { + await sendProgressMessage(text); + } + return; + } + if (text) { await channel.sendMessage(chatJid, text); - outputSentToUser = true; + finalOutputSentToUser = true; } } await channel.setTyping?.(chatJid, false); - resetIdleTimer(); - if (result.status === 'success') { + if (!poisonedSessionDetected) { + resetIdleTimer(); + } + + if (result.status === 'success' && !poisonedSessionDetected) { deps.queue.notifyIdle(chatJid, runId); } @@ -352,10 +429,10 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { if (idleTimer) clearTimeout(idleTimer); if (hadError) { - if (outputSentToUser) { + if (finalOutputSentToUser || progressOutputSentToUser) { logger.warn( { chatJid, group: group.name, groupFolder: group.folder, runId }, - 'Agent error after output was sent, skipping cursor rollback to prevent duplicates', + 'Agent error after conversational output was sent, skipping cursor rollback to prevent duplicates', ); return true; } @@ -374,7 +451,8 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { group: group.name, groupFolder: group.folder, runId, - outputSentToUser, + outputSentToUser: finalOutputSentToUser, + progressOutputSentToUser, }, 'Queued run completed successfully', ); @@ -395,18 +473,23 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { try { const registeredGroups = deps.getRegisteredGroups(); const jids = Object.keys(registeredGroups); - const { messages, newTimestamp } = getNewMessages( + const { messages: rawMessages, newTimestamp } = getNewMessages( jids, deps.getLastTimestamp(), deps.assistantName, ); + const messages = filterOwnChannelMessages(rawMessages); - if (messages.length > 0) { + if (rawMessages.length > 0) { logger.info({ count: messages.length }, 'New messages'); deps.setLastTimestamp(newTimestamp); deps.saveState(); + if (messages.length === 0) { + continue; + } + const messagesByGroup = new Map(); for (const msg of messages) { const existing = messagesByGroup.get(msg.chat_jid); @@ -484,10 +567,12 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { } const lastAgentTimestamps = deps.getLastAgentTimestamps(); - const allPending = getMessagesSince( - chatJid, - lastAgentTimestamps[chatJid] || '', - deps.assistantName, + const allPending = filterOwnChannelMessages( + getMessagesSince( + chatJid, + lastAgentTimestamps[chatJid] || '', + deps.assistantName, + ), ); const messagesToSend = allPending.length > 0 ? allPending : groupMessages; @@ -526,10 +611,12 @@ export function createMessageRuntime(deps: MessageRuntimeDeps): { const lastAgentTimestamps = deps.getLastAgentTimestamps(); for (const [chatJid, group] of Object.entries(registeredGroups)) { const sinceTimestamp = lastAgentTimestamps[chatJid] || ''; - const pending = getMessagesSince( - chatJid, - sinceTimestamp, - deps.assistantName, + const pending = filterOwnChannelMessages( + getMessagesSince( + chatJid, + sinceTimestamp, + deps.assistantName, + ), ); if (pending.length > 0) { logger.info( diff --git a/src/platform-prompts.test.ts b/src/platform-prompts.test.ts new file mode 100644 index 0000000..d4145a6 --- /dev/null +++ b/src/platform-prompts.test.ts @@ -0,0 +1,58 @@ +import fs from 'fs'; +import os from 'os'; +import path from 'path'; + +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; + +import { + getPairedRoomPromptPath, + getPlatformPromptPath, + readPairedRoomPrompt, + readPlatformPrompt, +} from './platform-prompts.js'; + +describe('platform-prompts', () => { + let tempDir: string; + + beforeEach(() => { + tempDir = fs.mkdtempSync(path.join(os.tmpdir(), 'nanoclaw-prompts-')); + vi.spyOn(process, 'cwd').mockReturnValue(tempDir); + }); + + afterEach(() => { + vi.restoreAllMocks(); + fs.rmSync(tempDir, { recursive: true, force: true }); + }); + + it('returns undefined when the prompt file is missing', () => { + expect(readPlatformPrompt('claude-code')).toBeUndefined(); + }); + + it('reads and trims provider-specific prompt files', () => { + const promptsDir = path.join(tempDir, 'prompts'); + fs.mkdirSync(promptsDir, { recursive: true }); + fs.writeFileSync( + path.join(promptsDir, 'codex-platform.md'), + '\nCodex platform prompt\n', + ); + + expect(getPlatformPromptPath('codex')).toBe( + path.join(promptsDir, 'codex-platform.md'), + ); + expect(readPlatformPrompt('codex')).toBe('Codex platform prompt'); + }); + + it('reads and trims paired-room prompt files', () => { + const promptsDir = path.join(tempDir, 'prompts'); + fs.mkdirSync(promptsDir, { recursive: true }); + fs.writeFileSync( + path.join(promptsDir, 'claude-paired-room.md'), + '\nClaude paired prompt\n', + ); + + expect(getPairedRoomPromptPath('claude-code')).toBe( + path.join(promptsDir, 'claude-paired-room.md'), + ); + expect(readPairedRoomPrompt('claude-code')).toBe('Claude paired prompt'); + }); +}); diff --git a/src/platform-prompts.ts b/src/platform-prompts.ts new file mode 100644 index 0000000..e3c15c7 --- /dev/null +++ b/src/platform-prompts.ts @@ -0,0 +1,60 @@ +import fs from 'fs'; +import path from 'path'; + +import type { AgentType } from './types.js'; + +const PLATFORM_PROMPT_FILES: Record = { + 'claude-code': 'claude-platform.md', + codex: 'codex-platform.md', +}; + +const PAIRED_ROOM_PROMPT_FILES: Record = { + 'claude-code': 'claude-paired-room.md', + codex: 'codex-paired-room.md', +}; + +export function getPlatformPromptsDir(projectRoot = process.cwd()): string { + return path.join(projectRoot, 'prompts'); +} + +export function getPlatformPromptPath( + agentType: AgentType, + projectRoot = process.cwd(), +): string { + return path.join( + getPlatformPromptsDir(projectRoot), + PLATFORM_PROMPT_FILES[agentType], + ); +} + +export function readPlatformPrompt( + agentType: AgentType, + projectRoot = process.cwd(), +): string | undefined { + const promptPath = getPlatformPromptPath(agentType, projectRoot); + if (!fs.existsSync(promptPath)) return undefined; + + const prompt = fs.readFileSync(promptPath, 'utf-8').trim(); + return prompt || undefined; +} + +export function getPairedRoomPromptPath( + agentType: AgentType, + projectRoot = process.cwd(), +): string { + return path.join( + getPlatformPromptsDir(projectRoot), + PAIRED_ROOM_PROMPT_FILES[agentType], + ); +} + +export function readPairedRoomPrompt( + agentType: AgentType, + projectRoot = process.cwd(), +): string | undefined { + const promptPath = getPairedRoomPromptPath(agentType, projectRoot); + if (!fs.existsSync(promptPath)) return undefined; + + const prompt = fs.readFileSync(promptPath, 'utf-8').trim(); + return prompt || undefined; +} diff --git a/src/session-commands.test.ts b/src/session-commands.test.ts index f6a2d1d..abbc7e4 100644 --- a/src/session-commands.test.ts +++ b/src/session-commands.test.ts @@ -2,6 +2,7 @@ import { describe, it, expect, vi } from 'vitest'; import { extractSessionCommand, handleSessionCommand, + isSessionCommandControlMessage, isSessionCommandAllowed, } from './session-commands.js'; import type { NewMessage } from './types.js'; @@ -67,6 +68,28 @@ describe('isSessionCommandAllowed', () => { }); }); +describe('isSessionCommandControlMessage', () => { + it('matches clear confirmation output', () => { + expect( + isSessionCommandControlMessage( + 'Current session cleared. The next message will start a new conversation.', + ), + ).toBe(true); + }); + + it('matches admin denial output', () => { + expect( + isSessionCommandControlMessage('Session commands require admin access.'), + ).toBe(true); + }); + + it('does not match regular bot conversation', () => { + expect(isSessionCommandControlMessage('좋네요. 필요해지면 바로 말씀드리겠습니다.')).toBe( + false, + ); + }); +}); + function makeMsg( content: string, overrides: Partial = {}, diff --git a/src/session-commands.ts b/src/session-commands.ts index 170c54b..5928ecd 100644 --- a/src/session-commands.ts +++ b/src/session-commands.ts @@ -1,5 +1,14 @@ import type { NewMessage } from './types.js'; import { logger } from './logger.js'; +import { formatOutbound } from './router.js'; + +const SESSION_COMMAND_CONTROL_PATTERNS = [ + /^Current session cleared\. The next message will start a new conversation\.$/, + /^Session commands require admin access\.$/, + /^Failed to process messages before \/compact\. Try again\.$/, + /^\/compact failed\. The session is unchanged\.$/, + /^Conversation compacted\.$/, +]; /** * Extract a session slash command from a message, stripping the trigger prefix if present. @@ -28,10 +37,18 @@ export function isSessionCommandAllowed( return isMainGroup || isFromMe || isAdminSender; } +export function isSessionCommandControlMessage(content: string): boolean { + const trimmed = content.trim(); + return SESSION_COMMAND_CONTROL_PATTERNS.some((pattern) => + pattern.test(trimmed), + ); +} + /** Minimal agent result interface — matches the subset of AgentOutput used here. */ export interface AgentResult { status: 'success' | 'error'; result?: string | object | null; + phase?: 'progress' | 'final'; } /** Dependencies injected by the orchestrator. */ @@ -54,7 +71,7 @@ export interface SessionCommandDeps { function resultToText(result: string | object | null | undefined): string { if (!result) return ''; const raw = typeof result === 'string' ? result : JSON.stringify(result); - return raw.replace(/[\s\S]*?<\/internal>/g, '').trim(); + return formatOutbound(raw); } /** @@ -133,6 +150,7 @@ export async function handleSessionCommand(opts: { const preResult = await deps.runAgent(prePrompt, async (result) => { if (result.status === 'error') hadPreError = true; + if (result.phase === 'progress') return; const text = resultToText(result.result); if (text) { await deps.sendMessage(text); @@ -169,6 +187,7 @@ export async function handleSessionCommand(opts: { let hadCmdError = false; const cmdOutput = await deps.runAgent(command, async (result) => { if (result.status === 'error') hadCmdError = true; + if (result.phase === 'progress') return; const text = resultToText(result.result); if (text) await deps.sendMessage(text); }); 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/task-scheduler.ts b/src/task-scheduler.ts index a0a9dee..2be69a8 100644 --- a/src/task-scheduler.ts +++ b/src/task-scheduler.ts @@ -183,6 +183,9 @@ async function runTask( (proc, processName) => deps.onProcess(task.chat_jid, proc, processName, task.group_folder), async (streamedOutput: AgentOutput) => { + if (streamedOutput.phase === 'progress') { + return; + } if (streamedOutput.result) { result = streamedOutput.result; // Forward result to user (sendMessage handles formatting) diff --git a/src/types.ts b/src/types.ts index 316b284..60d2fd2 100644 --- a/src/types.ts +++ b/src/types.ts @@ -78,6 +78,8 @@ export interface Channel { sendMessage(jid: string, text: string): Promise; isConnected(): boolean; ownsJid(jid: string): boolean; + // Optional: whether a stored inbound message was authored by this channel's own bot/user. + isOwnMessage?(msg: NewMessage): boolean; disconnect(): Promise; // Optional: typing indicator. Channels that support it implement it. setTyping?(jid: string, isTyping: boolean): Promise; diff --git a/vitest.config.ts b/vitest.config.ts index a456d1c..a887ef0 100644 --- a/vitest.config.ts +++ b/vitest.config.ts @@ -2,6 +2,10 @@ import { defineConfig } from 'vitest/config'; export default defineConfig({ test: { - include: ['src/**/*.test.ts', 'setup/**/*.test.ts'], + include: [ + 'src/**/*.test.ts', + 'setup/**/*.test.ts', + 'runners/**/test/**/*.test.ts', + ], }, });