Compare commits
2
Commits
154454ac6d
...
0079d11d94
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
0079d11d94 | ||
|
|
ab3cb50eb4 |
Generated
+2337
File diff suppressed because it is too large
Load Diff
@@ -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
@@ -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");
|
||||
}
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
Vendored
+34
@@ -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>;
|
||||
}
|
||||
Reference in New Issue
Block a user