feat: major telegram bot rewrite - security, UX, smart commands

- SECURITY: cwd=/tmp, strip ANTHROPIC_API_KEY from Pi env - FIX: recursive retry now has max retries counter - FIX: await processMessage, proper error handling - FIX: ESM imports only (no require) - FIX: full stderr drain, check SSH exit codes - UX: single progress message edited in-place (no spam) - UX: HTML formatting for commands output - UX: session warning at 80KB, auto-clear at 100KB - UX: block concurrent tasks per user - SMART: /health command - check pi-worker status - SMART: /asana command - list open tasks - SMART: /status uses edit-in-place - SMART: /ssh shows exit code - CLEAN: removed dead deps reference
2026-03-06 21:15:18 +08:00
parent ea0ba4ee10
commit ccfa969261
1 changed files with 418 additions and 322 deletions
@@ -1,151 +1,196 @@
-import { Bot, Context } from "grammy";
-import { readFileSync, existsSync, mkdirSync } from "fs";
-import { resolve } from "path";
+import {
+  readFileSync,
+  existsSync,
+  mkdirSync,
+  statSync,
+  unlinkSync,
+} from "fs";
+import { Bot } from "grammy";
+import { join } from "path";

 // ─── Config ───
 const BOT_TOKEN = process.env.TELEGRAM_BOT_TOKEN!;
+const ALLOWED_USERS = (process.env.TELEGRAM_ALLOWED_USERS || "")
+  .split(",")
+  .map((s) => s.trim())
+  .filter(Boolean);
 const SSH_USER = process.env.SSH_USER || "root";
-const SSH_HOST = process.env.SSH_HOST!;
+const SSH_HOST = process.env.SSH_HOST || "159.195.60.33";
 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";
+const PI_MODEL = process.env.PI_MODEL || "claude-sonnet-4-6";
+const PI_TIMEOUT_MS = 10 * 60 * 1000; // 10 min
+const SESSION_DIR = join(import.meta.dir, "sessions");
+const SESSION_WARN_KB = 80;
+const SESSION_MAX_KB = 100;
+const MAX_RETRY = 1;
+const RATE_LIMIT_WINDOW = 60_000;
+const RATE_LIMIT_MAX = 15;

-if (!BOT_TOKEN) throw new Error("TELEGRAM_BOT_TOKEN missing from .env");
-if (!SSH_HOST) throw new Error("SSH_HOST missing from .env");
+// ─── State ───
+const activeProcesses = new Map<number, AbortController>();
+const rateLimits = new Map<number, number[]>();

