Something went wrong. Try again.
I'm going to build my own OpenClaw, with blackjack... and bun!
Something went wrong. Try again.
TypeScript
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600/** * ipc.ts – Inter-Process Communication watcher for file-based task commands. * * External processes (skills, scripts, the reload flow) communicate with the * running piclaw instance by dropping JSON files into the IPC directories: * - `<DATA_DIR>/ipc/messages/*.json` – outbound messages to send. * - `<DATA_DIR>/ipc/tasks/*.json` – task lifecycle commands (schedule, * pause, resume, cancel, update, cleanup, resume_chat, resume_pending). * * The watcher polls these directories every IPC_POLL_INTERVAL ms, processes * each file, and deletes it on success (or renames it to `error-*` on failure). * * Consumers: * - runtime.ts calls startIpcWatcher() at startup, passing in the required * dependency functions (sendMessage, sendNudge, resolveModel, etc.). * - The `schedule` skill writes task files to trigger scheduled tasks. * - The `send-message` skill writes message files for immediate delivery. * - The reload script writes `resume_pending` tasks to resume after restart. */
import { existsSync, mkdirSync, readFileSync, readdirSync, renameSync, unlinkSync } from "fs";import { basename, join } from "path";import { CronExpressionParser } from "cron-parser";import { DATA_DIR, getRuntimeTimingConfig } from "./core/config.js";import { MediaService } from "./channels/web/media/media-service.js";import { createTask, deleteTask, getTaskById, updateTask } from "./db.js";import type { ScheduledTask } from "./types.js";import { createUuid } from "./utils/ids.js";import { createLogger, debugSuppressedError } from "./utils/logger.js";import { validateShellCommand, validateShellCwd } from "./utils/task-validation.js";
const log = createLogger("ipc");
/** * Options bag accepted by the injected `sendMessage` callback. * Kept local to IPC so runtime wiring can evolve without cross-module coupling. */interface IpcMessageOptions { forceRoot?: boolean; threadId?: number | null; source?: string; mediaIds?: number[]; contentBlocks?: Array<Record<string, unknown>>;}
/** * Dependency injection contract for IPC command handlers. * * Provided by runtime wiring so IPC can send messages, resolve models, * and resume chats without importing runtime modules directly. */export interface IpcDeps { /** Send a text message to a specific chat JID. */ sendMessage: (jid: string, text: string, options?: IpcMessageOptions) => Promise<void>; /** Send a push notification nudge (Pushover). */ sendNudge?: (text: string) => Promise<void>; /** Validate and resolve a model identifier string. */ resolveModel?: (input: string) => { model?: string; error?: string }; /** Resume processing a chat (after restart or pause). */ resumeChat?: (data: Record<string, unknown>) => Promise<void>; /** Resume any pending agent turns after a restart. */ resumePending?: (data?: Record<string, unknown>) => Promise<void>; /** Run a Dream/AutoDream cycle out of band. */ runDream?: (data: Record<string, unknown>) => Promise<void>;}
/** Guard to prevent starting the watcher more than once. */let running = false;let pollTimer: ReturnType<typeof setTimeout> | null = null;
type IpcScheduleType = "cron" | "interval" | "once";type JsonRecord = Record<string, unknown>;type TaskUpdates = Parameters<typeof updateTask>[1];
function isJsonRecord(value: unknown): value is JsonRecord { return typeof value === "object" && value !== null && !Array.isArray(value);}
function getStringField(data: JsonRecord, key: string): string | undefined { const value = data[key]; return typeof value === "string" ? value : undefined;}
function getFiniteNumberField(data: JsonRecord, key: string): number | undefined { const value = data[key]; return Number.isFinite(value) ? Number(value) : undefined;}
function getBooleanField(data: JsonRecord, key: string): boolean | undefined { const value = data[key]; if (typeof value === "boolean") return value; if (typeof value === "number") return value === 1; if (typeof value === "string") { const normalized = value.trim().toLowerCase(); if (normalized === "true" || normalized === "1" || normalized === "yes" || normalized === "on") return true; if (normalized === "false" || normalized === "0" || normalized === "no" || normalized === "off") return false; } return undefined;}
function normalizeMediaContentType(value: string | undefined): string { if (!value) return ""; return value.split(";")[0].trim().toLowerCase();}
function getArrayField(data: JsonRecord, key: string): unknown[] | undefined { const value = data[key]; return Array.isArray(value) ? value : undefined;}
function resolveIpcChatJid(data: JsonRecord): string { return ( getStringField(data, "chatJid") || getStringField(data, "chat_jid") || process.env.PICLAW_CHAT_JID || "web:default" );}
function isScheduleType(value: string): value is IpcScheduleType { return value === "cron" || value === "interval" || value === "once";}
const ipcMediaService = new MediaService();
async function buildMediaPayloadFromIpcEntries(media: unknown[]): Promise<{ mediaIds: number[]; contentBlocks: Array<Record<string, unknown>>; warnings: string[];}> { const mediaEntries = media; const mediaIds: number[] = []; const contentBlocks: Array<Record<string, unknown>> = []; const warnings: string[] = [];
for (const item of mediaEntries) { if (!item || typeof item !== "object") { warnings.push("Skipped invalid media entry."); continue; }
const normalized = item as Record<string, unknown>; const mediaPath = getStringField(normalized, "path"); if (!mediaPath) { warnings.push("Skipped media entry with missing path."); continue; }
const filename = getStringField(normalized, "filename") || basename(mediaPath); const contentType = getStringField(normalized, "content_type"); const inline = getBooleanField(normalized, "inline");
const result = await ipcMediaService.createFromPath(mediaPath, contentType, filename); if (result.status !== 200) { warnings.push( `Failed to attach ${filename}: ${(result.body as { error?: string }).error || `HTTP ${result.status}`}`, ); continue; }
const body = result.body as { id?: number; contentType?: string }; const mediaId = Number(body.id); const mediaContentType = normalizeMediaContentType( (typeof body.contentType === "string" ? body.contentType : contentType) || "" ); if (!Number.isFinite(mediaId) || mediaId <= 0) { warnings.push(`Failed to attach ${filename}: invalid media id`); continue; }
const isImage = mediaContentType.startsWith("image/"); const finalInline = inline === undefined ? isImage : inline && isImage; mediaIds.push(mediaId); contentBlocks.push({ type: finalInline ? "image" : "file", media_id: mediaId, name: filename, mime_type: mediaContentType, }); }
return { mediaIds, contentBlocks, warnings };}
async function processIpcDir( dirPath: string, ipcDir: string, kind: "message" | "task", handler: (data: JsonRecord) => Promise<void>): Promise<void> { if (!existsSync(dirPath)) return;
for (const file of readdirSync(dirPath).filter((f) => f.endsWith(".json"))) { const fp = join(dirPath, file); try { const parsed = JSON.parse(readFileSync(fp, "utf-8")); if (!isJsonRecord(parsed)) { throw new Error("IPC payload must be a JSON object."); } await handler(parsed); unlinkSync(fp); } catch (e) { log.error("Failed to process IPC payload", { operation: "process_ipc_dir", kind, file, err: e, }); try { renameSync(fp, join(ipcDir, `error-${file}`)); } catch (renameErr) { debugSuppressedError(log, "Failed to rename a broken IPC payload into the error bucket; leaving the original file in place.", renameErr, { operation: "process_ipc_dir.rename_failed_payload", kind, file, }); } } }}
function computeScheduledNextRun( scheduleType: IpcScheduleType, scheduleValue: string): string | undefined { if (scheduleType === "cron") { try { const timezone = process.env.TZ || getRuntimeTimingConfig().timezone; const next = CronExpressionParser.parse(scheduleValue, { tz: timezone }).next().toISOString(); return next ?? undefined; } catch { return undefined; } }
if (scheduleType === "interval") { const ms = parseInt(scheduleValue, 10); if (isNaN(ms) || ms <= 0) return undefined; return new Date(Date.now() + ms).toISOString(); }
const d = new Date(scheduleValue); if (isNaN(d.getTime())) return undefined; return d.toISOString();}
/** * Start the IPC directory watcher. Polls for new JSON files in the messages * and tasks directories on a recurring timer. * * Called once by runtime.ts during application startup. */export function startIpcWatcher(deps: IpcDeps): () => void { if (running) return stopIpcWatcher; running = true;
const ipcDir = join(DATA_DIR, "ipc"); const tasksDir = join(ipcDir, "tasks"); const messagesDir = join(ipcDir, "messages"); mkdirSync(tasksDir, { recursive: true }); mkdirSync(messagesDir, { recursive: true });
const poll = async () => { // --- Process outbound message files --- try { await processIpcDir(messagesDir, ipcDir, "message", (data) => processMessageCommand(data, deps)); } catch (e) { log.error("Failed to read IPC messages directory", { operation: "start_ipc_watcher.poll_messages", err: e, }); }
// --- Process task command files --- try { await processIpcDir(tasksDir, ipcDir, "task", (data) => processTaskCommand(data, deps)); } catch (e) { log.error("Failed to read IPC tasks directory", { operation: "start_ipc_watcher.poll_tasks", err: e, }); }
if (!running) return; pollTimer = setTimeout(poll, getRuntimeTimingConfig().ipcPollIntervalMs); };
poll(); log.info("IPC watcher started", { operation: "start_ipc_watcher" }); return stopIpcWatcher;}
/** Stop the active IPC watcher and clear associated runtime state. */export function stopIpcWatcher(): void { running = false; if (pollTimer) { clearTimeout(pollTimer); pollTimer = null; }}
/** * Dispatch a single IPC message command. */export async function processMessageCommand(data: JsonRecord, deps: IpcDeps): Promise<void> { const type = getStringField(data, "type"); const chatJid = resolveIpcChatJid(data); const text = getStringField(data, "text") || "";
if (type !== "message") return;
const media = getArrayField(data, "media") || []; const { mediaIds, contentBlocks, warnings } = await buildMediaPayloadFromIpcEntries(media);
const warningSuffix = warnings.length > 0 ? `\n\n⚠️ Media attachment warnings:\n${warnings.map((warning) => `- ${warning}`).join("\n")}` : ""; const finalText = `${text}${warningSuffix}`; const hasPayload = Boolean(text) || warnings.length > 0 || mediaIds.length > 0;
if (!hasPayload) return;
const options: IpcMessageOptions = {}; if (mediaIds.length > 0) options.mediaIds = mediaIds; if (contentBlocks.length > 0) options.contentBlocks = contentBlocks;
await deps.sendMessage(chatJid, finalText, options); if (data.noNudge !== true) { await deps.sendNudge?.(finalText || "Message with attachment(s)"); }}
/** * Dispatch a single IPC task command. The `data.type` field determines * which operation is performed (schedule, pause, resume, cancel, etc.). */export async function processTaskCommand(data: JsonRecord, deps: IpcDeps): Promise<void> { const commandType = getStringField(data, "type");
switch (commandType) { // --- Create a new scheduled task --- case "schedule_task": { const scheduleTypeValue = getStringField(data, "schedule_type"); const scheduleValue = getStringField(data, "schedule_value"); const chatJid = resolveIpcChatJid(data); if (!scheduleTypeValue || !scheduleValue || !isScheduleType(scheduleTypeValue)) return;
const taskKind = data.task_kind === "shell" || Boolean(data.command) ? "shell" : "agent"; const nextRun = computeScheduledNextRun(scheduleTypeValue, String(scheduleValue)); if (nextRun === undefined) return;
if (taskKind === "shell") { const validated = validateShellCommand(data.command); if (!validated.ok) { await deps.sendMessage(chatJid, `Cannot schedule shell task: ${validated.error || "Invalid command."}`); return; }
const cwdResult = validateShellCwd(data.cwd); if (!cwdResult.ok) { await deps.sendMessage(chatJid, `Cannot schedule shell task: ${cwdResult.error || "Invalid cwd."}`); return; }
if (data.model) { await deps.sendMessage(chatJid, "Cannot schedule shell task with a model override."); return; }
createTask({ id: createUuid("task"), chat_jid: chatJid, prompt: validated.command || "", model: null, task_kind: "shell", command: validated.command || null, cwd: cwdResult.cwd, timeout_sec: getFiniteNumberField(data, "timeout_sec") ?? null, schedule_type: scheduleTypeValue, schedule_value: String(scheduleValue), next_run: nextRun, status: "active", created_at: new Date().toISOString(), }); break; }
const prompt = getStringField(data, "prompt"); if (!prompt?.trim()) return;
// Validate the model override if one was requested. const modelInput = getStringField(data, "model"); const requested = modelInput?.trim() ? modelInput.trim() : null; let model: string | null = null;
if (requested) { if (!deps.resolveModel) { await deps.sendMessage(chatJid, `Cannot schedule task: model validation unavailable for "${requested}".`); return; } const resolved = deps.resolveModel(requested); if (!resolved.model) { await deps.sendMessage(chatJid, `Cannot schedule task: ${resolved.error || "Invalid model."}`); return; } model = resolved.model; }
createTask({ id: createUuid("task"), chat_jid: chatJid, prompt, model, task_kind: "agent", command: null, cwd: null, timeout_sec: null, schedule_type: scheduleTypeValue, schedule_value: String(scheduleValue), next_run: nextRun, status: "active", created_at: new Date().toISOString(), }); break; }
// --- Pause an active task --- case "pause_task": { const taskId = getStringField(data, "taskId"); if (!taskId) return; const t = getTaskById(taskId); if (t) updateTask(taskId, { status: "paused" }); break; }
// --- Resume a paused task --- case "resume_task": { const taskId = getStringField(data, "taskId"); if (!taskId) return; const t = getTaskById(taskId); if (t) updateTask(taskId, { status: "active" }); break; }
// --- Delete a task and its run logs --- case "cancel_task": { const taskId = getStringField(data, "taskId"); if (!taskId) return; const t = getTaskById(taskId); if (t) deleteTask(taskId); break; }
// --- Partially update a task (prompt, schedule, model) --- case "update_task": { const taskId = getStringField(data, "taskId"); const chatJid = getStringField(data, "chatJid"); if (!taskId) return;
const t = getTaskById(taskId); if (!t) return;
const updates: TaskUpdates = {};
const prompt = getStringField(data, "prompt"); if (typeof prompt === "string") updates.prompt = prompt;
const scheduleType = getStringField(data, "schedule_type"); if (typeof scheduleType === "string") { updates.schedule_type = scheduleType as ScheduledTask["schedule_type"]; }
const scheduleValue = getStringField(data, "schedule_value"); if (typeof scheduleValue === "string") updates.schedule_value = scheduleValue;
const taskKind = getStringField(data, "task_kind"); if (typeof taskKind === "string") { updates.task_kind = taskKind as ScheduledTask["task_kind"]; }
const command = getStringField(data, "command"); if (typeof command === "string") updates.command = command;
const cwd = getStringField(data, "cwd"); if (typeof cwd === "string") updates.cwd = cwd;
const timeoutSec = getFiniteNumberField(data, "timeout_sec"); if (timeoutSec !== undefined) updates.timeout_sec = timeoutSec;
const modelInput = getStringField(data, "model"); if (typeof modelInput === "string") { if (modelInput === "") { updates.model = null; } else if (deps.resolveModel) { const resolved = deps.resolveModel(modelInput.trim()); if (!resolved.model) { if (chatJid) await deps.sendMessage(chatJid, `Cannot update task: ${resolved.error || "Invalid model."}`); return; } updates.model = resolved.model; } else { updates.model = modelInput.trim(); } }
// Validate shell command changes if applicable const kind = updates.task_kind ?? t.task_kind ?? (t.command ? "shell" : "agent"); if (kind === "shell") { if (updates.command !== undefined) { const validated = validateShellCommand(updates.command); if (!validated.ok) { if (chatJid) await deps.sendMessage(chatJid, `Cannot update task: ${validated.error || "Invalid command."}`); return; } const validatedCommand = validated.command || ""; updates.command = validatedCommand; updates.prompt = validatedCommand; }
if (updates.cwd !== undefined) { const cwdResult = validateShellCwd(updates.cwd); if (!cwdResult.ok) { if (chatJid) await deps.sendMessage(chatJid, `Cannot update task: ${cwdResult.error || "Invalid cwd."}`); return; } updates.cwd = cwdResult.cwd; }
if (updates.model !== undefined && updates.model) { if (chatJid) await deps.sendMessage(chatJid, "Cannot set model override on shell tasks."); return; } }
// Recalculate next_run if schedule changed. if (updates.schedule_type || updates.schedule_value) { const currentScheduleType = String(updates.schedule_type ?? t.schedule_type); const currentScheduleValue = String(updates.schedule_value ?? t.schedule_value); if (isScheduleType(currentScheduleType)) { const nextRun = computeScheduledNextRun(currentScheduleType, currentScheduleValue); if (nextRun !== undefined) { updates.next_run = nextRun; } } }
if (Object.keys(updates).length > 0) updateTask(taskId, updates); break; }
// --- Bulk-delete all completed tasks --- case "cleanup_tasks": { const chatJid = getStringField(data, "chatJid"); const db = (await import("./db/connection.js")).getDb(); const completed = db.prepare("SELECT id FROM scheduled_tasks WHERE status = 'completed'").all() as { id: string }[];
for (const { id } of completed) { db.prepare("DELETE FROM task_run_logs WHERE task_id = ?").run(id); }
const result = db.prepare("DELETE FROM scheduled_tasks WHERE status = 'completed'").run(); const countValue = (result as { changes?: unknown }).changes; const count = typeof countValue === "number" ? countValue : 0; if (chatJid) await deps.sendMessage(chatJid, `Cleaned up ${count} completed task(s).`); break; }
// --- Resume a specific chat after restart --- case "resume_chat": { if (deps.resumeChat) { await deps.resumeChat(data); } break; }
// --- Resume any pending agent turns after restart --- case "resume_pending": { log.info("Processing resume_pending IPC task", { operation: "process_task_command.resume_pending", chatJid: getStringField(data, "chatJid") || "all", }); if (deps.resumePending) { await deps.resumePending(data); } break; }
case "run_dream": { log.info("Processing run_dream IPC task", { operation: "process_task_command.run_dream", chatJid: getStringField(data, "chatJid") || getStringField(data, "chat_jid") || "web:default", }); if (deps.runDream) { await deps.runDream(data); } break; } }}