telegram-bot: rewrite to use Pi CLI with streaming JSON output

- Replaced direct Anthropic API with Pi CLI (uses platform auth, no API key needed)
- Model: claude-sonnet-4-6
- Streaming JSON mode with live progress updates in Telegram
- Per-user session files via Pi's --session/--continue
- Auto-recovery from corrupted sessions
- 5 min timeout, not-lazy system prompt

telegram-bot/index.ts

@@ -1,19 +1,17 @@
 import { Bot, Context } from "grammy";
-import { run } from "@grammyjs/runner";
-import Anthropic from "@anthropic-ai/sdk";
-import { readFileSync } from "fs";
+import { readFileSync, existsSync, mkdirSync } from "fs";
 import { resolve } from "path";
 
 // ─── Config ───
 const BOT_TOKEN = process.env.TELEGRAM_BOT_TOKEN!;
-const ANTHROPIC_KEY = process.env.ANTHROPIC_API_KEY!;
 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 (!ANTHROPIC_KEY) throw new Error("ANTHROPIC_API_KEY missing from .env");
 if (!SSH_HOST) throw new Error("SSH_HOST missing from .env");
 
 // ─── Load infra context ───
@@ -26,119 +24,237 @@ try {
 
 // ─── Init ───
 const bot = new Bot(BOT_TOKEN);
-const anthropic = new Anthropic({ apiKey: ANTHROPIC_KEY });
 
-// Per-user conversation history
-const conversations = new Map<number, Anthropic.MessageParam[]>();
-const MAX_HISTORY = 30; // raw entries before compression triggers
-const MAX_TOKENS_ESTIMATE = 12000; // compress when estimated tokens exceed this
+// ─── Session directory ───
+const SESSIONS_DIR = resolve(__dirname, "sessions");
+if (!existsSync(SESSIONS_DIR)) mkdirSync(SESSIONS_DIR, { recursive: true });
 
-// Per-user abort controllers — cancel stale requests
-const activeRequests = new Map<number, AbortController>();
-
-// ─── Context compression ───
-function estimateTokens(messages: Anthropic.MessageParam[]): number {
-  let chars = 0;
-  for (const msg of messages) {
-    if (typeof msg.content === "string") {
-      chars += msg.content.length;
-    } else if (Array.isArray(msg.content)) {
-      for (const block of msg.content) {
-        if ("text" in block) chars += (block as any).text.length;
-        else if ("content" in block) chars += String((block as any).content).length;
-        else chars += 100; // tool_use blocks etc
-      }
-    }
-  }
-  return Math.ceil(chars / 4); // rough chars-to-tokens
+function sessionPath(userId: number): string {
+  return resolve(SESSIONS_DIR, `user-${userId}.jsonl`);
 }
 
-async function compressHistory(userId: number, history: Anthropic.MessageParam[]): Promise<void> {
-  if (history.length < 10 || estimateTokens(history) < MAX_TOKENS_ESTIMATE) return;
+// ─── Active requests ───
+const activeRequests = new Map<number, AbortController>();
 
-  // Keep last 6 messages intact (current conversation), compress the rest
-  const toCompress = history.slice(0, -6);
-  const toKeep = history.slice(-6);
+// ─── 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.
 
-  // Build a text summary of old messages
-  const summaryParts: string[] = [];
-  for (const msg of toCompress) {
-    if (typeof msg.content === "string") {
-      const role = msg.role === "user" ? "User" : "Assistant";
-      summaryParts.push(`${role}: ${msg.content.slice(0, 200)}`);
-    } else if (Array.isArray(msg.content)) {
-      for (const block of msg.content) {
-        if ("text" in block && (block as any).text) {
-          const role = msg.role === "user" ? "User" : "Assistant";
-          summaryParts.push(`${role}: ${(block as any).text.slice(0, 200)}`);
-        } else if ("content" in block && (block as any).type === "tool_result") {
-          summaryParts.push(`[cmd output: ${String((block as any).content).slice(0, 100)}]`);
-        } else if ((block as any).type === "tool_use") {
-          summaryParts.push(`[ran: ${JSON.stringify((block as any).input).slice(0, 100)}]`);
+${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.`;
+
+// ─── 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);
+  const hasSession = existsSync(session);
+
+  const args = [
+    PI_BIN, "-p",
+    "--model", PI_MODEL,
+    "--tools", "bash",
+    "--mode", "json",
+    "--system-prompt", SYSTEM_PROMPT,
+    "--session", session,
+  ];
+
+  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",
+    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 300s`);
+    proc.kill();
+  }, 300_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 {}
+  }
+
+  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 call started — show what command is running
+          if (event.type === "tool_call") {
+            const input = event.tool_call?.input;
+            if (input?.command) {
+              currentToolCmd = input.command;
+              const short = currentToolCmd.length > 80 ? currentToolCmd.slice(0, 80) + "..." : currentToolCmd;
+              console.log(`🔧 [${userId}] ${short}`);
+              await sendProgress(short);
+            }
+          }
+
+          // Tool result — show brief output
+          if (event.type === "tool_result") {
+            const output = event.tool_result?.stdout || event.tool_result?.content || "";
+            if (output && typeof output === "string") {
+              const brief = output.split("\n")[0].slice(0, 60);
+              if (brief) await sendProgress(`→ ${brief}`);
+            }
+          }
+
+          // Accumulate assistant text from message_end or text_end events
+          if (event.type === "message_end" && event.message?.role === "assistant") {
+            const content = event.message.content;
+            if (Array.isArray(content)) {
+              for (const block of content) {
+                if (block.type === "text" && block.text) {
+                  finalText += (finalText ? "\n" : "") + block.text;
+                }
+              }
+            } else if (typeof content === "string") {
+              finalText += (finalText ? "\n" : "") + content;
+            }
+          }
+
+          // Also catch turn_end which has the final message
+          if (event.type === "turn_end" && event.message?.role === "assistant") {
+            const content = event.message.content;
+            if (Array.isArray(content)) {
+              // Reset finalText to avoid duplication from message_end
+              const turnText = content
+                .filter((b: any) => b.type === "text" && b.text)
+                .map((b: any) => b.text)
+                .join("\n");
+              if (turnText) finalText = turnText;
+            }
+          }
+
+          // agent_end — extract final text from all assistant messages
+          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) {
+              // Use last assistant text block as final answer
+              finalText = allText[allText.length - 1];
+            }
+          }
+
+        } catch {
+          // Not valid JSON, skip
         }
       }
     }