-// ─── Load infra context ───
-let infraContext = "";
-try {
-  infraContext = readFileSync(resolve(__dirname, "../.pi/infra.md"), "utf-8");
-} catch {
-  console.warn("⚠️  Could not load .pi/infra.md");
+// Ensure session dir
+if (!existsSync(SESSION_DIR)) mkdirSync(SESSION_DIR, { recursive: true });
+
+// ─── Infrastructure context ───
+function loadInfraContext(): string {
+  try {
+    const infraPath = join(import.meta.dir, "../.pi/infra.md");
+    if (existsSync(infraPath)) return readFileSync(infraPath, "utf-8");
+  } catch {}
+  return "";
 }
+const INFRA_CONTEXT = loadInfraContext();

-// ─── Init ───
-const bot = new Bot(BOT_TOKEN);
+// ─── System Prompt ───
+const SYSTEM_PROMPT = `You are a DevOps management bot for CharityRight, operated via Telegram.

-// ─── 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
+INFRASTRUCTURE:
+- Primary server: ${SSH_USER}@${SSH_HOST}:${SSH_PORT}
+- Containers: cr-server-new (CharityRight), qc-server-new (QuikCue) via Incus
+- Services: Docker, Dokploy, Postgres, Redis, sync services
+${INFRA_CONTEXT ? `\nDETAILED INFRA:\n${INFRA_CONTEXT}` : ""}

 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.`;
+1. ALL server commands MUST use: ssh ${SSH_USER}@${SSH_HOST} -p ${SSH_PORT} "command"
+2. For containers: ssh ${SSH_USER}@${SSH_HOST} "incus exec cr-server-new -- command"
+3. NEVER run commands directly on the local machine
+4. NEVER read local files, .env files, or secrets
+5. NEVER run destructive commands: rm -rf /, mkfs, dd, DROP DATABASE
+6. Format responses clearly with status indicators
+7. Be concise — this is a mobile chat interface`;

-// ─── Stream Pi JSON output with live Telegram updates ───
+// ─── Dangerous command patterns ───
+const DANGEROUS_PATTERNS = [
+  /rm\s+-rf\s+\/(?!\w)/,
+  /mkfs\./,
+  /dd\s+if=/,
+  /:(){ :\|:& };:/,
+  /shutdown/,
+  /reboot/,
+  /init\s+0/,
+  /DROP\s+(DATABASE|TABLE)/i,
+  /TRUNCATE\s+TABLE/i,
+  /DELETE\s+FROM\s+\w+\s*;/i,
+];
+
+// ─── Bot ───
+const bot = new Bot(BOT_TOKEN);
+
+function isAllowed(userId: number): boolean {
+  if (ALLOWED_USERS.length === 0) return false;
+  return ALLOWED_USERS.includes(String(userId));
+}
+
+function checkRateLimit(userId: number): boolean {
+  const now = Date.now();
+  const timestamps = rateLimits.get(userId) || [];
+  const recent = timestamps.filter((t) => now - t < RATE_LIMIT_WINDOW);
+  recent.push(now);
+  rateLimits.set(userId, recent);
+  return recent.length <= RATE_LIMIT_MAX;
+}
+
+function getSessionPath(userId: number): string {
+  return join(SESSION_DIR, `user-${userId}.jsonl`);
+}
+
+function checkSession(userId: number): { exists: boolean; sizeKB: number } {
+  const path = getSessionPath(userId);
+  if (!existsSync(path)) return { exists: false, sizeKB: 0 };
+  const stat = statSync(path);
+  return { exists: true, sizeKB: Math.round(stat.size / 1024) };
+}
+
+// ─── SSH helper ───
+async function sshExec(
+  command: string,
+  timeoutMs = 30_000
+): Promise<{ stdout: string; stderr: string; exitCode: number }> {
+  const proc = Bun.spawn(
+    [
+      "ssh",
+      "-o", "StrictHostKeyChecking=no",
+      "-o", "ConnectTimeout=10",
+      "-p", SSH_PORT,
+      `${SSH_USER}@${SSH_HOST}`,
+      command,
+    ],
+    { stdout: "pipe", stderr: "pipe" }
+  );
+
+  const timer = setTimeout(() => proc.kill(), timeoutMs);
+
+  const [stdout, stderr] = await Promise.all([
+    new Response(proc.stdout).text(),
+    new Response(proc.stderr).text(),
+  ]);
+
+  clearTimeout(timer);
+  const exitCode = await proc.exited;
+
+  return { stdout: stdout.trim(), stderr: stderr.trim(), exitCode };
+}
+
+// ─── Pi Streaming ───
 async function callPiStreaming(
  userId: number,
  message: string,
  signal: AbortSignal,
-  ctx: Context,
+  ctx: any,
+  retries = MAX_RETRY
 ): 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 session = getSessionPath(userId);
+  const hasSession = existsSync(session);

  const args = [
-    PI_BIN, "-p",
+    "pi",
+    "--mode", "json",
+    "-p",
+    "--no-extensions",
    "--model", PI_MODEL,
    "--tools", "bash",
-    "--mode", "json",
-    "--system-prompt", SYSTEM_PROMPT,
-    "--session", session,
-    "--no-extensions",
+    "--thinking", "off",
+    "--append-system-prompt", SYSTEM_PROMPT,
  ];

-  if (hasSession) args.push("--continue");
+  if (hasSession) {
+    args.push("--session", session, "--continue");
+  } else {
+    args.push("--session", session);
+  }
  args.push(message);

-  console.log(`🤖 [${userId}] Calling Pi (${hasSession ? "continue" : "new"} session)...`);
+  // Strip secrets from child env
+  const { ANTHROPIC_API_KEY: _removed, ...cleanEnv } = process.env;

-  const proc = Bun.spawn(args, {
+  const proc = Bun.spawn({
+    cmd: args,
+    cwd: "/tmp",
    stdout: "pipe",
    stderr: "pipe",
-    cwd: "/tmp",
-    env: { ...process.env, TERM: "dumb" },
+    env: cleanEnv,
  });

+  const timeout = setTimeout(() => {
+    console.log(`⏰ [${userId}] Pi timed out`);
+    proc.kill();
+  }, PI_TIMEOUT_MS);
+
  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 === '<' ? '&lt;' : c === '>' ? '&gt;' : '&amp;');
-      await ctx.reply(`⚙️ <code>${escaped}</code>`, { parse_mode: "HTML" });
-    } catch {}
-  }
+  let toolCalls = 0;
+  let progressMsgId: number | null = null;
+  let lastProgressUpdate = 0;

  try {
+    const reader = proc.stdout.getReader();
+    const decoder = new TextDecoder();
+    let buffer = "";
+
    while (true) {
      const { done, value } = await reader.read();
      if (done) break;
@@ -158,30 +203,39 @@ async function callPiStreaming(
        if (!line.trim()) continue;
        try {
          const event = JSON.parse(line);
-          
-          // Tool execution started — show what command is running
+
+          // Tool execution — update single progress message
          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);
+            toolCalls++;
+            const input = event.tool_input;
+            let preview = "";
+
+            if (input?.command) {
+              preview = input.command.substring(0, 100);
+            } else if (input?.path) {
+              preview = `📄 ${input.path}`;
            } else {
-              console.log(`🔧 [${userId}] ${event.toolName}(${JSON.stringify(event.args).slice(0, 60)})`);
+              preview = event.tool_name || "working";
+            }
+
+            const progressText = `⚙️ [${toolCalls}] ${preview}...`;
+            const now = Date.now();
+
+            // Throttle to every 2s
+            if (now - lastProgressUpdate > 2000) {
+              try {
+                if (progressMsgId) {
+                  await ctx.api.editMessageText(ctx.chat.id, progressMsgId, progressText);
+                } else {
+                  const sent = await ctx.reply(progressText);
+                  progressMsgId = sent.message_id;
+                }
+                lastProgressUpdate = now;
+              } catch {}
            }
          }

-          // 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
+          // Agent end — extract final assistant response
          if (event.type === "agent_end" && event.messages) {
            const allText: string[] = [];
            for (const msg of event.messages) {
@@ -195,241 +249,283 @@ async function callPiStreaming(
              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 {}
      }
    }
-  } catch (err: any) {
-    if (!signal.aborted) console.error(`❌ [${userId}] Stream error:`, err.message);
-  }

-  clearTimeout(timer);
-  signal.removeEventListener("abort", abortHandler);
+    // Drain stderr
+    const stderrText = await new Response(proc.stderr).text();

-  const stderr = await stderrPromise;
-  const exitCode = await proc.exited;
+    clearTimeout(timeout);
+    signal.removeEventListener("abort", abortHandler);
+    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);
+    // Delete progress message
+    if (progressMsgId) {
+      try {
+        await ctx.api.deleteMessage(ctx.chat.id, progressMsgId);
+      } catch {}
    }
-    // 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.");
+
+    // Handle corrupted session — retry with counter
+    if (
+      exitCode !== 0 &&
+      hasSession &&
+      retries > 0 &&
+      (stderrText.includes("session") || stderrText.includes("corrupt") || stderrText.includes("parse"))
+    ) {
+      console.log(`🔄 [${userId}] Corrupted session — clearing and retrying (${retries} left)...`);
+      try {
+        unlinkSync(session);
+      } catch {}
+      return callPiStreaming(userId, message, signal, ctx, retries - 1);
    }
-    // 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}`);
+
+    if (!finalText.trim()) {
+      if (stderrText) return `⚠️ Pi error: ${stderrText.substring(0, 500)}`;
+      return "⚠️ No response from Pi agent.";
+    }
+
+    return finalText;
+  } catch (error: any) {
+    clearTimeout(timeout);
+    signal.removeEventListener("abort", abortHandler);
+    if (progressMsgId) {
+      try {
+        await ctx.api.deleteMessage(ctx.chat.id, progressMsgId);
+      } catch {}
+    }
+    if (signal.aborted) return "🛑 Task cancelled.";
+    return `❌ Error: ${error.message}`;
  }
