From 88f8b3006ce0ad0b21db6865aa5bf004d1638a99 Mon Sep 17 00:00:00 2001 From: ejclaw Date: Sat, 11 Apr 2026 05:49:40 +0900 Subject: [PATCH] runtime: split ipc queue and runner helpers --- runners/agent-runner/src/index.ts | 562 ++-------------- runners/agent-runner/src/ipc-input.ts | 52 ++ runners/agent-runner/src/output-protocol.ts | 174 +++++ runners/agent-runner/src/runner-hooks.ts | 304 +++++++++ src/group-queue-state.ts | 203 ++++++ src/group-queue.ts | 214 +----- src/ipc-file-claims.ts | 65 ++ src/ipc-message-forwarding.ts | 45 ++ src/ipc-task-processor.ts | 502 ++++++++++++++ src/ipc-types.ts | 71 ++ src/ipc.ts | 695 +------------------- 11 files changed, 1498 insertions(+), 1389 deletions(-) create mode 100644 runners/agent-runner/src/ipc-input.ts create mode 100644 runners/agent-runner/src/output-protocol.ts create mode 100644 runners/agent-runner/src/runner-hooks.ts create mode 100644 src/group-queue-state.ts create mode 100644 src/ipc-file-claims.ts create mode 100644 src/ipc-message-forwarding.ts create mode 100644 src/ipc-task-processor.ts create mode 100644 src/ipc-types.ts diff --git a/runners/agent-runner/src/index.ts b/runners/agent-runner/src/index.ts index 092e23e..68d10ad 100644 --- a/runners/agent-runner/src/index.ts +++ b/runners/agent-runner/src/index.ts @@ -16,20 +16,11 @@ import fs from 'fs'; import path from 'path'; +import { query } from '@anthropic-ai/claude-agent-sdk'; import { - query, - HookCallback, - PreCompactHookInput, - PreToolUseHookInput, -} from '@anthropic-ai/claude-agent-sdk'; -import { - extractImageTagPaths, IPC_CLOSE_SENTINEL, IPC_INPUT_SUBDIR, IPC_POLL_MS, - normalizePublicTextOutput, - type RunnerStructuredOutput, - writeProtocolOutput, } from 'ejclaw-runners-shared'; import { fileURLToPath } from 'url'; @@ -44,11 +35,24 @@ import { getClaudeReadonlySandboxMode, isArbiterRuntimeEnvEnabled, isClaudeReadonlyReviewerRuntime, - isReviewerMutatingShellCommand, isReviewerRuntime, isReviewerRuntimeEnvEnabled, } from './reviewer-runtime.js'; -import { selectCompactMemoriesFromSummary } from './memory-selection.js'; +import { drainIpcInput, shouldClose } from './ipc-input.js'; +import { + MessageStream, + buildMultimodalContent, + extractAssistantText, + normalizeStructuredOutput, + readStdin, + type RunnerOutput, + writeOutput, +} from './output-protocol.js'; +import { + createPreCompactHook, + createReviewerBashGuardHook, + createSanitizeBashHook, +} from './runner-hooks.js'; interface RunnerInput { prompt: string; @@ -62,53 +66,6 @@ interface RunnerInput { roomRoleContext?: RoomRoleContext; } -/** Mirrors AgentOutput in src/agent-runner.ts (separate package, can't import directly). */ -interface RunnerOutput { - status: 'success' | 'error'; - phase?: 'progress' | 'final' | 'tool-activity' | 'intermediate'; - agentId?: string; - agentLabel?: string; - agentDone?: boolean; - result: string | null; - output?: RunnerStructuredOutput; - newSessionId?: string; - error?: string; -} - -interface SessionEntry { - sessionId: string; - fullPath: string; - summary: string; - firstPrompt: string; -} - -interface SessionsIndex { - entries: SessionEntry[]; -} - -type ContentBlock = - | { type: 'text'; text: string } - | { - type: 'image'; - source: { - type: 'base64'; - media_type: 'image/jpeg' | 'image/png' | 'image/gif' | 'image/webp'; - data: string; - }; - }; - -interface SDKUserMessage { - type: 'user'; - message: { role: 'user'; content: string | ContentBlock[] }; - parent_tool_use_id: null; - session_id: string; -} - -interface AssistantContentBlock { - type?: string; - text?: string; -} - // Paths configurable via env vars. const GROUP_DIR = process.env.EJCLAW_GROUP_DIR || '/workspace/group'; const IPC_DIR = process.env.EJCLAW_IPC_DIR || '/workspace/ipc'; @@ -120,116 +77,6 @@ const GROUP_FOLDER = process.env.EJCLAW_GROUP_FOLDER || ''; const IPC_INPUT_DIR = path.join(IPC_DIR, IPC_INPUT_SUBDIR); const IPC_INPUT_CLOSE_SENTINEL = path.join(IPC_INPUT_DIR, IPC_CLOSE_SENTINEL); const HOST_TASKS_DIR = path.join(HOST_IPC_DIR, 'tasks'); -const MIME_TYPES: Record = { - '.jpg': 'image/jpeg', - '.jpeg': 'image/jpeg', - '.png': 'image/png', - '.gif': 'image/gif', - '.webp': 'image/webp', -}; - -/** - * Parse [Image: /absolute/path] tags from text and build multimodal content. - * Returns a plain string if no images found, or ContentBlock[] with text + image blocks. - */ -function buildMultimodalContent(text: string): string | ContentBlock[] { - const { cleanText, imagePaths } = extractImageTagPaths(text); - if (imagePaths.length === 0) return text; - - const blocks: ContentBlock[] = []; - if (cleanText) { - blocks.push({ type: 'text', text: cleanText }); - } - - for (const imgPath of imagePaths) { - try { - if (!fs.existsSync(imgPath)) { - log(`Image not found, skipping: ${imgPath}`); - continue; - } - const data = fs.readFileSync(imgPath).toString('base64'); - const ext = path.extname(imgPath).toLowerCase(); - const mediaType = (MIME_TYPES[ext] || 'image/png') as - | 'image/jpeg' - | 'image/png' - | 'image/gif' - | 'image/webp'; - blocks.push({ - type: 'image', - source: { type: 'base64', media_type: mediaType, data }, - }); - log(`Added image block: ${imgPath} (${mediaType})`); - } catch (err) { - log( - `Failed to read image ${imgPath}: ${err instanceof Error ? err.message : String(err)}`, - ); - } - } - - return blocks.length > 0 ? blocks : text; -} - -/** - * Push-based async iterable for streaming user messages to the SDK. - * Keeps the iterable alive until end() is called, preventing isSingleUserTurn. - */ -class MessageStream { - private queue: SDKUserMessage[] = []; - private waiting: (() => void) | null = null; - private done = false; - - push(text: string): void { - const content = buildMultimodalContent(text); - this.queue.push({ - type: 'user', - message: { role: 'user', content }, - parent_tool_use_id: null, - session_id: '', - }); - this.waiting?.(); - } - - end(): void { - this.done = true; - this.waiting?.(); - } - - async *[Symbol.asyncIterator](): AsyncGenerator { - while (true) { - while (this.queue.length > 0) { - yield this.queue.shift()!; - } - if (this.done) return; - await new Promise((r) => { - this.waiting = r; - }); - this.waiting = null; - } - } -} - -async function readStdin(): Promise { - return new Promise((resolve, reject) => { - let data = ''; - process.stdin.setEncoding('utf8'); - process.stdin.on('data', (chunk) => { - data += chunk; - }); - process.stdin.on('end', () => resolve(data)); - process.stdin.on('error', reject); - }); -} - -function writeOutput(output: RunnerOutput): void { - writeProtocolOutput(output); -} - -function normalizeStructuredOutput(result: string | null): { - result: string | null; - output?: RunnerOutput['output']; -} { - return normalizePublicTextOutput(result); -} function log(message: string): void { console.error(`[agent-runner] ${message}`); @@ -242,349 +89,6 @@ process.on('SIGTERM', () => { agentAbortController.abort(); }); -function extractAssistantText(message: unknown): string | null { - const assistant = message as { - message?: { - content?: AssistantContentBlock[]; - }; - }; - const blocks = assistant.message?.content; - if (!Array.isArray(blocks)) return null; - - const text = blocks - .filter((block) => block?.type === 'text' && typeof block.text === 'string') - .map((block) => block.text!.trim()) - .filter(Boolean) - .join('\n\n') - .trim(); - - return text || null; -} - -function getSessionSummary( - sessionId: string, - transcriptPath: string, -): string | null { - const projectDir = path.dirname(transcriptPath); - const indexPath = path.join(projectDir, 'sessions-index.json'); - - if (!fs.existsSync(indexPath)) { - log(`Sessions index not found at ${indexPath}`); - return null; - } - - try { - const index: SessionsIndex = JSON.parse( - fs.readFileSync(indexPath, 'utf-8'), - ); - const entry = index.entries.find((e) => e.sessionId === sessionId); - if (entry?.summary) { - return entry.summary; - } - } catch (err) { - log( - `Failed to read sessions index: ${err instanceof Error ? err.message : String(err)}`, - ); - } - - return null; -} - -function trimSummary(summary: string, maxChars: number): string { - if (summary.length <= maxChars) return summary; - return summary.slice(0, Math.max(0, maxChars - 1)).trimEnd() + '…'; -} - -function writeHostTaskIpcFile(data: object): string { - fs.mkdirSync(HOST_TASKS_DIR, { recursive: true }); - const filename = `${Date.now()}-${Math.random().toString(36).slice(2, 8)}.json`; - const filepath = path.join(HOST_TASKS_DIR, filename); - const tempPath = `${filepath}.tmp`; - fs.writeFileSync(tempPath, JSON.stringify(data, null, 2)); - fs.renameSync(tempPath, filepath); - return filename; -} - -async function persistCompactMemory( - summary: string, - sessionId: string, -): Promise { - const normalized = summary.trim(); - if (!normalized || !GROUP_FOLDER) return; - - try { - const scopeKey = `room:${GROUP_FOLDER}`; - const selected = selectCompactMemoriesFromSummary(normalized, scopeKey); - if (selected.length === 0) { - log('Skipped compact memory persist - no salient room memory found'); - return; - } - - for (const memory of selected) { - const file = writeHostTaskIpcFile({ - type: 'persist_memory', - scopeKind: 'room', - scopeKey, - content: trimSummary(memory.content, 300), - keywords: memory.keywords, - memory_kind: memory.memoryKind, - source_kind: 'compact', - source_ref: sessionId ? `compact:${sessionId}` : null, - timestamp: new Date().toISOString(), - }); - log(`Persisted compact memory via IPC (${file})`); - } - } catch (err) { - log( - `Failed to persist compact memory via IPC: ${ - err instanceof Error ? err.message : String(err) - }`, - ); - } -} - -/** - * Archive the full transcript to conversations/ before compaction. - */ -function createPreCompactHook(assistantName?: string): HookCallback { - return async (input, _toolUseId, _context) => { - const preCompact = input as PreCompactHookInput; - const transcriptPath = preCompact.transcript_path; - const sessionId = preCompact.session_id; - const trigger = preCompact.trigger || 'auto'; - - // Show compact status in chat so users know it's not just slow loading - writeOutput({ - status: 'success', - phase: 'progress', - result: trigger === 'auto' ? '대화 요약 중...' : '컴팩트 중...', - }); - - if (!transcriptPath || !fs.existsSync(transcriptPath)) { - log('No transcript found for archiving'); - return {}; - } - - try { - const content = fs.readFileSync(transcriptPath, 'utf-8'); - const messages = parseTranscript(content); - - if (messages.length === 0) { - log('No messages to archive'); - return {}; - } - - const summary = getSessionSummary(sessionId, transcriptPath); - const name = summary ? sanitizeFilename(summary) : generateFallbackName(); - - const conversationsDir = path.join(GROUP_DIR, 'conversations'); - fs.mkdirSync(conversationsDir, { recursive: true }); - - const date = new Date().toISOString().split('T')[0]; - const filename = `${date}-${name}.md`; - const filePath = path.join(conversationsDir, filename); - - const markdown = formatTranscriptMarkdown( - messages, - summary, - assistantName, - ); - fs.writeFileSync(filePath, markdown); - - log(`Archived conversation to ${filePath}`); - - if (summary) { - await persistCompactMemory(summary, sessionId); - } - } catch (err) { - log( - `Failed to archive transcript: ${err instanceof Error ? err.message : String(err)}`, - ); - } - - return {}; - }; -} - -// Secrets to strip from Bash tool subprocess environments. -// These are needed by claude-code for API auth but should never -// be visible to commands Kit runs. -const SECRET_ENV_VARS = ['ANTHROPIC_API_KEY', 'CLAUDE_CODE_OAUTH_TOKEN']; - -function createSanitizeBashHook(): HookCallback { - return async (input, _toolUseId, _context) => { - const preInput = input as PreToolUseHookInput; - const command = (preInput.tool_input as { command?: string })?.command; - if (!command) return {}; - - const unsetPrefix = `unset ${SECRET_ENV_VARS.join(' ')} 2>/dev/null; `; - return { - hookSpecificOutput: { - hookEventName: 'PreToolUse', - updatedInput: { - ...(preInput.tool_input as Record), - command: unsetPrefix + command, - }, - }, - }; - }; -} - -function createReviewerBashGuardHook(): HookCallback { - return async (input, _toolUseId, _context) => { - const preInput = input as PreToolUseHookInput; - const command = (preInput.tool_input as { command?: string })?.command; - if (!command || !isReviewerMutatingShellCommand(command)) { - return {}; - } - - return { - hookSpecificOutput: { - hookEventName: 'PreToolUse', - permissionDecision: 'deny', - permissionDecisionReason: - 'EJClaw reviewer runtime blocks mutating shell commands in paired review mode.', - }, - }; - }; -} - -function sanitizeFilename(summary: string): string { - return summary - .toLowerCase() - .replace(/[^a-z0-9]+/g, '-') - .replace(/^-+|-+$/g, '') - .slice(0, 50); -} - -function generateFallbackName(): string { - const time = new Date(); - return `conversation-${time.getHours().toString().padStart(2, '0')}${time.getMinutes().toString().padStart(2, '0')}`; -} - -interface ParsedMessage { - role: 'user' | 'assistant'; - content: string; -} - -function parseTranscript(content: string): ParsedMessage[] { - const messages: ParsedMessage[] = []; - - for (const line of content.split('\n')) { - if (!line.trim()) continue; - try { - const entry = JSON.parse(line); - if (entry.type === 'user' && entry.message?.content) { - const text = - typeof entry.message.content === 'string' - ? entry.message.content - : entry.message.content - .map((c: { text?: string }) => c.text || '') - .join(''); - if (text) messages.push({ role: 'user', content: text }); - } else if (entry.type === 'assistant' && entry.message?.content) { - const textParts = entry.message.content - .filter((c: { type: string }) => c.type === 'text') - .map((c: { text: string }) => c.text); - const text = textParts.join(''); - if (text) messages.push({ role: 'assistant', content: text }); - } - } catch {} - } - - return messages; -} - -function formatTranscriptMarkdown( - messages: ParsedMessage[], - title?: string | null, - assistantName?: string, -): string { - const now = new Date(); - const formatDateTime = (d: Date) => - d.toLocaleString('en-US', { - month: 'short', - day: 'numeric', - hour: 'numeric', - minute: '2-digit', - hour12: true, - }); - - const lines: string[] = []; - lines.push(`# ${title || 'Conversation'}`); - lines.push(''); - lines.push(`Archived: ${formatDateTime(now)}`); - lines.push(''); - lines.push('---'); - lines.push(''); - - for (const msg of messages) { - const sender = msg.role === 'user' ? 'User' : assistantName || 'Assistant'; - const content = - msg.content.length > 2000 - ? msg.content.slice(0, 2000) + '...' - : msg.content; - lines.push(`**${sender}**: ${content}`); - lines.push(''); - } - - return lines.join('\n'); -} - -/** - * Check for _close sentinel. - */ -function shouldClose(): boolean { - if (fs.existsSync(IPC_INPUT_CLOSE_SENTINEL)) { - try { - fs.unlinkSync(IPC_INPUT_CLOSE_SENTINEL); - } catch { - /* ignore */ - } - return true; - } - return false; -} - -/** - * Drain all pending IPC input messages. - * Returns messages found, or empty array. - */ -function drainIpcInput(): string[] { - try { - fs.mkdirSync(IPC_INPUT_DIR, { recursive: true }); - const files = fs - .readdirSync(IPC_INPUT_DIR) - .filter((f) => f.endsWith('.json')) - .sort(); - - const messages: string[] = []; - for (const file of files) { - const filePath = path.join(IPC_INPUT_DIR, file); - try { - const data = JSON.parse(fs.readFileSync(filePath, 'utf-8')); - fs.unlinkSync(filePath); - if (data.type === 'message' && data.text) { - 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 */ - } - } - } - return messages; - } catch (err) { - log(`IPC drain error: ${err instanceof Error ? err.message : String(err)}`); - return []; - } -} - /** * Run a single query and stream results via writeOutput. * Uses MessageStream (AsyncIterable) to keep isSingleUserTurn=false, @@ -605,7 +109,7 @@ async function runQuery( closedDuringQuery: boolean; terminalResultObserved: boolean; }> { - const stream = new MessageStream(); + const stream = new MessageStream((text) => buildMultimodalContent(text, log)); stream.push(prompt); // Poll IPC for follow-up messages and _close sentinel during the query @@ -613,7 +117,7 @@ async function runQuery( let closedDuringQuery = false; const pollIpcDuringQuery = () => { if (!ipcPolling) return; - if (shouldClose()) { + if (shouldClose(IPC_INPUT_CLOSE_SENTINEL)) { log('Close sentinel detected during query, ending stream'); // Flush any buffered text before closing — the for-await loop may not // reach the post-loop flush code after stream.end(). @@ -635,7 +139,7 @@ async function runQuery( ipcPolling = false; return; } - const messages = drainIpcInput(); + const messages = drainIpcInput(IPC_INPUT_DIR, log); for (const text of messages) { log(`Piping IPC message into active query (${text.length} chars)`); stream.push(text); @@ -794,7 +298,18 @@ async function runQuery( }, hooks: { PreCompact: [ - { hooks: [createPreCompactHook(runnerInput.assistantName)] }, + { + hooks: [ + createPreCompactHook({ + assistantName: runnerInput.assistantName, + groupDir: GROUP_DIR, + groupFolder: GROUP_FOLDER, + hostTasksDir: HOST_TASKS_DIR, + log, + writeOutput, + }), + ], + }, ], PreToolUse: [ { @@ -1163,7 +678,7 @@ async function main(): Promise { if (runnerInput.isScheduledTask) { prompt = `[SCHEDULED TASK - The following message was sent automatically and is not coming directly from the user or group.]\n\n${prompt}`; } - const pending = drainIpcInput(); + const pending = drainIpcInput(IPC_INPUT_DIR, log); if (pending.length > 0) { log(`Draining ${pending.length} pending IPC messages into initial prompt`); prompt += '\n' + pending.join('\n'); @@ -1202,7 +717,18 @@ async function main(): Promise { abortController: agentAbortController, hooks: { PreCompact: [ - { hooks: [createPreCompactHook(runnerInput.assistantName)] }, + { + hooks: [ + createPreCompactHook({ + assistantName: runnerInput.assistantName, + groupDir: GROUP_DIR, + groupFolder: GROUP_FOLDER, + hostTasksDir: HOST_TASKS_DIR, + log, + writeOutput, + }), + ], + }, ], }, }, diff --git a/runners/agent-runner/src/ipc-input.ts b/runners/agent-runner/src/ipc-input.ts new file mode 100644 index 0000000..3349822 --- /dev/null +++ b/runners/agent-runner/src/ipc-input.ts @@ -0,0 +1,52 @@ +import fs from 'fs'; +import path from 'path'; + +export function shouldClose(closeSentinelPath: string): boolean { + if (fs.existsSync(closeSentinelPath)) { + try { + fs.unlinkSync(closeSentinelPath); + } catch { + // Ignore cleanup errors on shutdown signal files. + } + return true; + } + return false; +} + +export function drainIpcInput( + inputDir: string, + log: (message: string) => void, +): string[] { + try { + fs.mkdirSync(inputDir, { recursive: true }); + const files = fs + .readdirSync(inputDir) + .filter((file) => file.endsWith('.json')) + .sort(); + + const messages: string[] = []; + for (const file of files) { + const filePath = path.join(inputDir, file); + try { + const data = JSON.parse(fs.readFileSync(filePath, 'utf-8')); + fs.unlinkSync(filePath); + if (data.type === 'message' && data.text) { + 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 best-effort cleanup failures. + } + } + } + return messages; + } catch (err) { + log(`IPC drain error: ${err instanceof Error ? err.message : String(err)}`); + return []; + } +} diff --git a/runners/agent-runner/src/output-protocol.ts b/runners/agent-runner/src/output-protocol.ts new file mode 100644 index 0000000..8d38f0d --- /dev/null +++ b/runners/agent-runner/src/output-protocol.ts @@ -0,0 +1,174 @@ +import fs from 'fs'; +import path from 'path'; + +import { + extractImageTagPaths, + normalizePublicTextOutput, + type RunnerStructuredOutput, + writeProtocolOutput, +} from 'ejclaw-runners-shared'; + +export interface RunnerOutput { + status: 'success' | 'error'; + phase?: 'progress' | 'final' | 'tool-activity' | 'intermediate'; + agentId?: string; + agentLabel?: string; + agentDone?: boolean; + result: string | null; + output?: RunnerStructuredOutput; + newSessionId?: string; + error?: string; +} + +type ContentBlock = + | { type: 'text'; text: string } + | { + type: 'image'; + source: { + type: 'base64'; + media_type: 'image/jpeg' | 'image/png' | 'image/gif' | 'image/webp'; + data: string; + }; + }; + +interface SDKUserMessage { + type: 'user'; + message: { role: 'user'; content: string | ContentBlock[] }; + parent_tool_use_id: null; + session_id: string; +} + +interface AssistantContentBlock { + type?: string; + text?: string; +} + +export type LogFn = (message: string) => void; +type StreamContent = string | ContentBlock[]; + +const MIME_TYPES: Record = { + '.jpg': 'image/jpeg', + '.jpeg': 'image/jpeg', + '.png': 'image/png', + '.gif': 'image/gif', + '.webp': 'image/webp', +}; + +export function buildMultimodalContent( + text: string, + log: LogFn, +): StreamContent { + const { cleanText, imagePaths } = extractImageTagPaths(text); + if (imagePaths.length === 0) return text; + + const blocks: ContentBlock[] = []; + if (cleanText) { + blocks.push({ type: 'text', text: cleanText }); + } + + for (const imgPath of imagePaths) { + try { + if (!fs.existsSync(imgPath)) { + log(`Image not found, skipping: ${imgPath}`); + continue; + } + const data = fs.readFileSync(imgPath).toString('base64'); + const ext = path.extname(imgPath).toLowerCase(); + const mediaType = (MIME_TYPES[ext] || 'image/png') as + | 'image/jpeg' + | 'image/png' + | 'image/gif' + | 'image/webp'; + blocks.push({ + type: 'image', + source: { type: 'base64', media_type: mediaType, data }, + }); + log(`Added image block: ${imgPath} (${mediaType})`); + } catch (err) { + log( + `Failed to read image ${imgPath}: ${err instanceof Error ? err.message : String(err)}`, + ); + } + } + + return blocks.length > 0 ? blocks : text; +} + +export class MessageStream { + private queue: SDKUserMessage[] = []; + private waiting: (() => void) | null = null; + private done = false; + + constructor(private readonly buildContent: (text: string) => StreamContent) {} + + push(text: string): void { + const content = this.buildContent(text); + this.queue.push({ + type: 'user', + message: { role: 'user', content }, + parent_tool_use_id: null, + session_id: '', + }); + this.waiting?.(); + } + + end(): void { + this.done = true; + this.waiting?.(); + } + + async *[Symbol.asyncIterator](): AsyncGenerator { + while (true) { + while (this.queue.length > 0) { + yield this.queue.shift()!; + } + if (this.done) return; + await new Promise((resolve) => { + this.waiting = resolve; + }); + this.waiting = null; + } + } +} + +export async function readStdin(): Promise { + return new Promise((resolve, reject) => { + let data = ''; + process.stdin.setEncoding('utf8'); + process.stdin.on('data', (chunk) => { + data += chunk; + }); + process.stdin.on('end', () => resolve(data)); + process.stdin.on('error', reject); + }); +} + +export function writeOutput(output: RunnerOutput): void { + writeProtocolOutput(output); +} + +export function normalizeStructuredOutput(result: string | null): { + result: string | null; + output?: RunnerOutput['output']; +} { + return normalizePublicTextOutput(result); +} + +export function extractAssistantText(message: unknown): string | null { + const assistant = message as { + message?: { + content?: AssistantContentBlock[]; + }; + }; + const blocks = assistant.message?.content; + if (!Array.isArray(blocks)) return null; + + const text = blocks + .filter((block) => block?.type === 'text' && typeof block.text === 'string') + .map((block) => block.text!.trim()) + .filter(Boolean) + .join('\n\n') + .trim(); + + return text || null; +} diff --git a/runners/agent-runner/src/runner-hooks.ts b/runners/agent-runner/src/runner-hooks.ts new file mode 100644 index 0000000..accc8ba --- /dev/null +++ b/runners/agent-runner/src/runner-hooks.ts @@ -0,0 +1,304 @@ +import fs from 'fs'; +import path from 'path'; + +import { + HookCallback, + PreCompactHookInput, + PreToolUseHookInput, +} from '@anthropic-ai/claude-agent-sdk'; + +import { selectCompactMemoriesFromSummary } from './memory-selection.js'; +import type { RunnerOutput } from './output-protocol.js'; +import { isReviewerMutatingShellCommand } from './reviewer-runtime.js'; + +interface SessionEntry { + sessionId: string; + fullPath: string; + summary: string; + firstPrompt: string; +} + +interface SessionsIndex { + entries: SessionEntry[]; +} + +interface ParsedMessage { + role: 'user' | 'assistant'; + content: string; +} + +interface PreCompactHookOptions { + assistantName?: string; + groupDir: string; + groupFolder: string; + hostTasksDir: string; + log: (message: string) => void; + writeOutput: (output: RunnerOutput) => void; +} + +const SECRET_ENV_VARS = ['ANTHROPIC_API_KEY', 'CLAUDE_CODE_OAUTH_TOKEN']; + +function getSessionSummary( + sessionId: string, + transcriptPath: string, + log: (message: string) => void, +): string | null { + const projectDir = path.dirname(transcriptPath); + const indexPath = path.join(projectDir, 'sessions-index.json'); + + if (!fs.existsSync(indexPath)) { + log(`Sessions index not found at ${indexPath}`); + return null; + } + + try { + const index: SessionsIndex = JSON.parse( + fs.readFileSync(indexPath, 'utf-8'), + ); + const entry = index.entries.find( + (current) => current.sessionId === sessionId, + ); + if (entry?.summary) { + return entry.summary; + } + } catch (err) { + log( + `Failed to read sessions index: ${err instanceof Error ? err.message : String(err)}`, + ); + } + + return null; +} + +function trimSummary(summary: string, maxChars: number): string { + if (summary.length <= maxChars) return summary; + return summary.slice(0, Math.max(0, maxChars - 1)).trimEnd() + '…'; +} + +function writeHostTaskIpcFile(hostTasksDir: string, data: object): string { + fs.mkdirSync(hostTasksDir, { recursive: true }); + const filename = `${Date.now()}-${Math.random().toString(36).slice(2, 8)}.json`; + const filepath = path.join(hostTasksDir, filename); + const tempPath = `${filepath}.tmp`; + fs.writeFileSync(tempPath, JSON.stringify(data, null, 2)); + fs.renameSync(tempPath, filepath); + return filename; +} + +async function persistCompactMemory( + summary: string, + sessionId: string, + options: Pick, +): Promise { + const normalized = summary.trim(); + if (!normalized || !options.groupFolder) return; + + try { + const scopeKey = `room:${options.groupFolder}`; + const selected = selectCompactMemoriesFromSummary(normalized, scopeKey); + if (selected.length === 0) { + options.log( + 'Skipped compact memory persist - no salient room memory found', + ); + return; + } + + for (const memory of selected) { + const file = writeHostTaskIpcFile(options.hostTasksDir, { + type: 'persist_memory', + scopeKind: 'room', + scopeKey, + content: trimSummary(memory.content, 300), + keywords: memory.keywords, + memory_kind: memory.memoryKind, + source_kind: 'compact', + source_ref: sessionId ? `compact:${sessionId}` : null, + timestamp: new Date().toISOString(), + }); + options.log(`Persisted compact memory via IPC (${file})`); + } + } catch (err) { + options.log( + `Failed to persist compact memory via IPC: ${ + err instanceof Error ? err.message : String(err) + }`, + ); + } +} + +function sanitizeFilename(summary: string): string { + return summary + .toLowerCase() + .replace(/[^a-z0-9]+/g, '-') + .replace(/^-+|-+$/g, '') + .slice(0, 50); +} + +function generateFallbackName(): string { + const time = new Date(); + return `conversation-${time.getHours().toString().padStart(2, '0')}${time.getMinutes().toString().padStart(2, '0')}`; +} + +function parseTranscript(content: string): ParsedMessage[] { + const messages: ParsedMessage[] = []; + for (const line of content.split('\n')) { + if (!line.trim()) continue; + try { + const entry = JSON.parse(line); + if (entry.type === 'user' && entry.message?.content) { + const text = + typeof entry.message.content === 'string' + ? entry.message.content + : entry.message.content + .map((current: { text?: string }) => current.text || '') + .join(''); + if (text) messages.push({ role: 'user', content: text }); + } else if (entry.type === 'assistant' && entry.message?.content) { + const textParts = entry.message.content + .filter((current: { type: string }) => current.type === 'text') + .map((current: { text: string }) => current.text); + const text = textParts.join(''); + if (text) messages.push({ role: 'assistant', content: text }); + } + } catch { + // Ignore malformed transcript rows and keep best-effort archive behavior. + } + } + return messages; +} + +function formatTranscriptMarkdown( + messages: ParsedMessage[], + summary: string | null, + assistantName?: string, +): string { + const now = new Date(); + const formatDateTime = (date: Date) => + date.toLocaleString('en-US', { + month: 'short', + day: 'numeric', + hour: 'numeric', + minute: '2-digit', + hour12: true, + }); + + const lines: string[] = []; + lines.push(`# ${summary || 'Conversation'}`); + lines.push(''); + lines.push(`Archived: ${formatDateTime(now)}`); + lines.push(''); + lines.push('---'); + lines.push(''); + + for (const message of messages) { + const sender = + message.role === 'user' ? 'User' : assistantName || 'Assistant'; + const content = + message.content.length > 2000 + ? message.content.slice(0, 2000) + '...' + : message.content; + lines.push(`**${sender}**: ${content}`); + lines.push(''); + } + + return lines.join('\n'); +} + +export function createPreCompactHook( + options: PreCompactHookOptions, +): HookCallback { + return async (input, _toolUseId, _context) => { + const preCompact = input as PreCompactHookInput; + const transcriptPath = preCompact.transcript_path; + const sessionId = preCompact.session_id; + const trigger = preCompact.trigger || 'auto'; + + options.writeOutput({ + status: 'success', + phase: 'progress', + result: trigger === 'auto' ? '대화 요약 중...' : '컴팩트 중...', + }); + + if (!transcriptPath || !fs.existsSync(transcriptPath)) { + options.log('No transcript found for archiving'); + return {}; + } + + try { + const content = fs.readFileSync(transcriptPath, 'utf-8'); + const messages = parseTranscript(content); + + if (messages.length === 0) { + options.log('No messages to archive'); + return {}; + } + + const summary = getSessionSummary(sessionId, transcriptPath, options.log); + const name = summary ? sanitizeFilename(summary) : generateFallbackName(); + + const conversationsDir = path.join(options.groupDir, 'conversations'); + fs.mkdirSync(conversationsDir, { recursive: true }); + + const date = new Date().toISOString().split('T')[0]; + const filename = `${date}-${name}.md`; + const filePath = path.join(conversationsDir, filename); + + const markdown = formatTranscriptMarkdown( + messages, + summary, + options.assistantName, + ); + fs.writeFileSync(filePath, markdown); + + options.log(`Archived conversation to ${filePath}`); + + if (summary) { + await persistCompactMemory(summary, sessionId, options); + } + } catch (err) { + options.log( + `Failed to archive transcript: ${err instanceof Error ? err.message : String(err)}`, + ); + } + + return {}; + }; +} + +export function createSanitizeBashHook(): HookCallback { + return async (input, _toolUseId, _context) => { + const preInput = input as PreToolUseHookInput; + const command = (preInput.tool_input as { command?: string })?.command; + if (!command) return {}; + + const unsetPrefix = `unset ${SECRET_ENV_VARS.join(' ')} 2>/dev/null; `; + return { + hookSpecificOutput: { + hookEventName: 'PreToolUse', + updatedInput: { + ...(preInput.tool_input as Record), + command: unsetPrefix + command, + }, + }, + }; + }; +} + +export function createReviewerBashGuardHook(): HookCallback { + return async (input, _toolUseId, _context) => { + const preInput = input as PreToolUseHookInput; + const command = (preInput.tool_input as { command?: string })?.command; + if (!command || !isReviewerMutatingShellCommand(command)) { + return {}; + } + + return { + hookSpecificOutput: { + hookEventName: 'PreToolUse', + permissionDecision: 'deny', + permissionDecisionReason: + 'EJClaw reviewer runtime blocks mutating shell commands in paired review mode.', + }, + }; + }; +} diff --git a/src/group-queue-state.ts b/src/group-queue-state.ts new file mode 100644 index 0000000..e99b2ff --- /dev/null +++ b/src/group-queue-state.ts @@ -0,0 +1,203 @@ +import type { ChildProcess } from 'child_process'; + +import { logger } from './logger.js'; + +export interface QueuedTask { + id: string; + groupJid: string; + fn: () => Promise; +} + +export interface GroupRunContext { + runId: string; + reason: 'messages' | 'drain'; +} + +export type RunPhase = + | 'idle' + | 'running_messages' + | 'running_task' + | 'closing_messages'; + +const VALID_TRANSITIONS: Record = { + idle: ['running_messages', 'running_task'], + running_messages: ['closing_messages', 'idle'], + closing_messages: ['idle'], + running_task: ['idle'], +}; + +export interface GroupState { + runPhase: RunPhase; + runningTaskId: string | null; + currentRunId: string | null; + directTerminalDeliveries: Map; + recentDirectTerminalDeliveries: Map>; + pendingMessages: boolean; + pendingTasks: QueuedTask[]; + process: ChildProcess | null; + processName: string | null; + ipcDir: string | null; + retryCount: number; + retryTimer: ReturnType | null; + retryScheduledAt: number | null; + postCloseTermTimer: ReturnType | null; + postCloseKillTimer: ReturnType | null; + startedAt: number | null; +} + +const MAX_RECORDED_DIRECT_TERMINAL_RUNS = 16; + +export function createGroupState(): GroupState { + return { + runPhase: 'idle', + runningTaskId: null, + currentRunId: null, + directTerminalDeliveries: new Map(), + recentDirectTerminalDeliveries: new Map(), + pendingMessages: false, + pendingTasks: [], + process: null, + processName: null, + ipcDir: null, + retryCount: 0, + retryTimer: null, + retryScheduledAt: null, + postCloseTermTimer: null, + postCloseKillTimer: null, + startedAt: null, + }; +} + +export function recordRecentDirectTerminalDelivery( + state: GroupState, + runId: string, + senderRole: string, + text: string, +): void { + const existing = state.recentDirectTerminalDeliveries.get(runId) ?? new Map(); + existing.set(senderRole, text); + state.recentDirectTerminalDeliveries.delete(runId); + state.recentDirectTerminalDeliveries.set(runId, existing); + + while ( + state.recentDirectTerminalDeliveries.size > + MAX_RECORDED_DIRECT_TERMINAL_RUNS + ) { + const oldestRunId = + state.recentDirectTerminalDeliveries.keys().next().value ?? null; + if (!oldestRunId) { + break; + } + state.recentDirectTerminalDeliveries.delete(oldestRunId); + } +} + +export function transitionRunPhase( + state: GroupState, + groupJid: string, + nextPhase: RunPhase, + metadata?: { + reason?: string; + runId?: string | null; + taskId?: string | null; + }, +): void { + const fromPhase = state.runPhase; + if (fromPhase === nextPhase) return; + + const validNextPhases = VALID_TRANSITIONS[fromPhase]; + if (!validNextPhases.includes(nextPhase)) { + logger.error( + { + groupJid, + fromPhase, + toPhase: nextPhase, + validNextPhases, + reason: metadata?.reason, + runId: metadata?.runId, + taskId: metadata?.taskId, + }, + 'Invalid group run phase transition', + ); + } + + state.runPhase = nextPhase; + logger.info( + { + groupJid, + fromPhase, + toPhase: nextPhase, + transition: `${fromPhase} → ${nextPhase}`, + reason: metadata?.reason, + runId: metadata?.runId, + taskId: metadata?.taskId, + }, + 'Group run phase changed', + ); +} + +export function resetRunState(state: GroupState, groupJid: string): void { + state.currentRunId = null; + state.runningTaskId = null; + state.startedAt = null; + state.process = null; + state.processName = null; + state.ipcDir = null; + state.directTerminalDeliveries.clear(); + transitionRunPhase(state, groupJid, 'idle'); +} + +export function assertRunPhaseInvariants( + state: GroupState, + groupJid: string, +): void { + switch (state.runPhase) { + case 'idle': + if ( + state.currentRunId != null || + state.runningTaskId != null || + state.process != null || + state.processName != null + ) { + logger.error( + { + groupJid, + runPhase: state.runPhase, + currentRunId: state.currentRunId, + runningTaskId: state.runningTaskId, + hasProcess: state.process != null, + processName: state.processName, + }, + 'Invariant violation: idle phase has stale run/task ID or process', + ); + } + break; + case 'running_messages': + case 'closing_messages': + if (state.currentRunId == null || state.runningTaskId != null) { + logger.error( + { + groupJid, + runPhase: state.runPhase, + currentRunId: state.currentRunId, + runningTaskId: state.runningTaskId, + }, + 'Invariant violation: messages phase has missing runId or stale taskId', + ); + } + break; + case 'running_task': + if (state.runningTaskId == null || state.currentRunId != null) { + logger.error( + { + groupJid, + runPhase: state.runPhase, + runningTaskId: state.runningTaskId, + currentRunId: state.currentRunId, + }, + 'Invariant violation: task phase has no taskId or has stale currentRunId', + ); + } + break; + } +} diff --git a/src/group-queue.ts b/src/group-queue.ts index 87969b3..f8a4c11 100644 --- a/src/group-queue.ts +++ b/src/group-queue.ts @@ -6,18 +6,21 @@ import { RECOVERY_DURATION_MS, } from './config.js'; import { queueFollowUpMessage, writeCloseSentinel } from './group-queue-ipc.js'; +import { + assertRunPhaseInvariants, + createGroupState, + recordRecentDirectTerminalDelivery, + resetRunState, + transitionRunPhase, +} from './group-queue-state.js'; import { logger } from './logger.js'; +import type { + GroupRunContext, + GroupState, + QueuedTask, +} from './group-queue-state.js'; -interface QueuedTask { - id: string; - groupJid: string; - fn: () => Promise; -} - -export interface GroupRunContext { - runId: string; - reason: 'messages' | 'drain'; -} +export type { GroupRunContext } from './group-queue-state.js'; const MAX_RETRIES = 5; const BASE_RETRY_MS = 5000; @@ -26,178 +29,6 @@ const MAX_CONCURRENT_TASKS = const POST_CLOSE_SIGTERM_DELAY_MS = 60_000; const POST_CLOSE_SIGKILL_DELAY_MS = 75_000; -/** - * Run lifecycle phase — single axis for what the group is currently executing. - * Message retry backoff is tracked separately (retryCount / retryScheduledAt) - * because tasks can run independently of message retry state. - */ -type RunPhase = - | 'idle' - | 'running_messages' - | 'running_task' - | 'closing_messages'; - -const VALID_TRANSITIONS: Record = { - idle: ['running_messages', 'running_task'], - running_messages: ['closing_messages', 'idle'], - closing_messages: ['idle'], - running_task: ['idle'], -}; - -interface GroupState { - runPhase: RunPhase; - runningTaskId: string | null; - currentRunId: string | null; - directTerminalDeliveries: Map; - recentDirectTerminalDeliveries: Map>; - pendingMessages: boolean; - pendingTasks: QueuedTask[]; - process: ChildProcess | null; - processName: string | null; - ipcDir: string | null; - retryCount: number; - retryTimer: ReturnType | null; - retryScheduledAt: number | null; - postCloseTermTimer: ReturnType | null; - postCloseKillTimer: ReturnType | null; - startedAt: number | null; -} - -const MAX_RECORDED_DIRECT_TERMINAL_RUNS = 16; - -function recordRecentDirectTerminalDelivery( - state: GroupState, - runId: string, - senderRole: string, - text: string, -): void { - const existing = state.recentDirectTerminalDeliveries.get(runId) ?? new Map(); - existing.set(senderRole, text); - state.recentDirectTerminalDeliveries.delete(runId); - state.recentDirectTerminalDeliveries.set(runId, existing); - - while ( - state.recentDirectTerminalDeliveries.size > - MAX_RECORDED_DIRECT_TERMINAL_RUNS - ) { - const oldestRunId = - state.recentDirectTerminalDeliveries.keys().next().value ?? null; - if (!oldestRunId) { - break; - } - state.recentDirectTerminalDeliveries.delete(oldestRunId); - } -} - -function transitionRunPhase( - state: GroupState, - groupJid: string, - nextPhase: RunPhase, - metadata?: { - reason?: string; - runId?: string | null; - taskId?: string | null; - }, -): void { - const fromPhase = state.runPhase; - if (fromPhase === nextPhase) return; - - const validNextPhases = VALID_TRANSITIONS[fromPhase]; - if (!validNextPhases.includes(nextPhase)) { - logger.error( - { - groupJid, - fromPhase, - toPhase: nextPhase, - validNextPhases, - reason: metadata?.reason, - runId: metadata?.runId, - taskId: metadata?.taskId, - }, - 'Invalid group run phase transition', - ); - } - - state.runPhase = nextPhase; - logger.info( - { - groupJid, - fromPhase, - toPhase: nextPhase, - transition: `${fromPhase} → ${nextPhase}`, - reason: metadata?.reason, - runId: metadata?.runId, - taskId: metadata?.taskId, - }, - 'Group run phase changed', - ); -} - -/** Reset all run-related fields, then transition back to idle. */ -function resetRunState(state: GroupState, groupJid: string): void { - state.currentRunId = null; - state.runningTaskId = null; - state.startedAt = null; - state.process = null; - state.processName = null; - state.ipcDir = null; - state.directTerminalDeliveries.clear(); - transitionRunPhase(state, groupJid, 'idle'); -} - -/** Validate that flat fields are consistent with runPhase. Called after every transition. */ -function assertRunPhaseInvariants(state: GroupState, groupJid: string): void { - switch (state.runPhase) { - case 'idle': - if ( - state.currentRunId != null || - state.runningTaskId != null || - state.process != null || - state.processName != null - ) { - logger.error( - { - groupJid, - runPhase: state.runPhase, - currentRunId: state.currentRunId, - runningTaskId: state.runningTaskId, - hasProcess: state.process != null, - processName: state.processName, - }, - 'Invariant violation: idle phase has stale run/task ID or process', - ); - } - break; - case 'running_messages': - case 'closing_messages': - if (state.currentRunId == null || state.runningTaskId != null) { - logger.error( - { - groupJid, - runPhase: state.runPhase, - currentRunId: state.currentRunId, - runningTaskId: state.runningTaskId, - }, - 'Invariant violation: messages phase has missing runId or stale taskId', - ); - } - break; - case 'running_task': - if (state.runningTaskId == null || state.currentRunId != null) { - logger.error( - { - groupJid, - runPhase: state.runPhase, - runningTaskId: state.runningTaskId, - currentRunId: state.currentRunId, - }, - 'Invariant violation: task phase has no taskId or has stale currentRunId', - ); - } - break; - } -} - export interface GroupStatus { jid: string; status: 'processing' | 'waiting' | 'inactive'; @@ -222,24 +53,7 @@ export class GroupQueue { private getGroup(groupJid: string): GroupState { let state = this.groups.get(groupJid); if (!state) { - state = { - runPhase: 'idle', - runningTaskId: null, - currentRunId: null, - directTerminalDeliveries: new Map(), - recentDirectTerminalDeliveries: new Map(), - pendingMessages: false, - pendingTasks: [], - process: null, - processName: null, - ipcDir: null, - retryCount: 0, - retryTimer: null, - retryScheduledAt: null, - postCloseTermTimer: null, - postCloseKillTimer: null, - startedAt: null, - }; + state = createGroupState(); this.groups.set(groupJid, state); } return state; diff --git a/src/ipc-file-claims.ts b/src/ipc-file-claims.ts new file mode 100644 index 0000000..2ca0f64 --- /dev/null +++ b/src/ipc-file-claims.ts @@ -0,0 +1,65 @@ +import fs from 'fs'; +import path from 'path'; + +const IPC_PROCESSING_DIRNAME = '.processing'; + +function buildIpcErrorPath( + errorDir: string, + prefix: string, + fileName: string, +): string { + return path.join(errorDir, `${prefix}-${Date.now()}-${fileName}`); +} + +export function claimIpcFile(filePath: string): string | null { + const processingDir = path.join( + path.dirname(filePath), + IPC_PROCESSING_DIRNAME, + ); + fs.mkdirSync(processingDir, { recursive: true }); + + const claimedPath = path.join(processingDir, path.basename(filePath)); + try { + fs.renameSync(filePath, claimedPath); + return claimedPath; + } catch (err) { + if ((err as NodeJS.ErrnoException).code === 'ENOENT') { + return null; + } + throw err; + } +} + +export function quarantineClaimedIpcFiles( + ipcDir: string, + errorDir: string, + prefix: string, +): string[] { + const processingDir = path.join(ipcDir, IPC_PROCESSING_DIRNAME); + if (!fs.existsSync(processingDir)) { + return []; + } + + const movedPaths: string[] = []; + for (const file of fs + .readdirSync(processingDir) + .filter((f) => f.endsWith('.json'))) { + const claimedPath = path.join(processingDir, file); + const errorPath = buildIpcErrorPath(errorDir, prefix, file); + fs.renameSync(claimedPath, errorPath); + movedPaths.push(errorPath); + } + + return movedPaths; +} + +export function moveClaimedIpcFileToError( + claimedPath: string, + errorDir: string, + prefix: string, +): void { + fs.renameSync( + claimedPath, + buildIpcErrorPath(errorDir, prefix, path.basename(claimedPath)), + ); +} diff --git a/src/ipc-message-forwarding.ts b/src/ipc-message-forwarding.ts new file mode 100644 index 0000000..bd5f2e4 --- /dev/null +++ b/src/ipc-message-forwarding.ts @@ -0,0 +1,45 @@ +import type { + IpcMessageForwardResult, + IpcMessagePayload, +} from './ipc-types.js'; +import type { RegisteredGroup } from './types.js'; + +export async function forwardAuthorizedIpcMessage( + msg: IpcMessagePayload, + sourceGroup: string, + isMain: boolean, + roomBindings: Record, + sendMessage: ( + jid: string, + text: string, + senderRole?: string, + runId?: string, + ) => Promise, +): Promise { + if (!(msg.type === 'message' && msg.chatJid && msg.text)) { + return { outcome: 'ignored', senderRole: msg.senderRole ?? null }; + } + + const targetGroup = roomBindings[msg.chatJid]; + const isMainOverride = isMain === true; + if ( + !(isMainOverride || (targetGroup && targetGroup.folder === sourceGroup)) + ) { + return { + outcome: 'blocked', + chatJid: msg.chatJid, + targetGroup: targetGroup?.folder ?? null, + isMainOverride, + senderRole: msg.senderRole ?? null, + }; + } + + await sendMessage(msg.chatJid, msg.text, msg.senderRole, msg.runId); + return { + outcome: 'sent', + chatJid: msg.chatJid, + targetGroup: targetGroup?.folder ?? null, + isMainOverride, + senderRole: msg.senderRole ?? null, + }; +} diff --git a/src/ipc-task-processor.ts b/src/ipc-task-processor.ts new file mode 100644 index 0000000..caf69c8 --- /dev/null +++ b/src/ipc-task-processor.ts @@ -0,0 +1,502 @@ +import { CronExpressionParser } from 'cron-parser'; + +import { TIMEZONE } from './config.js'; +import { + isHostEvidenceAction, + runHostEvidenceRequest, + writeHostEvidenceResponse, +} from './host-evidence.js'; +import { + createTask, + deleteTask, + findDuplicateCiWatcher, + getTaskById, + rememberMemory, + updateTask, +} from './db.js'; +import { isValidGroupFolder } from './group-folder.js'; +import { logger } from './logger.js'; +import { + DEFAULT_WATCH_CI_MAX_DURATION_MS, + isWatchCiTask, +} from './task-watch-status.js'; +import type { IpcDeps, TaskIpcPayload } from './ipc-types.js'; + +export async function processTaskIpc( + data: TaskIpcPayload, + sourceGroup: string, + isMain: boolean, + deps: IpcDeps, +): Promise { + const roomBindings = deps.roomBindings(); + + switch (data.type) { + case 'schedule_task': + if ( + data.prompt && + data.schedule_type && + data.schedule_value && + data.targetJid + ) { + const targetJid = data.targetJid; + const targetGroupEntry = + roomBindings[targetJid] || + Object.values(roomBindings).find( + (group) => group.folder === targetJid, + ); + + if (!targetGroupEntry) { + logger.warn( + { targetJid }, + 'Cannot schedule task: target group not registered', + ); + break; + } + + const targetFolder = targetGroupEntry.folder; + const resolvedTargetJid = + roomBindings[targetJid] !== undefined + ? targetJid + : Object.entries(roomBindings).find( + ([, group]) => group.folder === targetFolder, + )?.[0]; + + if (!resolvedTargetJid) { + logger.warn( + { targetJid, targetFolder }, + 'Cannot resolve scheduled task target JID from folder', + ); + break; + } + + if (!isMain && targetFolder !== sourceGroup) { + logger.warn( + { sourceGroup, targetFolder }, + 'Unauthorized schedule_task attempt blocked', + ); + break; + } + + const scheduleType = data.schedule_type as 'cron' | 'interval' | 'once'; + + let nextRun: string | null = null; + if (scheduleType === 'cron') { + try { + const interval = CronExpressionParser.parse(data.schedule_value, { + tz: TIMEZONE, + }); + nextRun = interval.next().toISOString(); + } catch { + logger.warn( + { scheduleValue: data.schedule_value }, + 'Invalid cron expression', + ); + break; + } + } else if (scheduleType === 'interval') { + const ms = parseInt(data.schedule_value, 10); + if (isNaN(ms) || ms <= 0) { + logger.warn( + { scheduleValue: data.schedule_value }, + 'Invalid interval', + ); + break; + } + nextRun = isWatchCiTask({ prompt: data.prompt }) + ? new Date().toISOString() + : new Date(Date.now() + ms).toISOString(); + } else if (scheduleType === 'once') { + const date = new Date(data.schedule_value); + if (isNaN(date.getTime())) { + logger.warn( + { scheduleValue: data.schedule_value }, + 'Invalid timestamp', + ); + break; + } + nextRun = date.toISOString(); + } + + if (data.ci_provider && data.ci_metadata) { + const existing = findDuplicateCiWatcher( + resolvedTargetJid, + data.ci_provider, + data.ci_metadata, + ); + if (existing) { + logger.info( + { + existingTaskId: existing.id, + existingAgentType: existing.agent_type, + ciProvider: data.ci_provider, + sourceGroup, + }, + 'Duplicate CI watcher skipped — another agent already watches this run', + ); + break; + } + } + + const taskId = + data.taskId || + `task-${Date.now()}-${Math.random().toString(36).slice(2, 8)}`; + const contextMode = + data.context_mode === 'group' || data.context_mode === 'isolated' + ? data.context_mode + : 'isolated'; + createTask({ + id: taskId, + group_folder: targetFolder, + chat_jid: resolvedTargetJid, + agent_type: targetGroupEntry.agentType || 'claude-code', + ci_provider: data.ci_provider ?? null, + ci_metadata: data.ci_metadata ?? null, + max_duration_ms: isWatchCiTask({ prompt: data.prompt }) + ? DEFAULT_WATCH_CI_MAX_DURATION_MS + : null, + prompt: data.prompt, + schedule_type: scheduleType, + schedule_value: data.schedule_value, + context_mode: contextMode, + next_run: nextRun, + status: 'active', + created_at: new Date().toISOString(), + }); + logger.info( + { + taskId, + sourceGroup, + targetFolder, + contextMode, + agentType: targetGroupEntry.agentType || 'claude-code', + }, + 'Task created via IPC', + ); + if (nextRun && new Date(nextRun).getTime() <= Date.now()) { + deps.nudgeScheduler?.(); + } + } + break; + + case 'pause_task': + if (data.taskId) { + const task = getTaskById(data.taskId); + if (task && (isMain || task.group_folder === sourceGroup)) { + updateTask(data.taskId, { status: 'paused' }); + logger.info( + { taskId: data.taskId, sourceGroup }, + 'Task paused via IPC', + ); + } else { + logger.warn( + { taskId: data.taskId, sourceGroup }, + 'Unauthorized task pause attempt', + ); + } + } + break; + + case 'resume_task': + if (data.taskId) { + const task = getTaskById(data.taskId); + if (task && (isMain || task.group_folder === sourceGroup)) { + updateTask(data.taskId, { status: 'active' }); + logger.info( + { taskId: data.taskId, sourceGroup }, + 'Task resumed via IPC', + ); + } else { + logger.warn( + { taskId: data.taskId, sourceGroup }, + 'Unauthorized task resume attempt', + ); + } + } + break; + + case 'cancel_task': + if (data.taskId) { + const task = getTaskById(data.taskId); + if (task && (isMain || task.group_folder === sourceGroup)) { + deleteTask(data.taskId); + logger.info( + { taskId: data.taskId, sourceGroup }, + 'Task cancelled via IPC', + ); + } else { + logger.warn( + { taskId: data.taskId, sourceGroup }, + 'Unauthorized task cancel attempt', + ); + } + } + break; + + case 'host_evidence_request': + if (!data.requestId) { + logger.warn( + { sourceGroup }, + 'Ignoring host_evidence_request without requestId', + ); + break; + } + + if (!isHostEvidenceAction(data.action)) { + writeHostEvidenceResponse(sourceGroup, { + requestId: data.requestId, + ok: false, + action: 'ejclaw_service_status', + command: '', + stdout: '', + stderr: '', + exitCode: 1, + error: `Unsupported host evidence action: ${String(data.action)}`, + }); + logger.warn( + { sourceGroup, requestId: data.requestId, action: data.action }, + 'Rejected unsupported host evidence action', + ); + break; + } + + { + const result = await runHostEvidenceRequest({ + requestId: data.requestId, + action: data.action, + tailLines: + typeof data.tail_lines === 'number' ? data.tail_lines : undefined, + }); + + writeHostEvidenceResponse(sourceGroup, { + requestId: data.requestId, + ...result, + }); + + logger.info( + { + sourceGroup, + requestId: data.requestId, + action: data.action, + ok: result.ok, + exitCode: result.exitCode, + }, + 'Processed host evidence request via IPC', + ); + } + break; + + case 'update_task': + if (data.taskId) { + const task = getTaskById(data.taskId); + if (!task) { + logger.warn( + { taskId: data.taskId, sourceGroup }, + 'Task not found for update', + ); + break; + } + if (!isMain && task.group_folder !== sourceGroup) { + logger.warn( + { taskId: data.taskId, sourceGroup }, + 'Unauthorized task update attempt', + ); + break; + } + + const updates: Parameters[1] = {}; + if (data.prompt !== undefined) updates.prompt = data.prompt; + if (data.schedule_type !== undefined) { + updates.schedule_type = data.schedule_type as + | 'cron' + | 'interval' + | 'once'; + } + if (data.schedule_value !== undefined) { + updates.schedule_value = data.schedule_value; + } + + if (data.schedule_type || data.schedule_value) { + const updatedTask = { + ...task, + ...updates, + }; + if (updatedTask.schedule_type === 'cron') { + try { + const interval = CronExpressionParser.parse( + updatedTask.schedule_value, + { tz: TIMEZONE }, + ); + updates.next_run = interval.next().toISOString(); + } catch { + logger.warn( + { taskId: data.taskId, value: updatedTask.schedule_value }, + 'Invalid cron in task update', + ); + break; + } + } else if (updatedTask.schedule_type === 'interval') { + const ms = parseInt(updatedTask.schedule_value, 10); + if (!isNaN(ms) && ms > 0) { + updates.next_run = new Date(Date.now() + ms).toISOString(); + } + } + } + + updateTask(data.taskId, updates); + logger.info( + { taskId: data.taskId, sourceGroup, updates }, + 'Task updated via IPC', + ); + } + break; + + case 'refresh_groups': + if (isMain) { + logger.info( + { sourceGroup }, + 'Group metadata refresh requested via IPC', + ); + await deps.syncGroups(true); + const availableGroups = deps.getAvailableGroups(); + deps.writeGroupsSnapshot(sourceGroup, true, availableGroups); + } else { + logger.warn( + { sourceGroup }, + 'Unauthorized refresh_groups attempt blocked', + ); + } + break; + + case 'assign_room': + if (!isMain) { + logger.warn( + { sourceGroup, type: data.type }, + `Unauthorized ${data.type} attempt blocked`, + ); + break; + } + if (data.jid && data.name) { + if (data.folder && !isValidGroupFolder(data.folder)) { + logger.warn( + { sourceGroup, folder: data.folder }, + `Invalid ${data.type} request - unsafe folder name`, + ); + break; + } + if ( + data.room_mode !== undefined && + data.room_mode !== 'single' && + data.room_mode !== 'tribunal' + ) { + logger.warn( + { sourceGroup, roomMode: data.room_mode }, + 'Invalid assign_room request - unknown room_mode', + ); + break; + } + if ( + data.owner_agent_type !== undefined && + data.owner_agent_type !== 'claude-code' && + data.owner_agent_type !== 'codex' + ) { + logger.warn( + { sourceGroup, ownerAgentType: data.owner_agent_type }, + 'Invalid assign_room request - unknown owner_agent_type', + ); + break; + } + + deps.assignRoom(data.jid, { + name: data.name, + roomMode: data.room_mode, + ownerAgentType: data.owner_agent_type, + folder: data.folder, + isMain: data.isMain, + workDir: data.workDir, + }); + } else { + logger.warn( + { data }, + `Invalid ${data.type} request - missing required fields`, + ); + } + break; + + case 'persist_memory': { + if ( + data.scopeKind !== 'room' || + typeof data.scopeKey !== 'string' || + typeof data.content !== 'string' + ) { + logger.warn( + { sourceGroup, data }, + 'Invalid persist_memory request - missing required fields', + ); + break; + } + + const expectedScopeKey = `room:${sourceGroup}`; + if (data.scopeKey !== expectedScopeKey) { + logger.warn( + { sourceGroup, scopeKey: data.scopeKey, expectedScopeKey }, + 'Unauthorized persist_memory attempt blocked', + ); + break; + } + + if ( + data.source_kind !== undefined && + data.source_kind !== 'compact' && + data.source_kind !== 'explicit' && + data.source_kind !== 'import' && + data.source_kind !== 'system' + ) { + logger.warn( + { sourceGroup, sourceKind: data.source_kind }, + 'Invalid persist_memory request - unknown source_kind', + ); + break; + } + + if ( + Array.isArray(data.keywords) && + !data.keywords.every((value) => typeof value === 'string') + ) { + logger.warn( + { sourceGroup, keywords: data.keywords }, + 'Invalid persist_memory request - keywords must be strings', + ); + break; + } + + rememberMemory({ + scopeKind: 'room', + scopeKey: data.scopeKey, + content: data.content, + keywords: data.keywords, + memoryKind: + typeof data.memory_kind === 'string' ? data.memory_kind : null, + sourceKind: + (data.source_kind as + | 'compact' + | 'explicit' + | 'import' + | 'system' + | undefined) ?? 'compact', + sourceRef: typeof data.source_ref === 'string' ? data.source_ref : null, + }); + logger.info( + { + sourceGroup, + scopeKey: data.scopeKey, + sourceKind: data.source_kind ?? 'compact', + }, + 'Memory persisted via IPC', + ); + break; + } + + default: + logger.warn({ type: data.type }, 'Unknown IPC task type'); + } +} diff --git a/src/ipc-types.ts b/src/ipc-types.ts new file mode 100644 index 0000000..853789d --- /dev/null +++ b/src/ipc-types.ts @@ -0,0 +1,71 @@ +import type { AvailableGroup } from './agent-runner.js'; +import type { AssignRoomInput } from './db.js'; +import type { AgentType, RegisteredGroup, RoomMode } from './types.js'; + +export interface IpcDeps { + sendMessage: ( + jid: string, + text: string, + senderRole?: string, + runId?: string, + ) => Promise; + nudgeScheduler?: () => void; + roomBindings: () => Record; + assignRoom: (jid: string, room: AssignRoomInput) => void; + syncGroups: (force: boolean) => Promise; + getAvailableGroups: () => AvailableGroup[]; + writeGroupsSnapshot: ( + groupFolder: string, + isMain: boolean, + availableGroups: AvailableGroup[], + ) => void; +} + +export interface IpcMessagePayload { + type?: string; + chatJid?: string; + text?: string; + senderRole?: string; + runId?: string; +} + +export interface IpcMessageForwardResult { + outcome: 'ignored' | 'sent' | 'blocked'; + chatJid?: string; + targetGroup?: string | null; + isMainOverride?: boolean; + senderRole?: string | null; +} + +export interface TaskIpcPayload { + type: string; + taskId?: string; + prompt?: string; + schedule_type?: string; + schedule_value?: string; + context_mode?: string; + ci_provider?: 'github'; + ci_metadata?: string; + groupFolder?: string; + chatJid?: string; + targetJid?: string; + jid?: string; + name?: string; + folder?: string; + room_mode?: RoomMode; + owner_agent_type?: AgentType; + isMain?: boolean; + workDir?: string; + scopeKind?: string; + scopeKey?: string; + content?: string; + keywords?: string[]; + memory_kind?: string | null; + source_kind?: string; + source_ref?: string | null; + requestId?: string; + action?: string; + tail_lines?: number; + profile?: string; + expected_snapshot_id?: string; +} diff --git a/src/ipc.ts b/src/ipc.ts index 3b89a06..04f2017 100644 --- a/src/ipc.ts +++ b/src/ipc.ts @@ -1,166 +1,36 @@ import fs from 'fs'; import path from 'path'; -import { CronExpressionParser } from 'cron-parser'; - import { DATA_DIR, IPC_POLL_INTERVAL, TIMEZONE } from './config.js'; -import { - isHostEvidenceAction, - runHostEvidenceRequest, - writeHostEvidenceResponse, -} from './host-evidence.js'; import { readJsonFile } from './utils.js'; -import { AvailableGroup } from './agent-runner.js'; -import { - type AssignRoomInput, - createTask, - deleteTask, - findDuplicateCiWatcher, - getTaskById, - rememberMemory, - updateTask, -} from './db.js'; -import { isValidGroupFolder } from './group-folder.js'; import { logger } from './logger.js'; import { - DEFAULT_WATCH_CI_MAX_DURATION_MS, - isWatchCiTask, -} from './task-watch-status.js'; -import { AgentType, RegisteredGroup, RoomMode } from './types.js'; + claimIpcFile, + moveClaimedIpcFileToError, + quarantineClaimedIpcFiles, +} from './ipc-file-claims.js'; +import { forwardAuthorizedIpcMessage } from './ipc-message-forwarding.js'; +import { processTaskIpc } from './ipc-task-processor.js'; +import type { + IpcDeps, + IpcMessagePayload, + TaskIpcPayload, +} from './ipc-types.js'; -export interface IpcDeps { - sendMessage: ( - jid: string, - text: string, - senderRole?: string, - runId?: string, - ) => Promise; - nudgeScheduler?: () => void; - roomBindings: () => Record; - assignRoom: (jid: string, room: AssignRoomInput) => void; - syncGroups: (force: boolean) => Promise; - getAvailableGroups: () => AvailableGroup[]; - writeGroupsSnapshot: ( - groupFolder: string, - isMain: boolean, - availableGroups: AvailableGroup[], - ) => void; -} - -export interface IpcMessagePayload { - type?: string; - chatJid?: string; - text?: string; - senderRole?: string; - runId?: string; -} - -export interface IpcMessageForwardResult { - outcome: 'ignored' | 'sent' | 'blocked'; - chatJid?: string; - targetGroup?: string | null; - isMainOverride?: boolean; - senderRole?: string | null; -} - -export async function forwardAuthorizedIpcMessage( - msg: IpcMessagePayload, - sourceGroup: string, - isMain: boolean, - roomBindings: Record, - sendMessage: IpcDeps['sendMessage'], -): Promise { - if (!(msg.type === 'message' && msg.chatJid && msg.text)) { - return { outcome: 'ignored', senderRole: msg.senderRole ?? null }; - } - - const targetGroup = roomBindings[msg.chatJid]; - const isMainOverride = isMain === true; - if ( - !(isMainOverride || (targetGroup && targetGroup.folder === sourceGroup)) - ) { - return { - outcome: 'blocked', - chatJid: msg.chatJid, - targetGroup: targetGroup?.folder ?? null, - isMainOverride, - senderRole: msg.senderRole ?? null, - }; - } - - await sendMessage(msg.chatJid, msg.text, msg.senderRole, msg.runId); - return { - outcome: 'sent', - chatJid: msg.chatJid, - targetGroup: targetGroup?.folder ?? null, - isMainOverride, - senderRole: msg.senderRole ?? null, - }; -} +export type { + IpcDeps, + IpcMessageForwardResult, + IpcMessagePayload, + TaskIpcPayload, +} from './ipc-types.js'; +export { + claimIpcFile, + forwardAuthorizedIpcMessage, + processTaskIpc, + quarantineClaimedIpcFiles, +}; let ipcWatcherRunning = false; -const IPC_PROCESSING_DIRNAME = '.processing'; - -function buildIpcErrorPath( - errorDir: string, - prefix: string, - fileName: string, -): string { - return path.join(errorDir, `${prefix}-${Date.now()}-${fileName}`); -} - -export function claimIpcFile(filePath: string): string | null { - const processingDir = path.join( - path.dirname(filePath), - IPC_PROCESSING_DIRNAME, - ); - fs.mkdirSync(processingDir, { recursive: true }); - - const claimedPath = path.join(processingDir, path.basename(filePath)); - try { - fs.renameSync(filePath, claimedPath); - return claimedPath; - } catch (err) { - if ((err as NodeJS.ErrnoException).code === 'ENOENT') { - return null; - } - throw err; - } -} - -export function quarantineClaimedIpcFiles( - ipcDir: string, - errorDir: string, - prefix: string, -): string[] { - const processingDir = path.join(ipcDir, IPC_PROCESSING_DIRNAME); - if (!fs.existsSync(processingDir)) { - return []; - } - - const movedPaths: string[] = []; - for (const file of fs - .readdirSync(processingDir) - .filter((f) => f.endsWith('.json'))) { - const claimedPath = path.join(processingDir, file); - const errorPath = buildIpcErrorPath(errorDir, prefix, file); - fs.renameSync(claimedPath, errorPath); - movedPaths.push(errorPath); - } - - return movedPaths; -} - -function moveClaimedIpcFileToError( - claimedPath: string, - errorDir: string, - prefix: string, -): void { - fs.renameSync( - claimedPath, - buildIpcErrorPath(errorDir, prefix, path.basename(claimedPath)), - ); -} export function startIpcWatcher(deps: IpcDeps): void { if (ipcWatcherRunning) { @@ -307,7 +177,7 @@ export function startIpcWatcher(deps: IpcDeps): void { throw new Error('Invalid JSON'); // Pass source group identity to processTaskIpc for authorization await processTaskIpc( - data as Parameters[0], + data as TaskIpcPayload, sourceGroup, isMain, deps, @@ -337,520 +207,3 @@ export function startIpcWatcher(deps: IpcDeps): void { processIpcFiles(); logger.info('IPC watcher started (per-group namespaces)'); } - -export async function processTaskIpc( - data: { - type: string; - taskId?: string; - prompt?: string; - schedule_type?: string; - schedule_value?: string; - context_mode?: string; - ci_provider?: 'github'; - ci_metadata?: string; - groupFolder?: string; - chatJid?: string; - targetJid?: string; - // For assign_room - jid?: string; - name?: string; - folder?: string; - room_mode?: RoomMode; - owner_agent_type?: AgentType; - isMain?: boolean; - workDir?: string; - scopeKind?: string; - scopeKey?: string; - content?: string; - keywords?: string[]; - memory_kind?: string | null; - source_kind?: string; - source_ref?: string | null; - requestId?: string; - action?: string; - tail_lines?: number; - profile?: string; - expected_snapshot_id?: string; - }, - sourceGroup: string, // Verified identity from IPC directory - isMain: boolean, // Verified from directory path - deps: IpcDeps, -): Promise { - const roomBindings = deps.roomBindings(); - - switch (data.type) { - case 'schedule_task': - if ( - data.prompt && - data.schedule_type && - data.schedule_value && - data.targetJid - ) { - // Resolve the target group from JID - const targetJid = data.targetJid as string; - const targetGroupEntry = - roomBindings[targetJid] || - Object.values(roomBindings).find( - (group) => group.folder === targetJid, - ); - - if (!targetGroupEntry) { - logger.warn( - { targetJid }, - 'Cannot schedule task: target group not registered', - ); - break; - } - - const targetFolder = targetGroupEntry.folder; - const resolvedTargetJid = - roomBindings[targetJid] !== undefined - ? targetJid - : Object.entries(roomBindings).find( - ([, group]) => group.folder === targetFolder, - )?.[0]; - - if (!resolvedTargetJid) { - logger.warn( - { targetJid, targetFolder }, - 'Cannot resolve scheduled task target JID from folder', - ); - break; - } - - // Authorization: non-main groups can only schedule for themselves - if (!isMain && targetFolder !== sourceGroup) { - logger.warn( - { sourceGroup, targetFolder }, - 'Unauthorized schedule_task attempt blocked', - ); - break; - } - - const scheduleType = data.schedule_type as 'cron' | 'interval' | 'once'; - - let nextRun: string | null = null; - if (scheduleType === 'cron') { - try { - const interval = CronExpressionParser.parse(data.schedule_value, { - tz: TIMEZONE, - }); - nextRun = interval.next().toISOString(); - } catch { - logger.warn( - { scheduleValue: data.schedule_value }, - 'Invalid cron expression', - ); - break; - } - } else if (scheduleType === 'interval') { - const ms = parseInt(data.schedule_value, 10); - if (isNaN(ms) || ms <= 0) { - logger.warn( - { scheduleValue: data.schedule_value }, - 'Invalid interval', - ); - break; - } - nextRun = isWatchCiTask({ prompt: data.prompt }) - ? new Date().toISOString() - : new Date(Date.now() + ms).toISOString(); - } else if (scheduleType === 'once') { - const date = new Date(data.schedule_value); - if (isNaN(date.getTime())) { - logger.warn( - { scheduleValue: data.schedule_value }, - 'Invalid timestamp', - ); - break; - } - nextRun = date.toISOString(); - } - - // Deduplicate CI watchers: if another agent already watches the same - // channel + provider + run, skip creation to avoid duplicate notifications. - if (data.ci_provider && data.ci_metadata) { - const existing = findDuplicateCiWatcher( - resolvedTargetJid, - data.ci_provider, - data.ci_metadata as string, - ); - if (existing) { - logger.info( - { - existingTaskId: existing.id, - existingAgentType: existing.agent_type, - ciProvider: data.ci_provider, - sourceGroup, - }, - 'Duplicate CI watcher skipped — another agent already watches this run', - ); - break; - } - } - - const taskId = - data.taskId || - `task-${Date.now()}-${Math.random().toString(36).slice(2, 8)}`; - const contextMode = - data.context_mode === 'group' || data.context_mode === 'isolated' - ? data.context_mode - : 'isolated'; - createTask({ - id: taskId, - group_folder: targetFolder, - chat_jid: resolvedTargetJid, - agent_type: targetGroupEntry.agentType || 'claude-code', - ci_provider: data.ci_provider ?? null, - ci_metadata: data.ci_metadata ?? null, - max_duration_ms: isWatchCiTask({ prompt: data.prompt }) - ? DEFAULT_WATCH_CI_MAX_DURATION_MS - : null, - prompt: data.prompt, - schedule_type: scheduleType, - schedule_value: data.schedule_value, - context_mode: contextMode, - next_run: nextRun, - status: 'active', - created_at: new Date().toISOString(), - }); - logger.info( - { - taskId, - sourceGroup, - targetFolder, - contextMode, - agentType: targetGroupEntry.agentType || 'claude-code', - }, - 'Task created via IPC', - ); - if (nextRun && new Date(nextRun).getTime() <= Date.now()) { - deps.nudgeScheduler?.(); - } - } - break; - - case 'pause_task': - if (data.taskId) { - const task = getTaskById(data.taskId); - if (task && (isMain || task.group_folder === sourceGroup)) { - updateTask(data.taskId, { status: 'paused' }); - logger.info( - { taskId: data.taskId, sourceGroup }, - 'Task paused via IPC', - ); - } else { - logger.warn( - { taskId: data.taskId, sourceGroup }, - 'Unauthorized task pause attempt', - ); - } - } - break; - - case 'resume_task': - if (data.taskId) { - const task = getTaskById(data.taskId); - if (task && (isMain || task.group_folder === sourceGroup)) { - updateTask(data.taskId, { status: 'active' }); - logger.info( - { taskId: data.taskId, sourceGroup }, - 'Task resumed via IPC', - ); - } else { - logger.warn( - { taskId: data.taskId, sourceGroup }, - 'Unauthorized task resume attempt', - ); - } - } - break; - - case 'cancel_task': - if (data.taskId) { - const task = getTaskById(data.taskId); - if (task && (isMain || task.group_folder === sourceGroup)) { - deleteTask(data.taskId); - logger.info( - { taskId: data.taskId, sourceGroup }, - 'Task cancelled via IPC', - ); - } else { - logger.warn( - { taskId: data.taskId, sourceGroup }, - 'Unauthorized task cancel attempt', - ); - } - } - break; - - case 'host_evidence_request': - if (!data.requestId) { - logger.warn( - { sourceGroup }, - 'Ignoring host_evidence_request without requestId', - ); - break; - } - - if (!isHostEvidenceAction(data.action)) { - writeHostEvidenceResponse(sourceGroup, { - requestId: data.requestId, - ok: false, - action: 'ejclaw_service_status', - command: '', - stdout: '', - stderr: '', - exitCode: 1, - error: `Unsupported host evidence action: ${String(data.action)}`, - }); - logger.warn( - { sourceGroup, requestId: data.requestId, action: data.action }, - 'Rejected unsupported host evidence action', - ); - break; - } - - { - const result = await runHostEvidenceRequest({ - requestId: data.requestId, - action: data.action, - tailLines: - typeof data.tail_lines === 'number' ? data.tail_lines : undefined, - }); - - writeHostEvidenceResponse(sourceGroup, { - requestId: data.requestId, - ...result, - }); - - logger.info( - { - sourceGroup, - requestId: data.requestId, - action: data.action, - ok: result.ok, - exitCode: result.exitCode, - }, - 'Processed host evidence request via IPC', - ); - } - break; - - case 'update_task': - if (data.taskId) { - const task = getTaskById(data.taskId); - if (!task) { - logger.warn( - { taskId: data.taskId, sourceGroup }, - 'Task not found for update', - ); - break; - } - if (!isMain && task.group_folder !== sourceGroup) { - logger.warn( - { taskId: data.taskId, sourceGroup }, - 'Unauthorized task update attempt', - ); - break; - } - - const updates: Parameters[1] = {}; - if (data.prompt !== undefined) updates.prompt = data.prompt; - if (data.schedule_type !== undefined) - updates.schedule_type = data.schedule_type as - | 'cron' - | 'interval' - | 'once'; - if (data.schedule_value !== undefined) - updates.schedule_value = data.schedule_value; - - // Recompute next_run if schedule changed - if (data.schedule_type || data.schedule_value) { - const updatedTask = { - ...task, - ...updates, - }; - if (updatedTask.schedule_type === 'cron') { - try { - const interval = CronExpressionParser.parse( - updatedTask.schedule_value, - { tz: TIMEZONE }, - ); - updates.next_run = interval.next().toISOString(); - } catch { - logger.warn( - { taskId: data.taskId, value: updatedTask.schedule_value }, - 'Invalid cron in task update', - ); - break; - } - } else if (updatedTask.schedule_type === 'interval') { - const ms = parseInt(updatedTask.schedule_value, 10); - if (!isNaN(ms) && ms > 0) { - updates.next_run = new Date(Date.now() + ms).toISOString(); - } - } - } - - updateTask(data.taskId, updates); - logger.info( - { taskId: data.taskId, sourceGroup, updates }, - 'Task updated via IPC', - ); - } - break; - - case 'refresh_groups': - // Only main group can request a refresh - if (isMain) { - logger.info( - { sourceGroup }, - 'Group metadata refresh requested via IPC', - ); - await deps.syncGroups(true); - // Write updated snapshot immediately - const availableGroups = deps.getAvailableGroups(); - deps.writeGroupsSnapshot(sourceGroup, true, availableGroups); - } else { - logger.warn( - { sourceGroup }, - 'Unauthorized refresh_groups attempt blocked', - ); - } - break; - - case 'assign_room': - // Only main group can assign rooms - if (!isMain) { - logger.warn( - { sourceGroup, type: data.type }, - `Unauthorized ${data.type} attempt blocked`, - ); - break; - } - if (data.jid && data.name) { - if (data.folder && !isValidGroupFolder(data.folder)) { - logger.warn( - { sourceGroup, folder: data.folder }, - `Invalid ${data.type} request - unsafe folder name`, - ); - break; - } - if ( - data.room_mode !== undefined && - data.room_mode !== 'single' && - data.room_mode !== 'tribunal' - ) { - logger.warn( - { sourceGroup, roomMode: data.room_mode }, - 'Invalid assign_room request - unknown room_mode', - ); - break; - } - if ( - data.owner_agent_type !== undefined && - data.owner_agent_type !== 'claude-code' && - data.owner_agent_type !== 'codex' - ) { - logger.warn( - { sourceGroup, ownerAgentType: data.owner_agent_type }, - 'Invalid assign_room request - unknown owner_agent_type', - ); - break; - } - - deps.assignRoom(data.jid, { - name: data.name, - roomMode: data.room_mode, - ownerAgentType: data.owner_agent_type, - folder: data.folder, - isMain: data.isMain, - workDir: data.workDir, - }); - } else { - logger.warn( - { data }, - `Invalid ${data.type} request - missing required fields`, - ); - } - break; - - case 'persist_memory': { - if ( - data.scopeKind !== 'room' || - typeof data.scopeKey !== 'string' || - typeof data.content !== 'string' - ) { - logger.warn( - { sourceGroup, data }, - 'Invalid persist_memory request - missing required fields', - ); - break; - } - - const expectedScopeKey = `room:${sourceGroup}`; - if (data.scopeKey !== expectedScopeKey) { - logger.warn( - { sourceGroup, scopeKey: data.scopeKey, expectedScopeKey }, - 'Unauthorized persist_memory attempt blocked', - ); - break; - } - - if ( - data.source_kind !== undefined && - data.source_kind !== 'compact' && - data.source_kind !== 'explicit' && - data.source_kind !== 'import' && - data.source_kind !== 'system' - ) { - logger.warn( - { sourceGroup, sourceKind: data.source_kind }, - 'Invalid persist_memory request - unknown source_kind', - ); - break; - } - - if ( - Array.isArray(data.keywords) && - !data.keywords.every((v) => typeof v === 'string') - ) { - logger.warn( - { sourceGroup, keywords: data.keywords }, - 'Invalid persist_memory request - keywords must be strings', - ); - break; - } - - rememberMemory({ - scopeKind: 'room', - scopeKey: data.scopeKey, - content: data.content, - keywords: data.keywords, - memoryKind: - typeof data.memory_kind === 'string' ? data.memory_kind : null, - sourceKind: - (data.source_kind as - | 'compact' - | 'explicit' - | 'import' - | 'system' - | undefined) ?? 'compact', - sourceRef: typeof data.source_ref === 'string' ? data.source_ref : null, - }); - logger.info( - { - sourceGroup, - scopeKey: data.scopeKey, - sourceKind: data.source_kind ?? 'compact', - }, - 'Memory persisted via IPC', - ); - break; - } - - default: - logger.warn({ type: data.type }, 'Unknown IPC task type'); - } -}