+  } catch (err: any) {
+    if (!signal.aborted) console.error(`❌ [${userId}] Stream error:`, err.message);
   }
 
-  const compressedSummary = summaryParts.join("\n").slice(0, 2000);
+  clearTimeout(timer);
+  signal.removeEventListener("abort", abortHandler);
 
-  // Replace history with compressed summary + recent messages
-  history.length = 0;
-  history.push({
-    role: "user",
-    content: `[Previous conversation summary]\n${compressedSummary}\n[End summary — continue from here]`,
-  });
-  history.push({
-    role: "assistant",
-    content: "Understood, I have the context from our previous conversation. How can I help?",
-  });
-  history.push(...toKeep);
+  const stderr = await stderrPromise;
+  const exitCode = await proc.exited;
 
-  console.log(`📦 [${userId}] Compressed history: ${toCompress.length + toKeep.length} → ${history.length} entries (~${estimateTokens(history)} tokens)`);
+  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"))) {
+      console.log(`🔄 [${userId}] Corrupted session — clearing and retrying...`);
+      try { require("fs").unlinkSync(session); } catch {}
+      return callPiStreaming(userId, message, signal, ctx);
+    }
+    const exitMsg = exitCode === 143 ? "timed out (5 min limit)" : (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 ───
+// ─── Auth / Rate limit ───
 function isAuthorized(ctx: Context): boolean {
   if (ALLOWED_USERS.length === 0) return false;
   return ALLOWED_USERS.includes(ctx.from?.id || 0);
 }
 
-// ─── Rate limiting ───
 const rateLimits = new Map<number, number[]>();
-const RATE_LIMIT_WINDOW = 60_000;
-const RATE_LIMIT_MAX = 15;
-
 function isRateLimited(userId: number): boolean {
   const now = Date.now();
-  const timestamps = (rateLimits.get(userId) || []).filter(t => now - t < RATE_LIMIT_WINDOW);
-  timestamps.push(now);
-  rateLimits.set(userId, timestamps);
-  return timestamps.length > RATE_LIMIT_MAX;
+  const ts = (rateLimits.get(userId) || []).filter(t => now - t < 60_000);
+  ts.push(now);
+  rateLimits.set(userId, ts);
+  return ts.length > 15;
 }
 
-// ─── Blocked commands ───
-const BLOCKED_PATTERNS = [
-  /rm\s+-rf\s+\/(?!\w)/i,
-  /mkfs\./i,
-  /dd\s+if=.*of=\/dev/i,
-  /:\(\)\s*\{\s*:\|:/,
-];
-
-function isDangerous(cmd: string): string | null {
-  for (const pattern of BLOCKED_PATTERNS) {
-    if (pattern.test(cmd)) return `⛔ Blocked dangerous pattern`;
-  }
-  return null;
-}
-
-// ─── SSH exec ───
+// ─── 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(),
@@ -146,12 +262,10 @@ async function sshExec(command: string, timeoutSec = 30): Promise<string> {
   ]);
   clearTimeout(timer);
   await proc.exited;
-
   const output = (stdout + stderr).trim();
   return output.length > 2000 ? output.slice(0, 2000) + "\n...(truncated)" : output;
 }
 