-
-  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" }
+// ─── Command Handlers ───
+bot.command("start", async (ctx) => {
+  if (!isAllowed(ctx.from?.id || 0)) return;
+  await ctx.reply(
+    `🤖 <b>CharityRight DevOps Bot</b>\n\n` +
+      `<b>Commands:</b>\n` +
+      `/status — Server health check\n` +
+      `/ssh &lt;cmd&gt; — Run SSH command\n` +
+      `/health — Pi Worker status\n` +
+      `/asana — List Asana tasks\n` +
+      `/clear — Reset conversation\n` +
+      `/stop — Cancel running task\n\n` +
+      `Or just type naturally — I'll figure it out.`,
+    { parse_mode: "HTML" }
  );
-  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...");
+  if (!isAllowed(ctx.from?.id || 0)) return;
+  const msg = await ctx.reply("🔍 Checking server status...");
  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}`); }
+    const { stdout, exitCode } = await sshExec(
+      "uptime && echo '---' && df -h / | tail -1 && echo '---' && free -h | grep Mem && echo '---' && incus list -f compact 2>/dev/null || true"
+    );
+    const status = exitCode === 0 ? "✅" : "⚠️";
+    await ctx.api.editMessageText(
+      ctx.chat.id,
+      msg.message_id,
+      `${status} <b>Server Status</b>\n<pre>${stdout.substring(0, 3500)}</pre>`,
+      { parse_mode: "HTML" }
+    );
+  } catch (error: any) {
+    await ctx.api.editMessageText(ctx.chat.id, msg.message_id, `❌ SSH failed: ${error.message}`);
+  }
 });

 bot.command("ssh", async (ctx) => {
+  if (!isAllowed(ctx.from?.id || 0)) return;
  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");
+  if (!cmd) {
+    await ctx.reply("Usage: /ssh <command>");
+    return;
  }
+
+  if (DANGEROUS_PATTERNS.some((p) => p.test(cmd))) {
+    await ctx.reply("🚫 Blocked — dangerous command detected.");
+    return;
+  }
+
+  const msg = await ctx.reply(`⚙️ Running: <code>${cmd.substring(0, 200)}</code>`, { parse_mode: "HTML" });
  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}`); }
