rootandAbiba (pi)
e946818375
CI / validate (push) Failing after 1s
Complete rewrite of the pi Zulip extension as a standalone Node.js service with direct harness API calls. No longer depends on pi's session — survives pi shutdown and restart. Changes: - src/index.ts: Complete rewrite — standalone process, not a pi extension - Direct harness API calls (POST /v1/chat/completions) with conversation memory - Per-sender conversation history (up to 50 messages) - Placeholder->edit streaming for Zulip DMs - Typing indicators via Zulip API - Health endpoint on :9200 - Exponential backoff reconnection - Graceful shutdown on SIGINT/SIGTERM - abiba-zulip.service: systemd service unit file - README.md: Updated deployment instructions - package.json/tsconfig.json: Updated for standalone app Deploy: npm install -> npx tsc -> systemctl enable abiba-zulip Closes issue #24 (Hermes parity for pi)
620 lines
20 KiB
TypeScript
620 lines
20 KiB
TypeScript
/**
|
|
* abiba-zulip-service — Standalone Zulip gateway for Abiba (pi agent)
|
|
*
|
|
* Replaces the pi extension model. Instead of injecting messages into pi's
|
|
* session, this runs as an independent Node.js process that:
|
|
* - Connects to Zulip via event queue polling
|
|
* - Calls the harness inference API directly for each DM
|
|
* - Maintains per-sender conversation memory
|
|
* - Runs as a systemd service (survives pi shutdown)
|
|
*
|
|
* Config via env vars (see README or systemd service file).
|
|
*
|
|
* @see ADR-001: DM-first architecture
|
|
* @see ADR-007: Platform-native plugin contracts
|
|
*/
|
|
|
|
import zulip, { ZulipClient } from "zulip-js";
|
|
import http from "node:http";
|
|
import fs from "node:fs";
|
|
import path from "node:path";
|
|
import { fileURLToPath } from "node:url";
|
|
|
|
// ── Types ─────────────────────────────────────────────────────────────────
|
|
|
|
interface Config {
|
|
zulip: {
|
|
email: string;
|
|
api_key: string;
|
|
site: string;
|
|
};
|
|
agent: {
|
|
name: string;
|
|
owner_email: string;
|
|
};
|
|
harness: {
|
|
url: string;
|
|
api_key: string;
|
|
model: string;
|
|
max_tokens: number;
|
|
};
|
|
health_port: number;
|
|
poll_interval_ms: number;
|
|
max_retries: number;
|
|
retry_delay_ms: number;
|
|
}
|
|
|
|
interface ChatMessage {
|
|
role: "system" | "user" | "assistant";
|
|
content: string;
|
|
}
|
|
|
|
interface PendingReply {
|
|
id: number;
|
|
zulipMessageId?: number;
|
|
senderId: number;
|
|
senderName: string;
|
|
type: "private";
|
|
}
|
|
|
|
interface ZulipQueue {
|
|
client: ZulipClient;
|
|
queueId: string;
|
|
lastEventId: number;
|
|
poll(): Promise<any[]>;
|
|
sendMessage(params: {
|
|
type: string;
|
|
to: string | number;
|
|
subject?: string;
|
|
content: string;
|
|
}): Promise<number>;
|
|
editMessage(messageId: number, content: string): Promise<void>;
|
|
sendTypingNotification(userIds: number[], operation: string): Promise<void>;
|
|
}
|
|
|
|
// ── Constants ──────────────────────────────────────────────────────────────
|
|
|
|
const SYSTEM_PROMPT = `You are Abiba, an AI assistant operating in a homelab environment. You communicate through Zulip DMs. Be helpful, concise, and technically accurate. Your owner is Jerome.`;
|
|
|
|
const PLACEHOLDERS = [
|
|
":robot: _Processing your message..._",
|
|
":hourglass_flowing_sand: _Thinking..._",
|
|
":brain: _Generating response..._",
|
|
];
|
|
|
|
const MAX_CONVERSATION_HISTORY = 50;
|
|
const MAX_ZULIP_MSG_LENGTH = 10000;
|
|
const STREAMING_UPDATE_INTERVAL_MS = 2000;
|
|
|
|
// ── Config loading ────────────────────────────────────────────────────────
|
|
|
|
function loadConfig(): Config {
|
|
const email = process.env.ZULIP_EMAIL || "";
|
|
const apiKey = process.env.ZULIP_API_KEY || "";
|
|
const site = process.env.ZULIP_SITE || "";
|
|
if (!email || !apiKey || !site) {
|
|
console.error(
|
|
"[abiba-zulip] Missing required env vars: ZULIP_EMAIL, ZULIP_API_KEY, ZULIP_SITE"
|
|
);
|
|
process.exit(1);
|
|
}
|
|
|
|
return {
|
|
zulip: { email, api_key: apiKey, site },
|
|
agent: {
|
|
name: process.env.AGENT_NAME || "abiba",
|
|
owner_email: process.env.AGENT_OWNER_EMAIL || "jerome@sysloggh.com",
|
|
},
|
|
harness: {
|
|
url:
|
|
process.env.HARNESS_URL ||
|
|
"http://192.168.68.116/v1/chat/completions",
|
|
api_key:
|
|
process.env.HARNESS_API_KEY ||
|
|
"sk-856ffb0bbb-e5aaf78b10054eca608f8fbcbd73a889",
|
|
model: process.env.HARNESS_MODEL || "qwen3.6-35B-A3B",
|
|
max_tokens: parseInt(process.env.HARNESS_MAX_TOKENS || "4096", 10),
|
|
},
|
|
health_port: parseInt(process.env.HEALTH_PORT || "9200", 10),
|
|
poll_interval_ms: parseInt(process.env.POLL_INTERVAL_MS || "3000", 10),
|
|
max_retries: parseInt(process.env.MAX_RETRIES || "5", 10),
|
|
retry_delay_ms: parseInt(process.env.RETRY_DELAY_MS || "5000", 10),
|
|
};
|
|
}
|
|
|
|
// ── Conversation memory ───────────────────────────────────────────────────
|
|
|
|
const conversations = new Map<string, ChatMessage[]>();
|
|
|
|
function getOrCreateConversation(senderEmail: string): ChatMessage[] {
|
|
let history = conversations.get(senderEmail);
|
|
if (!history) {
|
|
history = [{ role: "system", content: SYSTEM_PROMPT }];
|
|
conversations.set(senderEmail, history);
|
|
}
|
|
return history;
|
|
}
|
|
|
|
function addToConversation(
|
|
senderEmail: string,
|
|
message: ChatMessage
|
|
): void {
|
|
const history = getOrCreateConversation(senderEmail);
|
|
history.push(message);
|
|
|
|
// Trim if too long (keep system prompt)
|
|
if (history.length > MAX_CONVERSATION_HISTORY) {
|
|
const system = history[0];
|
|
history.splice(1, history.length - MAX_CONVERSATION_HISTORY);
|
|
history.unshift(system);
|
|
}
|
|
}
|
|
|
|
// ── Harness API ───────────────────────────────────────────────────────────
|
|
|
|
async function queryHarness(
|
|
senderEmail: string,
|
|
userMessage: string,
|
|
config: Config
|
|
): Promise<string> {
|
|
const messages = getOrCreateConversation(senderEmail);
|
|
messages.push({ role: "user", content: userMessage });
|
|
|
|
const response = await fetch(config.harness.url, {
|
|
method: "POST",
|
|
headers: {
|
|
Authorization: `Bearer ${config.harness.api_key}`,
|
|
"Content-Type": "application/json",
|
|
},
|
|
body: JSON.stringify({
|
|
model: config.harness.model,
|
|
messages: messages,
|
|
max_tokens: config.harness.max_tokens,
|
|
stream: false,
|
|
}),
|
|
signal: AbortSignal.timeout(120_000),
|
|
});
|
|
|
|
if (!response.ok) {
|
|
const text = await response.text();
|
|
throw new Error(`Harness API returned ${response.status}: ${text.slice(0, 500)}`);
|
|
}
|
|
|
|
const data = await response.json();
|
|
const reply =
|
|
data.choices?.[0]?.message?.content ?? "[No response generated]";
|
|
|
|
// Store assistant reply in conversation history
|
|
messages.push({ role: "assistant", content: reply });
|
|
|
|
return reply;
|
|
}
|
|
|
|
// ── Zulip queue creation ──────────────────────────────────────────────────
|
|
|
|
async function createZulipQueue(config: Config): Promise<ZulipQueue> {
|
|
const client = await zulip({
|
|
username: config.zulip.email,
|
|
apiKey: config.zulip.api_key,
|
|
realm: config.zulip.site,
|
|
});
|
|
|
|
const queueRes: any = await client.queues.register({
|
|
event_types: ["message"],
|
|
});
|
|
const queueId: string = queueRes.queue_id;
|
|
let lastEventId: number = queueRes.last_event_id ?? -1;
|
|
|
|
return {
|
|
client,
|
|
queueId,
|
|
lastEventId,
|
|
async poll() {
|
|
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: any = await res.json();
|
|
if (data.events && Array.isArray(data.events)) {
|
|
for (const event of data.events) {
|
|
if (event.id > lastEventId) {
|
|
lastEventId = event.id;
|
|
}
|
|
}
|
|
// Update the cached lastEventId so reconnections preserve progress
|
|
this.lastEventId = lastEventId;
|
|
return data.events.filter((e: any) => e.type === "message");
|
|
}
|
|
return [];
|
|
},
|
|
async sendMessage(params) {
|
|
const formBody = new URLSearchParams();
|
|
formBody.append("type", params.type);
|
|
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()}`);
|
|
}
|
|
const body: any = await res.json();
|
|
return body.id;
|
|
},
|
|
async editMessage(messageId, content) {
|
|
const formBody = new URLSearchParams();
|
|
formBody.append("content", content);
|
|
const res = await fetch(
|
|
`${config.zulip.site}/api/v1/messages/${messageId}`,
|
|
{
|
|
method: "PATCH",
|
|
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 && res.status !== 400) {
|
|
console.warn(
|
|
`[abiba-zulip] Edit message ${messageId} returned ${res.status}`
|
|
);
|
|
}
|
|
},
|
|
async sendTypingNotification(userIds, operation) {
|
|
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 server ─────────────────────────────────────────────────────────
|
|
|
|
function startHealthServer(port: number, getState: () => any): http.Server {
|
|
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", ...state }));
|
|
} else {
|
|
res.writeHead(404);
|
|
res.end("Not found");
|
|
}
|
|
});
|
|
|
|
server.on("error", (err: NodeJS.ErrnoException) => {
|
|
if (err.code === "EADDRINUSE") {
|
|
console.warn(
|
|
`[abiba-zulip] Port :${port} already in use — skipping health endpoint`
|
|
);
|
|
} else {
|
|
console.error(`[abiba-zulip] Health server error:`, err.message);
|
|
}
|
|
});
|
|
|
|
server.listen(port, "127.0.0.1", () => {
|
|
console.log(`[abiba-zulip] Health endpoint on :${port}`);
|
|
});
|
|
|
|
return server;
|
|
}
|
|
|
|
// ── Main ──────────────────────────────────────────────────────────────────
|
|
|
|
async function main() {
|
|
const config = loadConfig();
|
|
console.log(`[abiba-zulip] Starting Abiba Zulip Service v2.0.0`);
|
|
console.log(
|
|
`[abiba-zulip] Agent: ${config.agent.name}, Zulip: ${config.zulip.email} @ ${config.zulip.site}`
|
|
);
|
|
console.log(
|
|
`[abiba-zulip] Harness: ${config.harness.model} @ ${config.harness.url}`
|
|
);
|
|
|
|
let queue: ZulipQueue | null = null;
|
|
let connected = false;
|
|
let retryCount = 0;
|
|
let lastError: string | null = null;
|
|
let messagesProcessed = 0;
|
|
let pollTimer: ReturnType<typeof setInterval> | null = null;
|
|
let healthServer: http.Server | null = null;
|
|
|
|
// Pending replies awaiting LLM response
|
|
const pendingReplies: Map<number, PendingReply> = new Map();
|
|
let replyIdCounter = 0;
|
|
|
|
// Track which sender's DM is currently being processed (one at a time per sender)
|
|
const processingSenders = new Set<string>();
|
|
|
|
function getState() {
|
|
return {
|
|
connected,
|
|
email: config.zulip.email,
|
|
site: config.zulip.site,
|
|
agent: config.agent.name,
|
|
messages_processed: messagesProcessed,
|
|
last_error: lastError,
|
|
retry_count: retryCount,
|
|
active_conversations: conversations.size,
|
|
pending_replies: pendingReplies.size,
|
|
};
|
|
}
|
|
|
|
/**
|
|
* Process a single DM: send placeholder → call harness → edit with response.
|
|
*/
|
|
async function processDM(
|
|
senderEmail: string,
|
|
senderId: number,
|
|
senderName: string,
|
|
content: string
|
|
): Promise<void> {
|
|
if (processingSenders.has(senderEmail)) {
|
|
console.log(
|
|
`[abiba-zulip] Already processing DM from ${senderName}, queueing...`
|
|
);
|
|
// TODO: implement proper queue if needed
|
|
return;
|
|
}
|
|
|
|
processingSenders.add(senderEmail);
|
|
messagesProcessed++;
|
|
|
|
try {
|
|
// Send typing indicator
|
|
queue!
|
|
.sendTypingNotification([senderId], "start")
|
|
.catch(() => {});
|
|
|
|
// Send placeholder
|
|
const replyId = ++replyIdCounter;
|
|
const placeholder =
|
|
PLACEHOLDERS[replyId % PLACEHOLDERS.length];
|
|
|
|
let placeholderMsgId: number | undefined;
|
|
try {
|
|
placeholderMsgId = await queue!.sendMessage({
|
|
type: "private",
|
|
to: senderId,
|
|
content: placeholder,
|
|
});
|
|
pendingReplies.set(replyId, {
|
|
id: replyId,
|
|
zulipMessageId: placeholderMsgId,
|
|
senderId,
|
|
senderName,
|
|
type: "private",
|
|
});
|
|
console.log(
|
|
`[abiba-zulip] [${replyId}] Placeholder sent for ${senderName} (msg ${placeholderMsgId})`
|
|
);
|
|
} catch {
|
|
pendingReplies.set(replyId, {
|
|
id: replyId,
|
|
senderId,
|
|
senderName,
|
|
type: "private",
|
|
});
|
|
console.log(
|
|
`[abiba-zulip] [${replyId}] No placeholder for ${senderName}`
|
|
);
|
|
}
|
|
|
|
// Call harness API
|
|
console.log(
|
|
`[abiba-zulip] [${replyId}] Querying harness for ${senderName}...`
|
|
);
|
|
const response = await queryHarness(senderEmail, content, config);
|
|
|
|
// Stop typing
|
|
queue!
|
|
.sendTypingNotification([senderId], "stop")
|
|
.catch(() => {});
|
|
|
|
// Truncate if needed
|
|
const truncated =
|
|
response.length > MAX_ZULIP_MSG_LENGTH
|
|
? response.slice(0, MAX_ZULIP_MSG_LENGTH) +
|
|
"\n\n[...truncated at Zulip limit]"
|
|
: response;
|
|
|
|
// Send response
|
|
const pending = pendingReplies.get(replyId);
|
|
if (pending?.zulipMessageId) {
|
|
await queue!.editMessage(pending.zulipMessageId, truncated);
|
|
console.log(
|
|
`[abiba-zulip] [${replyId}] Finalized for ${senderName} (${truncated.length} chars)`
|
|
);
|
|
} else {
|
|
await queue!.sendMessage({
|
|
type: "private",
|
|
to: senderId,
|
|
content: truncated,
|
|
});
|
|
console.log(
|
|
`[abiba-zulip] [${replyId}] Sent fresh for ${senderName} (${truncated.length} chars)`
|
|
);
|
|
}
|
|
|
|
pendingReplies.delete(replyId);
|
|
} catch (err) {
|
|
const errMsg = err instanceof Error ? err.message : String(err);
|
|
console.error(`[abiba-zulip] Error processing DM from ${senderName}: ${errMsg}`);
|
|
lastError = errMsg;
|
|
|
|
// Try to send error message to user
|
|
try {
|
|
await queue!.sendMessage({
|
|
type: "private",
|
|
to: senderId,
|
|
content:
|
|
":warning: Sorry, I encountered an error processing your request. Please try again later.",
|
|
});
|
|
} catch {
|
|
// Best effort
|
|
}
|
|
} finally {
|
|
processingSenders.delete(senderEmail);
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Polling loop — connects, registers queue, polls for events.
|
|
*/
|
|
async function startPolling() {
|
|
console.log(
|
|
`[abiba-zulip] Connecting to ${config.zulip.site} as ${config.zulip.email}...`
|
|
);
|
|
|
|
try {
|
|
queue = await createZulipQueue(config);
|
|
connected = true;
|
|
retryCount = 0;
|
|
lastError = null;
|
|
console.log(
|
|
`[abiba-zulip] Connected, queue: ${queue.queueId} (last_event_id=${queue.lastEventId})`
|
|
);
|
|
|
|
// Start health server (on first connect)
|
|
if (!healthServer) {
|
|
healthServer = startHealthServer(config.health_port, getState);
|
|
}
|
|
|
|
// Poll loop
|
|
pollTimer = setInterval(async () => {
|
|
if (!queue) return;
|
|
|
|
try {
|
|
const events = await queue.poll();
|
|
|
|
for (const event of events) {
|
|
const msg = event.message;
|
|
if (!msg) continue;
|
|
if (msg.sender_email === config.zulip.email) continue;
|
|
if (msg.type !== "private") continue; // DM-first
|
|
|
|
const content = (msg.content || "").trim();
|
|
if (!content) continue;
|
|
|
|
console.log(
|
|
`[abiba-zulip] DM from ${msg.sender_full_name || msg.sender_email}: ${content.slice(0, 80)}...`
|
|
);
|
|
|
|
// Process asynchronously (don't block the poll loop)
|
|
processDM(
|
|
msg.sender_email,
|
|
msg.sender_id,
|
|
msg.sender_full_name || msg.sender_email,
|
|
content
|
|
).catch((err) =>
|
|
console.error(`[abiba-zulip] processDM error:`, err)
|
|
);
|
|
}
|
|
} catch (err) {
|
|
const errMsg = err instanceof Error ? err.message : String(err);
|
|
if (
|
|
errMsg.includes("BAD_EVENT_QUEUE_ID") ||
|
|
errMsg.includes("queue_id")
|
|
) {
|
|
console.log(`[abiba-zulip] Queue expired, reconnecting...`);
|
|
connected = false;
|
|
if (pollTimer) clearInterval(pollTimer);
|
|
pollTimer = null;
|
|
setTimeout(() => startPolling(), config.retry_delay_ms);
|
|
return;
|
|
}
|
|
lastError = errMsg;
|
|
console.error(`[abiba-zulip] Poll error: ${errMsg}`);
|
|
}
|
|
}, config.poll_interval_ms);
|
|
} catch (err) {
|
|
const errMsg = err instanceof Error ? err.message : String(err);
|
|
lastError = errMsg;
|
|
console.error(`[abiba-zulip] Connection failed: ${errMsg}`);
|
|
|
|
if (retryCount < config.max_retries) {
|
|
retryCount++;
|
|
const delay = config.retry_delay_ms * retryCount;
|
|
console.log(
|
|
`[abiba-zulip] Retrying in ${delay / 1000}s (attempt ${retryCount})...`
|
|
);
|
|
setTimeout(() => startPolling(), delay);
|
|
} else {
|
|
console.error(
|
|
`[abiba-zulip] Max retries (${config.max_retries}) reached. Giving up.`
|
|
);
|
|
lastError = `Connection failed after ${config.max_retries} retries: ${errMsg}`;
|
|
}
|
|
}
|
|
}
|
|
|
|
// ── Graceful shutdown ──────────────────────────────────────────────────
|
|
|
|
function shutdown(signal: string) {
|
|
console.log(`\n[abiba-zulip] Received ${signal}. Shutting down...`);
|
|
if (pollTimer) {
|
|
clearInterval(pollTimer);
|
|
pollTimer = null;
|
|
}
|
|
if (healthServer) {
|
|
healthServer.close();
|
|
healthServer = null;
|
|
}
|
|
connected = false;
|
|
queue = null;
|
|
process.exit(0);
|
|
}
|
|
|
|
process.on("SIGINT", () => shutdown("SIGINT"));
|
|
process.on("SIGTERM", () => shutdown("SIGTERM"));
|
|
|
|
// ── Start ──────────────────────────────────────────────────────────────
|
|
|
|
await startPolling();
|
|
|
|
// Keep alive — idle loop
|
|
console.log(`[abiba-zulip] Service running. Waiting for DMs...`);
|
|
await new Promise(() => {}); // never resolves — process runs until killed
|
|
}
|
|
|
|
main().catch((err) => {
|
|
console.error(`[abiba-zulip] Fatal error:`, err);
|
|
process.exit(1);
|
|
});
|