-// ─── Promise with timeout helper ───
 function withTimeout<T>(promise: Promise<T>, ms: number, label: string): Promise<T> {
   return Promise.race([
     promise,
@@ -161,312 +275,85 @@ function withTimeout<T>(promise: Promise<T>, ms: number, label: string): Promise
   ]);
 }
 
-// ─── System prompt ───
-const SYSTEM_PROMPT = `You are a DevOps assistant on Telegram. Connected to a live server via SSH.
-
-${infraContext}
-
-ARCHITECTURE:
-- The bot runs on a LOCAL LAPTOP (macOS). It is NOT on the server.
-- run_command executes via SSH on the PRIMARY HOST (${SSH_HOST}).
-- 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, you MUST wrap with: incus exec cr-server-new -- <command>
-- Example: incus exec cr-server-new -- docker ps
-- Example: incus exec cr-server-new -- bash /opt/enthuse-db-sync-v2/run-sync.sh
-- To query Postgres: 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 — they won't work.
-
-KEY PATHS (inside cr-server-new):
-- /opt/enthuse-db-sync-v2/ — Enthuse sync
-- /opt/launchgood-sync/ — LaunchGood sync  
-- /opt/charityright-sync/ — CharityRight sync
-- Postgres container: dokploy-migrated-cr-postgres-data
-- Docker network: dokploy-network
-
-RULES:
-- Be VERY concise — this is Telegram, not a terminal.
-- Keep command output short. Use | head, | tail, grep to limit output.
-- For destructive commands, WARN and ask for confirmation first.
-- ALWAYS end your response with a text message to the user. Never end on a tool call with no text summary.`;
-
-// ─── Claude tools ───
-const tools: Anthropic.Tool[] = [
-  {
-    name: "run_command",
-    description: "Run a shell command on the server via SSH. Keep commands short and focused. Always pipe through head/tail/grep to limit output.",
-    input_schema: {
-      type: "object" as const,
-      properties: {
-        command: { type: "string", description: "Shell command" },
-      },
-      required: ["command"],
-    },
-  },
-];
-
-// ─── Chat with Claude — with cancellation + timeout ───
-// Global typing context so SSH exec can trigger typing too
-let activeCtx: Context | null = null;
-
-async function chat(userId: number, message: string, signal: AbortSignal): Promise<string> {
-  if (!conversations.has(userId)) conversations.set(userId, []);
-  const history = conversations.get(userId)!;
-
-  // Snapshot history length before we start — rollback to here on error
-  const historySnapshot = history.length;
-
-  history.push({ role: "user", content: message });
-
-  // Compress old history if getting too large
-  await compressHistory(userId, history);
-  while (history.length > MAX_HISTORY) history.shift();
-
-  try {
-    return await _chatInner(userId, history, signal);
-  } catch (err) {
-    // Rollback history to prevent corruption (dangling tool_use without tool_result)
-    history.splice(historySnapshot);
-    throw err;
-  }
-}
-
-async function _chatInner(userId: number, history: Anthropic.MessageParam[], signal: AbortSignal): Promise<string> {
-  let rounds = 0;
-  const MAX_ROUNDS = 10;
-
-  let response = await withTimeout(
-    anthropic.messages.create({
-      model: "claude-sonnet-4-20250514",
-      max_tokens: 1024,
-      system: SYSTEM_PROMPT,
-      tools,
-      messages: history,
-    }),
-    90_000,
-    "Claude API"
-  );
-
-  while (response.stop_reason === "tool_use" && rounds < MAX_ROUNDS) {
-    if (signal.aborted) throw new Error("Cancelled — new message received");
-    rounds++;
-
-    const assistantContent = response.content;
-    history.push({ role: "assistant", content: assistantContent });
-
-    const toolResults: Anthropic.ToolResultBlockParam[] = [];
-
-    for (const block of assistantContent) {
-      if (block.type === "tool_use") {
-        if (signal.aborted) throw new Error("Cancelled");
-        const input = block.input as { command: string };
-        console.log(`🔧 [${userId}] ${input.command}`);
-
-        const danger = isDangerous(input.command);
-        if (danger) {
-          toolResults.push({ type: "tool_result", tool_use_id: block.id, content: danger, is_error: true });
-          continue;
-        }
-        try {
-          // Show what's running
-          if (activeCtx) {
-            const shortCmd = input.command.length > 80 ? input.command.slice(0, 80) + "..." : input.command;
-            await activeCtx.reply(`⚙️ \`${shortCmd}\``, { parse_mode: "Markdown" }).catch(() => {});
-          }
-          const output = await sshExec(input.command);
-          toolResults.push({ type: "tool_result", tool_use_id: block.id, content: output || "(no output)" });
-        } catch (err: any) {
-          toolResults.push({ type: "tool_result", tool_use_id: block.id, content: `ERROR: ${err.message}`, is_error: true });
-        }
-      }
-    }
-
-    // Always push tool_results immediately after assistant tool_use — keeps history valid
-    history.push({ role: "user", content: toolResults });
-
-    response = await withTimeout(
-      anthropic.messages.create({
-        model: "claude-sonnet-4-20250514",
-        max_tokens: 1024,
-        system: SYSTEM_PROMPT,
-        tools,
-        messages: history,
-      }),
-      90_000,
-      "Claude API"
-    );
-  }
-
-  if (response.stop_reason === "tool_use") {
-    // Max rounds — force summary
-    const assistantContent = response.content;
-    history.push({ role: "assistant", content: assistantContent });
-
-    const toolResults: Anthropic.ToolResultBlockParam[] = [];
-    for (const block of assistantContent) {
-      if (block.type === "tool_use") {
-        toolResults.push({ type: "tool_result", tool_use_id: block.id, content: "Skipped — max rounds reached. Summarize what you have so far.", is_error: true });
-      }
-    }
-    history.push({ role: "user", content: toolResults });
-
-    const finalResponse = await withTimeout(
-      anthropic.messages.create({
-        model: "claude-sonnet-4-20250514",
-        max_tokens: 1024,
-        system: SYSTEM_PROMPT,
-        messages: history,
-      }),
-      60_000,
-      "Claude summary"
-    );
-
-    const text = finalResponse.content
-      .filter((b): b is Anthropic.TextBlock => b.type === "text")
-      .map((b) => b.text)
-      .join("\n")
-      .trim();
-    history.push({ role: "assistant", content: text || "Done." });
-    return text || "Done (max rounds reached).";
-  }
-
-  const text = response.content
-    .filter((b): b is Anthropic.TextBlock => b.type === "text")
-    .map((b) => b.text)
-    .join("\n")
-    .trim();
-
-  if (!text) {
-    // Claude gave no text — ask it to summarize
-    history.push({ role: "assistant", content: response.content });
-    history.push({ role: "user", content: "Now give me a brief summary of what you found. Text only, no tool calls." });
-    const followUp = await withTimeout(
-      anthropic.messages.create({
-        model: "claude-sonnet-4-20250514",
-        max_tokens: 1024,
-        system: SYSTEM_PROMPT,
-        messages: history,
-      }),
-      60_000,
-      "Claude follow-up"
-    );
-    const followUpText = followUp.content
-      .filter((b): b is Anthropic.TextBlock => b.type === "text")
-      .map((b) => b.text)
-      .join("\n")
-      .trim();
-    history.push({ role: "assistant", content: followUpText || "Done." });
-    return followUpText || "Done.";
-  }
-
-  history.push({ role: "assistant", content: text });
-  return text;
-}
-
-// ─── Security middleware ───
+// ─── 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.");
-  }
+  if (ctx.from && isRateLimited(ctx.from.id)) return ctx.reply("⏳ Slow down.");
   await next();
 });
 
 // ─── Commands ───
 bot.command("start", (ctx) =>
-  ctx.reply(
-    "🤖 CR Management Bot\n\n" +
-    "• Chat naturally — I understand your infra\n" +
-    "• /status — server health\n" +
-    "• /ssh <cmd> — run command directly\n" +
-    "• /cancel — stop current task\n" +
-    "• /clear — reset conversation"
-  )
+  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• /cancel — stop current task\n• /clear — reset conversation")
 );
 
 bot.command("status", async (ctx) => {
   await ctx.reply("🔍 Checking...");
   try {
-    const output = 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(output);
-  } catch (err: any) {
-    return ctx.reply(`❌ ${err.message}`);
-  }
+    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 danger = isDangerous(cmd);
-  if (danger) return ctx.reply(danger);
-  try {
-    const output = await withTimeout(sshExec(cmd, 10), 15_000, "SSH");
-    return ctx.reply(output || "(no output)");
-  } catch (err: any) {
-    return ctx.reply(`❌ ${err.message}`);
+  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) => {
-  conversations.delete(ctx.from!.id);
+  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", (ctx) => {
-  const controller = activeRequests.get(ctx.from!.id);
-  if (controller) {
-    controller.abort();
-    activeRequests.delete(ctx.from!.id);
-    return ctx.reply("🛑 Cancelled.");
-  }
+  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.");
 });
 
-// ─── Free text → Claude (fire-and-forget, never blocks bot) ───
+// ─── Message handler ───
 bot.on("message:text", (ctx) => {
   const userId = ctx.from!.id;
   const msg = ctx.message!.text;
 
-  // Cancel any in-flight request for this user
   const prev = activeRequests.get(userId);
   if (prev) {
     prev.abort();
-    // Clear corrupted history to avoid tool_result mismatch
-    conversations.delete(userId);
-    console.log(`⏹️ [${userId}] cancelled previous + cleared history`);
+    console.log(`⏹️ [${userId}] cancelled previous request`);
   }
 
   const controller = new AbortController();
   activeRequests.set(userId, controller);
-
-  // Fire and forget — bot handler returns immediately
   processMessage(ctx, userId, msg, controller);
 });
 
 async function processMessage(ctx: Context, userId: number, msg: string, controller: AbortController) {
-  const startTime = Date.now();
-  activeCtx = ctx;
+  const start = Date.now();
 
-  // Keep typing every 3s — Telegram typing indicator expires after ~5s
-  const typingInterval = setInterval(async () => {
-    if (controller.signal.aborted) return;
-    ctx.replyWithChatAction("typing").catch(() => {});
+  const typingInterval = setInterval(() => {
+    if (!controller.signal.aborted) ctx.replyWithChatAction("typing").catch(() => {});
   }, 3000);
   ctx.replyWithChatAction("typing").catch(() => {});
 
   try {
-    const reply = await chat(userId, msg, controller.signal);
+    const reply = await callPiStreaming(userId, msg, controller.signal, ctx);
     if (controller.signal.aborted) return;
 
-    const elapsed = ((Date.now() - startTime) / 1000).toFixed(1);
+    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)"];
@@ -480,17 +367,54 @@ async function processMessage(ctx: Context, userId: number, msg: string, control
     }
   } finally {
     clearInterval(typingInterval);
-    activeCtx = null;
     if (activeRequests.get(userId) === controller) activeRequests.delete(userId);
   }
 }
 
-// ─── Launch with concurrent runner (never blocks on slow handlers) ───
-console.log("🚀 Bot starting...");
-const runner = run(bot);
-console.log(`✅ Bot is live (concurrent mode)`);
+// ─── Polling loop ───
+bot.catch((err) => {
+  console.error("❌ Middleware error:", err.message || err);
+});
 
-// Graceful shutdown
-const stop = () => { runner.isRunning() && runner.stop(); };
-process.on("SIGINT", stop);
-process.on("SIGTERM", stop);
+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); });