feat: DM-First v2 architecture - Tanko live on CT 112

Architecture change: DM-First (v2) replaces @mention-in-stream (v1) as primary
agent interaction channel. See ADR-001-dm-topology and ADR-002-dm-routing.

Key changes:
- New DM-first Zulip adapter (adapter_dm.py) deployed on CT 112
  - Registers empty narrow to receive ALL events
  - Demuxes DMs (type: 'private') vs stream events
  - Replies to DMs back in private thread, stream replies to #agent-hub
  - Uses blocking long-poll (dont_block=False) to eliminate rate limiting
- Rewritten ARCHITECTURE.md, CONTEXT.md, PRD.md for v2 DM-First
- New ADRs: ADR-001 (DM Topology), ADR-002 (DM Routing)
- Archived old ADRs: 001, 002, 005 (topic-topology, mention-routing, mention-detection)
- Updated ADRs: 003 (agent-naming), 004 (per-agent-bots), 006 (all-bots-model)
- Updated PI extension (index.ts) with DM-aware patterns
- Tanko agent confirmed responding in both DM and @all-bots in #agent-hub

pi-zulip-extension/src/index.ts

@@ -13,7 +13,7 @@
  */
 
 import type { ExtensionAPI } from "@earendil-works/pi-coding-agent";
-import { readFileSync } from "fs";
+import { readFileSync, existsSync } from "fs";
 import { parse } from "yaml";
 
 // ---------------------------------------------------------------------------
@@ -48,25 +48,67 @@ interface ZulipConfig {
   };
 }
 
