claude-bot/EJClaw
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

runtime: split ipc queue and runner helpers

Browse Source
This commit is contained in:
ejclaw
2026-04-11 05:49:40 +09:00
parent 30dda74621
commit 88f8b3006c
11 changed files with 1498 additions and 1389 deletions
Download Patch File Download Diff File Expand all files Collapse all files

562
runners/agent-runner/src/index.ts
View File

@@ -16,20 +16,11 @@
import fs from 'fs'; import fs from 'fs';
import path from 'path'; import path from 'path';
import { query } from '@anthropic-ai/claude-agent-sdk';
import { import {
query,
HookCallback,
PreCompactHookInput,
PreToolUseHookInput,
} from '@anthropic-ai/claude-agent-sdk';
import {
extractImageTagPaths,
IPC_CLOSE_SENTINEL, IPC_CLOSE_SENTINEL,
IPC_INPUT_SUBDIR, IPC_INPUT_SUBDIR,
IPC_POLL_MS, IPC_POLL_MS,
normalizePublicTextOutput,
type RunnerStructuredOutput,
writeProtocolOutput,
} from 'ejclaw-runners-shared'; } from 'ejclaw-runners-shared';
import { fileURLToPath } from 'url'; import { fileURLToPath } from 'url';
@@ -44,11 +35,24 @@ import {
getClaudeReadonlySandboxMode, getClaudeReadonlySandboxMode,
isArbiterRuntimeEnvEnabled, isArbiterRuntimeEnvEnabled,
isClaudeReadonlyReviewerRuntime, isClaudeReadonlyReviewerRuntime,
isReviewerMutatingShellCommand,
isReviewerRuntime, isReviewerRuntime,
isReviewerRuntimeEnvEnabled, isReviewerRuntimeEnvEnabled,
} from './reviewer-runtime.js'; } 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 { interface RunnerInput {
prompt: string; prompt: string;
@@ -62,53 +66,6 @@ interface RunnerInput {
roomRoleContext?: RoomRoleContext; 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. // Paths configurable via env vars.
const GROUP_DIR = process.env.EJCLAW_GROUP_DIR || '/workspace/group'; const GROUP_DIR = process.env.EJCLAW_GROUP_DIR || '/workspace/group';
const IPC_DIR = process.env.EJCLAW_IPC_DIR || '/workspace/ipc'; 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_DIR = path.join(IPC_DIR, IPC_INPUT_SUBDIR);
const IPC_INPUT_CLOSE_SENTINEL = path.join(IPC_INPUT_DIR, IPC_CLOSE_SENTINEL); const IPC_INPUT_CLOSE_SENTINEL = path.join(IPC_INPUT_DIR, IPC_CLOSE_SENTINEL);
const HOST_TASKS_DIR = path.join(HOST_IPC_DIR, 'tasks'); const HOST_TASKS_DIR = path.join(HOST_IPC_DIR, 'tasks');
const MIME_TYPES: Record<string, string> = {
'.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<SDKUserMessage> {
while (true) {
while (this.queue.length > 0) {
yield this.queue.shift()!;
}
if (this.done) return;
await new Promise<void>((r) => {
this.waiting = r;
});
this.waiting = null;
}
}
}
async function readStdin(): Promise<string> {
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 { function log(message: string): void {
console.error(`[agent-runner] ${message}`); console.error(`[agent-runner] ${message}`);
@@ -242,349 +89,6 @@ process.on('SIGTERM', () => {
agentAbortController.abort(); 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<void> {
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<string, unknown>),
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. * Run a single query and stream results via writeOutput.
* Uses MessageStream (AsyncIterable) to keep isSingleUserTurn=false, * Uses MessageStream (AsyncIterable) to keep isSingleUserTurn=false,
@@ -605,7 +109,7 @@ async function runQuery(
closedDuringQuery: boolean; closedDuringQuery: boolean;
terminalResultObserved: boolean; terminalResultObserved: boolean;
}> { }> {
const stream = new MessageStream(); const stream = new MessageStream((text) => buildMultimodalContent(text, log));
stream.push(prompt); stream.push(prompt);
// Poll IPC for follow-up messages and _close sentinel during the query // Poll IPC for follow-up messages and _close sentinel during the query
@@ -613,7 +117,7 @@ async function runQuery(
let closedDuringQuery = false; let closedDuringQuery = false;
const pollIpcDuringQuery = () => { const pollIpcDuringQuery = () => {
if (!ipcPolling) return; if (!ipcPolling) return;
if (shouldClose()) { if (shouldClose(IPC_INPUT_CLOSE_SENTINEL)) {
log('Close sentinel detected during query, ending stream'); log('Close sentinel detected during query, ending stream');
// Flush any buffered text before closing — the for-await loop may not // Flush any buffered text before closing — the for-await loop may not
// reach the post-loop flush code after stream.end(). // reach the post-loop flush code after stream.end().
@@ -635,7 +139,7 @@ async function runQuery(
ipcPolling = false; ipcPolling = false;
return; return;
} }
const messages = drainIpcInput(); const messages = drainIpcInput(IPC_INPUT_DIR, log);
for (const text of messages) { for (const text of messages) {
log(`Piping IPC message into active query (${text.length} chars)`); log(`Piping IPC message into active query (${text.length} chars)`);
stream.push(text); stream.push(text);
@@ -794,7 +298,18 @@ async function runQuery(
}, },
hooks: { hooks: {
PreCompact: [ PreCompact: [
{ hooks: [createPreCompactHook(runnerInput.assistantName)] }, {
hooks: [
createPreCompactHook({
assistantName: runnerInput.assistantName,
groupDir: GROUP_DIR,
groupFolder: GROUP_FOLDER,
hostTasksDir: HOST_TASKS_DIR,
log,
writeOutput,
}),
],
},
], ],
PreToolUse: [ PreToolUse: [
{ {
@@ -1163,7 +678,7 @@ async function main(): Promise<void> {
if (runnerInput.isScheduledTask) { 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}`; 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) { if (pending.length > 0) {
log(`Draining ${pending.length} pending IPC messages into initial prompt`); log(`Draining ${pending.length} pending IPC messages into initial prompt`);
prompt += '\n' + pending.join('\n'); prompt += '\n' + pending.join('\n');
@@ -1202,7 +717,18 @@ async function main(): Promise<void> {
abortController: agentAbortController, abortController: agentAbortController,
hooks: { hooks: {
PreCompact: [ PreCompact: [
{ hooks: [createPreCompactHook(runnerInput.assistantName)] }, {
hooks: [
createPreCompactHook({
assistantName: runnerInput.assistantName,
groupDir: GROUP_DIR,
groupFolder: GROUP_FOLDER,
hostTasksDir: HOST_TASKS_DIR,
log,
writeOutput,
}),
],
},
], ],
}, },
}, },

52
runners/agent-runner/src/ipc-input.ts Normal file
View File

@@ -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 [];
}
}

174
runners/agent-runner/src/output-protocol.ts Normal file
View File

@@ -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<string, string> = {
'.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<SDKUserMessage> {
while (true) {
while (this.queue.length > 0) {
yield this.queue.shift()!;
}
if (this.done) return;
await new Promise<void>((resolve) => {
this.waiting = resolve;
});
this.waiting = null;
}
}
}
export async function readStdin(): Promise<string> {
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;
}

304
runners/agent-runner/src/runner-hooks.ts Normal file
View File

@@ -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<PreCompactHookOptions, 'groupFolder' | 'hostTasksDir' | 'log'>,
): Promise<void> {
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<string, unknown>),
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.',
},
};
};
}

203
src/group-queue-state.ts Normal file
View File

@@ -0,0 +1,203 @@
import type { ChildProcess } from 'child_process';
import { logger } from './logger.js';
export interface QueuedTask {
id: string;
groupJid: string;
fn: () => Promise<void>;
}
export interface GroupRunContext {
runId: string;
reason: 'messages' | 'drain';
}
export type RunPhase =
| 'idle'
| 'running_messages'
| 'running_task'
| 'closing_messages';
const VALID_TRANSITIONS: Record<RunPhase, readonly RunPhase[]> = {
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<string, string>;
recentDirectTerminalDeliveries: Map<string, Map<string, string>>;
pendingMessages: boolean;
pendingTasks: QueuedTask[];
process: ChildProcess | null;
processName: string | null;
ipcDir: string | null;
retryCount: number;
retryTimer: ReturnType<typeof setTimeout> | null;
retryScheduledAt: number | null;
postCloseTermTimer: ReturnType<typeof setTimeout> | null;
postCloseKillTimer: ReturnType<typeof setTimeout> | 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;
}
}

214
src/group-queue.ts
View File

@@ -6,18 +6,21 @@ import {
RECOVERY_DURATION_MS, RECOVERY_DURATION_MS,
} from './config.js'; } from './config.js';
import { queueFollowUpMessage, writeCloseSentinel } from './group-queue-ipc.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 { logger } from './logger.js';
import type {
GroupRunContext,
GroupState,
QueuedTask,
} from './group-queue-state.js';
interface QueuedTask { export type { GroupRunContext } from './group-queue-state.js';
id: string;
groupJid: string;
fn: () => Promise<void>;
}
export interface GroupRunContext {
runId: string;
reason: 'messages' | 'drain';
}
const MAX_RETRIES = 5; const MAX_RETRIES = 5;
const BASE_RETRY_MS = 5000; const BASE_RETRY_MS = 5000;
@@ -26,178 +29,6 @@ const MAX_CONCURRENT_TASKS =
const POST_CLOSE_SIGTERM_DELAY_MS = 60_000; const POST_CLOSE_SIGTERM_DELAY_MS = 60_000;
const POST_CLOSE_SIGKILL_DELAY_MS = 75_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<RunPhase, readonly RunPhase[]> = {
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<string, string>;
recentDirectTerminalDeliveries: Map<string, Map<string, string>>;
pendingMessages: boolean;
pendingTasks: QueuedTask[];
process: ChildProcess | null;
processName: string | null;
ipcDir: string | null;
retryCount: number;
retryTimer: ReturnType<typeof setTimeout> | null;
retryScheduledAt: number | null;
postCloseTermTimer: ReturnType<typeof setTimeout> | null;
postCloseKillTimer: ReturnType<typeof setTimeout> | 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 { export interface GroupStatus {
jid: string; jid: string;
status: 'processing' | 'waiting' | 'inactive'; status: 'processing' | 'waiting' | 'inactive';
@@ -222,24 +53,7 @@ export class GroupQueue {
private getGroup(groupJid: string): GroupState { private getGroup(groupJid: string): GroupState {
let state = this.groups.get(groupJid); let state = this.groups.get(groupJid);
if (!state) { if (!state) {
state = { state = createGroupState();
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,
};
this.groups.set(groupJid, state); this.groups.set(groupJid, state);
} }
return state; return state;

65
src/ipc-file-claims.ts Normal file
View File

@@ -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)),
);
}

45
src/ipc-message-forwarding.ts Normal file
View File

@@ -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<string, RegisteredGroup>,
sendMessage: (
jid: string,
text: string,
senderRole?: string,
runId?: string,
) => Promise<void>,
): Promise<IpcMessageForwardResult> {
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,
};
}

502
src/ipc-task-processor.ts Normal file
View File

@@ -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<void> {
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<typeof updateTask>[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');
}
}

71
src/ipc-types.ts Normal file
View File

@@ -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<void>;
nudgeScheduler?: () => void;
roomBindings: () => Record<string, RegisteredGroup>;
assignRoom: (jid: string, room: AssignRoomInput) => void;
syncGroups: (force: boolean) => Promise<void>;
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;
}

693
src/ipc.ts
View File

@@ -1,166 +1,36 @@
import fs from 'fs'; import fs from 'fs';
import path from 'path'; import path from 'path';
import { CronExpressionParser } from 'cron-parser';
import { DATA_DIR, IPC_POLL_INTERVAL, TIMEZONE } from './config.js'; import { DATA_DIR, IPC_POLL_INTERVAL, TIMEZONE } from './config.js';
import {
isHostEvidenceAction,
runHostEvidenceRequest,
writeHostEvidenceResponse,
} from './host-evidence.js';
import { readJsonFile } from './utils.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 { logger } from './logger.js';
import { import {
DEFAULT_WATCH_CI_MAX_DURATION_MS, claimIpcFile,
isWatchCiTask, moveClaimedIpcFileToError,
} from './task-watch-status.js'; quarantineClaimedIpcFiles,
import { AgentType, RegisteredGroup, RoomMode } from './types.js'; } 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 { export type {
sendMessage: ( IpcDeps,
jid: string, IpcMessageForwardResult,
text: string, IpcMessagePayload,
senderRole?: string, TaskIpcPayload,
runId?: string, } from './ipc-types.js';
) => Promise<void>; export {
nudgeScheduler?: () => void; claimIpcFile,
roomBindings: () => Record<string, RegisteredGroup>; forwardAuthorizedIpcMessage,
assignRoom: (jid: string, room: AssignRoomInput) => void; processTaskIpc,
syncGroups: (force: boolean) => Promise<void>; quarantineClaimedIpcFiles,
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<string, RegisteredGroup>,
sendMessage: IpcDeps['sendMessage'],
): Promise<IpcMessageForwardResult> {
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,
};
}
let ipcWatcherRunning = false; 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 { export function startIpcWatcher(deps: IpcDeps): void {
if (ipcWatcherRunning) { if (ipcWatcherRunning) {
@@ -307,7 +177,7 @@ export function startIpcWatcher(deps: IpcDeps): void {
throw new Error('Invalid JSON'); throw new Error('Invalid JSON');
// Pass source group identity to processTaskIpc for authorization // Pass source group identity to processTaskIpc for authorization
await processTaskIpc( await processTaskIpc(
data as Parameters<typeof processTaskIpc>[0], data as TaskIpcPayload,
sourceGroup, sourceGroup,
isMain, isMain,
deps, deps,
@@ -337,520 +207,3 @@ export function startIpcWatcher(deps: IpcDeps): void {
processIpcFiles(); processIpcFiles();
logger.info('IPC watcher started (per-group namespaces)'); 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<void> {
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<typeof updateTask>[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');
}
}