Azreen Jamal
436 lines
16 KiB
TypeScript
436 lines
16 KiB
TypeScript
import { Bot, Context } from "grammy";
|
|
import { readFileSync, existsSync, mkdirSync } from "fs";
|
|
import { resolve } from "path";
|
|
|
|
// ─── Config ───
|
|
const BOT_TOKEN = process.env.TELEGRAM_BOT_TOKEN!;
|
|
const SSH_USER = process.env.SSH_USER || "root";
|
|
const SSH_HOST = process.env.SSH_HOST!;
|
|
const SSH_PORT = process.env.SSH_PORT || "22";
|
|
const ALLOWED_USERS = process.env.TELEGRAM_ALLOWED_USERS?.split(",").map(Number) || [];
|
|
const PI_MODEL = "claude-sonnet-4-6";
|
|
const PI_BIN = "/Users/azreenjamal/.bun/bin/pi";
|
|
|
|
if (!BOT_TOKEN) throw new Error("TELEGRAM_BOT_TOKEN missing from .env");
|
|
if (!SSH_HOST) throw new Error("SSH_HOST missing from .env");
|
|
|
|
// ─── Load infra context ───
|
|
let infraContext = "";
|
|
try {
|
|
infraContext = readFileSync(resolve(__dirname, "../.pi/infra.md"), "utf-8");
|
|
} catch {
|
|
console.warn("⚠️ Could not load .pi/infra.md");
|
|
}
|
|
|
|
// ─── Init ───
|
|
const bot = new Bot(BOT_TOKEN);
|
|
|
|
// ─── Session directory ───
|
|
const SESSIONS_DIR = resolve(__dirname, "sessions");
|
|
if (!existsSync(SESSIONS_DIR)) mkdirSync(SESSIONS_DIR, { recursive: true });
|
|
|
|
function sessionPath(userId: number): string {
|
|
return resolve(SESSIONS_DIR, `user-${userId}.jsonl`);
|
|
}
|
|
|
|
// ─── Active requests ───
|
|
const activeRequests = new Map<number, AbortController>();
|
|
|
|
// ─── System prompt ───
|
|
const SYSTEM_PROMPT = `You are Pi, a DevOps management assistant on Telegram. You are NOT Claude — your name is Pi. If anyone asks who you are, say "I'm Pi, your DevOps assistant." You are connected to a live server via SSH.
|
|
|
|
${infraContext}
|
|
|
|
ARCHITECTURE:
|
|
- You (Pi) run on a local laptop (macOS). You are NOT on the server.
|
|
- You have a bash tool. To run commands on the server, use: ssh -o ConnectTimeout=5 -o StrictHostKeyChecking=accept-new -p ${SSH_PORT} ${SSH_USER}@${SSH_HOST} "<command>"
|
|
- The primary host runs HAProxy and Incus (LXC containers).
|
|
- Docker runs INSIDE the Incus container "cr-server-new" (IP 10.213.16.224), NOT on the primary host.
|
|
- To run Docker/sync commands: ssh -p ${SSH_PORT} ${SSH_USER}@${SSH_HOST} "incus exec cr-server-new -- <command>"
|
|
- To query Postgres: ssh -p ${SSH_PORT} ${SSH_USER}@${SSH_HOST} "incus exec cr-server-new -- docker exec dokploy-migrated-cr-postgres-data psql -U postgres -d donation_warehouse -c \\"SELECT ...\\""
|
|
- NEVER run docker commands directly on the primary host.
|
|
|
|
KEY PATHS (inside cr-server-new):
|
|
- /opt/enthuse-db-sync-v2/, /opt/launchgood-sync/, /opt/charityright-sync/
|
|
- Postgres container: dokploy-migrated-cr-postgres-data
|
|
- Docker network: dokploy-network
|
|
|
|
RULES:
|
|
- Be VERY concise — this is Telegram, not a blog post.
|
|
- Use | head, | tail, grep to limit output.
|
|
- For destructive commands (rm, drop, delete, restart), WARN and ask confirmation first.
|
|
- ALWAYS respond with text. Never return an empty response.
|
|
- DO NOT be lazy. When asked to check something, actually run the commands. When asked to fix something, do it immediately. Don't say "you could try X" — just do X.
|
|
- Take initiative. If you see a problem while investigating, fix it or flag it.
|
|
- Chain commands when possible to get answers faster.
|
|
- If a command fails, try an alternative approach immediately. Don't just report the error.
|
|
- NEVER run local commands like "just", "pi", "npm", "bun" etc. on the local machine. You only interact with the REMOTE server via SSH.
|
|
- NEVER run "cd /Users" or any local filesystem commands. ALL commands must go through SSH.
|
|
- The ONLY use of your bash tool is: ssh -p ${SSH_PORT} ${SSH_USER}@${SSH_HOST} "..." or curl for APIs (like Asana).
|
|
- Be EFFICIENT with tool calls. Combine multiple checks into ONE ssh command using && or ;
|
|
- Example: ssh ... "uptime && df -h / && free -h && docker ps" — NOT 4 separate SSH calls.
|
|
- Aim for max 5-8 tool calls per task. If a task needs more, break it into phases and report after each phase.`;
|
|
|
|
// ─── Stream Pi JSON output with live Telegram updates ───
|
|
async function callPiStreaming(
|
|
userId: number,
|
|
message: string,
|
|
signal: AbortSignal,
|
|
ctx: Context,
|
|
): Promise<string> {
|
|
const session = sessionPath(userId);
|
|
let hasSession = existsSync(session);
|
|
|
|
// Auto-clear session if too large (>500KB = context bloat)
|
|
if (hasSession) {
|
|
try {
|
|
const stat = require("fs").statSync(session);
|
|
if (stat.size > 500_000) {
|
|
console.log(`🧹 [${userId}] Session too large (${(stat.size / 1024).toFixed(0)}KB) — clearing`);
|
|
require("fs").unlinkSync(session);
|
|
hasSession = false;
|
|
}
|
|
} catch {}
|
|
}
|
|
|
|
const args = [
|
|
PI_BIN, "-p",
|
|
"--model", PI_MODEL,
|
|
"--tools", "bash",
|
|
"--mode", "json",
|
|
"--system-prompt", SYSTEM_PROMPT,
|
|
"--session", session,
|
|
"--no-extensions",
|
|
];
|
|
|
|
if (hasSession) args.push("--continue");
|
|
args.push(message);
|
|
|
|
console.log(`🤖 [${userId}] Calling Pi (${hasSession ? "continue" : "new"} session)...`);
|
|
|
|
const proc = Bun.spawn(args, {
|
|
stdout: "pipe",
|
|
stderr: "pipe",
|
|
cwd: "/tmp",
|
|
env: { ...process.env, TERM: "dumb" },
|
|
});
|
|
|
|
const abortHandler = () => proc.kill();
|
|
signal.addEventListener("abort", abortHandler, { once: true });
|
|
|
|
const timer = setTimeout(() => {
|
|
console.log(`⏰ [${userId}] Pi timed out after 600s`);
|
|
proc.kill();
|
|
}, 600_000);
|
|
|
|
// Read stderr in background
|
|
const stderrPromise = new Response(proc.stderr).text();
|
|
|
|
// Stream stdout line by line
|
|
const reader = proc.stdout.getReader();
|
|
const decoder = new TextDecoder();
|
|
let buffer = "";
|
|
let finalText = "";
|
|
let currentToolCmd = "";
|
|
let lastProgressTime = 0;
|
|
const MIN_PROGRESS_INTERVAL = 3000; // Don't spam Telegram
|
|
|
|
async function sendProgress(msg: string) {
|
|
const now = Date.now();
|
|
if (now - lastProgressTime < MIN_PROGRESS_INTERVAL) return;
|
|
lastProgressTime = now;
|
|
try {
|
|
const escaped = msg.replace(/[<>&]/g, c => c === '<' ? '<' : c === '>' ? '>' : '&');
|
|
await ctx.reply(`⚙️ <code>${escaped}</code>`, { parse_mode: "HTML" });
|
|
} catch {}
|
|
}
|
|
|
|
try {
|
|
while (true) {
|
|
const { done, value } = await reader.read();
|
|
if (done) break;
|
|
|
|
buffer += decoder.decode(value, { stream: true });
|
|
const lines = buffer.split("\n");
|
|
buffer = lines.pop() || "";
|
|
|
|
for (const line of lines) {
|
|
if (!line.trim()) continue;
|
|
try {
|
|
const event = JSON.parse(line);
|
|
|
|
// Tool execution started — show what command is running
|
|
if (event.type === "tool_execution_start") {
|
|
const cmd = event.args?.command;
|
|
if (cmd) {
|
|
currentToolCmd = cmd;
|
|
const short = cmd.length > 80 ? cmd.slice(0, 80) + "..." : cmd;
|
|
console.log(`🔧 [${userId}] ${short}`);
|
|
await sendProgress(short);
|
|
} else {
|
|
console.log(`🔧 [${userId}] ${event.toolName}(${JSON.stringify(event.args).slice(0, 60)})`);
|
|
}
|
|
}
|
|
|
|
// Tool execution finished — show brief result
|
|
if (event.type === "tool_execution_end") {
|
|
const output = typeof event.result === "string" ? event.result : "";
|
|
if (output) {
|
|
const brief = output.split("\n")[0].slice(0, 60);
|
|
if (brief) await sendProgress(`→ ${brief}`);
|
|
}
|
|
}
|
|
|
|
// agent_end — extract ALL assistant text blocks
|
|
if (event.type === "agent_end" && event.messages) {
|
|
const allText: string[] = [];
|
|
for (const msg of event.messages) {
|
|
if (msg.role === "assistant" && Array.isArray(msg.content)) {
|
|
for (const block of msg.content) {
|
|
if (block.type === "text" && block.text) allText.push(block.text);
|
|
}
|
|
}
|
|
}
|
|
if (allText.length > 0) {
|
|
finalText = allText[allText.length - 1];
|
|
}
|
|
}
|
|
|
|
// Fallback: turn_end for intermediate text
|
|
if (event.type === "turn_end" && event.message?.role === "assistant") {
|
|
const content = event.message.content;
|
|
if (Array.isArray(content)) {
|
|
const turnText = content
|
|
.filter((b: any) => b.type === "text" && b.text)
|
|
.map((b: any) => b.text)
|
|
.join("\n");
|
|
if (turnText) finalText = turnText;
|
|
}
|
|
}
|
|
|
|
} catch {
|
|
// Not valid JSON, skip
|
|
}
|
|
}
|
|
}
|
|
} catch (err: any) {
|
|
if (!signal.aborted) console.error(`❌ [${userId}] Stream error:`, err.message);
|
|
}
|
|
|
|
clearTimeout(timer);
|
|
signal.removeEventListener("abort", abortHandler);
|
|
|
|
const stderr = await stderrPromise;
|
|
const exitCode = await proc.exited;
|
|
|
|
if (signal.aborted) throw new Error("Cancelled");
|
|
|
|
if (exitCode !== 0) {
|
|
console.error(`❌ [${userId}] Pi exited ${exitCode}: ${stderr.slice(0, 500)}`);
|
|
if (hasSession && (stderr.includes("session") || stderr.includes("tool_use_id") || stderr.includes("tool_result") || stderr.includes("invalid_request") || stderr.includes("Could not parse"))) {
|
|
console.log(`🔄 [${userId}] Corrupted session — clearing and retrying...`);
|
|
try { require("fs").unlinkSync(session); } catch {}
|
|
return callPiStreaming(userId, message, signal, ctx);
|
|
}
|
|
// Auth failure — don't delete session, just report
|
|
if (stderr.includes("Authentication failed")) {
|
|
throw new Error("Pi auth expired. Run `pi /login anthropic` in terminal.");
|
|
}
|
|
// On timeout (143) or crash, the session file may be incomplete — don't delete it,
|
|
// but next call with --continue will retry. If THAT fails, the corruption handler above catches it.
|
|
const exitMsg = exitCode === 143 ? "timed out (10 min limit). Send your message again to continue." : (stderr.slice(0, 200) || `exited with code ${exitCode}`);
|
|
throw new Error(`Pi error: ${exitMsg}`);
|
|
}
|
|
|
|
return finalText.trim() || "🤷 Got empty response. Try rephrasing or /clear.";
|
|
}
|
|
|
|
// ─── Auth / Rate limit ───
|
|
function isAuthorized(ctx: Context): boolean {
|
|
if (ALLOWED_USERS.length === 0) return false;
|
|
return ALLOWED_USERS.includes(ctx.from?.id || 0);
|
|
}
|
|
|
|
const rateLimits = new Map<number, number[]>();
|
|
function isRateLimited(userId: number): boolean {
|
|
const now = Date.now();
|
|
const ts = (rateLimits.get(userId) || []).filter(t => now - t < 60_000);
|
|
ts.push(now);
|
|
rateLimits.set(userId, ts);
|
|
return ts.length > 15;
|
|
}
|
|
|
|
// ─── SSH exec (for direct /ssh command) ───
|
|
async function sshExec(command: string, timeoutSec = 30): Promise<string> {
|
|
const proc = Bun.spawn(
|
|
["ssh", "-o", "ConnectTimeout=5", "-o", "StrictHostKeyChecking=accept-new",
|
|
"-p", SSH_PORT, `${SSH_USER}@${SSH_HOST}`, command],
|
|
{ stdout: "pipe", stderr: "pipe" }
|
|
);
|
|
const timer = setTimeout(() => proc.kill(), timeoutSec * 1000);
|
|
const [stdout, stderr] = await Promise.all([
|
|
new Response(proc.stdout).text(),
|
|
new Response(proc.stderr).text(),
|
|
]);
|
|
clearTimeout(timer);
|
|
await proc.exited;
|
|
const output = (stdout + stderr).trim();
|
|
return output.length > 2000 ? output.slice(0, 2000) + "\n...(truncated)" : output;
|
|
}
|
|
|
|
function withTimeout<T>(promise: Promise<T>, ms: number, label: string): Promise<T> {
|
|
return Promise.race([
|
|
promise,
|
|
new Promise<never>((_, reject) =>
|
|
setTimeout(() => reject(new Error(`${label} timed out after ${ms / 1000}s`)), ms)
|
|
),
|
|
]);
|
|
}
|
|
|
|
// ─── Middleware ───
|
|
bot.use(async (ctx, next) => {
|
|
if (ctx.chat?.type !== "private") return;
|
|
if (!isAuthorized(ctx)) {
|
|
console.log(`🚫 Blocked: ${ctx.from?.id} (@${ctx.from?.username})`);
|
|
return;
|
|
}
|
|
if (ctx.from && isRateLimited(ctx.from.id)) return ctx.reply("⏳ Slow down.");
|
|
await next();
|
|
});
|
|
|
|
// ─── Commands ───
|
|
bot.command("start", (ctx) =>
|
|
ctx.reply("🤖 Pi DevOps Bot (Sonnet 4.6)\n\n• Chat naturally — I'll run commands\n• /status — quick health check\n• /ssh <cmd> — direct SSH command\n• /stop or /cancel — stop current task\n• /clear — reset conversation")
|
|
);
|
|
|
|
bot.command("status", async (ctx) => {
|
|
await ctx.reply("🔍 Checking...");
|
|
try {
|
|
const out = await withTimeout(sshExec("uptime && echo --- && df -h / | tail -1 && echo --- && free -h | grep Mem && echo --- && incus list -f csv -c ns 2>/dev/null", 10), 15_000, "SSH");
|
|
return ctx.reply(out);
|
|
} catch (err: any) { return ctx.reply(`❌ ${err.message}`); }
|
|
});
|
|
|
|
bot.command("ssh", async (ctx) => {
|
|
const cmd = ctx.match;
|
|
if (!cmd) return ctx.reply("Usage: /ssh <command>");
|
|
const BLOCKED = [/rm\s+-rf\s+\/(?!\w)/i, /mkfs\./i, /dd\s+if=.*of=\/dev/i, /:\(\)\s*\{\s*:\|:/];
|
|
for (const p of BLOCKED) {
|
|
if (p.test(cmd)) return ctx.reply("⛔ Blocked dangerous pattern");
|
|
}
|
|
try {
|
|
const out = await withTimeout(sshExec(cmd, 10), 15_000, "SSH");
|
|
return ctx.reply(out || "(no output)");
|
|
} catch (err: any) { return ctx.reply(`❌ ${err.message}`); }
|
|
});
|
|
|
|
bot.command("clear", (ctx) => {
|
|
const session = sessionPath(ctx.from!.id);
|
|
try { require("fs").unlinkSync(session); } catch {}
|
|
activeRequests.get(ctx.from!.id)?.abort();
|
|
return ctx.reply("🧹 Cleared.");
|
|
});
|
|
|
|
bot.command("cancel", handleCancel);
|
|
bot.command("stop", handleCancel);
|
|
|
|
function handleCancel(ctx: Context) {
|
|
const c = activeRequests.get(ctx.from!.id);
|
|
if (c) { c.abort(); activeRequests.delete(ctx.from!.id); return ctx.reply("🛑 Cancelled."); }
|
|
return ctx.reply("Nothing running.");
|
|
}
|
|
|
|
// ─── Message handler ───
|
|
bot.on("message:text", (ctx) => {
|
|
const userId = ctx.from!.id;
|
|
const msg = ctx.message!.text;
|
|
|
|
const prev = activeRequests.get(userId);
|
|
if (prev) {
|
|
prev.abort();
|
|
console.log(`⏹️ [${userId}] cancelled previous request`);
|
|
}
|
|
|
|
const controller = new AbortController();
|
|
activeRequests.set(userId, controller);
|
|
processMessage(ctx, userId, msg, controller);
|
|
});
|
|
|
|
async function processMessage(ctx: Context, userId: number, msg: string, controller: AbortController) {
|
|
const start = Date.now();
|
|
|
|
const typingInterval = setInterval(() => {
|
|
if (!controller.signal.aborted) ctx.replyWithChatAction("typing").catch(() => {});
|
|
}, 3000);
|
|
ctx.replyWithChatAction("typing").catch(() => {});
|
|
|
|
try {
|
|
const reply = await callPiStreaming(userId, msg, controller.signal, ctx);
|
|
if (controller.signal.aborted) return;
|
|
|
|
const elapsed = ((Date.now() - start) / 1000).toFixed(1);
|
|
console.log(`📤 [${userId}] ${reply.length} chars in ${elapsed}s`);
|
|
|
|
const chunks = reply.match(/[\s\S]{1,4000}/g) || ["(empty)"];
|
|
for (const chunk of chunks) {
|
|
await ctx.reply(chunk).catch(() => ctx.reply("⚠️ Send failed."));
|
|
}
|
|
} catch (err: any) {
|
|
if (!controller.signal.aborted) {
|
|
console.error(`❌ [${userId}]`, err.message);
|
|
await ctx.reply(`❌ ${err.message}`).catch(() => {});
|
|
}
|
|
} finally {
|
|
clearInterval(typingInterval);
|
|
if (activeRequests.get(userId) === controller) activeRequests.delete(userId);
|
|
}
|
|
}
|
|
|
|
// ─── Polling loop ───
|
|
bot.catch((err) => {
|
|
console.error("❌ Middleware error:", err.message || err);
|
|
});
|
|
|
|
let polling = true;
|
|
let offset = 0;
|
|
|
|
async function pollLoop() {
|
|
try {
|
|
await bot.api.deleteWebhook({ drop_pending_updates: true });
|
|
const stale = await bot.api.getUpdates({ offset: -1, limit: 1, timeout: 0 });
|
|
if (stale.length > 0) offset = stale[stale.length - 1].update_id + 1;
|
|
} catch {}
|
|
|
|
await bot.init();
|
|
console.log(`✅ Bot is live — polling started (model: ${PI_MODEL})`);
|
|
|
|
while (polling) {
|
|
try {
|
|
const updates = await bot.api.getUpdates({ offset, limit: 100, timeout: 30 });
|
|
for (const update of updates) {
|
|
offset = update.update_id + 1;
|
|
bot.handleUpdate(update).catch(err => {
|
|
console.error("❌ Update error:", err.message);
|
|
});
|
|
}
|
|
} catch (err: any) {
|
|
if (err?.error_code === 409) {
|
|
await new Promise(r => setTimeout(r, 5000));
|
|
} else if (err?.error_code === 429) {
|
|
const wait = err?.parameters?.retry_after || 10;
|
|
await new Promise(r => setTimeout(r, wait * 1000));
|
|
} else {
|
|
console.error("❌ Poll error:", err.message);
|
|
await new Promise(r => setTimeout(r, 3000));
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
console.log("🚀 Bot starting...");
|
|
pollLoop();
|
|
|
|
const stop = () => { polling = false; };
|
|
process.on("SIGINT", () => { stop(); process.exit(0); });
|
|
process.on("SIGTERM", () => { stop(); process.exit(0); });
|