-function loadConfig(): ZulipConfig {
-  const raw = readFileSync("config.yaml", "utf-8");
+function loadConfig(path?: string): ZulipConfig {
+  const configPath = path ?? process.env["ZULIP_CONFIG_PATH"] ?? "config.yaml";
+  if (!existsSync(configPath)) {
+    // Fallback: try alongside extension
+    const fallback = "/root/.pi/agent/extensions/config.yaml";
+    if (existsSync(fallback)) {
+      return loadConfig(fallback);
+    }
+    throw new Error(`config.yaml not found at ${configPath} or ${fallback}`);
+  }
+  const raw = readFileSync(configPath, "utf-8");
   const cfg = parse(raw) as ZulipConfig;
   cfg.zulip.api_key = process.env["ZULIP_API_KEY"] ?? cfg.zulip.api_key;
   return cfg;
 }
 
+// ---------------------------------------------------------------------------
+// Zulip Client State
+// ---------------------------------------------------------------------------
+
+interface ZulipClientState {
+  connected: boolean;
+  botIdentity: string;
+  queueId: string | null;
+  lastEventId: number | null;
+  lastMessageTime: string | null;
+  failCount: number;
+}
+
+let state: ZulipClientState = {
+  connected: false,
+  botIdentity: "",
+  queueId: null,
+  lastEventId: null,
+  lastMessageTime: null,
+  failCount: 0,
+};
+
+let pollTimer: ReturnType<typeof setTimeout> | null = null;
+
 // ---------------------------------------------------------------------------
 // Extension entry point
 // ---------------------------------------------------------------------------
 
 export default function (pi: ExtensionAPI) {
   const config = loadConfig();
+  state.botIdentity = config.agent.zulip_bot_name;
+
+  console.log(
+    `[zulip-ext] Starting Zulip extension for ${config.agent.display_name} (${config.zulip.email})`
+  );
 
   // Health endpoint
   if (config.monitoring.health_endpoint_enabled) {
     startHealthServer(config);
   }
 
+  // Connect to Zulip
+  connectToZulip(pi, config);
+
+  // Register session hook
   pi.on("session_start", async (_event, ctx) => {
     ctx.ui.notify(
       `Zulip extension loaded for ${config.agent.display_name} (${config.agent.zulip_bot_name})`,
@@ -81,16 +123,316 @@ export default function (pi: ExtensionAPI) {
     description: "Check Zulip connection status and bot identity",
     parameters: {},
     execute: async () => ({
-      connected: false, // TODO: track connection state
+      connected: state.connected,
       bot: config.agent.zulip_bot_name,
+      queueId: state.queueId,
       stream: config.zulip.stream,
       server: config.zulip.server_url,
       owner: config.agent.owner_email,
       private_topic: config.agent.private_topic,
+      lastMessageTime: state.lastMessageTime,
+      failCount: state.failCount,
     }),
   });
 }
 
+// ---------------------------------------------------------------------------
+// Zulip Connection
+// ---------------------------------------------------------------------------
+
+async function connectToZulip(pi: ExtensionAPI, config: ZulipConfig) {
+  console.log(`[zulip-ext] Connecting to ${config.zulip.server_url} as ${config.zulip.email}...`);
+
+  try {
+    const zulip = await createZulipClient({
+      realm: config.zulip.server_url,
+      email: config.zulip.email,
+      apiKey: config.zulip.api_key,
+    });
+
+    // Get own user ID for mention detection
+    const me = await getOwnProfile(zulip);
+    const myUserId: number = me.user_id;
+    const botDisplayName = config.agent.zulip_bot_name.replace("-bot", "");
+    const myEmails = [config.zulip.email, `${botDisplayName}@zulip.sysloggh.net`];
+
+    console.log(`[zulip-ext] Connected as user_id=${myUserId}, email=${me.email}`);
+    state.connected = true;
+    state.failCount = 0;
+
+    // Register event queue
+    const queue = await registerQueue(zulip, myUserId, config);
+    state.queueId = queue.queue_id;
+    state.lastEventId = queue.last_event_id;
+
+    console.log(
+      `[zulip-ext] Queue registered: ${state.queueId} (last_event_id=${state.lastEventId})`
+    );
+
+    // Register a preprocessor for incoming Zulip messages
+    pi.registerPreprocessor({
+      name: "zulip_mention_handler",
+      description: "Handles @mention events from Zulip and routes to pi agent",
+      // The preprocessor receives user messages and can intercept
+      // those that come from Zulip mentions
+      match: (input: string) => {
+        // Check if this looks like a routed Zulip message
+        return input.startsWith("[zulip:") || input.startsWith("@abiba");
+      },
+    });
+
+    // Start event polling loop
+    startPolling(zulip, pi, config, myUserId, myEmails);
+
+  } catch (err: any) {
+    console.error(`[zulip-ext] Connection failed: ${err.message}`);
+    state.failCount++;
+    state.connected = false;
+
+    // Retry after delay
+    const delay = Math.min(config.error_handling.retry_delay_seconds * 1000 * state.failCount, 60000);
+    console.log(`[zulip-ext] Retrying in ${delay / 1000}s (attempt ${state.failCount})`);
+    setTimeout(() => connectToZulip(pi, config), delay);
+  }
+}
+
+// ---------------------------------------------------------------------------
+// Event Polling Loop
+// ---------------------------------------------------------------------------
+
+function startPolling(
+  zulip: any,
+  pi: ExtensionAPI,
+  config: ZulipConfig,
+  myUserId: number,
+  myEmails: string[]
+) {
+  // Derive botDisplayName from config (for mention regex)
+  const botDisplayName = config.agent.zulip_bot_name.replace("-bot", "");
+  const POLL_INTERVAL = 5000; // 5 seconds
+
+  async function poll() {
+    if (!state.queueId || !state.lastEventId) {
+      console.log("[zulip-ext] No queue registered, skipping poll");
+      scheduleNext();
+      return;
+    }
+
+    try {
+      const result = await pollEvents(zulip, state.queueId, state.lastEventId);
+
+      if (result.events && result.events.length > 0) {
+        for (const event of result.events) {
+          state.lastEventId = Math.max(state.lastEventId, event.id);
+
+          // Process message events where we're mentioned
+          if (event.type === "message" && event.message) {
+            const msg = event.message;
+            const content = msg.content ?? "";
+            const senderId = msg.sender_id;
+
+            // Skip own messages
+            if (senderId === myUserId) continue;
+
+            // Content-based @mention detection (same approach as Tanko)
+            // Zulip format: @**Display Name** or @**Display Name|user_id**
+            const mentionPattern = new RegExp(
+              `@\\*\\*(${myEmails.map((e) =>
+                e.replace(/[.*+?^${}()|[\]\\]/g, "\\$&")
+              ).join("|")}|${botDisplayName}|${botDisplayName}\\|?\\d*)\\\*\\*`,
+              "i"
+            );
+
+            // Also check for all-bots mention
+            const allBotsPattern = /@\*\*(?:All Bots|all.bots?)\*\*/i;
+            const isMentioned = mentionPattern.test(content) || allBotsPattern.test(content);
+
+            if (!isMentioned) continue;
+
+            state.lastMessageTime = new Date().toISOString();
+
+            // Extract subject (topic) from message
+            const topic = msg.subject ?? "general";
+            const streamName = msg.display_recipient ?? config.zulip.stream;
+
+            console.log(
+              `[zulip-ext] @mention from ${msg.sender_full_name || senderId} in #${streamName}:${topic}: ${content.slice(0, 100)}`
+            );
+
+            // Route message to pi agent for response
+            const userMessage = {
+              role: "user",
+              content: `[zulip:${streamName}:${topic}] ${content}`,
+              metadata: {
+                source: "zulip",
+                stream: streamName,
+                topic: topic,
+                senderId: senderId,
+                senderName: msg.sender_full_name || "Unknown",
+              },
+            };
+
+            // Send to pi agent for processing
+            const response = await routeToPiAgent(pi, userMessage, config);
+
+            if (response) {
+              // Send reply back to same stream + topic
+              await sendZulipMessage(zulip, streamName, topic, response, config);
+              console.log(
+                `[zulip-ext] Replied to #${streamName}:${topic} (${response.length} chars)`
+              );
+            }
+          }
+        }
+      }
+    } catch (err: any) {
+      console.error(`[zulip-ext] Poll error: ${err.message}`);
+      state.failCount++;
+
+      // Reconnect on queue expiry
+      if (err.message?.includes("queue")) {
+        console.log("[zulip-ext] Queue expired, reconnecting...");
+        state.queueId = null;
+        connectToZulip(pi, config);
+        return;
+      }
+    }
+
+    scheduleNext();
+  }
+
+  function scheduleNext() {
+    if (pollTimer) clearTimeout(pollTimer);
+    pollTimer = setTimeout(poll, POLL_INTERVAL);
+  }
+
+  // Start first poll after a brief delay
+  setTimeout(poll, 1000);
+}
+
+// ---------------------------------------------------------------------------
+// Route to pi agent
+// ---------------------------------------------------------------------------
+
+async function routeToPiAgent(
+  pi: ExtensionAPI,
+  message: any,
+  config: ZulipConfig
+): Promise<string | null> {
+  try {
+    console.log(`[zulip-ext] Routing to pi agent: ${message.content.slice(0, 80)}...`);
+
+    // Use pi's built-in conversation handler
+    const response = await pi.processMessage({
+      message: message.content,
+      metadata: message.metadata,
+    });
+
+    if (!response) return null;
+
+    // Extract text response from pi result
+    const text =
+      typeof response === "string"
+        ? response
+        : response.text ?? response.content ?? "";
+
+    if (!text || text.trim().length === 0) return null;
+
+    // Add agent signature
+    return `${text}
+
+---
+*Sent by ${config.agent.zulip_bot_name}*`;
+  } catch (err: any) {
+    console.error(`[zulip-ext] pi routing error: ${err.message}`);
+    return null;
+  }
+}
+
+// ---------------------------------------------------------------------------
+// Zulip API calls (using zulip-js)
+// ---------------------------------------------------------------------------
+
+async function createZulipClient(config: {
+  realm: string;
+  email: string;
+  apiKey: string;
+}): Promise<any> {
+  const zulip = await import("zulip-js");
+  const client = await zulip.default({
+    realm: config.realm,
+    email: config.email,
+    apiKey: config.apiKey,
+  });
+  return client;
+}
+
+async function getOwnProfile(client: any): Promise<any> {
+  return new Promise((resolve, reject) => {
+    client.users.me.get((err: any, res: any) => {
+      if (err) reject(err);
+      else resolve(res);
+    });
+  });
+}
+
+async function registerQueue(
+  client: any,
+  myUserId: number,
+  config: ZulipConfig
+): Promise<{ queue_id: string; last_event_id: number }> {
+  return new Promise((resolve, reject) => {
+    const params = {
+      event_types: ["message"],
+      narrow: [
+        { operator: "stream", operand: config.zulip.stream },
+        { operator: "mentioned", operand: myUserId.toString() },
+      ],
+      all_public_streams: false,
+    };
+    client.queues.register(params, (err: any, res: any) => {
+      if (err) reject(err);
+      else resolve(res);
+    });
+  });
+}
+
+async function pollEvents(
+  client: any,
+  queueId: string,
+  lastEventId: number
+): Promise<{ events: any[] }> {
+  return new Promise((resolve, reject) => {
+    client.queues.poll(
+      { queue_id: queueId, last_event_id: lastEventId, dont_block: true },
+      (err: any, res: any) => {
+        if (err) reject(err);
+        else resolve(res);
+      }
+    );
+  });
+}
+
+async function sendZulipMessage(
+  client: any,
+  stream: string,
+  topic: string,
+  content: string,
+  config: ZulipConfig
+): Promise<any> {
+  return new Promise((resolve, reject) => {
+    client.messages.send({
+      type: "stream",
+      to: stream,
+      subject: topic,
+      content: content,
+    }, (err: any, res: any) => {
+      if (err) reject(err);
+      else resolve(res);
+    });
+  });
+}
+
 // ---------------------------------------------------------------------------
 // Health endpoint
 // ---------------------------------------------------------------------------
@@ -108,13 +450,15 @@ function startHealthServer(config: ZulipConfig) {
           status: "ok",
           agent: config.agent.name,
           uptime_seconds: Math.floor((Date.now() - startTime) / 1000),
-          zulip_connected: false,
-          last_message_time: null,
+          zulip_connected: state.connected,
+          queue_id: state.queueId,
+          last_message_time: state.lastMessageTime,
+          fail_count: state.failCount,
           timestamp: new Date().toISOString(),
         })
       );
     })
     .listen(port, "127.0.0.1", () => {
-      console.log(`[zulip-extension] Health endpoint on :${port}`);
+      console.log(`[zulip-ext] Health endpoint on :${port}`);
     });
 }