fix: replace subprocess spawn with direct session injection for Zulip extension #22

Merged
jerome merged 1 commits from fix/zulip-session-injection into main 2026-06-24 20:25:56 +00:00
4 changed files with 2959 additions and 82 deletions
Showing only changes of commit ab3cb50eb4 - Show all commits
File diff suppressed because it is too large Load Diff
+10 -4
View File
@@ -9,14 +9,20 @@
"check": "tsc --noEmit"
},
"dependencies": {
"zulip-js": "^2.0.0",
"yaml": "^2.0.0"
"yaml": "^2.9.0",
"zulip-js": "^2.0.0"
},
"devDependencies": {
"@earendil-works/pi-coding-agent": "*",
"@earendil-works/pi-coding-agent": "^0.80.2",
"typescript": "^5.0.0"
},
"keywords": ["pi", "zulip", "agent", "abiba", "sysloggh"],
"keywords": [
"pi",
"zulip",
"agent",
"abiba",
"sysloggh"
],
"license": "UNLICENSED",
"private": true
}
+578 -78
View File
@@ -1,120 +1,620 @@
/**
* pi-zulip-extension — Zulip agent communication plugin for pi agents
*
* Deploy to: ~/.pi/agent/extensions/zulip.ts
* Config: config.yaml (alongside extension, or symlinked)
* Deploy to: ~/.pi/agent/extensions/zulip/
* Config: config.yaml (alongside extension, or env vars)
*
* Connects a pi agent (Abiba) to the Sysloggh Zulip agent mesh.
* Listens for @mentions of abiba-bot in #agent-hub and forwards
* to the pi agent. Responses are posted back to the same Zulip topic.
* DM-first architecture per ADR-001/ADR-002.
* Background Zulip event queue poller that injects messages directly into
* the current pi session via pi.sendUserMessage(), then captures the LLM
* response from agent_end and sends it back to Zulip. No subprocess overhead.
*
* @see ADR-007: Platform-native plugin contracts
* @see docs/ARCHITECTURE.md
*/
import type { ExtensionAPI } from "@earendil-works/pi-coding-agent";
import { readFileSync } from "fs";
import zulip from "zulip-js";
import http from "node:http";
import fs from "node:fs";
import path from "node:path";
import url from "node:url";
import { parse } from "yaml";
// zulip-js types are declared in types.d.ts in this directory
// ---------------------------------------------------------------------------
// Config
// Configuration
// ---------------------------------------------------------------------------
interface ZulipConfig {
agent: {
name: string;
display_name: string;
zulip_bot_name: string;
owner_email: string;
private_topic: string;
};
zulip: {
server_url: string;
email: string;
api_key: string;
stream: string;
all_bots_user_id: number;
site: string;
stream?: string;
all_bots_user_id?: number;
};
error_handling: {
timeout_seconds: number;
retry_count: number;
retry_delay_seconds: number;
graceful_message: string;
};
monitoring: {
health_endpoint_enabled: boolean;
health_port: number;
log_level: string;
agent: {
name: string;
owner_email: string;
display_name?: string;
zulip_bot_name?: string;
private_topic?: string;
};
health_port?: number;
poll_interval_ms?: number;
max_retries?: number;
retry_delay_ms?: number;
pi_command?: string;
}
function loadConfig(): ZulipConfig {
const raw = readFileSync("config.yaml", "utf-8");
const cfg = parse(raw) as ZulipConfig;
cfg.zulip.api_key = process.env["ZULIP_API_KEY"] ?? cfg.zulip.api_key;
return cfg;
// Check env vars first
const email = process.env.ZULIP_EMAIL;
const apiKey = process.env.ZULIP_API_KEY;
const site = process.env.ZULIP_SITE;
if (email && apiKey && site) {
return {
zulip: { email, api_key: apiKey, site },
agent: {
name: process.env.AGENT_NAME ?? "abiba",
owner_email: process.env.AGENT_OWNER_EMAIL ?? "jerome@sysloggh.com",
},
health_port: parseInt(process.env.HEALTH_PORT ?? "9200", 10),
poll_interval_ms: 3000,
max_retries: 3,
retry_delay_ms: 5000,
pi_command: "pi",
};
}
// Fall back to config.yaml alongside the extension
const configDir = path.dirname(url.fileURLToPath(import.meta.url));
const configPath = path.join(configDir, "config.yaml");
if (!fs.existsSync(configPath)) {
throw new Error(
`No config.yaml found at ${configPath}. Set ZULIP_EMAIL, ZULIP_API_KEY, ZULIP_SITE env vars or provide config.yaml.`,
);
}
const raw = fs.readFileSync(configPath, "utf-8");
const cfg = parse(raw) as Record<string, unknown>;
// Support both flat (env-style) and nested config formats
function getField(key: string): unknown {
if (cfg[key] !== undefined) return cfg[key];
// Try nested lookup: "zulip.email" -> cfg.zulip?.email
const parts = key.split(".");
let cur: Record<string, unknown> = cfg;
for (const p of parts) {
if (cur && typeof cur === "object" && p in cur) {
cur = cur[p] as Record<string, unknown>;
} else {
return undefined;
}
}
return cur;
}
const parsed: ZulipConfig = {
zulip: {
email: String(getField("zulip.email") ?? getField("zulip.email") ?? ""),
api_key: String(getField("zulip.api_key") ?? getField("zulip.api_key") ?? ""),
site: String(getField("zulip.site") ?? getField("zulip.server_url") ?? ""),
stream: String(getField("zulip.stream") ?? ""),
all_bots_user_id: Number(getField("zulip.all_bots_user_id") ?? 1),
},
agent: {
name: String(getField("agent.name") ?? "abiba"),
owner_email: String(getField("agent.owner_email") ?? "jerome@sysloggh.com"),
display_name: String(getField("agent.display_name") ?? ""),
zulip_bot_name: String(getField("agent.zulip_bot_name") ?? ""),
private_topic: String(getField("agent.private_topic") ?? ""),
},
health_port: parseInt(String(getField("health_port") ?? getField("monitoring.health_port") ?? "9200"), 10),
poll_interval_ms: 3000,
max_retries: 3,
retry_delay_ms: 5000,
pi_command: "pi",
};
// Override api_key from env if set
parsed.zulip.api_key = process.env.ZULIP_API_KEY ?? parsed.zulip.api_key;
return parsed;
}
// ---------------------------------------------------------------------------
// Extension entry point
// Zulip API helpers
// ---------------------------------------------------------------------------
export default function (pi: ExtensionAPI) {
const config = loadConfig();
interface ZulipMessage {
id: number;
type: "stream" | "private";
display_recipient?: string | Array<{ id: number; email: string; full_name: string }>;
sender_id: number;
sender_email: string;
sender_full_name: string;
content: string;
subject: string;
stream_id?: number;
}
// Health endpoint
if (config.monitoring.health_endpoint_enabled) {
startHealthServer(config);
}
interface ZulipEvent {
id: number;
type: "message";
message: ZulipMessage;
}
pi.on("session_start", async (_event, ctx) => {
ctx.ui.notify(
`Zulip extension loaded for ${config.agent.display_name} (${config.agent.zulip_bot_name})`,
"info"
);
/**
* Register a Zulip event queue and poll for events.
*/
async function createZulipQueue(config: ZulipConfig) {
const client = await zulip({
username: config.zulip.email,
apiKey: config.zulip.api_key,
realm: config.zulip.site,
});
// Diagnostic tool
pi.registerTool({
name: "zulip_status",
label: "Zulip Status",
description: "Check Zulip connection status and bot identity",
parameters: {},
execute: async () => ({
connected: false, // TODO: track connection state
bot: config.agent.zulip_bot_name,
stream: config.zulip.stream,
server: config.zulip.server_url,
owner: config.agent.owner_email,
private_topic: config.agent.private_topic,
}),
// Register queue
const queueRes = await client.queues.register({
event_types: ["message"],
});
const queueId = queueRes.queue_id;
let lastEventId: number = queueRes.last_event_id ?? -1;
return {
client,
queueId,
lastEventId,
/**
* Poll the event queue and return any new events.
*/
async poll(): Promise<ZulipEvent[]> {
const res = await fetch(
`${config.zulip.site}/api/v1/events?queue_id=${encodeURIComponent(queueId)}&last_event_id=${lastEventId}`,
{
headers: {
Authorization:
"Basic " +
Buffer.from(`${config.zulip.email}:${config.zulip.api_key}`).toString("base64"),
},
},
);
if (!res.ok) {
throw new Error(`Events API returned ${res.status}: ${await res.text()}`);
}
const data = (await res.json()) as { events?: ZulipEvent[] };
if (data.events && Array.isArray(data.events)) {
for (const event of data.events) {
if (event.id > lastEventId) {
lastEventId = event.id;
}
}
return data.events.filter((e) => e.type === "message");
}
return [];
},
/**
* Send a message to Zulip via the API.
*/
async sendMessage(params: {
type: "private" | "stream";
to: number | string;
subject?: string;
content: string;
}): Promise<void> {
const formBody = new URLSearchParams();
formBody.append("type", params.type);
// Zulip API: private messages need JSON array of user IDs, stream messages accept string/number
formBody.append("to", params.type === "private" ? JSON.stringify([params.to]) : String(params.to));
if (params.subject) formBody.append("subject", params.subject);
formBody.append("content", params.content);
const res = await fetch(`${config.zulip.site}/api/v1/messages`, {
method: "POST",
headers: {
Authorization:
"Basic " +
Buffer.from(`${config.zulip.email}:${config.zulip.api_key}`).toString("base64"),
"Content-Type": "application/x-www-form-urlencoded",
},
body: formBody.toString(),
});
if (!res.ok) {
throw new Error(`Send message API returned ${res.status}: ${await res.text()}`);
}
},
/**
* Send typing indicator.
*/
async sendTypingNotification(userIds: number[], operation: "start" | "stop"): Promise<void> {
const formBody = new URLSearchParams();
formBody.append("to", JSON.stringify(userIds));
formBody.append("op", operation);
await fetch(`${config.zulip.site}/api/v1/typing`, {
method: "POST",
headers: {
Authorization:
"Basic " +
Buffer.from(`${config.zulip.email}:${config.zulip.api_key}`).toString("base64"),
"Content-Type": "application/x-www-form-urlencoded",
},
body: formBody.toString(),
});
},
};
}
// ---------------------------------------------------------------------------
// Health endpoint
// ---------------------------------------------------------------------------
function startHealthServer(config: ZulipConfig) {
const http = require("http") as typeof import("http");
const port = config.monitoring.health_port;
const startTime = Date.now();
http
.createServer((_req: any, res: any) => {
function startHealthServer(port: number, getState: () => Record<string, unknown>) {
const server = http.createServer((req, res) => {
if (req.url === "/health" || req.url === "/") {
const state = getState();
res.writeHead(200, { "Content-Type": "application/json" });
res.end(
JSON.stringify({
status: "ok",
agent: config.agent.name,
uptime_seconds: Math.floor((Date.now() - startTime) / 1000),
zulip_connected: false,
last_message_time: null,
timestamp: new Date().toISOString(),
})
);
})
.listen(port, "127.0.0.1", () => {
console.log(`[zulip-extension] Health endpoint on :${port}`);
});
res.end(JSON.stringify({ status: "ok", ...state }));
} else {
res.writeHead(404);
res.end("Not found");
}
});
server.on("error", (err: NodeJS.ErrnoException) => {
if (err.code === "EADDRINUSE") {
console.warn(`[zulip-ext] Port :${port} already in use — skipping health endpoint`);
} else {
console.error(`[zulip-ext] Health server error:`, err.message);
}
});
server.listen(port, "127.0.0.1", () => {
console.log(`[zulip-ext] Health endpoint on :${port}`);
});
return server;
}
// ---------------------------------------------------------------------------
// Detect @all-bots content
// ---------------------------------------------------------------------------
const ALL_BOTS_RE = /@\*\*(?:all-bots|All Bots)\*\*/i;
function isAllBotsMention(content: string): boolean {
return ALL_BOTS_RE.test(content);
}
// ---------------------------------------------------------------------------
// Main extension
// ---------------------------------------------------------------------------
export default function (pi: ExtensionAPI) {
const config = loadConfig();
let queue: Awaited<ReturnType<typeof createZulipQueue>> | null = null;
let pollTimer: ReturnType<typeof setInterval> | null = null;
let retryCount = 0;
let connected = false;
let lastError: string | null = null;
let processedCount = 0;
let healthServer: ReturnType<typeof startHealthServer> | null = null;
// Track state for health endpoint
function getState() {
return {
connected,
email: config.zulip.email,
site: config.zulip.site,
agent: config.agent.name,
is_owner: config.agent.owner_email,
queue_id: queue?.queueId ?? null,
last_event_id: queue?.lastEventId ?? null,
messages_processed: processedCount,
last_error: lastError,
retry_count: retryCount,
};
}
// Pending Zulip replies awaiting the LLM response from agent_end.
const pendingZulipReplies: Array<{
id: number;
type: "private" | "stream";
senderId?: number;
senderName?: string;
streamName?: string;
topic?: string;
streamId?: number;
}> = [];
let zulipReplyIdCounter = 0;
/**
* Queue a Zulip message into the current pi session via sendUserMessage
* and schedule the reply to be sent back on the next agent_end event.
*/
function injectIntoSession(
msg: ZulipMessage,
senderName: string,
senderId: number,
replyInfo: {
type: "private" | "stream";
senderId?: number;
senderName?: string;
streamName?: string;
topic?: string;
streamId?: number;
},
): void {
// Send typing indicator (fire-and-forget)
queue!.sendTypingNotification([senderId], "start").catch(() => {});
// Track the pending reply so agent_end knows where to send the response
const replyId = ++zulipReplyIdCounter;
pendingZulipReplies.push({ id: replyId, ...replyInfo });
// Inject directly into this pi session — no subprocess overhead
const label = replyInfo.type === "stream"
? `@all-bots in #${replyInfo.streamName!}`
: `DM from @${senderName}`;
pi.sendUserMessage(`[Zulip ${label}]: ${msg.content}`);
console.log(`[zulip-ext] Injected into session [${replyId}]: from ${senderName}`);
}
/**
* Process a single Zulip event — injects DMs and @all-bots into the
* current pi session instead of spawning a subprocess.
*/
async function processEvent(event: ZulipEvent): Promise<void> {
const msg = event.message;
// Ignore non-message events
if (event.type !== "message") return;
// Ignore own messages (sent by this bot)
if (msg.sender_email === config.zulip.email) return;
// DM-first architecture (ADR-001, ADR-002): process private messages
if (msg.type === "private") {
processedCount++;
const senderName = msg.sender_full_name || msg.sender_email;
const senderId = msg.sender_id;
console.log(`[zulip-ext] DM from ${senderName} (id=${senderId}): ${msg.content.slice(0, 80)}...`);
injectIntoSession(msg, senderName, senderId, { type: "private", senderId, senderName });
return;
}
// Stream messages: only respond to @all-bots (ADR-006)
if (msg.type === "stream" && isAllBotsMention(msg.content)) {
processedCount++;
const streamName = typeof msg.display_recipient === "string"
? msg.display_recipient
: "unknown";
const topic = msg.subject || "general";
console.log(`[zulip-ext] @all-bots in #${streamName} > ${topic}`);
injectIntoSession(msg, msg.sender_full_name || msg.sender_email, msg.sender_id, {
type: "stream",
streamName,
topic,
streamId: msg.stream_id,
});
}
}
/**
* Polling loop.
*/
async function startPolling(): Promise<void> {
console.log(`[zulip-ext] Connecting to ${config.zulip.site} as ${config.zulip.email}...`);
try {
queue = await createZulipQueue(config);
connected = true;
retryCount = 0;
lastError = null;
console.log(`[zulip-ext] Connected, queue: ${queue.queueId} (last_event_id=${queue.lastEventId})`);
// Start polling
pollTimer = setInterval(async () => {
if (!queue) return;
try {
const events = await queue.poll();
for (const event of events) {
await processEvent(event);
}
} catch (err) {
const errMsg = err instanceof Error ? err.message : String(err);
// Check if queue was deregistered (need to reconnect)
if (errMsg.includes("BAD_EVENT_QUEUE_ID") || errMsg.includes("queue_id")) {
console.log(`[zulip-ext] Queue expired, reconnecting...`);
connected = false;
if (pollTimer) clearInterval(pollTimer);
pollTimer = null;
setTimeout(() => startPolling(), config.retry_delay_ms ?? 5000);
return;
}
// Network errors — retry on next poll interval
lastError = errMsg;
console.error(`[zulip-ext] Poll error: ${errMsg}`);
}
}, config.poll_interval_ms ?? 3000);
} catch (err) {
const errMsg = err instanceof Error ? err.message : String(err);
lastError = errMsg;
console.error(`[zulip-ext] Connection failed: ${errMsg}`);
if (retryCount < (config.max_retries ?? 3)) {
retryCount++;
const delay = (config.retry_delay_ms ?? 5000) * retryCount;
console.log(`[zulip-ext] Retrying in ${delay / 1000}s (attempt ${retryCount})...`);
setTimeout(() => startPolling(), delay);
} else {
console.error(`[zulip-ext] Max retries (${config.max_retries}) reached. Giving up.`);
lastError = `Connection failed after ${config.max_retries} retries: ${errMsg}`;
}
}
}
// -------------------------------------------------------------------------
// Extension lifecycle
// -------------------------------------------------------------------------
pi.on("session_start", async (_event) => {
console.log(`[zulip-ext] Starting Zulip extension for ${config.agent.name} (${config.zulip.email})`);
// Start health endpoint
if (!healthServer) {
healthServer = startHealthServer(config.health_port ?? 9200, getState);
}
// Begin Zulip event polling
startPolling();
});
/**
* agent_end captures the LLM response from this session and relays it
* back to the Zulip user/stream that sent the original message.
*/
pi.on("agent_end", async (event) => {
if (pendingZulipReplies.length === 0) return;
const reply = pendingZulipReplies.shift()!;
// Extract assistant response text from the completed turn
const allMsgs = event.messages as unknown as Array<Record<string, unknown>>;
const assistantMsgs = allMsgs.filter((m) => m.role === "assistant");
const lastAssistant = assistantMsgs[assistantMsgs.length - 1];
let responseText = "";
if (lastAssistant) {
const content = lastAssistant.content;
if (typeof content === "string") {
responseText = content;
} else if (Array.isArray(content)) {
responseText = (content as Array<{ type: string; text: string }>)
.filter((c) => c.type === "text")
.map((c) => c.text)
.join("\n");
}
}
// Stop typing indicator
if (reply.senderId) {
queue!.sendTypingNotification([reply.senderId], "stop").catch(() => {});
}
if (responseText.trim()) {
const truncated = responseText.length > 1500
? responseText.slice(0, 1500) + "\n\n[...truncated, see session for full output]"
: responseText;
const target = reply.type === "private"
? { type: "private" as const, to: reply.senderId! }
: { type: "stream" as const, to: reply.streamId ?? reply.streamName!, subject: reply.topic! };
try {
await queue!.sendMessage({ ...target, content: truncated });
console.log(`[zulip-ext] Replied to [${reply.id}] ${reply.senderName} (${truncated.length} chars)`);
} catch (err) {
const errMsg = err instanceof Error ? err.message : String(err);
console.error(`[zulip-ext] Failed to send reply [${reply.id}]: ${errMsg}`);
lastError = errMsg;
}
} else {
console.log(`[zulip-ext] No assistant response for [${reply.id}] from ${reply.senderName}`);
}
});
pi.on("session_shutdown", async () => {
console.log(`[zulip-ext] Shutting down...`);
if (pollTimer) {
clearInterval(pollTimer);
pollTimer = null;
}
if (healthServer) {
healthServer.close();
healthServer = null;
}
connected = false;
queue = null;
});
// Register /zulip-status command for diagnostics
pi.registerCommand("zulip-status", {
description: "Show Zulip extension status and stats",
handler: async (_args: unknown, ctx) => {
const state = getState();
const lines: string[] = [
"=== Zulip Extension Status ===",
"",
`Agent: ${state.agent}`,
`Email: ${state.email}`,
`Server: ${state.site}`,
`Owner: ${state.is_owner}`,
`Connected: ${state.connected ? "✅ Yes" : "❌ No"}`,
`Queue ID: ${state.queue_id ?? "N/A"}`,
`Last Event ID: ${state.last_event_id ?? "N/A"}`,
`Messages: ${state.messages_processed} processed`,
`Retries: ${state.retry_count}`,
`Health Port: :${config.health_port ?? 9200}`,
];
if (state.last_error) {
lines.push("", `Last Error: ${state.last_error}`);
}
lines.push("", "ADRs followed: DM-first (ADR-001), DM-routing (ADR-002), @all-bots content (ADR-006)");
ctx.ui.notify(lines.join("\n"), "info");
},
});
// Register zulip-send-test command
pi.registerCommand("zulip-send-test", {
description: "Send a test message to Zulip user: /zulip-send-test <email> <message>",
handler: async (args: string, ctx) => {
const parts = args.trim().match(/^([^\s]+)\s+(.+)$/);
const targetEmail = parts?.[1] ?? config.agent.owner_email;
const message = parts?.[2] ?? "Test message from pi Zulip extension";
if (!queue) {
ctx.ui.notify("Zulip queue not connected. Use /zulip-status to check.", "error");
return;
}
try {
await queue.sendMessage({
type: "private",
to: targetEmail,
content: message,
});
ctx.ui.notify(`Test message sent to ${targetEmail}`, "info");
} catch (err) {
const errMsg = err instanceof Error ? err.message : String(err);
ctx.ui.notify(`Failed to send: ${errMsg}`, "error");
}
},
});
}
+34
View File
@@ -0,0 +1,34 @@
// Type declarations for zulip-js (no @types available)
declare module "zulip-js" {
interface ZulipClient {
queues: {
register(params: {
event_types: string[];
narrow?: Array<Array<string | number>>;
}): Promise<{ queue_id: string; last_event_id: number }>;
};
users: {
me: {
get(): Promise<any>;
};
};
messages: {
store: {
send(rawContent: string): Promise<any>;
};
};
events: {
retrieve(params: {
queue_id: string;
last_event_id: number;
dont_block?: boolean;
}): Promise<{ events: any[]; result?: string; msg?: string }>;
};
}
export default function zulip(config: {
username: string;
apiKey: string;
realm: string;
}): Promise<ZulipClient>;
}