+    const { stdout, stderr, exitCode } = await sshExec(cmd);
+    const output = stdout || stderr || "(no output)";
+    const status = exitCode === 0 ? "✅" : `⚠️ Exit ${exitCode}`;
+    await ctx.api.editMessageText(
+      ctx.chat.id,
+      msg.message_id,
+      `${status}\n<pre>${output.substring(0, 3500)}</pre>`,
+      { parse_mode: "HTML" }
+    );
+  } catch (error: any) {
+    await ctx.api.editMessageText(ctx.chat.id, msg.message_id, `❌ ${error.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("health", async (ctx) => {
+  if (!isAllowed(ctx.from?.id || 0)) return;
+  const msg = await ctx.reply("🔍 Checking Pi Worker...");
+  try {
+    const { stdout } = await sshExec("incus exec cr-server-new -- curl -s http://localhost:8787/health");
+    const health = JSON.parse(stdout);
+    const statusEmoji = health.status === "ok" ? "✅" : "⚠️";
+    await ctx.api.editMessageText(
+      ctx.chat.id,
+      msg.message_id,
+      `${statusEmoji} <b>Pi Worker</b>\n` +
+        `Status: ${health.status}\n` +
+        `Uptime: ${Math.floor(health.uptime / 60)}m\n` +
+        `Task Loop: ${health.taskLoop ? "✅" : "❌"}\n` +
+        `Improvement Loop: ${health.improvementLoop ? "✅" : "❌"}`,
+      { parse_mode: "HTML" }
+    );
+  } catch (error: any) {
+    await ctx.api.editMessageText(ctx.chat.id, msg.message_id, `❌ Pi Worker unreachable: ${error.message}`);
+  }
 });

-bot.command("cancel", handleCancel);
-bot.command("stop", handleCancel);
+bot.command("asana", async (ctx) => {
+  if (!isAllowed(ctx.from?.id || 0)) return;
+  const msg = await ctx.reply("🔍 Fetching Asana tasks...");
+  try {
+    const token = process.env.ASANA_ACCESS_TOKEN;
+    if (!token) {
+      await ctx.api.editMessageText(ctx.chat.id, msg.message_id, "❌ No ASANA_ACCESS_TOKEN");
+      return;
+    }

-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.");
-}
+    const res = await fetch(
+      "https://app.asana.com/api/1.0/user_task_lists/1212299150915258/tasks?opt_fields=name,completed,assignee_status&limit=20",
+      { headers: { Authorization: `Bearer ${token}` } }
+    );
+    const data = (await res.json()) as any;
+    const tasks = data.data || [];
+    const open = tasks.filter((t: any) => !t.completed);

-// ─── Message handler ───
-bot.on("message:text", (ctx) => {
+    let text = `📋 <b>My Tasks</b> (${open.length} open)\n\n`;
+    for (const t of open.slice(0, 15)) {
+      text += `• ${t.name.substring(0, 60)}\n`;
+    }
+    if (open.length > 15) text += `\n... and ${open.length - 15} more`;
+
+    await ctx.api.editMessageText(ctx.chat.id, msg.message_id, text, { parse_mode: "HTML" });
+  } catch (error: any) {
+    await ctx.api.editMessageText(ctx.chat.id, msg.message_id, `❌ Asana error: ${error.message}`);
+  }
+});
+
+bot.command("clear", async (ctx) => {
+  if (!isAllowed(ctx.from?.id || 0)) return;
  const userId = ctx.from!.id;
-  const msg = ctx.message!.text;
+  const session = getSessionPath(userId);
+  try {
+    unlinkSync(session);
+  } catch {}
+  await ctx.reply("🧹 Session cleared.");
+});

-  const prev = activeRequests.get(userId);
-  if (prev) {
-    prev.abort();
-    console.log(`⏹️ [${userId}] cancelled previous request`);
+bot.command("stop", async (ctx) => {
+  const userId = ctx.from?.id || 0;
+  const controller = activeProcesses.get(userId);
+  if (controller) {
+    controller.abort();
+    activeProcesses.delete(userId);
+    await ctx.reply("🛑 Stopped.");
+  } else {
+    await ctx.reply("ℹ️ Nothing running.");
+  }
+});
+
+bot.command("cancel", async (ctx) => {
+  const userId = ctx.from?.id || 0;
+  const controller = activeProcesses.get(userId);
+  if (controller) {
+    controller.abort();
+    activeProcesses.delete(userId);
+    await ctx.reply("🛑 Cancelled.");
+  } else {
+    await ctx.reply("ℹ️ Nothing running.");
+  }
+});
+
+// ─── Natural language messages ───
+bot.on("message:text", async (ctx) => {
+  const userId = ctx.from?.id || 0;
+  if (!isAllowed(userId)) return;
+  if (!checkRateLimit(userId)) {
+    await ctx.reply("⏳ Slow down — rate limit reached.");
+    return;
+  }
+
+  const msg = ctx.message.text;
+
+  // Block concurrent tasks per user
+  if (activeProcesses.has(userId)) {
+    await ctx.reply("⏳ Already processing a task. Use /stop to cancel it first.");
+    return;
+  }
+
+  // Session size management
+  const sessionInfo = checkSession(userId);
+  if (sessionInfo.sizeKB > SESSION_WARN_KB) {
+    if (sessionInfo.sizeKB > SESSION_MAX_KB) {
+      try {
+        unlinkSync(getSessionPath(userId));
+      } catch {}
+      await ctx.reply(
+        `🧹 Session auto-cleared (${sessionInfo.sizeKB}KB exceeded ${SESSION_MAX_KB}KB limit). Starting fresh.`
+      );
+    } else {
+      await ctx.reply(`⚠️ Session is ${sessionInfo.sizeKB}KB/${SESSION_MAX_KB}KB. Consider /clear soon.`);
+    }
  }

  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(() => {});
+  activeProcesses.set(userId, controller);

  try {
-    const reply = await callPiStreaming(userId, msg, controller.signal, ctx);
-    if (controller.signal.aborted) return;
+    const response = await callPiStreaming(userId, msg, controller.signal, ctx);

-    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."));
+    // Send response in chunks (Telegram 4096 char limit)
+    const maxLen = 4000;
+    for (let i = 0; i < response.length; i += maxLen) {
+      const chunk = response.substring(i, i + maxLen);
+      await ctx.reply(chunk);
    }
-  } catch (err: any) {
+  } catch (error: any) {
    if (!controller.signal.aborted) {
-      console.error(`❌ [${userId}]`, err.message);
-      await ctx.reply(`❌ ${err.message}`).catch(() => {});
+      await ctx.reply(`❌ Error: ${error.message}`);
    }
  } finally {
-    clearInterval(typingInterval);
-    if (activeRequests.get(userId) === controller) activeRequests.delete(userId);
+    activeProcesses.delete(userId);
  }
-}
-
-// ─── Polling loop ───
-bot.catch((err) => {
-  console.error("❌ Middleware error:", err.message || err);
 });

-let polling = true;
-let offset = 0;
+// ─── Error handler ───
+bot.catch((err) => {
+  console.error("Bot error:", err.message || err);
+});

-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 {}
+// ─── Startup ───
+console.log("🤖 CharityRight DevOps Bot starting...");

-  await bot.init();
-  console.log(`✅ Bot is live — polling started (model: ${PI_MODEL})`);
+// Clear stale updates
+await bot.api.getUpdates({ offset: -1, limit: 1 }).catch(() => {});

-  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));
-      }
-    }
-  }
-}
+bot.start({
+  onStart: () => console.log("✅ Bot online — polling for messages"),
+  drop_pending_update: true,
+});

-console.log("🚀 Bot starting...");
-pollLoop();
-
-const stop = () => { polling = false; };
-process.on("SIGINT", () => { stop(); process.exit(0); });
-process.on("SIGTERM", () => { stop(); process.exit(0); });
+// Cleanup on exit
+const cleanup = () => {
+  console.log("🛑 Shutting down...");
+  for (const [, controller] of activeProcesses) controller.abort();
+  bot.stop();
+  process.exit(0);
+};
+process.on("SIGINT", cleanup);
+process.on("SIGTERM", cleanup);