Abiba
b6fa3da540
hermes-zulip-plugin (Tanko/Mumuni):
- Add _extract_event_attachments() to handle message.attachments from
Zulip UI file uploads (previously only inline markdown links worked)
- Add shared _cache_attachment() method for images/audio/documents
- Refactor _extract_inline_media() to use _cache_attachment()
- Fix connect() signature for hermes-agent 0.18.0 compat (add is_reconnect)
pi-zulip-extension (Abiba):
- Extend attachment handling beyond images only: text files (decoded
inline), PDFs (pdftotext extraction), binary files (metadata)
- 50K char cap on text/PDF extraction to prevent context flooding
- Classify attachments by extension (image/text/pdf/binary)
pi-mcp-extension (Abiba):
- Detect bridge-side text truncation (… ellipsis marker)
- Rebuild relay message display from structuredContent when truncated
- Add rebuildRelayTextFromStructuredContent() helper
Config:
- Add ZULIP_ROLE=router to ecosystem.abiba.config.cjs (was missing)
1256 lines
44 KiB
JavaScript
1256 lines
44 KiB
JavaScript
/**
|
||
* pi-zulip-extension v2 — Router-worker architecture for per-sender session isolation
|
||
*
|
||
* Deploy to: ~/.pi/agent/extensions/zulip/
|
||
* Config: config.yaml (alongside extension)
|
||
*
|
||
* Architecture:
|
||
* Router (this process, PM2 abiba-zulip, ZULIP_ROLE=router):
|
||
* Polls Zulip for events, maintains a pool of per-sender pi RPC workers.
|
||
* Each sender gets their own pi --mode rpc process with a dedicated session file.
|
||
* Messages are routed to the correct worker; worker responses are relayed
|
||
* back to Zulip (with streaming edits).
|
||
* Worker (child pi --mode rpc --session-dir ...):
|
||
* Extension loads but is NO-OP (ZULIP_ROLE != router). Just runs the agent
|
||
* with per-sender persistent sessions. No Zulip logic in the worker.
|
||
*/
|
||
|
||
import fs from "node:fs";
|
||
import path from "node:path";
|
||
import url from "node:url";
|
||
import http from "node:http";
|
||
import { spawn } from "node:child_process";
|
||
import { execSync } from "node:child_process";
|
||
import { parse } from "yaml";
|
||
|
||
// ---------------------------------------------------------------------------
|
||
// Constants
|
||
// ---------------------------------------------------------------------------
|
||
const EXT_DIR = path.dirname(url.fileURLToPath(import.meta.url));
|
||
const SESSION_DIR = path.join(process.env.HOME, ".pi", "agent", "sessions", "zulip");
|
||
const SESSIONS_FILE = path.join(SESSION_DIR, "worker-sessions.json");
|
||
const POLL_INTERVAL_MS = 3000;
|
||
const WORKER_IDLE_TIMEOUT_MS = 30 * 60 * 1000; // kill idle workers after 30 min
|
||
const STREAMING_EDIT_MS = 8000; // throttle Zulip message edits during streaming
|
||
const MAX_ZULIP_MSG = 10000;
|
||
const HEARTBEAT_INTERVAL_MS = 5 * 60 * 1000;
|
||
|
||
// ---------------------------------------------------------------------------
|
||
// Config loading (cleaned — flat yaml format expected)
|
||
// ---------------------------------------------------------------------------
|
||
function loadConfig() {
|
||
const configPath = path.join(EXT_DIR, "config.yaml");
|
||
if (!fs.existsSync(configPath)) {
|
||
throw new Error(`No config.yaml at ${configPath}`);
|
||
}
|
||
const raw = fs.readFileSync(configPath, "utf-8");
|
||
const cfg = parse(raw);
|
||
|
||
function get(key) {
|
||
if (cfg[key] !== undefined) return cfg[key];
|
||
const parts = key.split(".");
|
||
let cur = cfg;
|
||
for (const p of parts) {
|
||
if (cur && typeof cur === "object" && p in cur) cur = cur[p];
|
||
else return undefined;
|
||
}
|
||
return cur;
|
||
}
|
||
|
||
return {
|
||
zulip: {
|
||
email: String(get("zulip.email") ?? ""),
|
||
api_key: process.env.ZULIP_API_KEY ?? String(get("zulip.api_key") ?? ""),
|
||
site: String(get("zulip.site") ?? ""),
|
||
all_bots_user_id: Number(get("zulip.all_bots_user_id") ?? 1),
|
||
},
|
||
agent: {
|
||
name: String(get("agent.name") ?? "abiba"),
|
||
owner_email: String(get("agent.owner_email") ?? "jerome@sysloggh.com"),
|
||
},
|
||
health_port: parseInt(String(get("health_port") ?? "9200"), 10),
|
||
};
|
||
}
|
||
|
||
const config = loadConfig();
|
||
|
||
// ---------------------------------------------------------------------------
|
||
// Zulip client (raw fetch — no zulip-js dependency)
|
||
// ---------------------------------------------------------------------------
|
||
function authHeader() {
|
||
return "Basic " + Buffer.from(`${config.zulip.email}:${config.zulip.api_key}`).toString("base64");
|
||
}
|
||
|
||
/**
|
||
* Create a Zulip event queue and resolve bot + all-bots user IDs.
|
||
*/
|
||
async function createZulipQueue() {
|
||
const auth = authHeader();
|
||
|
||
// Register event queue
|
||
const regForm = new URLSearchParams();
|
||
regForm.append("event_types", JSON.stringify(["message"]));
|
||
const regRes = await fetch(`${config.zulip.site}/api/v1/register`, {
|
||
method: "POST",
|
||
headers: { Authorization: auth, "Content-Type": "application/x-www-form-urlencoded" },
|
||
body: regForm,
|
||
});
|
||
if (!regRes.ok) throw new Error(`Queue registration failed: ${regRes.status} ${await regRes.text().catch(() => '').then(t => t.slice(0,100))}`);
|
||
const qr = await regRes.json();
|
||
|
||
let lastEventId = qr.last_event_id ?? -1;
|
||
let botUserId = null;
|
||
let allBotsUserId = config.zulip.all_bots_user_id ?? 1;
|
||
|
||
// Resolve bot's own user_id
|
||
try {
|
||
const meRes = await fetch(`${config.zulip.site}/api/v1/users/me`, { headers: { Authorization: auth } });
|
||
if (meRes.ok) {
|
||
const md = await meRes.json();
|
||
if (md.user_id) botUserId = md.user_id;
|
||
}
|
||
} catch { /* non-critical */ }
|
||
|
||
// Resolve @all-bots user_id dynamically
|
||
try {
|
||
const usersRes = await fetch(`${config.zulip.site}/api/v1/users`, { headers: { Authorization: auth } });
|
||
if (usersRes.ok) {
|
||
const ud = await usersRes.json();
|
||
for (const m of ud.members || []) {
|
||
if ((m.email || "").toLowerCase().includes("all-bots")) {
|
||
allBotsUserId = m.user_id;
|
||
break;
|
||
}
|
||
}
|
||
}
|
||
} catch { /* fall back to config */ }
|
||
|
||
console.log(`[zulip-ext] Bot user_id=${botUserId}, all-bots user_id=${allBotsUserId}`);
|
||
|
||
return {
|
||
queueId: qr.queue_id,
|
||
lastEventId,
|
||
botUserId,
|
||
allBotsUserId,
|
||
async poll() {
|
||
const res = await fetch(
|
||
`${config.zulip.site}/api/v1/events?queue_id=${encodeURIComponent(qr.queue_id)}&last_event_id=${lastEventId}`,
|
||
{ headers: { Authorization: authHeader() } }
|
||
);
|
||
if (!res.ok) {
|
||
if (res.status === 429) {
|
||
const ra = parseFloat(res.headers.get("retry-after") || "1");
|
||
await new Promise(r => setTimeout(r, ra * 1000 + 100));
|
||
return [];
|
||
}
|
||
const errBody = await res.text().catch(() => '');
|
||
throw new Error(`Events API ${res.status}: ${errBody.slice(0, 200)}`);
|
||
}
|
||
const data = await res.json();
|
||
if (data.events) {
|
||
for (const e of data.events) {
|
||
if (e.id > lastEventId) lastEventId = e.id;
|
||
}
|
||
return data.events.filter(e => e.type === "message");
|
||
}
|
||
return [];
|
||
},
|
||
async sendMessage(params) {
|
||
const fb = new URLSearchParams();
|
||
fb.append("type", params.type);
|
||
fb.append("to", params.type === "private" ? JSON.stringify([params.to]) : String(params.to));
|
||
if (params.subject) fb.append("subject", params.subject);
|
||
fb.append("content", params.content);
|
||
const res = await fetch(`${config.zulip.site}/api/v1/messages`, {
|
||
method: "POST",
|
||
headers: { Authorization: authHeader(), "Content-Type": "application/x-www-form-urlencoded" },
|
||
body: fb.toString(),
|
||
});
|
||
if (!res.ok) { const errBody = await res.text().catch(() => ''); throw new Error(`Send failed ${res.status}: ${(errBody || '').slice(0, 200)}`); }
|
||
return (await res.json()).id;
|
||
},
|
||
async editMessage(messageId, content) {
|
||
const fb = new URLSearchParams();
|
||
fb.append("content", content);
|
||
const res = await fetch(`${config.zulip.site}/api/v1/messages/${messageId}`, {
|
||
method: "PATCH",
|
||
headers: { Authorization: authHeader(), "Content-Type": "application/x-www-form-urlencoded" },
|
||
body: fb.toString(),
|
||
});
|
||
if (!res.ok) {
|
||
console.warn(`[zulip-ext] Edit ${messageId} failed: ${res.status}`);
|
||
}
|
||
},
|
||
async sendTyping(userIds, op) {
|
||
const fb = new URLSearchParams();
|
||
fb.append("to", JSON.stringify(userIds));
|
||
fb.append("op", op);
|
||
await fetch(`${config.zulip.site}/api/v1/typing`, {
|
||
method: "POST",
|
||
headers: { Authorization: authHeader(), "Content-Type": "application/x-www-form-urlencoded" },
|
||
body: fb.toString(),
|
||
}).catch(() => {});
|
||
},
|
||
};
|
||
}
|
||
|
||
// ---------------------------------------------------------------------------
|
||
// Dynamic echo prevention — fetch bot users from Zulip API
|
||
// ---------------------------------------------------------------------------
|
||
let BOT_EMAILS = new Set([config.zulip.email]); // always include self
|
||
|
||
async function loadBotEmails(queue) {
|
||
try {
|
||
const res = await fetch(`${config.zulip.site}/api/v1/users`, {
|
||
headers: { Authorization: authHeader() },
|
||
});
|
||
if (res.ok) {
|
||
const data = await res.json();
|
||
const count = BOT_EMAILS.size;
|
||
for (const m of data.members || []) {
|
||
if (m.is_bot && m.email) BOT_EMAILS.add(m.email);
|
||
}
|
||
console.log(`[zulip-ext] Echo prevention: ${BOT_EMAILS.size} bot emails (${BOT_EMAILS.size - count} dynamic)`);
|
||
}
|
||
} catch (e) {
|
||
console.warn(`[zulip-ext] Could not fetch bot list: ${e.message}`);
|
||
}
|
||
}
|
||
|
||
function isBotSender(email) {
|
||
return BOT_EMAILS.has(email);
|
||
}
|
||
|
||
const ALL_BOTS_RE = /@\*\*(?:all-bots|All Bots)\*\*/i;
|
||
function isAllBotsMention(content) {
|
||
return ALL_BOTS_RE.test(content);
|
||
}
|
||
|
||
// ---------------------------------------------------------------------------
|
||
// Worker pool manager
|
||
// ---------------------------------------------------------------------------
|
||
/** @type {Map<string, Worker>} */
|
||
const workers = new Map();
|
||
let workerSessions = {}; // senderId → sessionFilePath (persisted to disk)
|
||
let cmdIdCounter = 0;
|
||
let idleCleanupTimer = null;
|
||
|
||
function loadWorkerSessions() {
|
||
try {
|
||
if (fs.existsSync(SESSIONS_FILE)) {
|
||
return JSON.parse(fs.readFileSync(SESSIONS_FILE, "utf-8"));
|
||
}
|
||
} catch (e) {
|
||
console.warn(`[zulip-ext] Failed to load worker sessions: ${e.message}`);
|
||
}
|
||
return {};
|
||
}
|
||
|
||
function saveWorkerSessions() {
|
||
try {
|
||
if (!fs.existsSync(SESSION_DIR)) fs.mkdirSync(SESSION_DIR, { recursive: true });
|
||
fs.writeFileSync(SESSIONS_FILE, JSON.stringify(workerSessions, null, 2));
|
||
} catch (e) {
|
||
console.warn(`[zulip-ext] Failed to save worker sessions: ${e.message}`);
|
||
}
|
||
}
|
||
|
||
/**
|
||
* Attach a JSONL line reader to a stream.
|
||
*/
|
||
function attachJsonlReader(stream, onLine, onError) {
|
||
let buf = "";
|
||
const td = new TextDecoder();
|
||
stream.on("data", (chunk) => {
|
||
buf += typeof chunk === "string" ? chunk : td.decode(chunk, { stream: true });
|
||
while (true) {
|
||
const idx = buf.indexOf("\n");
|
||
if (idx === -1) break;
|
||
const line = buf.slice(0, idx).replace(/\r$/, "");
|
||
buf = buf.slice(idx + 1);
|
||
if (!line) continue;
|
||
try {
|
||
onLine(JSON.parse(line));
|
||
} catch (e) {
|
||
if (onError) onError(e, line);
|
||
}
|
||
}
|
||
});
|
||
}
|
||
|
||
/**
|
||
* Spawn a pi RPC worker process for a sender.
|
||
*/
|
||
function spawnWorker(senderId) {
|
||
const sessionDir = SESSION_DIR;
|
||
if (!fs.existsSync(sessionDir)) fs.mkdirSync(sessionDir, { recursive: true });
|
||
|
||
const proc = spawn("pi", [
|
||
"--mode", "rpc",
|
||
"--session-dir", sessionDir,
|
||
], {
|
||
env: {
|
||
...process.env,
|
||
ZULIP_ROLE: "worker",
|
||
ZULIP_EXTENSION_ACTIVE: "", // ensure NOT active
|
||
},
|
||
stdio: ["pipe", "pipe", "pipe"],
|
||
cwd: process.env.HOME,
|
||
});
|
||
|
||
const worker = {
|
||
process: proc,
|
||
senderId,
|
||
busy: false,
|
||
streamingBuffer: "",
|
||
lastStreamEdit: 0,
|
||
pendingReplies: [], // [{ zulipMsgId, senderId, senderName, replyInfo, createdAt }]
|
||
lastMessageContent: null,
|
||
lastReplyInfo: null,
|
||
lastActivity: Date.now(),
|
||
createdAt: Date.now(),
|
||
sessionFile: workerSessions[senderId] || null,
|
||
pendingCommands: new Map(),
|
||
rpcCommandQueue: [],
|
||
};
|
||
|
||
// Read stdout events (JSONL)
|
||
attachJsonlReader(
|
||
proc.stdout,
|
||
(event) => handleWorkerEvent(worker, event),
|
||
(err, line) => console.warn(`[zulip-ext] Worker ${senderId} parse error: ${err.message} | ${line.slice(0, 100)}`)
|
||
);
|
||
|
||
// Pipe stderr for diagnostics
|
||
proc.stderr.on("data", (d) => {
|
||
const text = d.toString().trim();
|
||
if (text) process.stderr.write(`[worker-${senderId}] ${text}\n`);
|
||
});
|
||
|
||
// Handle exit
|
||
proc.on("exit", (code, signal) => {
|
||
console.log(`[zulip-ext] Worker ${senderId} exited: code=${code} signal=${signal || "none"}`);
|
||
// Fail any pending replies
|
||
while (worker.pendingReplies.length > 0) {
|
||
const reply = worker.pendingReplies.shift();
|
||
if (reply.zulipMsgId && zulipQueue) {
|
||
zulipQueue.editMessage(reply.zulipMsgId,
|
||
":warning: Response interrupted — worker process exited. Please try again."
|
||
).catch(() => {});
|
||
}
|
||
}
|
||
workers.delete(senderId);
|
||
});
|
||
|
||
// Resolve session file
|
||
(async () => {
|
||
if (worker.sessionFile) {
|
||
// Resume existing session
|
||
const cmdId = String(++cmdIdCounter);
|
||
sendRpc(worker, { id: cmdId, type: "switch_session", sessionPath: worker.sessionFile });
|
||
console.log(`[zulip-ext] Worker ${senderId}: resuming session ${worker.sessionFile}`);
|
||
}
|
||
// If no session file, the auto-created session is fine. Discover it.
|
||
if (!worker.sessionFile) {
|
||
const cmdId = String(++cmdIdCounter);
|
||
sendRpc(worker, { id: cmdId, type: "get_state" });
|
||
}
|
||
})();
|
||
|
||
return worker;
|
||
}
|
||
|
||
function sendRpc(worker, cmd) {
|
||
worker.process.stdin.write(JSON.stringify(cmd) + "\n");
|
||
}
|
||
|
||
/**
|
||
* Handle all RPC events from a worker.
|
||
*/
|
||
function handleWorkerEvent(worker, event) {
|
||
// Debug: log all worker events to diagnose processing
|
||
if (event.type !== 'message_update' || !event.assistantMessageEvent || event.assistantMessageEvent.type === 'text_start' || event.assistantMessageEvent.type === 'text_end') {
|
||
console.log(`[zulip-ext] Worker ${worker.senderId} event: ${event.type}${event.assistantMessageEvent ? ':' + event.assistantMessageEvent.type : ''}${event.command ? ' cmd=' + event.command : ''}`);
|
||
}
|
||
switch (event.type) {
|
||
case "response":
|
||
handleWorkerResponse(worker, event);
|
||
break;
|
||
case "message_update":
|
||
handleWorkerStreaming(worker, event);
|
||
break;
|
||
case "agent_end":
|
||
handleWorkerAgentEnd(worker, event);
|
||
break;
|
||
case "agent_start":
|
||
break;
|
||
case "turn_start":
|
||
break;
|
||
case "turn_end":
|
||
break;
|
||
case "compaction_start":
|
||
console.log(`[zulip-ext] Worker ${worker.senderId}: compaction started`);
|
||
break;
|
||
case "compaction_end":
|
||
console.log(`[zulip-ext] Worker ${worker.senderId}: compaction ended`);
|
||
// Track new session file path if it changed
|
||
break;
|
||
case "extension_error":
|
||
console.error(`[zulip-ext] Worker ${worker.senderId} ext error: ${event.error?.slice(0, 200)}`);
|
||
break;
|
||
case "message_start":
|
||
case "message_end":
|
||
case "tool_execution_start":
|
||
case "tool_execution_update":
|
||
case "tool_execution_end":
|
||
case "queue_update":
|
||
case "auto_retry_start":
|
||
case "auto_retry_end":
|
||
break; // informational, not needed by router
|
||
}
|
||
}
|
||
|
||
function handleWorkerResponse(worker, event) {
|
||
// Resolve pending command callbacks
|
||
if (event.id) {
|
||
const cb = worker.pendingCommands.get(event.id);
|
||
if (cb) {
|
||
worker.pendingCommands.delete(event.id);
|
||
cb.resolve(event);
|
||
return;
|
||
}
|
||
}
|
||
|
||
// Handle responses without explicit callbacks
|
||
if (event.command === "get_state" && event.success) {
|
||
const sessionFile = event.data?.sessionFile;
|
||
if (sessionFile && !worker.sessionFile) {
|
||
worker.sessionFile = sessionFile;
|
||
workerSessions[worker.senderId] = sessionFile;
|
||
saveWorkerSessions();
|
||
console.log(`[zulip-ext] Worker ${worker.senderId}: session file = ${sessionFile}`);
|
||
}
|
||
}
|
||
|
||
if (event.command === "switch_session" && event.success) {
|
||
console.log(`[zulip-ext] Worker ${worker.senderId}: session switched, cancelled=${event.data?.cancelled}`);
|
||
}
|
||
|
||
if (event.command === "new_session" && event.success) {
|
||
// Track the new session file (get_state will follow if needed)
|
||
if (!event.data?.cancelled) {
|
||
const cmdId = String(++cmdIdCounter);
|
||
worker.pendingCommands.set(cmdId, {
|
||
resolve: (ev) => {
|
||
const sf = ev.data?.sessionFile;
|
||
if (sf) {
|
||
worker.sessionFile = sf;
|
||
workerSessions[worker.senderId] = sf;
|
||
saveWorkerSessions();
|
||
console.log(`[zulip-ext] Worker ${worker.senderId}: new session = ${sf}`);
|
||
}
|
||
},
|
||
});
|
||
sendRpc(worker, { id: cmdId, type: "get_state" });
|
||
}
|
||
}
|
||
}
|
||
|
||
function handleWorkerStreaming(worker, event) {
|
||
const delta = event.assistantMessageEvent;
|
||
if (!delta || delta.type !== "text_delta") return;
|
||
|
||
worker.streamingBuffer += delta.delta;
|
||
const now = Date.now();
|
||
if (worker.streamingBuffer.length > 10 && (!worker.lastStreamEdit || now - worker.lastStreamEdit >= STREAMING_EDIT_MS)) {
|
||
worker.lastStreamEdit = now;
|
||
const reply = worker.pendingReplies[0];
|
||
if (reply?.zulipMsgId && zulipQueue) {
|
||
const preview = worker.streamingBuffer.length > 9500
|
||
? worker.streamingBuffer.slice(0, 9500) + "\n\n_… still generating…_"
|
||
: worker.streamingBuffer;
|
||
zulipQueue.editMessage(reply.zulipMsgId, preview).catch(() => {});
|
||
}
|
||
}
|
||
}
|
||
|
||
async function handleWorkerAgentEnd(worker, event) {
|
||
worker.busy = false;
|
||
worker.streamingBuffer = "";
|
||
worker.lastStreamEdit = 0;
|
||
worker.lastActivity = Date.now();
|
||
|
||
if (worker.pendingReplies.length === 0) return;
|
||
const reply = worker.pendingReplies.shift();
|
||
|
||
// Extract final assistant text — walk backwards to find the last message with text content
|
||
// (the final turn may only have tool calls with brief inline text like "On it. Let me")
|
||
const assistantMsgs = (event.messages || []).filter(m => m.role === "assistant");
|
||
let responseText = "";
|
||
for (let i = assistantMsgs.length - 1; i >= 0; i--) {
|
||
const content = assistantMsgs[i].content;
|
||
if (!content) continue;
|
||
if (typeof content === "string") {
|
||
responseText = content;
|
||
break;
|
||
}
|
||
if (Array.isArray(content)) {
|
||
const texts = content.filter(c => c.type === "text").map(c => c.text).join("\n");
|
||
if (texts.trim()) {
|
||
responseText = texts;
|
||
break;
|
||
}
|
||
}
|
||
}
|
||
|
||
// Stop typing
|
||
if (reply.senderId && zulipQueue) {
|
||
zulipQueue.sendTyping([reply.senderId], "stop").catch(() => {});
|
||
}
|
||
|
||
if (!responseText.trim()) {
|
||
// All assistant messages are tool-call-only. Send a summary.
|
||
console.log(`[zulip-ext] Tool-only response for ${reply.senderName} — sending activity note`);
|
||
responseText = "⚙️ _Running tools…_\n\nUse `/continue` to see results or send another message.";
|
||
}
|
||
|
||
const truncated = responseText.length > MAX_ZULIP_MSG
|
||
? responseText.slice(0, MAX_ZULIP_MSG) + "\n\n[...truncated at Zulip limit]"
|
||
: responseText;
|
||
|
||
const target = reply.replyInfo.type === "private"
|
||
? { type: "private", to: reply.senderId }
|
||
: { type: "stream", to: reply.replyInfo.streamId ?? reply.replyInfo.streamName, subject: reply.replyInfo.topic };
|
||
|
||
try {
|
||
if (reply.zulipMsgId && zulipQueue) {
|
||
await zulipQueue.editMessage(reply.zulipMsgId, truncated);
|
||
console.log(`[zulip-ext] Finalized reply for ${reply.senderName} on worker ${worker.senderId}`);
|
||
} else if (zulipQueue) {
|
||
await zulipQueue.sendMessage({ ...target, content: truncated });
|
||
console.log(`[zulip-ext] Sent reply for ${reply.senderName}`);
|
||
}
|
||
} catch (err) {
|
||
console.error(`[zulip-ext] Failed to send reply: ${err.message}`);
|
||
}
|
||
}
|
||
|
||
/**
|
||
* Kill a worker and clean up.
|
||
*/
|
||
function killWorker(senderId) {
|
||
const w = workers.get(senderId);
|
||
if (!w) return;
|
||
console.log(`[zulip-ext] Killing idle worker ${senderId}`);
|
||
workers.delete(senderId);
|
||
try { w.process.kill("SIGTERM"); } catch { /* already dead */ }
|
||
// Note: workerSessions entry is kept so session resumes on re-connect
|
||
}
|
||
|
||
/**
|
||
* Periodically kill idle workers.
|
||
*/
|
||
function cleanupIdleWorkers() {
|
||
const now = Date.now();
|
||
for (const [senderId, w] of workers) {
|
||
if (w.busy) continue;
|
||
if (now - w.lastActivity > WORKER_IDLE_TIMEOUT_MS) {
|
||
killWorker(senderId);
|
||
}
|
||
}
|
||
}
|
||
|
||
// ---------------------------------------------------------------------------
|
||
// Attachment download
|
||
// ---------------------------------------------------------------------------
|
||
|
||
/** Download as raw buffer (for text extraction / base64 image encoding). */
|
||
async function downloadAttachmentRaw(pathId) {
|
||
try {
|
||
const res = await fetch(`${config.zulip.site}/api/v1/user_uploads/${pathId}`, {
|
||
headers: { Authorization: authHeader() },
|
||
signal: AbortSignal.timeout(30000),
|
||
});
|
||
if (!res.ok) return null;
|
||
const buf = Buffer.from(await res.arrayBuffer());
|
||
return { contentType: res.headers.get("content-type") || "application/octet-stream", buf, size: buf.length };
|
||
} catch (e) {
|
||
console.error(`[zulip-ext] Download failed: ${e.message}`);
|
||
return null;
|
||
}
|
||
}
|
||
|
||
/** Download and base64-encode (for pi image API). */
|
||
async function downloadAttachment(pathId) {
|
||
const raw = await downloadAttachmentRaw(pathId);
|
||
if (!raw) return null;
|
||
return { contentType: raw.contentType, data: raw.buf.toString("base64"), size: raw.size };
|
||
}
|
||
|
||
// ---------------------------------------------------------------------------
|
||
// Attachment classification & text extraction
|
||
// ---------------------------------------------------------------------------
|
||
|
||
const IMAGE_EXTS = new Set(["png", "jpg", "jpeg", "gif", "webp", "bmp", "svg"]);
|
||
const PDF_EXTS = new Set(["pdf"]);
|
||
const TEXT_EXTS = new Set([
|
||
"txt", "md", "py", "js", "ts", "jsx", "tsx", "json", "yaml", "yml",
|
||
"log", "csv", "sh", "bash", "env", "cfg", "conf", "ini", "toml",
|
||
"xml", "html", "htm", "css", "scss", "sql", "rs", "go", "java",
|
||
"c", "cpp", "cc", "h", "hpp", "rb", "php", "swift", "kt", "scala",
|
||
"r", "lua", "pl", "ps1", "bat", "makefile", "dockerfile", "gitignore",
|
||
"editorconfig", "diff", "patch", "tex", "rst", "org", "nix",
|
||
]);
|
||
|
||
const MAX_TEXT_ATTACHMENT_CHARS = 50_000;
|
||
|
||
function classifyAttachment(att) {
|
||
const name = att.name || "";
|
||
const ext = name.split(".").pop()?.toLowerCase() || "";
|
||
if (IMAGE_EXTS.has(ext)) return "image";
|
||
if (PDF_EXTS.has(ext)) return "pdf";
|
||
if (TEXT_EXTS.has(ext)) return "text";
|
||
// MIME-based fallback
|
||
return "binary";
|
||
}
|
||
|
||
function formatSize(bytes) {
|
||
if (bytes < 1024) return `${bytes} B`;
|
||
if (bytes < 1024 * 1024) return `${(bytes / 1024).toFixed(1)} KB`;
|
||
return `${(bytes / (1024 * 1024)).toFixed(1)} MB`;
|
||
}
|
||
|
||
function tryDecodeText(buf) {
|
||
try { return buf.toString("utf-8"); } catch { return null; }
|
||
}
|
||
|
||
/** Try to extract text from a PDF using pdftotext (poppler-utils). */
|
||
function extractPdfText(buf) {
|
||
try {
|
||
const tmpIn = `/tmp/zulip-pdf-${Date.now()}.pdf`;
|
||
fs.writeFileSync(tmpIn, buf);
|
||
const text = execSync(`pdftotext -layout -nopgbrk "${tmpIn}" -`, {
|
||
timeout: 15000,
|
||
encoding: "utf-8",
|
||
maxBuffer: 5 * 1024 * 1024,
|
||
});
|
||
fs.unlinkSync(tmpIn);
|
||
return text.trim() || null;
|
||
} catch (e) {
|
||
console.error(`[zulip-ext] PDF extraction failed: ${e.message}`);
|
||
return null;
|
||
}
|
||
}
|
||
|
||
// ---------------------------------------------------------------------------
|
||
// Message routing
|
||
// ---------------------------------------------------------------------------
|
||
|
||
/**
|
||
* Send a message to a worker and track the pending reply.
|
||
*/
|
||
async function routeToWorker(senderId, senderName, replyInfo, content, attachments) {
|
||
let worker = workers.get(senderId);
|
||
if (!worker) {
|
||
worker = spawnWorker(senderId);
|
||
workers.set(senderId, worker);
|
||
}
|
||
|
||
// Send placeholder to Zulip
|
||
const placeholders = [":robot: _Processing…_", ":hourglass_flowing_sand: _Thinking…_", ":brain: _Generating…_"];
|
||
const placeholder = placeholders[Date.now() % placeholders.length];
|
||
const target = replyInfo.type === "private"
|
||
? { type: "private", to: senderId }
|
||
: { type: "stream", to: replyInfo.streamId ?? replyInfo.streamName, subject: replyInfo.topic };
|
||
|
||
let zulipMsgId = null;
|
||
if (zulipQueue) {
|
||
try {
|
||
zulipMsgId = await zulipQueue.sendMessage({ ...target, content: placeholder });
|
||
} catch (e) {
|
||
console.warn(`[zulip-ext] Placeholder send failed: ${e.message}`);
|
||
}
|
||
}
|
||
|
||
worker.pendingReplies.push({
|
||
zulipMsgId,
|
||
senderId,
|
||
senderName,
|
||
replyInfo,
|
||
createdAt: Date.now(),
|
||
});
|
||
|
||
// Build prompt text
|
||
const label = replyInfo.type === "stream"
|
||
? `@all-bots in #${replyInfo.streamName}`
|
||
: `DM from @${senderName}`;
|
||
const promptText = `[Zulip ${label}]: ${content}`;
|
||
|
||
// Handle attachments — download, classify, extract content
|
||
let images = [];
|
||
let attachmentTexts = [];
|
||
if (attachments?.length) {
|
||
for (const att of attachments) {
|
||
const category = classifyAttachment(att);
|
||
const raw = await downloadAttachmentRaw(att.path_id);
|
||
if (!raw) {
|
||
attachmentTexts.push(`[Attachment: ${att.name} — download failed]`);
|
||
continue;
|
||
}
|
||
|
||
switch (category) {
|
||
case "image": {
|
||
const b64 = raw.buf.toString("base64");
|
||
images.push({ type: "image", data: b64, mimeType: raw.contentType });
|
||
break;
|
||
}
|
||
case "text": {
|
||
const text = tryDecodeText(raw.buf);
|
||
if (text) {
|
||
const truncated = text.length > MAX_TEXT_ATTACHMENT_CHARS
|
||
? text.slice(0, MAX_TEXT_ATTACHMENT_CHARS) + `\n\n[...truncated at ${MAX_TEXT_ATTACHMENT_CHARS.toLocaleString()} chars, ${(text.length - MAX_TEXT_ATTACHMENT_CHARS).toLocaleString()} chars removed]`
|
||
: text;
|
||
attachmentTexts.push(
|
||
`[Attachment: ${att.name} (${formatSize(raw.size)})]\n\`\`\`${truncated}\n\`\`\``
|
||
);
|
||
} else {
|
||
attachmentTexts.push(`[Attachment: ${att.name} (${formatSize(raw.size)}, ${raw.contentType}) — binary content, could not decode as text]`);
|
||
}
|
||
break;
|
||
}
|
||
case "pdf": {
|
||
const pdfText = extractPdfText(raw.buf);
|
||
if (pdfText) {
|
||
const truncated = pdfText.length > MAX_TEXT_ATTACHMENT_CHARS
|
||
? pdfText.slice(0, MAX_TEXT_ATTACHMENT_CHARS) + `\n\n[...truncated at ${MAX_TEXT_ATTACHMENT_CHARS.toLocaleString()} chars]`
|
||
: pdfText;
|
||
attachmentTexts.push(
|
||
`[Attachment: ${att.name} (PDF, ${formatSize(raw.size)})]\n\`\`\`${truncated}\n\`\`\``
|
||
);
|
||
} else {
|
||
attachmentTexts.push(`[Attachment: ${att.name} (PDF, ${formatSize(raw.size)}) — text extraction failed (pdftotext unavailable or unreadable PDF)]`);
|
||
}
|
||
break;
|
||
}
|
||
default: {
|
||
attachmentTexts.push(`[Attachment: ${att.name} (${formatSize(raw.size)}, ${raw.contentType}) — binary file, content not extracted]`);
|
||
break;
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
// Merge attachment text into prompt
|
||
const fullText = attachmentTexts.length > 0
|
||
? promptText + "\n\n---\n" + attachmentTexts.join("\n\n")
|
||
: promptText;
|
||
|
||
// Send to worker
|
||
if (worker.busy) {
|
||
sendRpc(worker, { type: "steer", message: fullText });
|
||
console.log(`[zulip-ext] Steered to worker ${senderId} for ${senderName}${attachmentTexts.length ? ' with ' + attachmentTexts.length + ' attachment(s)' : ''}`);
|
||
} else {
|
||
const cmd = { type: "prompt", message: fullText };
|
||
if (images.length > 0) cmd.images = images;
|
||
sendRpc(worker, cmd);
|
||
const attInfo = [];
|
||
if (images.length > 0) attInfo.push(`${images.length} image(s)`);
|
||
if (attachmentTexts.length > 0) attInfo.push(`${attachmentTexts.length} file(s)`);
|
||
console.log(`[zulip-ext] Routed to worker ${senderId} for ${senderName}${attInfo.length ? ' with ' + attInfo.join(' + ') : ''}`);
|
||
}
|
||
|
||
worker.busy = true;
|
||
worker.lastMessageContent = content;
|
||
worker.lastReplyInfo = replyInfo;
|
||
worker.lastActivity = Date.now();
|
||
|
||
// Send typing indicator
|
||
if (zulipQueue) {
|
||
zulipQueue.sendTyping([senderId], "start").catch(() => {});
|
||
}
|
||
}
|
||
|
||
// ---------------------------------------------------------------------------
|
||
// Zulip command handlers
|
||
// ---------------------------------------------------------------------------
|
||
const ZULIP_COMMANDS = ["status", "retry", "continue", "new", "help", "stop", "model", "compact", "yolo", "verbose", "approve", "deny"];
|
||
|
||
function parseCommand(content) {
|
||
const m = content.match(/^\/(\w+)(?:\s+(.+))?$/);
|
||
return m ? { cmd: m[1].toLowerCase(), args: m[2] || "" } : null;
|
||
}
|
||
|
||
/**
|
||
* Handle a Zulip command. Returns true if handled, false to pass through as normal message.
|
||
*/
|
||
async function handleZulipCommand(cmd, args, senderId, senderName, replyInfo) {
|
||
const target = replyInfo.type === "private"
|
||
? { type: "private", to: senderId }
|
||
: { type: "stream", to: replyInfo.streamId ?? replyInfo.streamName, subject: replyInfo.topic };
|
||
|
||
const send = (content) => {
|
||
if (zulipQueue) zulipQueue.sendMessage({ ...target, content }).catch(() => {});
|
||
};
|
||
|
||
switch (cmd) {
|
||
// --- Global commands (no worker needed) ---
|
||
case "help": {
|
||
const cmds = {
|
||
status: "Show system status — PM2 processes, containers, uptime",
|
||
retry: "Retry the last response",
|
||
continue: "Continue the last response if cut off",
|
||
new: "Start fresh — clears conversation context",
|
||
help: "Show this help message",
|
||
stop: "Stop the current response",
|
||
model: "Show or change LLM model",
|
||
compact: "Compact conversation context",
|
||
yolo: "Acknowledge YOLO mode (no confirmations)",
|
||
verbose: "Toggle verbose output",
|
||
approve: "Approve pending actions (Hermes compat)",
|
||
deny: "Deny pending actions (Hermes compat)",
|
||
};
|
||
if (args && cmds[args]) {
|
||
send(`**/${args}** — ${cmds[args]}\n\nUsage: \`/${args}\``);
|
||
} else {
|
||
let helpText = "**Abiba Zulip Commands**\n\n";
|
||
for (const [name, desc] of Object.entries(cmds)) {
|
||
helpText += `• \`/${name}\` — ${desc}\n`;
|
||
}
|
||
helpText += `\nSend any message to chat.`;
|
||
send(helpText);
|
||
}
|
||
return true;
|
||
}
|
||
|
||
case "status": {
|
||
let resp = "**Abiba System Status**\n\n**PM2 Processes:**\n";
|
||
try {
|
||
const pm2 = execSync("pm2 status --no-color 2>/dev/null", { timeout: 10000, encoding: "utf-8" });
|
||
for (const line of pm2.split("\n")) {
|
||
if (line.includes("abiba-")) {
|
||
const cells = line.split("│").filter(c => c.trim());
|
||
if (cells.length >= 6) {
|
||
const clean = (s) => s.replace(/\x1b\[[0-9;]*m/g, "");
|
||
resp += `• ${clean(cells[1])}: ${clean(cells[5])} (uptime: ${clean(cells[3])})\n`;
|
||
}
|
||
}
|
||
}
|
||
} catch (e) {
|
||
resp += `_Error: ${e.message}_\n`;
|
||
}
|
||
resp += `\n**Worker Sessions:** ${workers.size} active\n`;
|
||
for (const [sid, w] of workers) {
|
||
resp += `• sender_${sid}: ${w.busy ? "busy" : "idle"}, replies pending: ${w.pendingReplies.length}\n`;
|
||
}
|
||
resp += `\n**Zulip:** connected=${!!zulipQueue} | processed=${msgCount} | echo prevented bots=${BOT_EMAILS.size}`;
|
||
send(resp);
|
||
return true;
|
||
}
|
||
|
||
// --- Per-sender commands (need worker) ---
|
||
case "retry": {
|
||
const worker = workers.get(senderId);
|
||
if (!worker?.lastMessageContent) {
|
||
send(":warning: No previous message to retry.");
|
||
return true;
|
||
}
|
||
send(":repeat: Retrying last message…");
|
||
routeToWorker(senderId, senderName, worker.lastReplyInfo, worker.lastMessageContent);
|
||
return true;
|
||
}
|
||
|
||
case "continue": {
|
||
send(":arrow_forward: Continuing…");
|
||
routeToWorker(senderId, senderName, replyInfo, "Continue your response from where you left off.");
|
||
return true;
|
||
}
|
||
|
||
case "new": {
|
||
const worker = workers.get(senderId);
|
||
if (worker) {
|
||
worker.lastMessageContent = null;
|
||
sendRpc(worker, { type: "new_session" });
|
||
}
|
||
const topicMsg = args ? `Start fresh, focusing on: ${args}` : "Start fresh.";
|
||
send(`:broom: ${topicMsg}\n\n_Previous context cleared._`);
|
||
routeToWorker(senderId, senderName, replyInfo, `[System: Session reset. ${topicMsg} Treat this as a fresh conversation.]`);
|
||
return true;
|
||
}
|
||
|
||
case "stop": {
|
||
const worker = workers.get(senderId);
|
||
if (worker?.busy) {
|
||
sendRpc(worker, { type: "abort" });
|
||
send("⏹️ **Stop signal sent.** Current processing interrupted.");
|
||
} else {
|
||
send(":information_source: No active processing to stop.");
|
||
}
|
||
return true;
|
||
}
|
||
|
||
case "model": {
|
||
const worker = workers.get(senderId);
|
||
if (!worker) {
|
||
send("`/model` requires an active conversation. Send a message first.");
|
||
return true;
|
||
}
|
||
sendRpc(worker, { type: "get_available_models" });
|
||
// We'll need to async-read the response. For now, just acknowledge.
|
||
// In a follow-up, the response can be captured and sent to Zulip.
|
||
send("🔍 Checking available models…");
|
||
return true;
|
||
}
|
||
|
||
case "compact": {
|
||
const worker = workers.get(senderId);
|
||
if (worker) sendRpc(worker, { type: "compact" });
|
||
send("🗜️ **Compaction triggered.** Context will be summarized.");
|
||
return true;
|
||
}
|
||
|
||
case "yolo":
|
||
send("🤘 YOLO mode acknowledged. All tools execute immediately.");
|
||
return true;
|
||
|
||
case "verbose":
|
||
send("📝 For detailed output, include 'be thorough' or 'show your work' in your message.");
|
||
return true;
|
||
|
||
case "approve":
|
||
case "deny":
|
||
send("ℹ️ Pi doesn't have pending approvals — commands execute immediately.");
|
||
return true;
|
||
|
||
default:
|
||
send(`:question: Unknown command \`/${cmd}\`. Try \`/help\`.`);
|
||
return true;
|
||
}
|
||
}
|
||
|
||
// ---------------------------------------------------------------------------
|
||
// Polling loop
|
||
// ---------------------------------------------------------------------------
|
||
let zulipQueue = null;
|
||
let pollTimer = null;
|
||
let connected = false;
|
||
let retryCount = 0;
|
||
let retryDelay = 5000;
|
||
let msgCount = 0;
|
||
let skippedCount = 0;
|
||
let lastHeartbeatTime = 0;
|
||
let lastError = null;
|
||
|
||
async function startPolling() {
|
||
console.log(`[zulip-ext] Connecting to ${config.zulip.site} as ${config.zulip.email}…`);
|
||
try {
|
||
zulipQueue = await createZulipQueue();
|
||
connected = true;
|
||
retryCount = 0;
|
||
retryDelay = 5000;
|
||
lastError = null;
|
||
await loadBotEmails(zulipQueue);
|
||
console.log(`[zulip-ext] Connected, queue=${zulipQueue.queueId}`);
|
||
|
||
pollTimer = setInterval(async () => {
|
||
if (!zulipQueue) return;
|
||
try {
|
||
const events = await zulipQueue.poll();
|
||
if (events.length > 0) {
|
||
for (const ev of events) {
|
||
await processEvent(ev);
|
||
}
|
||
}
|
||
|
||
// Heartbeat
|
||
const now = Date.now();
|
||
if (now - lastHeartbeatTime > HEARTBEAT_INTERVAL_MS) {
|
||
lastHeartbeatTime = now;
|
||
const workerInfo = Array.from(workers.values()).map(w => `${w.senderId}:${w.busy ? "busy" : "idle"}:${w.pendingReplies.length}`).join(",");
|
||
console.log(`[zulip-ext] Heartbeat — msgs=${msgCount} skipped=${skippedCount} workers=[${workerInfo}] queue=${zulipQueue.queueId}`);
|
||
}
|
||
} catch (err) {
|
||
const msg = err instanceof Error ? err.message : String(err);
|
||
const isQueueExpired = msg.includes("BAD_EVENT_QUEUE_ID") || msg.includes("deregistered");
|
||
if (isQueueExpired) {
|
||
console.log(`[zulip-ext] Queue expired, reconnecting… (${msg.slice(0, 80)})`);
|
||
connected = false;
|
||
clearInterval(pollTimer);
|
||
pollTimer = null;
|
||
setTimeout(startPolling, retryDelay);
|
||
} else {
|
||
lastError = msg;
|
||
console.error(`[zulip-ext] Poll error: ${msg}`);
|
||
}
|
||
}
|
||
}, POLL_INTERVAL_MS);
|
||
} catch (err) {
|
||
const msg = err instanceof Error ? err.message : String(err);
|
||
lastError = msg;
|
||
console.error(`[zulip-ext] Connection failed: ${msg}`);
|
||
|
||
// Exponential backoff (capped at ~5 min)
|
||
retryDelay = Math.min(5000 * Math.pow(2, retryCount), 300000);
|
||
console.log(`[zulip-ext] Reconnecting in ${Math.round(retryDelay / 1000)}s (attempt ${retryCount + 1})…`);
|
||
retryCount++;
|
||
setTimeout(startPolling, retryDelay);
|
||
}
|
||
}
|
||
|
||
async function processEvent(event) {
|
||
if (event.type !== "message") return;
|
||
const msg = event.message;
|
||
if (!msg) return;
|
||
|
||
// Ignore own messages
|
||
if (msg.sender_email === config.zulip.email) return;
|
||
|
||
// Echo prevention
|
||
if (isBotSender(msg.sender_email)) {
|
||
skippedCount++;
|
||
if (skippedCount % 10 === 1) {
|
||
console.log(`[zulip-ext] Echo skip: ${msg.sender_email} (total=${skippedCount})`);
|
||
}
|
||
return;
|
||
}
|
||
|
||
const senderName = msg.sender_full_name || msg.sender_email;
|
||
const senderId = msg.sender_id; // keep as number for correct API format
|
||
|
||
// Private messages (DM-first, ADR-001)
|
||
if (msg.type === "private") {
|
||
msgCount++;
|
||
console.log(`[zulip-ext] DM from ${senderName} (id=${senderId}): ${(msg.content || "").slice(0, 80)}`);
|
||
|
||
// Check for commands
|
||
const parsed = parseCommand(msg.content);
|
||
if (parsed && ZULIP_COMMANDS.includes(parsed.cmd)) {
|
||
await handleZulipCommand(parsed.cmd, parsed.args, senderId, senderName, {
|
||
type: "private", senderId, senderName,
|
||
});
|
||
return;
|
||
}
|
||
|
||
routeToWorker(senderId, senderName, { type: "private", senderId, senderName }, msg.content, msg.attachments);
|
||
return;
|
||
}
|
||
|
||
// Stream messages: @mention or @all-bots (ADR-005, ADR-006)
|
||
const mentionedUserIds = (msg.mentioned_users || []).map(u => u.user_id);
|
||
const isDirectMention = zulipQueue?.botUserId && mentionedUserIds.includes(zulipQueue.botUserId);
|
||
const isAllBots = isAllBotsMention(msg.content || "");
|
||
|
||
if (msg.type === "stream" && (isDirectMention || isAllBots)) {
|
||
msgCount++;
|
||
const streamName = typeof msg.display_recipient === "string" ? msg.display_recipient : "unknown";
|
||
const topic = msg.subject || "general";
|
||
console.log(`[zulip-ext] ${isDirectMention ? "@mention" : "@all-bots"} in #${streamName} > ${topic}`);
|
||
|
||
const replyInfo = { type: "stream", streamName, topic, streamId: msg.stream_id, senderId, senderName };
|
||
|
||
const parsed = parseCommand(msg.content);
|
||
if (parsed && ZULIP_COMMANDS.includes(parsed.cmd)) {
|
||
await handleZulipCommand(parsed.cmd, parsed.args, senderId, senderName, replyInfo);
|
||
return;
|
||
}
|
||
|
||
routeToWorker(senderId, senderName, replyInfo, msg.content, msg.attachments);
|
||
}
|
||
}
|
||
|
||
// ---------------------------------------------------------------------------
|
||
// Health endpoint
|
||
// ---------------------------------------------------------------------------
|
||
let healthServer = null;
|
||
|
||
function startHealthServer(port) {
|
||
const server = http.createServer((req, res) => {
|
||
if (req.url === "/health" || req.url === "/") {
|
||
const workerList = [];
|
||
for (const [sid, w] of workers) {
|
||
workerList.push({
|
||
sender_id: sid,
|
||
busy: w.busy,
|
||
pending_replies: w.pendingReplies.length,
|
||
last_activity: Math.floor((Date.now() - w.lastActivity) / 1000),
|
||
});
|
||
}
|
||
res.writeHead(200, { "Content-Type": "application/json" });
|
||
res.end(JSON.stringify({
|
||
status: connected && zulipQueue ? "ok" : "down",
|
||
platform: "pi",
|
||
agent: config.agent.name,
|
||
zulip: {
|
||
connected,
|
||
site: config.zulip.site,
|
||
email: config.zulip.email,
|
||
queue_id: zulipQueue?.queueId ?? null,
|
||
bot_user_id: zulipQueue?.botUserId ?? null,
|
||
messages_processed: msgCount,
|
||
skipped: skippedCount,
|
||
last_error: lastError,
|
||
},
|
||
workers: workerList,
|
||
worker_count: workers.size,
|
||
}));
|
||
} else {
|
||
res.writeHead(404);
|
||
res.end("Not found");
|
||
}
|
||
});
|
||
server.on("error", (err) => {
|
||
if (err.code === "EADDRINUSE") {
|
||
console.warn(`[zulip-ext] Port :${port} in use — skipping health`);
|
||
}
|
||
});
|
||
server.listen(port, "127.0.0.1", () => {
|
||
console.log(`[zulip-ext] Health endpoint on :${port}`);
|
||
});
|
||
return server;
|
||
}
|
||
|
||
// ---------------------------------------------------------------------------
|
||
// Extension lifecycle
|
||
// ---------------------------------------------------------------------------
|
||
export default function (pi) {
|
||
// Guard: only activate in the designated PM2 router process
|
||
if (process.env.ZULIP_ROLE !== "router") {
|
||
// Worker processes and TUI sessions: NO-OP
|
||
return;
|
||
}
|
||
|
||
pi.on("session_start", async () => {
|
||
console.log(`[zulip-ext] Router starting for ${config.agent.name}`);
|
||
|
||
// Load persisted worker session map
|
||
if (!fs.existsSync(SESSION_DIR)) fs.mkdirSync(SESSION_DIR, { recursive: true });
|
||
workerSessions = loadWorkerSessions();
|
||
|
||
// Start health server
|
||
if (!healthServer) {
|
||
healthServer = startHealthServer(config.health_port);
|
||
}
|
||
|
||
// Start Zulip poller
|
||
startPolling();
|
||
|
||
// Start idle worker cleanup (every 5 min)
|
||
idleCleanupTimer = setInterval(cleanupIdleWorkers, 5 * 60 * 1000);
|
||
|
||
console.log(`[zulip-ext] Router ready. Known sessions: ${Object.keys(workerSessions).length}`);
|
||
});
|
||
|
||
pi.on("session_shutdown", async () => {
|
||
console.log(`[zulip-ext] Router shutting down…`);
|
||
if (pollTimer) {
|
||
clearInterval(pollTimer);
|
||
pollTimer = null;
|
||
}
|
||
if (idleCleanupTimer) {
|
||
clearInterval(idleCleanupTimer);
|
||
idleCleanupTimer = null;
|
||
}
|
||
|
||
// Kill all workers gracefully
|
||
for (const [sid, w] of workers) {
|
||
try { w.process.kill("SIGTERM"); } catch {}
|
||
}
|
||
workers.clear();
|
||
|
||
// Delete Zulip event queue
|
||
if (zulipQueue?.queueId) {
|
||
try {
|
||
await fetch(`${config.zulip.site}/api/v1/queues/${zulipQueue.queueId}`, {
|
||
method: "DELETE",
|
||
headers: { Authorization: authHeader() },
|
||
});
|
||
console.log(`[zulip-ext] Queue ${zulipQueue.queueId} deregistered`);
|
||
} catch (e) {
|
||
console.warn(`[zulip-ext] Queue deregister failed: ${e.message}`);
|
||
}
|
||
}
|
||
|
||
if (healthServer) {
|
||
healthServer.close();
|
||
healthServer = null;
|
||
}
|
||
connected = false;
|
||
zulipQueue = null;
|
||
|
||
saveWorkerSessions();
|
||
});
|
||
|
||
// Register diagnostics commands
|
||
pi.registerCommand("zulip-status", {
|
||
description: "Show Zulip extension status and worker pool",
|
||
handler: async (_args, ctx) => {
|
||
const lines = [
|
||
"=== Zulip Extension v2 (router-worker) ===",
|
||
`Agent: ${config.agent.name}`,
|
||
`Server: ${config.zulip.site}`,
|
||
`Connected: ${connected ? "✅" : "❌"} Queue: ${zulipQueue?.queueId ?? "N/A"}`,
|
||
`Messages: ${msgCount} processed, ${skippedCount} echo-skipped`,
|
||
`Workers: ${workers.size} active`,
|
||
`Retries: ${retryCount} Last error: ${lastError ?? "none"}`,
|
||
"",
|
||
];
|
||
for (const [sid, w] of workers) {
|
||
lines.push(`• sender_${sid}: ${w.busy ? "busy" : "idle"}, ${w.pendingReplies.length} pending, session=${w.sessionFile ?? "new"}`);
|
||
}
|
||
ctx.ui.notify(lines.join("\n"), "info");
|
||
},
|
||
});
|
||
|
||
pi.registerCommand("zulip-send-test", {
|
||
description: "Send test DM: /zulip-send-test <email> <message>",
|
||
handler: async (args, ctx) => {
|
||
const parts = args.trim().match(/^([^\s]+)\s+(.+)$/);
|
||
const to = parts?.[1] ?? config.agent.owner_email;
|
||
const msg = parts?.[2] ?? "Test from pi Zulip extension v2";
|
||
if (!zulipQueue) {
|
||
ctx.ui.notify("Not connected. Check /zulip-status.", "error");
|
||
return;
|
||
}
|
||
try {
|
||
await zulipQueue.sendMessage({ type: "private", to, content: msg });
|
||
ctx.ui.notify(`Test sent to ${to}`, "info");
|
||
} catch (err) {
|
||
ctx.ui.notify(`Failed: ${err.message}`, "error");
|
||
}
|
||
},
|
||
});
|
||
|
||
pi.registerCommand("zulip-test", {
|
||
description: "Run comprehensive Zulip self-test",
|
||
handler: async (_args, ctx) => {
|
||
const results = [];
|
||
const passed = (n, d) => results.push({ name: n, ok: true, detail: d });
|
||
const failed = (n, d) => results.push({ name: n, ok: false, detail: d });
|
||
|
||
passed("queue_connected", `Queue: ${zulipQueue?.queueId ?? "N/A"}`);
|
||
zulipQueue?.botUserId ? passed("bot_identity", `Bot ID: ${zulipQueue.botUserId}`) : failed("bot_identity", "Bot ID not resolved");
|
||
zulipQueue && zulipQueue.allBotsUserId > 1 ? passed("all_bots", `ID: ${zulipQueue.allBotsUserId}`) : failed("all_bots", `ID: ${zulipQueue?.allBotsUserId ?? 1}`);
|
||
BOT_EMAILS.size > 1 ? passed("echo_prevention", `${BOT_EMAILS.size} bot emails`) : failed("echo_prevention", "Only self excluded");
|
||
workers.size >= 0 ? passed("worker_pool", `${workers.size} active workers`) : failed("worker_pool", "Pool broken");
|
||
|
||
// Loopback send
|
||
if (zulipQueue) {
|
||
try {
|
||
const id = await zulipQueue.sendMessage({ type: "private", to: config.agent.owner_email, content: `🔬 Zulip self-test at ${new Date().toISOString()}` });
|
||
passed("api_send", `Loopback DM sent (id=${id})`);
|
||
} catch (e) {
|
||
failed("api_send", e.message.slice(0, 80));
|
||
}
|
||
}
|
||
|
||
const ok = results.filter(r => r.ok).length;
|
||
const total = results.length;
|
||
const verdict = ok === total ? "HEALTHY" : ok >= total - 1 ? "DEGRADED" : "CRITICAL";
|
||
let report = `=== Zulip Self-Test: ${verdict} ===\n${ok}/${total} checks passed\n\n`;
|
||
for (const r of results) {
|
||
report += `${r.ok ? "✅" : "❌"} ${r.name}: ${r.detail}\n`;
|
||
}
|
||
ctx.ui.notify(report, verdict === "HEALTHY" ? "info" : "error");
|
||
},
|
||
});
|
||
}
|