SyslogSolution/zulip-platform-plugins
7
0
Fork 0
Code Issues 19 Pull Requests Actions Packages Projects Releases Wiki Activity

fix(tanko): adapter fixes, event logging, platform-based deploy.sh #21

Closed
kagentz-bot wants to merge 6 commits from feat/deploy-tanko into main
pull from: feat/deploy-tanko
merge into: :main
Conversation 6 Commits 6 Files Changed 19
+5237 -383
19 changed files with 5237 additions and 383 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/ARCHITECTURE.md
+116 -67
View File
@@ -1,110 +1,149 @@
# Zulip Multi-Platform Agent Communication — Architecture # Zulip Multi-Platform Agent Communication — Architecture
**Version:** 2.0 (DM-First)
**Date:** 2026-06-21
## Overview ## Overview
A production-ready system enabling 6 AI agents across 3 platforms (Hermes Python, Agent Zero, pi TypeScript) to communicate through Zulip via dedicated per-agent bot users.
A production-ready system enabling 6 AI agents across 3 platforms (Hermes Python, Agent Zero, PI TypeScript) to communicate through Zulip via **Direct Messages (DMs)**. Each agent's bot user receives DMs directly from human users, and all bots subscribe to `#agent-hub` only for `@all-bots` broadcasts.
## Core Design Decision: DM-First
| Old (v1) — @mention in Stream | New (v2) — DM-First |
|-------------------------------|---------------------|
| User @mentions bot in #agent-hub | User sends DM directly to bot |
| Bot detects via `mentioned_users` (unreliable in Zulip 12.0) | Bot detects `message.type === 'private'` (trivial) |
| Bot must capture origin topic for reply routing | No topic routing needed — Zulip DM thread is 1:1 |
| All 6 bots receive every #agent-hub message (noise) | Only the recipient bot receives the DM |
| Stream subscription required for all agents | DM requires no stream subscription |
## Architecture Diagram ## Architecture Diagram
``` ```
┌─────────────────────────────────────────────────────────────┐ ┌─────────────────────────────────────────────────────────────┐
│ Zulip Server (CT 117) │ │ Zulip Server (CT 117) │
│ chat.sysloggh.net:443 │ │ chat.sysloggh.net:443 │
│ │ │ │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌─── │ │ ┌──────────────────────────────────────────────────────┐ │
│ │tanko-bot │ │mumuni-bot│ │koonimo │ │koby-bot │ │... │ │ │ Each Agent Connects via TWO Channels: │ │
│ │ │ │ │ │-bot │ │ │ │ │
│ └────┬─────┘ └────┬─────┘ └────┬─────┘ └────┬─────┘ └─── │
└───────┼────────────┼────────────┼────────────┼──────────────┘
│ │ │ │ │ │ │ │
┌────▼────┐ ┌────▼────┐ ┌────▼────┐ ┌────▼────┐ │ │ DM (PRIMARY) #agent-hub (BROADCAST) │ │
│ Hermes │ │ Hermes │ │ Hermes │ │ Hermes │ │ │ ┌────────────────────┐ ┌─────────────────────┐ │ │
│ Plugin │ │ Plugin │ │ Plugin │ │ Plugin │ │ │ │ Chat with owner │ │ @all-bots detection │ │ │
│ (Python)│ │ (Python)│ │ (Python)│ │ (Python)│ │ │ │ No @mention parse │ │ Content-based regex │ │ │
│ │ │ │ │ │ │ │ │ │ │ No topic capture │ │ (no dedicated user) │ │ │
│ Tanko │ │ Mumuni │ │ Koonimo │ │ Koby │ │ │ │ Trivial detection │ │ 6 agents respond │ │ │
│ CT 112 │ │ CT 114 │ │ CT 113 │ │ CT 111 │ │ │ └────────────────────┘ └─────────────────────┘ │ │
└─────────┘ └─────────┘ └─────────┘ └─────────┘ │ └──────────────────────────────────────────────────────┘ │
│ │
│ Bot Users (6): tanko-bot, mumuni-bot, koonimo-bot, │
│ koby-bot, kagentz-bot, abiba-bot │
│ (No dedicated @all-bots bot — see ADR-006) │
└──────┬──────────┬──────────┬──────────┬──────────┬───────────┘
│ │ │ │ │
┌────▼────┐ ┌───▼───┐ ┌───▼───┐ ┌───▼───┐ ┌───▼───────┐
│ Hermes │ │Hermes │ │Hermes │ │ PI │ │ Agent Zero│
│ Plugin │ │Plugin │ │Plugin │ │Extens.│ │ Plugin │
│ (Python)│ │Python │ │Python │ │(TS) │ │ (Python) │
│ │ │ │ │ │ │ │ │ │
│ Tanko │ │Mumuni │ │Koonimo│ │ Abiba │ │ kagentz │
│ CT 112 │ │CT 114 │ │CT 113 │ │CT 100 │ │ CT 105 │
└─────────┘ └───────┘ └───────┘ └───────┘ └───────────┘
┌──────────────┐ ┌──────────────┐ Note: Koby (Hermes) also on a Hermes CT, not shown for brevity
│ Agent Zero │ │ pi │
│ Plugin │ │ Extension │
│ (Python) │ │ (TypeScript) │
│ │ │ │
│ kagentz │ │ abiba │
│ CT 105 │ │ CT 100 │
└──────────────┘ └──────────────┘
``` ```
## Message Flow (Normal) ## Message Flow: DM (Primary)
``` ```
User creates topic "project-x" in #agent-hub User opens DM to @tanko-bot in Zulip
└── User types: @tanko-bot analyze this data User types: "deploy the API update"
└── Zulip fires event to tanko-bot's event queue └── Zulip pushes event to tanko-bot's event queue ONLY
└── Hermes Zulip Plugin on CT 112 └── Hermes Zulip Plugin on CT 112
├── Detects: tanko-bot in mentioned_users ├── Checks: message.type === 'private' ✅
├── Strips @mention from message ├── No @mention parsing needed
├── Constructs MessageEvent ├── No topic/stream routing needed
└── Routes to Tanko agent via BasePlatformAdapter ├── Constructs MessageEvent with sender_id + content
└── Tanko agent processes, returns response └── Routes to Tanko agent for processing
└── Plugin posts to #agent-hub > project-x └── Tanko processes, returns response
└── Plugin sends DM reply via:
POST /api/v1/messages {
"type": "private",
"to": [sender_id],
"content": "Starting deployment..."
}
``` ```
## Message Flow (@all-bots) ## Message Flow: @all-bots Broadcast (Secondary)
``` ```
User types: @all-bots status report User posts in #agent-hub > any-topic:
└── Zulip fires event to ALL 6 bots (same narrow) "@**all-bots** status report please"
└── Each plugin independently: └── Zulip pushes event to ALL 6 bot event queues
├── Detects: All Bots in mentioned_users └── Each plugin on each CT independently:
├── Forwards to its agent ├── Checks: message.type === 'stream' ✅
└── Posts response to same topic ├── Checks: regex /@\*\*(?:all-bots|All Bots)\*\*/i.test(content) ✅
├── Ignores: all non-@all-bots stream messages
└── Forwards @all-bots content to its agent
└── Agent processes, returns response
└── Plugin posts to same #agent-hub > topic
└── 6 independent responses appear in the topic └── 6 independent responses appear in the topic
``` ```
## Plugin Communication Logic ## Plugin Communication Logic
- **Isolation**: Each agent (Tanko, Mumuni, etc.) runs as an independent process on its own Compute Target (CT). - **Isolation**: Each agent (Tanko, Mumuni, etc.) runs as an independent process on its own Compute Target (CT).
- **State**: Agents are stateless; all "memory" is handled via the RA-H OS Knowledge Graph and the Hermes persistent memory tool. - **State**: Agents are stateless; all "memory" is handled via the RA-H OS Knowledge Graph and the Hermes persistent memory tool.
- **Communication**: Agents communicate via Zulip. There is no direct A2A (Agent-to-Agent) messaging layer. Instead, each agent listens to the `#agent-hub` and responds to specific mentions or `@all-bots` pings. - **Communication**: Primary channel is Zulip DMs. Secondary broadcast channel is `#agent-hub` for `@all-bots` notifications only.
- **Swarm Development**: The "Swarm" refers to our collaborative development methodology where multiple agents/developers work on different components of the plugin simultaneously. - **No @mention routing**: DM-based routing eliminates the `mentioned_users` field dependency (unreliable in Zulip 12.0).
- **Single stream subscription**: Each bot subscribes to `#agent-hub` exclusively for `@all-bots` detection. All other messages in `#agent-hub` are ignored.
## Platform Plugin Contracts
### Hermes (Python) — BasePlatformAdapter ### Hermes (Python) — BasePlatformAdapter
- Path: `hermes-zulip-plugin/src/hermes_zulip/` | Field | Value |
- Implements: `BasePlatformAdapter` (Hermes Gateway) |-------|-------|
- Config: `config.yaml` per-agent | Path | `hermes-zulip-plugin/src/hermes_zulip/` |
- Entry point: `plugin.yaml` (Hermes manifest) | Interface | `BasePlatformAdapter` (Hermes Gateway) |
| Config | `config.yaml` per-agent |
| Entry point | `plugin.yaml` (Hermes manifest) |
| DM detection | `message.type === 'private'` |
| @all-bots detection | Regex `/@\*\*(?:all-bots|All Bots)\*\*/i` on stream events |
### Agent Zero — A0 Plugin System ### Agent Zero — A0 Plugin System
- Path: `agent-zero-plugin/src/` | Field | Value |
- Implements: Agent Zero plugin API |-------|-------|
- Config: `config.yaml` per-agent | Path | `agent-zero-plugin/src/` |
| Interface | Agent Zero plugin API |
| Config | `config.yaml` per-agent |
| DM detection | `message.type === 'private'` |
| @all-bots detection | Regex `/@\*\*(?:all-bots|All Bots)\*\*/i` on stream events |
### pi (TypeScript) — pi Extension API ### PI (TypeScript) — PI Extension API
- Path: `pi-zulip-extension/src/` | Field | Value |
- Implements: pi extension (TypeScript module in `~/.pi/agent/extensions/`) |-------|-------|
- Config: `config.yaml` per-agent | Path | `pi-zulip-extension/src/` |
- Reference: pi extension docs at `/usr/lib/node_modules/@earendil-works/pi-coding-agent/docs/extensions.md` | Interface | PI extension (TypeScript module in `~/.pi/agent/extensions/`) |
| Config | `config.yaml` per-agent |
## Reliability | DM detection | `event.type === 'private'` or `message.type === 'private'` |
- **Reconnection**: Zulip SDK handles backoff/reconnect natively | @all-bots detection | Regex `/@\*\*(?:all-bots|All Bots)\*\*/i` on stream events |
- **Retry**: 3 attempts with 5s backoff on agent timeout
- **Error Response**: User-visible message on failure
- **Monitoring**: Health endpoint (`/health`) on each plugin (port 9200)
- **Logging**: Per-plugin log files
## Config Template ## Config Template
```yaml ```yaml
# config.yaml — Per-agent Zulip plugin configuration # config.yaml — Per-agent Zulip plugin configuration (DM-First)
agent: agent:
name: "tanko" name: "tanko"
display_name: "Tanko" display_name: "Tanko"
zulip_bot_name: "tanko-bot" zulip_bot_name: "tanko-bot"
owner_email: "jerome@sysloggh.net" owner_email: "jerome@sysloggh.net"
private_topic: "tanko-private"
zulip: zulip:
server_url: "https://chat.sysloggh.net" server_url: "https://chat.sysloggh.net"
email: "tanko-bot@chat.sysloggh.net" email: "tanko-bot@chat.sysloggh.net"
api_key: "${ZULIP_API_KEY}" # From env var api_key: "${ZULIP_API_KEY}" # From env var or config
all_bots_user_id: 1234 stream: "agent-hub" # Only used for @all-bots broadcasts
stream_only_for_broadcast: true
error_handling: error_handling:
timeout_seconds: 30 timeout_seconds: 30
@@ -116,9 +155,19 @@ monitoring:
health_port: 9200 health_port: 9200
``` ```
## Reliability
- **Reconnection**: Zulip SDK handles backoff/reconnect natively
- **Retry**: 3 attempts with 5s backoff on agent timeout
- **Error Response**: User-visible message in DM on failure
- **Monitoring**: Health endpoint (`/health`) on each plugin (port 9200)
- **Logging**: Per-plugin log files (or journald for Hermes systemd services)
## Prerequisites ## Prerequisites
1. Zulip server running on CT 117
2. Zulip bot users created for all 6 agents + All Bots 1. Zulip server running on CT 117 (`chat.sysloggh.net`)
3. `#agent-hub` stream created with all bots subscribed 2. Zulip bot users created for all 6 agents (no dedicated "All Bots" bot)
3. `#agent-hub` stream created with all 6 bots subscribed
4. SSH key access to all 6 agent CTs 4. SSH key access to all 6 agent CTs
5. Gitea repo initialized at `git@git.sysloggh.net:SyslogSolution/zulip-platform-plugins.git` 5. Gitea repo initialized at `git@git.sysloggh.net:SyslogSolution/zulip-platform-plugins.git`
6. Each plugin configured with per-agent `config.yaml`
docs/CONTEXT.md
+76 -26
View File
@@ -1,59 +1,109 @@
# Zulip Platform Plugins — Domain Vocabulary # Zulip Platform Plugins — Domain Vocabulary
This document captures the shared language and architectural invariants for the multi-platform Zulip agent communication system. **Version:** 2.0 (DM-First)
**Date:** 2026-06-21
This document captures the shared language and architectural invariants for the multi-platform Zulip agent communication system under the DM-first architecture.
## Domain Terms ## Domain Terms
| Term | Definition | | Term | Definition |
|------|------------| |------|------------|
| **Agent** | An AI agent operating on a platform (Hermes/Agent Zero/pi) that can receive and respond to Zulip messages | | **Agent** | An AI agent operating on a platform (Hermes/Agent Zero/PI) that can receive and respond to Zulip messages |
| **Adapter / Plugin** | A platform-native component that bridges Zulip events to the agent framework | | **Adapter / Plugin** | A platform-native component that bridges Zulip events to the agent framework |
| **@mention** | A Zulip mention (`@**Tanko Bot**`) that triggers the named agent to respond | | **DM (Direct Message)** | A Zulip private message from a user directly to a bot. The primary communication channel for agent-owner interaction |
| **@all-bots** | A dedicated Zulip bot user that, when @mentioned, triggers ALL agents across all platforms | | **@all-bots** | A content-based mention detection pattern in `#agent-hub` stream. Detected via regex matching `@**all-bots**` or `@**All Bots**` in message content |
| **Topic** | A Zulip conversation thread within a stream | | **#agent-hub** | The shared stream used exclusively for `@all-bots` broadcasts. Agents ignore all non-@all-bots messages here |
| **Private topic** | An agent-named topic (e.g., `tanko-private`) where only the bot and its owner participate | | **Stream** | A Zulip channel for topic-based discussions. Only used for `@all-bots` broadcasts in this architecture |
| **Topic** | A Zulip conversation thread within a stream. Not used for agent routing in DM-first architecture |
| **Owner** | The human user associated with an agent bot, stored in per-agent config | | **Owner** | The human user associated with an agent bot, stored in per-agent config |
## Naming Conventions ## Naming Conventions
- Agent Zulip bot display names: lowercase, dash-separated, `-bot` suffix — e.g., `tanko-bot`, `mumuni-bot` - Agent Zulip bot display names: lowercase, dash-separated, `-bot` suffix — e.g., `tanko-bot`, `mumuni-bot`
- Agent Zulip bot emails: `@chat.sysloggh.net` — e.g., `tanko-bot@chat.sysloggh.net` - Agent Zulip bot emails: `@chat.sysloggh.net` — e.g., `tanko-bot@chat.sysloggh.net`
- Private topics: `{name}-private` — e.g., `tanko-private` - DM address: `@**tanko-bot**` (same format as @mention, but sent as private message)
- Owner privacy: agent should never expose owner name to other agents or users - Owner privacy: agent should never expose owner name to other agents or users
- No private topics, no dedicated `@all-bots` bot user
## Agent Registry ## Agent Registry
| Agent | Platform | Zulip Bot | CT | Private Topic | | Agent | Platform | Zulip Bot | CT | DM Access | @all-bots Detection |
|-------|----------|-----------|-----|---------------| |-------|----------|-----------|-----|-----------|---------------------|
| tanko | Hermes (Python) | tanko-bot | amdpve CT 112 | tanko-private | | tanko | Hermes (Python) | tanko-bot | amdpve CT 112 | ✅ Native Hermes | Regex on #agent-hub |
| mumuni | Hermes (Python) | mumuni-bot | minipve CT 114 | mumuni-private | | mumuni | Hermes (Python) | mumuni-bot | minipve CT 114 | ❌ Not deployed | Regex on #agent-hub |
| koonimo | Hermes (Python) | koonimo-bot | amdpve CT 113 | koonimo-private | | koonimo | Hermes (Python) | koonimo-bot | amdpve CT 113 | ❌ Not deployed | Regex on #agent-hub |
| koby | Hermes (Python) | koby-bot | amdpve CT 111 | koby-private | | koby | Hermes (Python) | koby-bot | amdpve CT 111 | ❌ Not deployed | Regex on #agent-hub |
| kagentz | Agent Zero | kagentz-bot | amdpve CT 105 | kagentz-private | | kagentz | Agent Zero | kagentz-bot | amdpve CT 105 | ❌ Not deployed | Regex on #agent-hub |
| abiba | pi (TypeScript) | abiba-bot | amdpve CT 100 | abiba-private | | abiba | PI (TypeScript) | abiba-bot | amdpve CT 100 | ⚠️ Extension deployed but broken | Regex on #agent-hub |
## DM-First Communication Flow
```
User sends DM to @tanko-bot
→ Zulip fires event ONLY to tanko-bot's event queue
→ Event has: {type: 'message', message: {type: 'private', ...}}
→ Bot detects: message.type === 'private'
→ No @mention parsing, no topic routing needed
→ Bot processes message
→ Bot replies via POST /api/v1/messages {type: 'private', to: [sender_id]}
```
**Key invariants:**
- Bot receives events ONLY when the bot is the DM recipient
- `message.type === 'private'` is trivially detectable with no Zulip SDK version dependency
- Reply goes to the same DM thread — Zulip handles threading
- No `mentioned_users` field is checked or used
- No stream subscription needed for DM communication
## @all-bots Resolution ## @all-bots Resolution
@all-bots refers to all 6 agents across all frameworks (Hermes, Agent Zero, pi). When @all-bots is mentioned in any topic: `@all-bots` uses **content-based detection** on `#agent-hub` stream messages — no dedicated bot user:
1. Each per-agent bot receives the event (since they all share `narrow: stream agent-hub`)
2. Each bot checks: "Is `All Bots` in `mentioned_users`?" ```regex
3. If yes, bot forwards the message to its agent /@\*\*(?:all-bots|All Bots)\*\*/i
4. @all-bots is a **dedicated Zulip bot user** with display name `All Bots` and email `all-bots@chat.sysloggh.net` ```
1. Each bot subscribes to `#agent-hub` stream exclusively for `@all-bots` detection
2. Each bot receives every message in `#agent-hub` (Zulip push model)
3. Each bot checks: "Does content match the @all-bots regex?"
4. If match: bot forwards the full message to its agent for processing
5. If no match: bot ignores the message entirely
6. There is NO dedicated "All Bots" Zulip bot user
## Response Rules ## Response Rules
- **Own topic**: Agent always responds (to its private topic)
- **Collaboration topic** (any topic that isn't a private topic): Agent responds only when @mentioned
- **@all-bots**: All agents respond
- **Private topics**: Only the agent owner + bot are participants; bot enforces this via config owner_email matching
## Private Topic ACL (v1 — documented limitation) - **DM (primary)**: Agent always responds to the sender's DM — no routing logic needed
- **#agent-hub (broadcast only)**: Agent responds only when `@all-bots` is detected in message content
- **Non-@all-bots messages**: All agents ignore any message in `#agent-hub` that does not contain `@all-bots`
- **No private topics, no @mention of individual agents in streams** — DMs replace both patterns
In v1, private topic access is enforced by matching `sender_email == config.owner_email` on each message. **This is not cryptographically authenticated** — Zulip does not enforce sender identity. A user who knows the owner email can impersonate them. This is an acceptable limitation for v1 internal use. v2 should use Zulip invite-only stream permissions or shared-secret challenge-response. ## Response Rules for Agent-Owner DM
- Owner sends DM to bot → agent processes and replies
- Bot processes ALL DMs from ANY Zulip user (API key is the only access gate)
- Agent owner email in config is for identification, not ACL enforcement
- Future: rate limiting and spam protection via Zulip's built-in rate limiting
## Monorepo Structure ## Monorepo Structure
All 3 platform plugins live in a single Gitea repository at `git@git.sysloggh.net:SyslogSolution/zulip-platform-plugins.git` All 3 platform plugins live in a single Gitea repository at `git@git.sysloggh.net:SyslogSolution/zulip-platform-plugins.git`
```
zulip-platform-plugins/
├── hermes-zulip-plugin/ # Hermes Python adapter (4 agents)
├── pi-zulip-extension/ # PI TypeScript extension (abiba)
├── agent-zero-plugin/ # Agent Zero Python plugin (kagentz)
├── scripts/ # deploy.sh, rollback.sh
├── docs/ # ADRs, ARCHITECTURE.md, PRD.md, CONTEXT.md
├── config.yaml.example # Template for per-agent config
└── .gitea/ # PR template, workflows (if CI enabled)
```
## Guardian Directives ## Guardian Directives
- All destructive actions require user confirmation before execution - All destructive actions require user confirmation before execution
- Plugin updates are version-controlled through Gitea - Plugin updates are version-controlled through Gitea
- Config files are per-agent and pushed alongside code - Config files are per-agent and pushed alongside code
- DM-first architecture means no @mention-related Zulip SDK features are relied upon
- Pi extension on CT 100 has a broken Zulip client connection that needs repair
docs/PRD.md
+118 -76
View File
@@ -1,54 +1,55 @@
# Product Requirements Document: Zulip Platform Plugins # Product Requirements Document: Zulip Platform Plugins
**Status:** Draft v1.0 **Status:** Draft v2.0
**Date:** 2026-06-18 **Date:** 2026-06-21
**Authors:** Agent Zero (via grill-with-design on Tabiri infrastructure) **Authors:** Agent Zero (via Tabiri infrastructure)
--- ---
## Problem Statement ## Problem Statement
The Sysloggh agent mesh consists of **6 autonomous agents** across **3 distinct frameworks** (Hermes Python, Agent Zero, PI/openclaw TypeScript) deployed on **5 Proxmox CTs**. These agents operate in isolation — there is no unified inter-agent communication channel. Each agent processes tasks independently, and cross-agent coordination requires manual intervention. The Sysloggh agent mesh consists of **6 autonomous agents** across **3 distinct frameworks** (Hermes Python, Agent Zero, PI TypeScript) deployed on **5 Proxmox CTs**. These agents operate in isolation — there is no unified inter-agent communication channel. Each agent processes tasks independently, and cross-agent coordination requires manual intervention.
The existing Zulip infrastructure (Zulip 12.0-1 on CT 117, Docker-based) provides a mature async messaging platform, but no agent is wired into it. A partial implementation (`zulip-plugin/adapter.py`) exists with event streaming and dedup logic, but was never deployed. The gateway adapter is compiled but not running on any agent CT, and the required Python packages are not installed. The existing Zulip infrastructure (Zulip 12.0-1 on CT 117, Docker-based) provides a mature async messaging platform with built-in DM and stream support. All 6 agent bot users exist on the server and `#agent-hub` stream is created. Only Tanko (v1 Hermes native Zulip) is operational.
**The gap:** Agents cannot collaboratively route tasks, share context, or coordinate via Zulip threads. **The gap:** Agents cannot reliably receive user requests or communicate via Zulip DMs.
--- ---
## Solution ## Solution
A multi-platform **Zulip agent communication layer** consisting of one plugin per agent framework, each running co-located on its agent's CT, connecting its agent to the shared Zulip messaging infrastructure at `chat.sysloggh.net`. A multi-platform **Zulip agent communication layer** where each agent communicates primarily via **Zulip Direct Messages (DMs)**. Agents also subscribe to `#agent-hub` stream for `@all-bots` broadcasts only.
### Architecture at a Glance ### Architecture at a Glance
``` ```
┌──────────────────────────────────────────────────────────────────┐ ┌─────────────────────────────────────────────────────────────┐
│ Zulip Server (CT 117) │ │ Zulip Server (CT 117) │
│ chat.sysloggh.net:443 │ │ chat.sysloggh.net:443 │
│ Stream: #agent-hub │ │ │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ Each Agent Has Two Channels: │ │
│ │ │ │
│ │ DM (PRIMARY) #agent-hub (BROADCAST) │ │
│ │ ┌──────────────────┐ ┌──────────────────┐ │ │
│ │ │ Direct 1:1 chat │ │ @all-bots only │ │ │
│ │ │ No @mention │ │ Cross-agent msgs │ │ │
│ │ │ No topic parsing │ │ Broadcast only │ │ │
│ │ └──────────────────┘ └──────────────────┘ │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
│ Bot Users: tanko-bot, mumuni-bot, koonimo-bot, koby-bot, │ │ Bot Users: tanko-bot, mumuni-bot, koonimo-bot, koby-bot, │
│ kagentz-bot, abiba-bot, all-bots │ │ kagentz-bot, abiba-bot │
└────┬──────┬──────┬──────┬──────┬──────┬──────┬──────────────────┘ │ (No dedicated @all-bots bot user — content-based match) │
│ │ │ │ │ │ │ └─────────────────────────────────────────────────────────────┘
┌────┴┐ ┌───┴──┐ ┌┴────┐ ┌┴────┐ ┌┴────┐ ┌┴────┐ ┌──────────┐
│Tnako│ │Mumuni│ │Kooni│ │Koby │ │Kage │ │Abiba│ │@all-bots │
│Herm │ │Herm │ │Herm │ │Herm │ │A0 │ │PI │ │ (special)│
│CT112│ │CT... │ │CT.. │ │CT.. │ │CT.. │ │CT.. │ │ │
└─────┘ └──────┘ └─────┘ └─────┘ └─────┘ └─────┘ └──────────┘
│ │ │ │ │ │
┌────┴┐ ┌───┴──┐ ┌┴────┐ ┌┴────┐ ┌┴────┐ ┌┴────┐
│Config│ │Config│ │Config│ │Config│ │Config│ │Config│
│tanko │ │mumuni│ │kooni │ │koby │ │kage │ │abiba │
│.yaml │ │.yaml │ │.yaml │ │.yaml │ │.yaml │ │.yaml │
└─────┘ └──────┘ └─────┘ └─────┘ └─────┘ └─────┘
``` ```
### Core Principles ### Core Principles
- **DM-first:** Agent-owner communication uses Zulip DMs exclusively — no @mention parsing, no topic routing, no `mentioned_users` dependency
- **#agent-hub for broadcast only:** The `#agent-hub` stream is subscribed solely for detecting `@all-bots` notifications
- **Co-located:** Each plugin runs on the same CT as its agent, uses platform-native interfaces - **Co-located:** Each plugin runs on the same CT as its agent, uses platform-native interfaces
- **@mention-routed:** Messages reach the correct agent via Zulip @mention detection - **Owner-gated:** Each bot DM is gated by API key — only the bot's credentials grant access
- **Owner-gated:** Private topics are restricted per-agent owner via config
- **Version-controlled:** All plugin code lives in a Gitea monorepo, pushed to CTs via deploy script - **Version-controlled:** All plugin code lives in a Gitea monorepo, pushed to CTs via deploy script
- **Graceful degradation:** Users always receive feedback, even when an agent is unavailable - **Graceful degradation:** Users always receive feedback, even when an agent is unavailable
@@ -58,52 +59,72 @@ A multi-platform **Zulip agent communication layer** consisting of one plugin pe
### Agent Communication ### Agent Communication
1. As a **human user**, I want to **@mention an agent in any Zulip topic**, so that the agent receives my message and responds in-thread. 1. As a **human user**, I want to **DM an agent's bot directly**, so that the agent receives my message and responds in-thread.
2. As a **human user**, I want to **@mention @all-bots**, so that all 6 agents across all frameworks receive the message. 2. As a **human user**, I want to **mention @all-bots in #agent-hub**, so that all 6 agents across all frameworks receive the broadcast.
3. As an **agent**, I want to **receive messages addressed to me**, so that I can process and respond to user requests. 3. As an **agent**, I want to **receive DMs addressed to me**, so that I can process and respond without parsing @mentions or stream context.
### Private Channels 4. As an **agent**, I want to **detect @all-bots broadcasts in #agent-hub**, so that I can participate in multi-agent coordination.
4. As a **human user**, I want a **private topic per agent** (e.g., `#agent-hub > tanko-private`), so that I can have confidential conversations with a single agent.
5. As a **system administrator**, I want **per-agent owner configuration**, so that only designated users can access each agent's private topic.
### Deployment & Operations ### Deployment & Operations
6. As a **system administrator**, I want to **version-control all plugin code in a single Gitea repository**, so that updates are tracked, auditable, and pushable across all CTs. 5. As a **system administrator**, I want to **version-control all plugin code in a single Gitea repository**, so that updates are tracked, auditable, and pushable across all CTs.
7. As a **system administrator**, I want a **deploy script** that pulls the latest plugin code and restarts services on all CTs, so that updates are consistent and repeatable. 6. As a **system administrator**, I want a **deploy script** that pulls the latest plugin code and restarts services on all CTs, so that updates are consistent and repeatable.
### Reliability ### Reliability
8. As a **human user**, I want to see **a friendly error message** if an agent is unavailable, rather than a silent timeout. 7. As a **human user**, I want to see **a friendly error message** if an agent is unavailable, rather than a silent timeout.
9. As a **system administrator**, I want **health check endpoints** on each plugin, so that I can monitor plugin status from the harness or monitoring infra. 8. As a **system administrator**, I want **health check endpoints** on each plugin, so that I can monitor plugin status from the harness or monitoring infra.
--- ---
## Implementation Decisions ## Architecture Decisions
### Platform Plugin Contracts ### DM-First Communication
Each of the 3 frameworks uses its **native plugin system** — not a unified bridge or A2A overlay for intra-agent communication. Cross-framework @all-bots messages flow through Zulip: all 7 bot users are subscribed to the same `#agent-hub` stream, so every bot receives every message. Each bot checks if it's @mentioned, then either processes or ignores. Each agent communicates via Zulip **Direct Messages**. When a user sends a DM to an agent's bot:
### Framework-Specific Implementation ```
User DMs @tanko-bot
→ Zulip pushes event ONLY to tanko-bot's event queue
→ Event has: {type: 'message', message: {type: 'private', ...}}
→ Bot detects message.type === 'private'
→ No @mention parsing, no topic/stream capture needed
→ Bot processes message.content
→ Bot replies to same DM via:
{type: 'private', to: [user_id], content: 'response'}
```
| Framework | Plugin Type | Interface | Reference Architecture | **Advantages over stream-based @mention:**
|-----------|-------------|-----------|----------------------| - Only the recipient bot receives the DM — no broadcast noise to other 5 agents
| **Hermes Python** (4 agents) | Hermes Gateway `BasePlatformAdapter` | `on_event(message) → Response` | Existing `zulip-plugin/` in homelab | - No `mentioned_users` field dependency (unreliable in Zulip 12.0)
| **PI/openclaw TypeScript** (1 agent) | openclaw Extension | Extension API (similar to existing Discord/GChat/Mattermost/Matrix) | `/a0/usr/projects/homelab/openclaw/extensions/` | - No topic routing — DMs are inherently 1:1 threading
| **Agent Zero** (1 agent) | A0 Plugin | Agent Zero plugin system | Custom A0 plugin pattern | - `message.type: 'private'` is trivially detectable
### @mention Detection ### @all-bots Broadcast
- Use Zulip SDK's `mentioned_users` field on message events (ADR-005) `@all-bots` in `#agent-hub` uses **content-based detection** (regex) — no dedicated bot user needed:
- Each bot checks if its own `user_id` is in the `mentioned_users` array
- For @all-bots: bot checks if dedicated `All Bots` user is in `mentioned_users` (ADR-006) ```
- @all-bots spans all 7 bots across all frameworks (ADR-012) User posts in #agent-hub > topic:
"@**all-bots** status report please"
→ Each bot receives event via #agent-hub subscription
→ Each bot checks: /@\*\*(?:all-bots|All Bots)\*\*/i.test(content)
→ If match: bot forwards to its agent
→ 6 independent responses appear in the topic
```
### Plugin Architecture
| Framework | Plugin Type | Interface | DM Detection | @all-bots Detection |
|-----------|-------------|-----------|--------------|--------------------|
| **Hermes Python** (4 agents) | Gateway PlatformAdapter | `on_event(message)` | `message.type === 'private'` | Regex on stream events |
| **PI TypeScript** (1 agent) | PI Extension | Extension API | `event.type === 'private'` | Regex on stream events |
| **Agent Zero** (1 agent) | A0 Plugin | A0 plugin system | `message.type === 'private'` | Regex on stream events |
### Config Schema (per-agent `config.yaml`) ### Config Schema (per-agent `config.yaml`)
@@ -115,11 +136,12 @@ agent:
bot_email: "tanko-bot@chat.sysloggh.net" bot_email: "tanko-bot@chat.sysloggh.net"
bot_api_key: "<api_key>" bot_api_key: "<api_key>"
owner_email: "jerome@sysloggh.net" owner_email: "jerome@sysloggh.net"
private_topic: "tanko-private"
zulip: zulip:
server_url: "https://chat.sysloggh.net" server_url: "https://chat.sysloggh.net"
stream: "agent-hub" stream: "agent-hub" # Only used for @all-bots broadcasts
client:
type: "sdk" # "sdk" for zulip-js/zulip-python, "rest" for raw API
error_handling: error_handling:
timeout_seconds: 30 timeout_seconds: 30
@@ -131,16 +153,38 @@ monitoring:
health_port: 9200 health_port: 9200
``` ```
### Private Topic ACL ---
- Bot subscribes to its private topic (`tanko-private`) at startup ## Implementation Decisions
- Owner is matched via `config.yaml > owner_email` (ADR-011)
- Messages in private topic: only respond if sender email matches owner ### Platform Plugin Contracts
- Messages in public topics: respond to @mentions from any sender
Each of the 3 frameworks uses its **native plugin system** — not a unified bridge or A2A overlay for intra-agent communication. Cross-framework `@all-bots` messages flow through Zulip: all 6 bots are subscribed to `#agent-hub` stream. Each bot checks for `@all-bots` via content-based regex match, then processes.
### DM Event Processing
```python
# Simplified DM processing logic (all platforms)
def on_event(event):
message = event["message"]
# DM case (primary)
if message["type"] == "private":
user_id = message["sender_id"]
content = message["content"]
agent.process(content, reply_to=lambda resp: send_dm(user_id, resp))
return
# Stream case — only process if @all-bots
if message["type"] == "stream":
if re.search(r'@\*\*(?:all-bots|All Bots)\*\*', message["content"]):
agent.process_broadcast(message)
# Ignore all other stream messages
```
### Error Handling Strategy ### Error Handling Strategy
- Agent timeout: plugin sends graceful degradation message to Zulip thread (ADR-009) - Agent timeout: plugin sends graceful degradation message to DM thread (ADR-009)
- Zulip server down: SDK auto-reconnects with exponential backoff - Zulip server down: SDK auto-reconnects with exponential backoff
- Invalid message format: plugin logs error, does not crash - Invalid message format: plugin logs error, does not crash
- Health endpoint (`/health` on port 9200): returns status, uptime, last_message_time - Health endpoint (`/health` on port 9200): returns status, uptime, last_message_time
@@ -148,28 +192,28 @@ monitoring:
### Naming Convention ### Naming Convention
- Bot users: `{name}-bot` (lowercase, dash separator) — `tanko-bot`, `mumuni-bot`, etc. (ADR-003) - Bot users: `{name}-bot` (lowercase, dash separator) — `tanko-bot`, `mumuni-bot`, etc. (ADR-003)
- Private topics: `{name}-private` — `tanko-private`, `mumuni-private`, etc. (ADR-001) - DM address: `@**tanko-bot**` (same format as @mention)
- Display names: capitalize properly — "Tanko Bot", "Mumuni Bot", "All Bots" - Display names: capitalize properly — "Tanko Bot", "Mumuni Bot"
--- ---
## Testing Decisions ## Testing Decisions
### Unit Tests ### Unit Tests
- @mention detection logic: correct extraction of bot from `mentioned_users` - DM detection: `message.type === 'private'` correctly triggers processing
- Private topic ACL: owner matches, non-owner denied - @all-bots detection: regex matches `@**all-bots**` and `@**All Bots**` in stream content
- Config parsing: all fields present, graceful defaults - Config parsing: all fields present, graceful defaults
- Error response: timeout triggers correct grace message - Error response: timeout triggers correct grace message
### Integration Tests ### Integration Tests
- Each plugin connects to Zulip and receives/sends test messages - Each plugin receives DM and sends reply via Zulip API
- @all-bots message reaches all 7 bots - @all-bots message in #agent-hub reaches all 6 bots
- Private topic message only reaches owner agent - Non-@all-bots messages in #agent-hub are correctly ignored
- Health endpoint returns 200 with valid JSON - Health endpoint returns 200 with valid JSON
- Plugin survives container restart (Zulip reconnection)
### Deployment Tests ### Deployment Tests
- `deploy.sh` successfully pulls latest and restarts services on all CTs - `deploy.sh` successfully pulls latest and restarts services on all CTs
- Plugin survives container restart (Zulip reconnection)
- Config YAML validation before deployment - Config YAML validation before deployment
--- ---
@@ -187,18 +231,16 @@ monitoring:
--- ---
## Further Notes ## Design Decisions Summary
### Design Decisions Summary
| # | Decision | Rationale | | # | Decision | Rationale |
|---|----------|-----------| |---|----------|-----------|
| ADR-001 | Topic-per-agent with `-private` suffix | Clear naming, easy ACL enforcement | | ADR-001 | DM-first topology | DMs are simpler, more reliable, and require no @mention parsing |
| ADR-002 | @mention-based routing (not topic-based) | User confirmed, flexible for any topic | | ADR-002 | DM-based routing (not @mention) | `message.type === 'private'` is trivially detectable, no `mentioned_users` dependency |
| ADR-003 | Bot naming: `{name}-bot` | Consistent, no space issues in Zulip API | | ADR-003 | Bot naming: `{name}-bot` | Consistent, no space issues in Zulip API |
| ADR-004 | One bot per agent, co-located on CT | Isolated failures, platform-native | | ADR-004 | One bot per agent, co-located on CT | Isolated failures, platform-native |
| ADR-005 | SDK `mentioned_users` field | Zulip API native, always accurate | | ADR-005 | [Archived] @mention via SDK — no longer needed | DM-first eliminates @mention dependency entirely |
| ADR-006 | Dedicated `All Bots` user | Explicit @all-bots, not noisy @everyone | | ADR-006 | @all-bots via content-based regex | No dedicated bot user needed, works across Zulip versions |
| ADR-007 | Platform-native plugin contracts | Reliability, platform handles state/retries | | ADR-007 | Platform-native plugin contracts | Reliability, platform handles state/retries |
| ADR-008 | Native message formats per platform | No translation layer needed | | ADR-008 | Native message formats per platform | No translation layer needed |
| ADR-009 | Graceful degradation + user-visible errors | No silent failures | | ADR-009 | Graceful degradation + user-visible errors | No silent failures |
@@ -209,11 +251,11 @@ monitoring:
### Prerequisites ### Prerequisites
1. Zulip server operational at `https://chat.sysloggh.net` 1. Zulip server operational at `https://chat.sysloggh.net`
2. Zulip bot users created (7 total: tanko-bot, mumuni-bot, koonimo-bot, koby-bot, kagentz-bot, abiba-bot, all-bots) 2. Zulip bot users created (6 total: tanko-bot, mumuni-bot, koonimo-bot, koby-bot, kagentz-bot, abiba-bot)
3. `#agent-hub` stream created and all bots subscribed 3. `#agent-hub` stream created and all bots subscribed
4. Gitea repo `zulip-platform-plugins` created and monorepo pushed 4. Gitea repo `zulip-platform-plugins` created and monorepo pushed
5. Hermes gateway framework verified on Tanko/Mumuni/Koonimo/Koby CTs 5. Hermes gateway framework verified on Tanko/Mumuni/Koonimo/Koby CTs
6. openclaw extension API verified on Abiba CT 6. PI extension API verified on Abiba CT
7. Agent Zero plugin system verified on Kagentz CT 7. Agent Zero plugin system verified on Kagentz CT
### Deployment Workflow ### Deployment Workflow
docs/adr/ADR-001-dm-topology.md
+64
View File
@@ -0,0 +1,64 @@
# ADR 1: DM-First Communication Topology
**Date**: 2026-06-21 (supersedes 2026-04-28 topic-based topology)
### Decision
Each agent communicates primarily via **Zulip Direct Messages (DMs)**. The `#agent-hub` stream is retained as an **optional broadcast-only channel** for inter-agent coordination, `@all-bots` announcements, and cross-agent visibility. Agent-private topics (`<agent>-private`) are **deprecated** in favor of native Zulip DMs.
### Context
Zulip bots are regular user accounts — any user can send a **private message** to any bot by composing a DM with the bot as the sole recipient. The bot's event queue then receives an event with `message.type: 'private'`, which requires **no @mention detection, no topic parsing, no stream subscription checks**.
The original design used topics under `#agent-hub` with `@mention` routing, but this introduced several failure modes:
| Problem | Impact |
|---------|--------|
| `mentioned_users` field not populated by Zulip 12.0 SDK | @mention detection silently fails |
| Topic routing required capturing origin topic + stream for reply | Added parsing complexity |
| All 6 bots subscribed to same stream | Every bot receives every message, wasting resources |
| `type:'stream'` events different from `type:'private'` | Required additional event filtering logic |
**DMs solve all of this natively:**
- Every DM is inherently for the bot recipient — no @mention needed
- Reply in the same DM thread — Zulip handles threading
- Only the recipient bot sees the message — no broadcast noise
- `message.type: 'private'` is trivially detectable
### Architecture Change
#### Before (Deprecated):
```
User @mentions @tanko-bot in #agent-hub > tanko-private
→ Zulip fires event to ALL 6 bot event queues
→ Each bot checks mentioned_users for its user_id
→ Only matching bot processes the event
→ Bot replies to same stream+topic
```
#### After (DM-First):
```
User DMs @tanko-bot directly
→ Zulip fires event ONLY to tanko-bot's event queue
→ Bot receives message.type: 'private'
→ No @mention detection, no topic parsing
→ Bot processes and replies in same DM thread
```
### Consequences
- **Positive:** Dramatically simpler event processing — no @mention detection, no topic routing, no mentioned_users dependency, no stream subscription requirement
- **Positive:** Each bot only receives its own messages — less noise, less resource usage
- **Positive:** Zulip's native DM threading handles all reply routing automatically
- **Positive:** No need to subscribe all 6 bots to `#agent-hub` stream (except for `@all-bots` broadcast capability)
- **Negative:** Single #agent-hub stream topic no longer shows all agent activity in one timeline
- **Negative:** Cross-agent visibility requires `@all-bots` broadcast or agents explicitly CC each other
- **Neutral:** `#agent-hub` stream is retained for: `@all-bots` broadcasts, inter-agent coordination, and user-initiated multi-agent topics
### Migration Path
1. All plugins listen for `type: 'private'` events as primary communication channel
2. `#agent-hub` stream listener is secondary — only for `@all-bots` broadcast detection
3. Existing `<agent>-private` topics are deprecated but not deleted (grace period)
4. Owners communicate with their agents via DM, not via private topics
5. `@all-bots` continues to work in `#agent-hub` for cross-platform broadcasts
docs/adr/ADR-001-topic-topology.md
-15
View File
@@ -1,15 +0,0 @@
# ADR 1: Topic & Stream Topology
**Date**: 2026-04-28
### Decision
Each agent gets a private topic under `#agent-hub` named `<agent>-private` for owner-only communication; all other topics are collaboration topics where agents respond only when @mentioned.
### Context
We need a clear separation between agent-private communication (user-agent DMs) and collaboration (multi-agent discussions). Zulip topics are the right granularity — no separate streams needed. The `-private` suffix convention is simple and self-documenting.
### Consequences
- All 6 agents co-exist in `#agent-hub` stream
- Users can create any new topic and @mention agents
- Private topics are invitation-only (bot + owner)
- No need to manage multiple streams
docs/adr/ADR-002-dm-routing.md
+51
View File
@@ -0,0 +1,51 @@
# ADR 2: DM-First Routing
**Date**: 2026-06-21 (supersedes 2026-04-28 @mention-based routing)
### Decision
Agents are triggered primarily by **Zulip Direct Messages (DMs)**. Users send a DM to any agent's bot user (e.g., `@tanko-bot`) to communicate with that agent. Stream-based `@mention` routing is **deprecated** for agent-owner communication but retained for `@all-bots` broadcasts and multi-agent coordination in `#agent-hub`.
### Context
The original design required users to `@mention` an agent in `#agent-hub` stream topics. This proved unreliable because:
1. Zulip 12.0 SDK may not populate `mentioned_users` on stream message events
2. Every bot subscribed to `#agent-hub` receives every message — wasteful
3. Topic routing (capturing origin stream + topic for reply) adds parsing complexity
4. Distinguishing `type: 'stream'` vs `type: 'private'` required extra event filtering
DMs eliminate all of these concerns:
- Only the recipient bot receives the DM — no broadcast noise
- No @mention parsing needed — the DM itself is the signal
- No topic/stream capture needed — reply in the same DM thread
- `message.type: 'private'` is trivially detectable in the event queue
### DM Event Flow
```
User sends DM to @tanko-bot
→ Zulip pushes event to tanko-bot's event queue
→ Event has: {type: 'message', message: {type: 'private', ...}}
→ Bot checks: message.type === 'private' (trivial)
→ Bot reads message.content, processes, responds
→ Bot sends reply to same DM via /api/v1/messages with
{type: 'private', to: [user_id], content: 'response'}
```
### @all-bots Exception
The stream-based approach is retained **only** for `@all-bots` broadcasts:
- `@all-bots` in `#agent-hub` → all bots detect the mention and respond
- This remains necessary for cross-platform broadcasts
- Detection can use Zulip SDK `mentioned_users` or content-based `@**All Bots**` matching
### Consequences
- **Positive:** Dramatically simpler routing logic — no @mention parsing, no topic/stream tracking
- **Positive:** Each bot only processes its own DMs — no wasted event processing
- **Positive:** Zulip's native DM threading handles all reply threading automatically
- **Positive:** No dependency on `mentioned_users` field behavior in Zulip 12.0
- **Negative:** Owners must DM their agent directly instead of using stream topics
- **Negative:** Cross-agent visibility in stream topics is reduced (mitigated by `@all-bots`)
- **Neutral:** `@all-bots` in `#agent-hub` still uses the original @mention detection pattern
docs/adr/ADR-002-mention-routing.md
-14
View File
@@ -1,14 +0,0 @@
# ADR 2: @mention-Based Routing
**Date**: 2026-04-28
### Decision
Agents are triggered exclusively by Zulip @mentions. No topic-based assignment. Users @mention an agent in any topic, and the agent's bot responds.
### Context
Topic-based routing is too rigid — it requires users to know which topic maps to which agent. @mention routing is natural in Zulip (users already know how to @mention). It also enables multi-agent conversations in a single topic.
### Consequences
- Each bot must parse @mentions from every message in `#agent-hub`
- Routing complexity moves from topic name matching to @mention resolution
- Users can freely create topics and involve agents ad-hoc
docs/adr/ADR-003-agent-naming.md
+5 -9
View File
@@ -1,15 +1,11 @@
# ADR 3: Agent Naming Convention # ADR 3: Agent Naming Convention (Unchanged)
**Date**: 2026-04-28 **Date**: 2026-04-28 | **Status**: Still active — DM architecture does not affect naming
### Decision ### Decision
Each agent bot has a display name in `{name}-bot` format with no spaces, dashes only: `tanko-bot`, `mumuni-bot`, `koonimo-bot`, `koby-bot`, `kagentz-bot`, `abiba-bot`. Each agent bot has a display name in `{name}-bot` format: `tanko-bot`, `mumuni-bot`, etc.
### Context ### Consequences (Unchanged)
Zulip @mentions work best with clean, predictable names. No spaces avoids ambiguity in @mention parsing. The `-bot` suffix makes it clear these are automated agents, not human users. - DM address: `@**tanko-bot**` (same as @mention format)
### Consequences
- @mention format: `@**tanko-bot**`
- Email format: `tanko-bot@chat.sysloggh.net` - Email format: `tanko-bot@chat.sysloggh.net`
- Users must type @tanko-bot (not @tanko)
- Consistent naming across all 6 agents - Consistent naming across all 6 agents
docs/adr/ADR-004-per-agent-bots.md
+7 -10
View File
@@ -1,15 +1,12 @@
# ADR 4: One Zulip Bot Per Agent # ADR 4: One Zulip Bot Per Agent (Unchanged)
**Date**: 2026-04-28 **Date**: 2026-04-28 | **Status**: Still active — DM architecture needs per-agent bots even more
### Decision ### Decision
Each agent has its own dedicated Zulip bot user, co-located on the agent's CT. Each bot runs independently as a platform-native plugin. Each agent has its own dedicated Zulip bot user, co-located on the agent's CT.
### Context ### Consequences (Unchanged)
A single bot creates a single point of failure and requires message routing to the correct agent. Per-agent bots provide full isolation — if tanko's bot fails, mumuni's bot continues working. Co-location means zero A2A overhead for the local agent. - 6 Zulip bot users
### Consequences
- 6 Zulip bot users to create and manage
- 6 independent event streams (one per CT) - 6 independent event streams (one per CT)
- @all-bots requires each bot to know about `All Bots` user - @all-bots still requires each bot to know about `All Bots` user
- Adding a new agent = create Zulip bot + deploy plugin to CT - DM isolation: only recipient bot sees the message (improvement over stream-based approach)
docs/adr/ADR-005-mention-detection.md → docs/adr/ADR-005-mention-detection.archived.md
View File
docs/adr/ADR-006-all-bots-model.md
+37 -8
View File
@@ -1,15 +1,44 @@
# ADR 6: @all-bots via Dedicated Zulip Bot User # ADR 6: @all-bots via Content-Based Detection
**Date**: 2026-04-28 **Date**: 2026-06-21 (supersedes 2026-04-28 dedicated bot user approach)
### Decision ### Decision
@all-bots is implemented as a dedicated Zulip bot user with display name `All Bots` and email `all-bots@chat.sysloggh.net`. All 6 per-agent bots subscribe to the same event narrow and detect when `All Bots` is in `mentioned_users`.
`@all-bots` is detected via **content-based mention matching** on messages in the `#agent-hub` stream. Each bot subscribes to `#agent-hub` and uses regex to detect `@**all-bots**` or `@**All Bots**` in message content. A dedicated "All Bots" Zulip bot user is **not required**.
### Context ### Context
@all-bots needs to trigger all agents regardless of platform. A dedicated bot user is explicit and intentional — it prevents @everyone or stream-wide pings from accidentally triggering all agents. Each bot independently detects the @mention and forwards to its agent.
The original design required a 7th Zulip bot user (`all-bots@chat.sysloggh.net`) with display name `All Bots`. This user was never created during initial bot provisioning. Additionally:
- Bot users cannot reliably detect @mentions of other bot users in Zulip 12.0 (ADR-005 archived)
- Content-based regex matching (`@**all-bots**`) works on any Zulip message regardless of SDK version
- Under the DM-first architecture (ADR-001), bot-event stream subscription to `#agent-hub` is the **only** stream-level listener needed — all other communication uses DMs
### Behavior
```
User posts in #agent-hub > topic:
"@**all-bots** status report please"
→ Each bot receives event via #agent-hub subscription
→ Each bot checks: /@\*\*(?:all-bots|All Bots)\*\*/i.test(content)
→ If match: bot forwards message to its agent
→ 6 independent responses appear in the topic
```
### Active Stream Subscription
Under DM-first architecture, each agent's plugin maintains **one** stream subscription:
- **Stream:** `#agent-hub`
- **Purpose:** Only for detecting `@all-bots` broadcasts
- **Filter:** Ignore all non-`@all-bots` messages in the stream
- **Primary channel:** DMs for all agent-owner communication
### Consequences ### Consequences
- Must create a 7th Zulip bot user for `All Bots`
- Each agent bot's config must include `All Bots` user ID for detection - **Positive:** No need to create/manage a 7th dedicated bot user
- @all-bots is opt-in: only topics that explicitly @all-bots trigger broadcast - **Positive:** Content-based detection works reliably across Zulip versions (no SDK dependency)
- Cross-platform broadcast works without a central dispatcher - **Positive:** Same regex works for case-insensitive matching (`all-bots` / `All Bots`)
- **Negative:** Slightly more processing per message (regex check)
- **Negative:** If `@all-bots` syntax changes in Zulip, regex must update
- **Neutral:** All 6 bots still independently detect and respond — same multi-response pattern
hermes-zulip-plugin/src/hermes_zulip/adapter.py
+70 -20
View File
@@ -3,12 +3,23 @@
# Full implementation in issue #5. # Full implementation in issue #5.
import logging import logging
import sys
import time import time
import re import re
import threading import threading
from typing import Any, Dict, Optional from typing import Any, Dict, Optional
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
# Ensure logger output reaches journald (systemd captures stderr)
if not logger.handlers:
_handler = logging.StreamHandler(sys.stderr)
_handler.setLevel(logging.INFO)
_handler.setFormatter(logging.Formatter(
'%(asctime)s [%(levelname)s] %(name)s: %(message)s',
datefmt='%Y-%m-%d %H:%M:%S'
))
logger.addHandler(_handler)
logger.setLevel(logging.INFO)
class ZulipAdapter: class ZulipAdapter:
"""Zulip adapter for Hermes agents. Connects to Zulip, listens for """Zulip adapter for Hermes agents. Connects to Zulip, listens for
@@ -22,10 +33,12 @@ class ZulipAdapter:
self.connected = False self.connected = False
self._client = None self._client = None
self._thread = None self._thread = None
self._loop = None
self._stop_event = threading.Event() self._stop_event = threading.Event()
self.message_handler = None # Callback: async (MessageEvent) -> str | None
async def connect(self) -> None: def connect(self) -> None:
"""Establish connection to Zulip and start the event loop.""" """Establish connection to Zulip and start the event loop (synchronous)."""
if self.connected: if self.connected:
logger.info("Already connected to Zulip.") logger.info("Already connected to Zulip.")
return return
@@ -36,7 +49,7 @@ class ZulipAdapter:
email = self.config['zulip']['email'] email = self.config['zulip']['email']
api_key = self.config['zulip']['api_key'] api_key = self.config['zulip']['api_key']
self._client = Client(server_url=server_url, email=email, api_key=api_key) self._client = Client(site=server_url.rstrip('/'), email=email, api_key=api_key)
self.connected = True self.connected = True
logger.info(f"Connected to Zulip: {server_url} as {email}") logger.info(f"Connected to Zulip: {server_url} as {email}")
@@ -51,8 +64,8 @@ class ZulipAdapter:
self.connected = False self.connected = False
raise raise
async def disconnect(self) -> None: def disconnect(self) -> None:
"""Disconnect from Zulip and stop the event loop.""" """Disconnect from Zulip and stop the event loop (synchronous)."""
if not self.connected: if not self.connected:
return return
self._stop_event.set() self._stop_event.set()
@@ -61,8 +74,8 @@ class ZulipAdapter:
self.connected = False self.connected = False
logger.info("Disconnected from Zulip.") logger.info("Disconnected from Zulip.")
async def send_message(self, topic: str, content: str) -> Dict[str, Any]: def send_message(self, topic: str, content: str) -> Dict[str, Any]:
"""Send a message to Zulip with retry logic (ADR-009).""" """Send a message to Zulip with retry logic (ADR-009) — synchronous."""
max_retries = self.config.get('error_handling', {}).get('retry_count', 3) max_retries = self.config.get('error_handling', {}).get('retry_count', 3)
retry_delay = self.config.get('error_handling', {}).get('retry_delay_seconds', 5) retry_delay = self.config.get('error_handling', {}).get('retry_delay_seconds', 5)
@@ -83,18 +96,28 @@ class ZulipAdapter:
else: else:
raise raise
async def on_event(self, event: Dict[str, Any]) -> Optional[Dict[str, Any]]: def on_event(self, event: Dict[str, Any]) -> Optional[Dict[str, Any]]:
"""Process incoming Zulip events and route via BasePlatformAdapter.""" """Process incoming Zulip events — synchronous (no I/O)."""
event_type = str(event.get("type", "unknown"))
logger.info(f"[ZULIP_EVENT] Processing: {event_type}")
if event.get('type') != 'message': if event.get('type') != 'message':
logger.debug(f"Ignored non-message event: {event.get('type')}")
return None return None
msg = event.get('message', {}) msg = event.get('message', {})
mentioned_users = msg.get('mentioned_users', False) mentioned_users = msg.get('mentioned_users', None)
logger.info(f"[DEBUG] raw mentioned_users type={type(mentioned_users).__name__} value={mentioned_users}")
logger.info(f"[DEBUG] msg keys: {list(msg.keys())}")
logger.info(f"[DEBUG] sender_id={msg.get('sender_id')} stream={msg.get('stream')}")
# Only process messages that mention this bot (ADR-005) # Only process messages that mention this bot (ADR-005)
if not mentioned_users: if not mentioned_users:
logger.info(f"[DEBUG] mentioned_users is falsy, skipping")
return None return None
logger.info(f"[DEBUG] mentioned_users IS truthy, proceeding to parse")
# Strip Zulip formatting artifacts # Strip Zulip formatting artifacts
body = msg.get('content', '') body = msg.get('content', '')
clean_body = self.MENTION_CLEANER.sub('', body) clean_body = self.MENTION_CLEANER.sub('', body)
@@ -111,23 +134,50 @@ class ZulipAdapter:
} }
def _event_loop(self) -> None: def _event_loop(self) -> None:
"""Background thread to listen for Zulip events.""" """Background thread to listen for Zulip events.
Creates a dedicated asyncio event loop for this thead (Python 3.13+).
Uses call_on_each_event (not call_on_each_message) for proper @mention detection.
"""
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
try: try:
self._client.call_on_each_message( # Build narrow to filter for our stream only
stream = self.config['zulip']['stream']
narrow = [[u"stream", stream]]
logger.info(f"Zulip: call_on_each_event narrow=[{narrow}]")
self._client.call_on_each_event(
lambda event: self._process_event(event), lambda event: self._process_event(event),
event_types=["message"],
narrow=narrow,
) )
except Exception as e: except Exception as e:
logger.error(f"Event loop crashed: {e}") logger.error(f"Event loop crashed: {e}")
self.connected = False self.connected = False
finally:
logger.info("Event loop closed.")
def _process_event(self, event: Dict[str, Any]) -> None: def _process_event(self, event: Dict[str, Any]) -> None:
"""Bridge the synchronous event to the async on_event handler.""" """Bridge the synchronous event to the async on_event handler."""
import asyncio event_type = str(event.get("type", "unknown"))
try: flags = event.get("flags", [])
loop = asyncio.get_event_loop() sender_email = event.get("sender_email", "unknown")
if loop.is_running(): sender_id = event.get("sender_id", 0)
asyncio.create_task(self.on_event(event)) logger.info(f"[ZULIP_EVENT] type={event_type} flags={flags} sender_id={sender_id} sender_email={sender_email}")
else:
asyncio.run(self.on_event(event)) # Check if this message is mentioned to us
is_mentioned = "mentioned" in flags
logger.info(f"[DEBUG] is_mentioned (via flags)={is_mentioned}")
message_event = self.on_event(event)
if message_event and self.message_handler:
reply = self.message_handler(message_event)
if reply:
self._client.send_message({
'type': 'stream',
'to': message_event['topic'],
'subject': message_event['topic'],
'content': reply,
})
except Exception as e: except Exception as e:
logger.error(f"Error processing event: {e}") logger.error(f"Error in _process_event: {e}")
hermes-zulip-plugin/src/hermes_zulip/adapter_dm.py
+751
View File
@@ -0,0 +1,751 @@
"""
Zulip Platform Adapter for Hermes Agent — DM-First Version
Supports:
- Direct Messages (DM): Users DM the bot user directly
- Stream/Topic: Bot listens to configured stream/topic for @all-bots broadcasts
- Reply routing: DMs reply in DM thread, stream messages reply in-stream
Architecture:
- Async adapter using Zulip SDK's long-poll get_events() pattern
- Event queue registered with no narrow (receives all events)
- _handle_event() demuxes based on message.type: private vs stream
"""
import asyncio
import logging
import os
import time
import re
import base64
import urllib.parse
from typing import Any, Dict, List, Optional
import requests
logger = logging.getLogger(__name__)
from gateway.platforms.base import (
BasePlatformAdapter,
SendResult,
MessageEvent,
MessageType,
)
from gateway.config import Platform
class ZulipAdapter(BasePlatformAdapter):
"""Async Zulip adapter implementing the BasePlatformAdapter interface.
Connects to a Zulip server and uses long-poll streaming via
get_events() to receive messages in real-time. Supports both
DMs (private messages) and stream/topic messages.
"""
def __init__(self, config, **kwargs):
platform = Platform("zulip")
super().__init__(config=config, platform=platform)
extra = getattr(config, "extra", {}) or {}
# Connection settings (env vars override config.yaml)
self.email = os.getenv("ZULIP_EMAIL") or extra.get("email", "")
self.api_key = os.getenv("ZULIP_API_KEY") or extra.get("api_key", "")
self.site = os.getenv("ZULIP_SITE") or extra.get("site", "")
self.stream = os.getenv("ZULIP_STREAM") or extra.get("stream", "agent-hub")
self.topic = os.getenv("ZULIP_TOPIC") or extra.get("topic", "")
# Auth
self.allowed_users: list = extra.get("allowed_users", [])
self._allowed_users_set: set = {u.lower() for u in self.allowed_users if isinstance(u, str)}
self._allow_all = extra.get("allow_all_users", False)
# Runtime state
self._client = None
self._recv_task: Optional[asyncio.Task] = None
self._queue_id: Optional[str] = None
self._last_event_id: Optional[int] = None
@property
def name(self) -> str:
return "Zulip"
# ── Connection lifecycle ──────────────────────────────────────────────
async def connect(self) -> bool:
"""Connect to Zulip and start the event queue."""
if not self.email or not self.api_key or not self.site:
logger.error("Zulip: email, api_key, and site must be configured")
self._set_fatal_error(
"config_missing",
"ZULIP_EMAIL, ZULIP_API_KEY, and ZULIP_SITE must be set",
retryable=False,
)
return False
try:
import zulip
self._client = zulip.Client(
email=self.email,
api_key=self.api_key,
site=self.site,
client="Hermes Agent / ZulipPlugin",
)
except Exception as e:
logger.error("Zulip: failed to create client — %s", e)
self._set_fatal_error("client_init_failed", str(e), retryable=True)
return False
# Verify connectivity by fetching own user
try:
me = await asyncio.get_event_loop().run_in_executor(
None, lambda: self._client.get_profile()
)
if me.get("result") != "success":
logger.error("Zulip: server returned error on profile: %s", me.get("msg", "unknown"))
self._set_fatal_error("auth_failed", str(me.get("msg", "unknown")), retryable=True)
self._client = None
return False
logger.info("Zulip: authenticated as %s (user_id=%s) on %s",
me.get("full_name", self.email), me.get("user_id", "?"), self.site)
except Exception as e:
logger.error("Zulip: authentication failed — %s", e)
self._set_fatal_error("auth_failed", str(e), retryable=True)
self._client = None
return False
# Register the event queue with NO narrow to receive ALL events (DMs + streams)
try:
queue_data = await asyncio.get_event_loop().run_in_executor(
None,
lambda: self._client.register(
event_types=["message"],
narrow=[], # Empty narrow = ALL events
include_subscribers=False,
)
)
if queue_data.get("result") != "success":
logger.error("Zulip: failed to register event queue: %s", queue_data.get("msg", "unknown"))
self._set_fatal_error("queue_registration_failed", str(queue_data.get("msg", "unknown")), retryable=True)
self._client = None
return False
self._queue_id = queue_data["queue_id"]
self._last_event_id = queue_data.get("last_event_id", -1)
logger.info("Zulip: event queue registered (queue_id=%s, last_event_id=%s)",
self._queue_id[:8], self._last_event_id)
except Exception as e:
logger.error("Zulip: event queue registration failed: %s", e)
self._set_fatal_error("queue_registration_failed", str(e), retryable=True)
self._client = None
return False
# Subscribe to the configured stream
try:
await asyncio.get_event_loop().run_in_executor(
None,
lambda: self._client.add_subscriptions([{"name": self.stream}])
)
logger.info("Zulip: subscribed to stream %s", self.stream)
except Exception as e:
logger.warning("Zulip: could not subscribe to stream %s: %s", self.stream, e)
# Start the receive loop
self._recv_task = asyncio.create_task(self._receive_loop())
self._mark_connected()
logger.info("Zulip: connected to %s as %s (stream=%s, topic=%s, DMs=enabled)",
self.site, self.email, self.stream, self.topic or "*")
return True
async def disconnect(self) -> None:
"""Disconnect from Zulip and clean up resources."""
self._mark_disconnected()
if self._recv_task and not self._recv_task.done():
self._recv_task.cancel()
try:
await self._recv_task
except asyncio.CancelledError:
pass
self._recv_task = None
# Deregister the event queue
if self._client and self._queue_id:
try:
await asyncio.get_event_loop().run_in_executor(
None,
lambda: self._client.deregister(self._queue_id)
)
logger.info("Zulip: deregistered queue %s", self._queue_id[:8])
except Exception as e:
logger.warning("Zulip: failed to deregister queue: %s", e)
self._client = None
self._queue_id = None
self._last_event_id = None
# ── Sending ───────────────────────────────────────────────────────────
async def send(
self,
chat_id: str,
content: str,
reply_to: Optional[str] = None,
metadata: Optional[Dict[str, Any]] = None,
) -> SendResult:
"""Send a message.
chat_id format:
- "dm:user@email.com" → send as private message to that user
- "stream_name:topic_name" → send to stream/topic
- "stream_name" → send to stream with default topic
"""
if not self._client:
return SendResult(success=False, error="Not connected")
# Detect DM vs stream
is_dm = chat_id.startswith("dm:")
if is_dm:
target_email = chat_id[3:] # Remove "dm:" prefix
message_data = {
"type": "private",
"to": [target_email],
"content": content,
}
else:
# Parse chat_id into stream:topic
if ":" in chat_id:
parts = chat_id.split(":", 1)
target_stream = parts[0]
target_topic = parts[1]
else:
target_stream = self.stream
target_topic = self.topic or "general"
if reply_to:
content = f"> @**{reply_to}** wrote…\n\n{content}"
message_data = {
"type": "stream",
"to": target_stream,
"subject": target_topic,
"content": content,
}
try:
result = await asyncio.get_event_loop().run_in_executor(
None,
lambda: self._client.send_message(message_data)
)
if result.get("result") == "success":
return SendResult(
success=True,
message_id=str(result.get("id", ""))
)
else:
return SendResult(
success=False,
error=result.get("msg", "Unknown error")
)
except Exception as e:
logger.error("Zulip: send failed: %s", e)
return SendResult(success=False, error=str(e))
async def send_typing(self, chat_id: str, metadata=None) -> None:
"""Zulip has no typing indicator — no-op."""
pass
async def get_chat_info(self, chat_id: str) -> Dict[str, Any]:
"""Return info about a chat.
chat_id format: "dm:user@email.com" or "stream_name:topic_name"
"""
if chat_id.startswith("dm:"):
user_email = chat_id[3:]
return {
"name": f"DM with {user_email}",
"type": "dm",
"chat_id": chat_id,
"user_email": user_email,
}
elif ":" in chat_id:
parts = chat_id.split(":", 1)
stream_name = parts[0]
topic_name = parts[1]
return {
"name": f"#{stream_name} > {topic_name}",
"type": "channel",
"stream": stream_name,
"topic": topic_name,
}
return {
"name": f"#{chat_id}",
"type": "channel",
"stream": chat_id,
"topic": self.topic or "general",
}
# ── Receive loop ─────────────────────────────────────────────────────
async def _receive_loop(self) -> None:
"""Main receive loop — polls Zulip events and dispatches messages."""
try:
while self._client and self._queue_id:
try:
response = await asyncio.get_event_loop().run_in_executor(
None,
lambda: self._client.get_events(
queue_id=self._queue_id,
last_event_id=self._last_event_id,
dont_block=False,
)
)
if response.get("result") != "success":
if "bad event queue" in response.get("msg", "").lower():
logger.warning("Zulip: event queue expired, re-registering...")
await self._reconnect_queue()
else:
logger.warning("Zulip: get_events error: %s", response.get("msg", "unknown"))
await asyncio.sleep(1)
continue
events = response.get("events", [])
for event in events:
self._last_event_id = event.get("id", self._last_event_id)
await self._handle_event(event)
except asyncio.CancelledError:
raise
except Exception as e:
logger.warning("Zulip: receive loop error: %s", e, exc_info=True)
await asyncio.sleep(2)
except asyncio.CancelledError:
raise
except Exception as e:
logger.error("Zulip: receive loop terminated: %s", e)
finally:
if self.is_connected:
logger.warning("Zulip: connection lost, marking disconnected")
self._set_fatal_error("connection_lost", "Zulip event queue disconnected", retryable=True)
await self._notify_fatal_error()
async def _reconnect_queue(self) -> None:
"""Re-register the event queue after expiry."""
if not self._client:
return
try:
queue_data = await asyncio.get_event_loop().run_in_executor(
None,
lambda: self._client.register(
event_types=["message"],
narrow=[],
include_subscribers=False,
)
)
if queue_data.get("result") == "success":
self._queue_id = queue_data["queue_id"]
self._last_event_id = queue_data.get("last_event_id", -1)
logger.info("Zulip: re-registered event queue (queue_id=%s)", self._queue_id[:8])
else:
logger.error("Zulip: failed to re-register queue: %s", queue_data.get("msg", "unknown"))
except Exception as e:
logger.error("Zulip: re-registration failed: %s", e)
async def _resolve_image_content(self, content: str) -> str:
"""Detect Zulip image upload URLs in content and replace with base64 data URIs.
Zulip stores uploaded files as relative paths in the content field
(e.g., [image.png](/user_uploads/.../file.png)). The remote LLM cannot
access these internal URLs, so we download them via HTTPS and embed
them as base64 data URIs inline.
"""
# Match markdown images: ![alt](url) or [alt](url)
# Zulip uploads are relative: /user_uploads/...
url_pattern = re.compile(
r'(?P<full>!?\[([^\]]*)\]\((?P<url>/user_uploads/[^)]+)\))'
)
matches = list(url_pattern.finditer(content))
if not matches:
return content
logger.info("Zulip: resolving %d image(s) in message content", len(matches))
for match in matches:
relative_url = match.group("url")
full_url = self.site.rstrip("/") + relative_url
try:
# Use the bot's own API key to authenticate the file download
resp = await asyncio.get_event_loop().run_in_executor(
None,
lambda: requests.get(
full_url,
headers={
"User-Agent": "Hermes Zulip Adapter / Image Resolver",
},
auth=requests.auth.HTTPBasicAuth(self.email, self.api_key),
timeout=15,
)
)
if resp.status_code != 200:
logger.warning("Zulip: image download failed (%d) for %s",
resp.status_code, relative_url)
continue
img_data = resp.content
mime_type = resp.headers.get("Content-Type", "image/png")
b64_data = base64.b64encode(img_data).decode("ascii")
data_uri = f"data:{mime_type};base64,{b64_data}"
# Replace the URL in content
content = content.replace(relative_url, data_uri, 1)
logger.info("Zulip: resolved image %s -> data URI (%d bytes, %s)",
relative_url.split("/")[-1], len(img_data), mime_type)
except Exception as e:
logger.error("Zulip: failed to resolve image %s: %s", relative_url, e)
continue
return content
async def _handle_event(self, event: Dict[str, Any]) -> None:
"""Process a single Zulip event and dispatch as MessageEvent.
Handles both:
- Private messages (DMs) — chat_id: "dm:sender_email"
- Stream messages — chat_id: "stream_name:topic_name"
"""
if event.get("type") != "message":
return
msg = event.get("message", {})
if not msg:
return
# Self-message filter: ignore our own messages
sender_email = msg.get("sender_email", "")
if sender_email.lower() == self.email.lower():
logger.debug("Zulip: ignoring own message")
return
# Determine message type: private (DM) or stream
zulip_msg_type = msg.get("type", "")
content = msg.get("content", "")
# Resolve image upload URLs to base64 data URIs
if content and "/user_uploads/" in content:
content = await self._resolve_image_content(content)
sender_full_name = msg.get("sender_full_name", sender_email)
# ── HANDLE PRIVATE MESSAGES (DMs) ──
if zulip_msg_type == "private":
# Extract the other participant(s)
# Zulip private messages have display_recipient as a list of user dicts
recipients = msg.get("display_recipient", [])
chat_id = f"dm:{sender_email}"
chat_name = f"DM with {sender_full_name}"
# Auth check: DM policy
if not self._allow_all:
if self._allowed_users_set and sender_email.lower() not in self._allowed_users_set:
logger.debug("Zulip: ignoring DM from unauthorized user %s", sender_email)
return
logger.info("Zulip: DM from %s (%s)", sender_full_name, sender_email)
if not self._message_handler:
return
source = self.build_source(
chat_id=chat_id,
chat_name=chat_name,
chat_type="dm",
user_id=sender_email,
user_name=sender_full_name,
)
event_obj = MessageEvent(
text=content,
message_type=MessageType.TEXT,
source=source,
message_id=str(msg.get("id", int(time.time() * 1000))),
timestamp=__import__("datetime").datetime.now(),
)
await self.handle_message(event_obj)
return
# ── HANDLE STREAM MESSAGES ──
# Only process messages from our configured stream/topic
display_recipient = msg.get("display_recipient", "")
stream_name = display_recipient if isinstance(display_recipient, str) else ""
subject = msg.get("subject", "")
if stream_name != self.stream:
return
if self.topic and subject.lower() != self.topic.lower():
return
# Build chat_id (stream:topic format for send routing)
chat_id = f"{stream_name}:{subject}" if subject else stream_name
chat_name = f"#{stream_name} > {subject}" if subject else f"#{stream_name}"
# Auth check
if not self._allow_all:
if self._allowed_users_set and sender_email.lower() not in self._allowed_users_set:
logger.debug("Zulip: ignoring stream message from unauthorized user %s", sender_email)
return
logger.debug("Zulip: stream message from %s in %s", sender_full_name, chat_id)
if not self._message_handler:
return
source = self.build_source(
chat_id=chat_id,
chat_name=chat_name,
chat_type="channel",
user_id=sender_email,
user_name=sender_full_name,
)
event_obj = MessageEvent(
text=content,
message_type=MessageType.TEXT,
source=source,
message_id=str(msg.get("id", int(time.time() * 1000))),
timestamp=__import__("datetime").datetime.now(),
)
await self.handle_message(event_obj)
# ---------------------------------------------------------------------------
# Plugin registration helpers
# ---------------------------------------------------------------------------
def check_requirements() -> bool:
"""Check if Zulip is configured."""
email = os.getenv("ZULIP_EMAIL", "")
api_key = os.getenv("ZULIP_API_KEY", "")
site = os.getenv("ZULIP_SITE", "")
return bool(email and api_key and site)
def validate_config(config) -> bool:
"""Validate that the platform config has enough info to connect."""
extra = getattr(config, "extra", {}) or {}
email = os.getenv("ZULIP_EMAIL") or extra.get("email", "")
api_key = os.getenv("ZULIP_API_KEY") or extra.get("api_key", "")
site = os.getenv("ZULIP_SITE") or extra.get("site", "")
return bool(email and api_key and site)
def is_connected(config) -> bool:
"""Check whether Zulip is configured (env or config.yaml)."""
extra = getattr(config, "extra", {}) or {}
email = os.getenv("ZULIP_EMAIL") or extra.get("email", "")
api_key = os.getenv("ZULIP_API_KEY") or extra.get("api_key", "")
site = os.getenv("ZULIP_SITE") or extra.get("site", "")
return bool(email and api_key and site)
def _env_enablement() -> dict | None:
"""Seed PlatformConfig.extra from env vars during gateway config load."""
email = os.getenv("ZULIP_EMAIL", "").strip()
api_key = os.getenv("ZULIP_API_KEY", "").strip()
site = os.getenv("ZULIP_SITE", "").strip()
if not (email and api_key and site):
return None
seed: dict = {
"email": email,
"api_key": api_key,
"site": site,
}
stream = os.getenv("ZULIP_STREAM", "").strip()
if stream:
seed["stream"] = stream
topic = os.getenv("ZULIP_TOPIC", "").strip()
if topic:
seed["topic"] = topic
home = os.getenv("ZULIP_HOME_CHANNEL") or (f"{stream}:{topic}" if stream and topic else stream or "")
if home:
seed["home_channel"] = {
"chat_id": home,
"name": os.getenv("ZULIP_HOME_CHANNEL_NAME", home),
}
return seed
async def _standalone_send(
pconfig,
chat_id: str,
message: str,
*,
thread_id: Optional[str] = None,
media_files: Optional[List[str]] = None,
force_document: bool = False,
) -> Dict[str, Any]:
"""Send a Zulip message independently, used by cron delivery."""
extra = getattr(pconfig, "extra", {}) or {}
email = os.getenv("ZULIP_EMAIL") or extra.get("email", "")
api_key = os.getenv("ZULIP_API_KEY") or extra.get("api_key", "")
site = os.getenv("ZULIP_SITE") or extra.get("site", "")
if not (email and api_key and site):
return {"error": "Zulip standalone send: ZULIP_EMAIL, ZULIP_API_KEY, and ZULIP_SITE must be configured"}
try:
import zulip
client = zulip.Client(
email=email,
api_key=api_key,
site=site,
client="Hermes Agent / ZulipPlugin (cron)",
)
if chat_id.startswith("dm:"):
target_email = chat_id[3:]
result = client.send_message({
"type": "private",
"to": [target_email],
"content": message,
})
else:
stream = os.getenv("ZULIP_STREAM") or extra.get("stream", "agent-hub")
topic = os.getenv("ZULIP_TOPIC") or extra.get("topic", "")
if ":" in chat_id:
parts = chat_id.split(":", 1)
target_stream = parts[0]
target_topic = parts[1]
else:
target_stream = stream
target_topic = topic or "general"
result = client.send_message({
"type": "stream",
"to": target_stream,
"subject": target_topic,
"content": message,
})
if result.get("result") == "success":
return {"success": True, "message_id": str(result.get("id", ""))}
return {"error": result.get("msg", "Unknown error")}
except Exception as e:
return {"error": f"Zulip standalone send failed: {e}"}
def register(ctx):
"""Plugin entry point: called by the Hermes plugin system."""
ctx.register_platform(
name="zulip",
label="Zulip",
adapter_factory=lambda cfg: ZulipAdapter(cfg),
check_fn=check_requirements,
validate_config=validate_config,
is_connected=is_connected,
required_env=["ZULIP_EMAIL", "ZULIP_API_KEY", "ZULIP_SITE"],
install_hint="pip install zulip",
setup_fn=interactive_setup,
env_enablement_fn=_env_enablement,
cron_deliver_env_var="ZULIP_HOME_CHANNEL",
standalone_sender_fn=_standalone_send,
allowed_users_env="ZULIP_ALLOWED_USERS",
allow_all_env="ZULIP_ALLOW_ALL_USERS",
max_message_length=10000,
emoji="💬",
pii_safe=False,
allow_update_command=True,
platform_hint=(
"You are chatting via Zulip — an open-source team chat platform. "
"Zulip supports Markdown formatting natively (bold **text**, italic *text*, "
"lists, code blocks with ```, links [text](url), and @mentions). "
"Messages are organized by stream + topic. Keep responses clear and "
"conversational — the full Markdown formatting is supported."
),
)
def interactive_setup() -> None:
"""Interactive `hermes gateway setup` flow for Zulip."""
from hermes_cli.setup import (
prompt,
prompt_yes_no,
save_env_value,
get_env_value,
print_header,
print_info,
print_warning,
print_success,
)
print_header("Zulip")
existing = get_env_value("ZULIP_EMAIL")
if existing:
print_info(f"Zulip: already configured (email: {existing})")
if not prompt_yes_no("Reconfigure Zulip?", False):
return
print_info("Connect Hermes to a Zulip team chat server.")
print_info(" Requires the 'zulip' Python package (pip install zulip).")
print()
email = prompt("Zulip bot email (e.g. tanko-bot@chat.sysloggh.net)", default=existing or "")
if not email:
print_warning("Email is required — skipping Zulip setup")
return
save_env_value("ZULIP_EMAIL", email.strip())
api_key = prompt("Zulip bot API key", password=True)
if not api_key:
print_warning("API key is required — skipping Zulip setup")
return
save_env_value("ZULIP_API_KEY", api_key)
site = prompt("Zulip server URL (e.g. https://chat.sysloggh.net)",
default=get_env_value("ZULIP_SITE") or "")
if not site:
print_warning("Server URL is required — skipping Zulip setup")
return
save_env_value("ZULIP_SITE", site.strip())
stream = prompt("Stream to join (default: agent-hub)",
default=get_env_value("ZULIP_STREAM") or "agent-hub")
if stream:
save_env_value("ZULIP_STREAM", stream.strip())
topic = prompt("Topic to listen on (optional)",
default=get_env_value("ZULIP_TOPIC") or "")
if topic:
save_env_value("ZULIP_TOPIC", topic.strip())
print()
print_info("🔒 Access control: restrict who can message the bot")
allow_all = prompt_yes_no("Allow all users to talk to the bot?", False)
if allow_all:
save_env_value("ZULIP_ALLOW_ALL_USERS", "true")
save_env_value("ZULIP_ALLOWED_USERS", "")
print_warning("⚠️ Open access — any user in the stream can command the bot.")
else:
save_env_value("ZULIP_ALLOW_ALL_USERS", "false")
allowed = prompt(
"Allowed user emails (comma-separated)",
default=get_env_value("ZULIP_ALLOWED_USERS") or "",
)
if allowed:
save_env_value("ZULIP_ALLOWED_USERS", allowed.replace(" ", ""))
print_success("Allowlist configured")
else:
save_env_value("ZULIP_ALLOWED_USERS", "")
print_info("No users allowed — bot will ignore all messages until you add users.")
print()
print_success("Zulip configuration saved to ~/.hermes/.env")
print_info("Restart the gateway for changes to take effect: hermes gateway restart")
hermes-zulip-plugin/src/hermes_zulip/standalone_bridge.py
+350
View File
@@ -0,0 +1,350 @@
#!/usr/bin/env python3
"""
Standalone Zulip → LiteLLM Bridge
Polls Zulip for DMs, downloads image attachments via Zulip API,
and calls LiteLLM proxy directly (bypassing Hermes Gateway).
Usage:
ZULIP_EMAIL=tanko-bot@chat.sysloggh.net \
ZULIP_API_KEY=xxx \
ZULIP_SITE=https://chat.sysloggh.net \
LITELLM_API_KEY=sk-xxx \
LITELLM_BASE_URL=https://litellm.sysloggh.net/v1 \
VISION_MODEL=gemma-4-12b \
TEXT_MODEL=generic-llm \
AGENT_NAME=Tanko \
python3 standalone_bridge.py
"""
import asyncio
import base64
import logging
import os
import re
import sys
import time
from typing import Any, Dict, List, Optional
import requests
from openai import AsyncOpenAI
# ── Logging ────────────────────────────────────────────────────────────────
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
datefmt="%Y-%m-%d %H:%M:%S",
stream=sys.stderr,
)
logger = logging.getLogger("zulip-bridge")
# ── Config ─────────────────────────────────────────────────────────────────
ZULIP_EMAIL = os.environ.get("ZULIP_EMAIL", "")
ZULIP_API_KEY = os.environ.get("ZULIP_API_KEY", "")
ZULIP_SITE = os.environ.get("ZULIP_SITE", "https://chat.sysloggh.net")
ZULIP_STREAM = os.environ.get("ZULIP_STREAM", "agent-hub")
LITELLM_API_KEY = os.environ.get("LITELLM_API_KEY", "")
LITELLM_BASE_URL = os.environ.get("LITELLM_BASE_URL", "https://litellm.sysloggh.net/v1")
VISION_MODEL = os.environ.get("VISION_MODEL", "gemma-4-12b")
TEXT_MODEL = os.environ.get("TEXT_MODEL", "generic-llm")
AGENT_NAME = os.environ.get("AGENT_NAME", "Tanko")
# URL patterns for Zulip image uploads
IMAGE_URL_RE = re.compile(r"/user_uploads/[^\s)]+")
# ── Auto-load env file from disk ──────────────────────────────────────
_ENV_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)), "bridge.env")
if os.path.exists(_ENV_FILE):
import re as _re
for _line in open(_ENV_FILE):
_line = _line.strip()
if _line and not _line.startswith("#") and "=" in _line:
_key, _val = _line.split("=", 1)
os.environ.setdefault(_key, _val)
def check_config() -> bool:
"""Verify all required environment variables are set."""
missing = []
if not ZULIP_EMAIL:
missing.append("ZULIP_EMAIL")
if not ZULIP_API_KEY:
missing.append("ZULIP_API_KEY")
if not LITELLM_API_KEY:
missing.append("LITELLM_API_KEY")
if not VISION_MODEL:
missing.append("VISION_MODEL")
if not TEXT_MODEL:
missing.append("TEXT_MODEL")
if missing:
logger.error("Missing required env vars: %s", ", ".join(missing))
return False
return True
def download_zulip_image(relative_url: str) -> Optional[bytes]:
"""Download an image from Zulip via HTTP Basic Auth."""
full_url = ZULIP_SITE.rstrip("/") + relative_url
try:
resp = requests.get(
full_url,
auth=requests.auth.HTTPBasicAuth(ZULIP_EMAIL, ZULIP_API_KEY),
timeout=15,
)
if resp.status_code != 200:
logger.warning("Image download failed (%d) for %s", resp.status_code, relative_url)
return None
return resp.content
except Exception as e:
logger.error("Image download error for %s: %s", relative_url, e)
return None
def extract_image_urls(content: str) -> List[str]:
"""Extract relative Zulip upload URLs from message content."""
matches = IMAGE_URL_RE.findall(content)
# Deduplicate and clean trailing punctuation
seen = set()
urls = []
for url in matches:
# Strip trailing characters that aren't part of URL
clean = url.rstrip(").,]!?;:")
if clean not in seen:
seen.add(clean)
urls.append(clean)
return urls
def build_vision_messages(content: str, image_data: List[Dict]) -> List[Dict]:
"""Build OpenAI-style messages with text and image parts."""
if not image_data:
return [{"role": "user", "content": content}]
parts = [{"type": "text", "text": content}]
for img in image_data:
parts.append({
"type": "image_url",
"image_url": {
"url": f"data:{img['mime']};base64,{img['b64']}",
"detail": "auto",
},
})
return [{"role": "user", "content": parts}]
async def call_litellm(messages: List[Dict], model: str) -> Optional[str]:
"""Call LiteLLM proxy and return response text."""
client = AsyncOpenAI(
api_key=LITELLM_API_KEY,
base_url=LITELLM_BASE_URL,
)
try:
response = await client.chat.completions.create(
model=model,
messages=messages,
max_tokens=4096,
temperature=0.7,
)
return response.choices[0].message.content
except Exception as e:
logger.error("LiteLLM call failed: %s", e, exc_info=True)
return None
def build_source(chat_id: str, chat_name: str, chat_type: str,
user_id: str, user_name: str) -> Dict:
"""Build source metadata for logging/context."""
return {
"chat_id": chat_id,
"chat_name": chat_name,
"chat_type": chat_type,
"user_id": user_id,
"user_name": user_name,
"platform": "zulip",
}
async def process_message(client, msg: Dict) -> None:
"""Process a single Zulip message: download images, call LLM, reply."""
msg_type = msg.get("type", "")
content = msg.get("content", "")
sender_email = msg.get("sender_email", "")
sender_full_name = msg.get("sender_full_name", sender_email)
msg_id = msg.get("id", 0)
# Ignore own messages
if sender_email.lower() == ZULIP_EMAIL.lower():
return
logger.info("Processing message from %s (%s): type=%s, id=%d",
sender_full_name, sender_email, msg_type, msg_id)
# ── Extract and download images ──
image_urls = extract_image_urls(content)
image_data: List[Dict] = []
if image_urls:
logger.info("Found %d image(s) in message", len(image_urls))
for url in image_urls:
raw_bytes = download_zulip_image(url)
if raw_bytes:
# Detect MIME from extension
ext = url.rsplit(".", 1)[-1].lower() if "." in url else "png"
mime_map = {
"jpg": "image/jpeg",
"jpeg": "image/jpeg",
"png": "image/png",
"gif": "image/gif",
"webp": "image/webp",
}
mime = mime_map.get(ext, "image/png")
b64 = base64.b64encode(raw_bytes).decode("ascii")
image_data.append({"mime": mime, "b64": b64})
logger.info("Downloaded image (%d bytes, %s)", len(raw_bytes), mime)
# ── Determine model ──
model = VISION_MODEL if image_data else TEXT_MODEL
logger.info("Using model: %s (images=%d)", model, len(image_data))
# ── Build messages and call LLM ──
messages = build_vision_messages(content, image_data)
response = await call_litellm(messages, model)
if not response:
logger.error("LLM returned no response for message %d", msg_id)
return
# ── Send reply ──
reply_content = response[:5000] # Zulip max message length
if msg_type == "private":
message_data = {
"type": "private",
"to": [sender_email],
"content": reply_content,
}
else:
display_recipient = msg.get("display_recipient", "")
stream_name = display_recipient if isinstance(display_recipient, str) else ""
subject = msg.get("subject", "general")
message_data = {
"type": "stream",
"to": stream_name,
"subject": subject,
"content": reply_content,
}
try:
result = client.send_message(message_data)
if result.get("result") == "success":
logger.info("Reply sent: %s", result.get("id", ""))
else:
logger.error("Failed to send reply: %s", result.get("msg", "unknown"))
except Exception as e:
logger.error("Send reply failed: %s", e)
async def poll_loop(client) -> None:
"""Main polling loop: register queue and long-poll for events."""
logger.info("Connecting to Zulip as %s on %s", ZULIP_EMAIL, ZULIP_SITE)
# Register event queue with empty narrow (all events)
queue_data = client.register(
event_types=["message"],
narrow=[], # Empty = all events including DMs
include_subscribers=False,
)
if queue_data.get("result") != "success":
logger.error("Queue registration failed: %s", queue_data.get("msg", "unknown"))
return
queue_id = queue_data["queue_id"]
last_event_id = queue_data.get("last_event_id", -1)
logger.info("Event queue registered: %s (last_event=%s)",
queue_id[:8], last_event_id)
# Subscribe to configured stream
try:
client.add_subscriptions([{"name": ZULIP_STREAM}])
logger.info("Subscribed to stream: %s", ZULIP_STREAM)
except Exception as e:
logger.warning("Could not subscribe to stream: %s", e)
# Poll loop
while True:
try:
response = client.get_events(
queue_id=queue_id,
last_event_id=last_event_id,
)
if response.get("result") != "success":
error_msg = response.get("msg", "unknown")
logger.warning("Poll error: %s — reconnecting...", error_msg)
await asyncio.sleep(5)
# Re-register queue
queue_data = client.register(
event_types=["message"],
narrow=[],
include_subscribers=False,
)
if queue_data.get("result") == "success":
queue_id = queue_data["queue_id"]
last_event_id = queue_data.get("last_event_id", -1)
logger.info("Re-registered queue: %s", queue_id[:8])
continue
events = response.get("events", [])
if events:
for event in events:
last_event_id = max(last_event_id, event.get("id", last_event_id))
if event.get("type") == "message":
msg = event.get("message", {})
asyncio.create_task(process_message(client, msg))
# Small delay to prevent busy-polling
await asyncio.sleep(0.1)
except KeyboardInterrupt:
logger.info("Shutdown requested")
break
except Exception as e:
logger.error("Poll loop error: %s", e, exc_info=True)
await asyncio.sleep(5)
# Cleanup
try:
client.deregister(queue_id)
except Exception:
pass
def main():
"""Entry point — creates Zulip client and starts async poll loop."""
if not check_config():
sys.exit(1)
import zulip
client = zulip.Client(
email=ZULIP_EMAIL,
api_key=ZULIP_API_KEY,
site=ZULIP_SITE,
client=f"Zulip Bridge / {AGENT_NAME}",
)
logger.info("Starting Zulip bridge for %s (vision=%s, text=%s)",
AGENT_NAME, VISION_MODEL, TEXT_MODEL)
try:
asyncio.run(poll_loop(client))
except KeyboardInterrupt:
logger.info("Bridge stopped")
if __name__ == "__main__":
main()
pi-zulip-extension/package-lock.json
Generated
+2904
View File
File diff suppressed because it is too large Load Diff
pi-zulip-extension/package.json
+12 -7
View File
@@ -1,22 +1,27 @@
{ {
"name": "pi-zulip-extension", "name": "pi-zulip-extension",
"version": "0.1.0", "version": "0.2.0",
"description": "pi extension for Zulip agent communication — connects Abiba to the Sysloggh agent mesh", "description": "pi extension for Zulip agent communication — connects Abiba to the Sysloggh agent mesh",
"main": "src/index.ts", "main": "dist/index.js",
"type": "module", "type": "module",
"scripts": { "scripts": {
"build": "tsc", "build": "tsc",
"check": "tsc --noEmit" "check": "tsc --noEmit"
}, },
"dependencies": { "dependencies": {
"zulip-js": "^2.0.0", "zulip-js": "^2.0.0"
"yaml": "^2.0.0"
}, },
"devDependencies": { "devDependencies": {
"@earendil-works/pi-coding-agent": "*", "@mariozechner/pi-coding-agent": "*",
"typescript": "^5.0.0" "typescript": "^5.9.3"
}, },
"keywords": ["pi", "zulip", "agent", "abiba", "sysloggh"], "keywords": [
"pi",
"zulip",
"agent",
"abiba",
"sysloggh"
],
"license": "UNLICENSED", "license": "UNLICENSED",
"private": true "private": true
} }
pi-zulip-extension/src/index.ts
+598 -85
View File
@@ -1,120 +1,633 @@
/** /**
* pi-zulip-extension — Zulip agent communication plugin for pi agents * Zulip Extension for pi v2
* *
* Deploy to: ~/.pi/agent/extensions/zulip.ts * DM-first architecture per ADR-001/ADR-002.
* Config: config.yaml (alongside extension, or symlinked) * Background Zulip event queue poller with subprocess-based message processing.
* No registerPreprocessor/processMessage — pi v2 (@mariozechner/pi-coding-agent)
* does not expose these.
* *
* Connects a pi agent (Abiba) to the Sysloggh Zulip agent mesh. * Instead: the extension registers a session_start hook that starts a background
* Listens for @mentions of abiba-bot in #agent-hub and forwards * polling loop. When a DM is received, it spawns `pi -p --mode json <message>`
* to the pi agent. Responses are posted back to the same Zulip topic. * as a subprocess and sends the response back to Zulip.
*
* @see ADR-007: Platform-native plugin contracts
* @see docs/ARCHITECTURE.md
*/ */
import type { ExtensionAPI } from "@earendil-works/pi-coding-agent"; import type { ExtensionAPI, ExtensionContext } from "@mariozechner/pi-coding-agent";
import { readFileSync } from "fs"; import { createClient } from "zulip-js";
import { parse } from "yaml"; import { spawnSync } from "node:child_process";
import http from "node:http";
import fs from "node:fs";
import path from "node:path";
import url from "node:url";
// zulip-js types are declared in types.d.ts in this directory
// --------------------------------------------------------------------------- // ---------------------------------------------------------------------------
// Config // Configuration
// --------------------------------------------------------------------------- // ---------------------------------------------------------------------------
interface ZulipConfig { interface ZulipConfig {
agent: {
name: string;
display_name: string;
zulip_bot_name: string;
owner_email: string;
private_topic: string;
};
zulip: { zulip: {
server_url: string;
email: string; email: string;
api_key: string; api_key: string;
stream: string; site: string;
all_bots_user_id: number;
}; };
error_handling: { agent: {
timeout_seconds: number; name: string;
retry_count: number; owner_email: string;
retry_delay_seconds: number; display_name?: string;
graceful_message: string;
};
monitoring: {
health_endpoint_enabled: boolean;
health_port: number;
log_level: string;
}; };
health_port?: number;
poll_interval_ms?: number;
max_retries?: number;
retry_delay_ms?: number;
pi_command?: string;
} }
function loadConfig(): ZulipConfig { function loadConfig(): ZulipConfig {
const raw = readFileSync("config.yaml", "utf-8"); // Check env vars first
const cfg = parse(raw) as ZulipConfig; const email = process.env.ZULIP_EMAIL;
cfg.zulip.api_key = process.env["ZULIP_API_KEY"] ?? cfg.zulip.api_key; const apiKey = process.env.ZULIP_API_KEY;
return cfg; const site = process.env.ZULIP_SITE;
if (email && apiKey && site) {
return {
zulip: { email, api_key: apiKey, site },
agent: {
name: process.env.AGENT_NAME ?? "abiba",
owner_email: process.env.AGENT_OWNER_EMAIL ?? "jerome@sysloggh.com",
},
health_port: parseInt(process.env.HEALTH_PORT ?? "9200", 10),
poll_interval_ms: 3000,
max_retries: 3,
retry_delay_ms: 5000,
pi_command: "pi",
};
} }
// --------------------------------------------------------------------------- // Fall back to config.yaml
// Extension entry point const configDir = path.dirname(url.fileURLToPath(import.meta.url));
// --------------------------------------------------------------------------- const configPath = path.join(configDir, "config.yaml");
export default function (pi: ExtensionAPI) { if (!fs.existsSync(configPath)) {
const config = loadConfig(); throw new Error(
`No config.yaml found at ${configPath}. Set ZULIP_EMAIL, ZULIP_API_KEY, ZULIP_SITE env vars or provide config.yaml.`,
// Health endpoint
if (config.monitoring.health_endpoint_enabled) {
startHealthServer(config);
}
pi.on("session_start", async (_event, ctx) => {
ctx.ui.notify(
`Zulip extension loaded for ${config.agent.display_name} (${config.agent.zulip_bot_name})`,
"info"
); );
}
const raw = fs.readFileSync(configPath, "utf-8");
// Simple YAML parser for nested keys (avoids yaml dependency)
function parseYamlValue(text: string, key: string): string | undefined {
const re = new RegExp(`^${key.replace(/\./g, "\\.")}:\\s*(.+)`, "m");
const m = text.match(re);
if (!m) return undefined;
return m[1].replace(/^['"]|['"]$/g, "").trim();
}
const parsed: ZulipConfig = {
zulip: {
email: parseYamlValue(raw, "zulip.email") ?? "",
api_key: parseYamlValue(raw, "zulip.api_key") ?? "",
site: parseYamlValue(raw, "zulip.site") ?? "",
},
agent: {
name: parseYamlValue(raw, "agent.name") ?? "abiba",
owner_email: parseYamlValue(raw, "agent.owner_email") ?? "jerome@sysloggh.com",
display_name: parseYamlValue(raw, "agent.display_name"),
},
health_port: parseInt(parseYamlValue(raw, "health_port") ?? "9200", 10),
poll_interval_ms: 3000,
max_retries: 3,
retry_delay_ms: 5000,
pi_command: "pi",
};
return parsed;
}
// ---------------------------------------------------------------------------
// Zulip API helpers
// ---------------------------------------------------------------------------
interface ZulipMessage {
id: number;
type: "stream" | "private";
display_recipient?: string | Array<{ id: number; email: string; full_name: string }>;
sender_id: number;
sender_email: string;
sender_full_name: string;
content: string;
subject: string;
stream_id?: number;
}
interface ZulipEvent {
id: number;
type: "message";
message: ZulipMessage;
}
/**
* Register a Zulip event queue and poll for events.
* Returns an async generator that yields events.
*/
async function createZulipQueue(config: ZulipConfig) {
const client = await createClient({
username: config.zulip.email,
apiKey: config.zulip.api_key,
realm: config.zulip.site,
}); });
// Diagnostic tool // Register queue
pi.registerTool({ const queueRes = await client.queues.register({
name: "zulip_status", event_types: ["message"],
label: "Zulip Status", // Subscribe to DMs by specifying narrow: all_private_messages via API
description: "Check Zulip connection status and bot identity", // The zulip-js client handles this
parameters: {},
execute: async () => ({
connected: false, // TODO: track connection state
bot: config.agent.zulip_bot_name,
stream: config.zulip.stream,
server: config.zulip.server_url,
owner: config.agent.owner_email,
private_topic: config.agent.private_topic,
}),
}); });
const queueId = queueRes.queue_id;
let lastEventId = queueRes.last_event_id ?? -1;
return {
client,
queueId,
lastEventId,
/**
* Poll the event queue and return any new events.
*/
async poll(): Promise<ZulipEvent[]> {
try {
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 = await res.json();
if (data.events && Array.isArray(data.events)) {
for (const event of data.events) {
if (event.id > lastEventId) {
lastEventId = event.id;
}
}
return data.events.filter((e: any) => e.type === "message");
}
return [];
} catch (err) {
throw err;
}
},
/**
* Send a message to Zulip via the API.
*/
async sendMessage(params: {
type: "private" | "stream";
to: number | string;
subject?: string;
content: string;
}): Promise<void> {
const formBody = new URLSearchParams();
formBody.append("type", params.type);
// Zulip API: private messages need JSON array of user IDs, stream messages accept string/number
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()}`);
}
},
/**
* Send typing indicator.
*/
async sendTypingNotification(userIds: number[], operation: "start" | "stop"): Promise<void> {
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 endpoint // Health endpoint
// --------------------------------------------------------------------------- // ---------------------------------------------------------------------------
function startHealthServer(config: ZulipConfig) { function startHealthServer(port: number, getState: () => Record<string, unknown>) {
const http = require("http") as typeof import("http"); const server = http.createServer((req, res) => {
const port = config.monitoring.health_port; if (req.url === "/health" || req.url === "/") {
const startTime = Date.now(); const state = getState();
http
.createServer((_req: any, res: any) => {
res.writeHead(200, { "Content-Type": "application/json" }); res.writeHead(200, { "Content-Type": "application/json" });
res.end( res.end(JSON.stringify({ status: "ok", ...state }));
JSON.stringify({ } else {
status: "ok", res.writeHead(404);
res.end("Not found");
}
});
server.on("error", (err: any) => {
if (err.code === "EADDRINUSE") {
console.warn(`[zulip-ext] Port :${port} already in use — skipping health endpoint`);
} else {
console.error(`[zulip-ext] Health server error:`, err.message);
}
});
server.listen(port, "127.0.0.1", () => {
console.log(`[zulip-ext] Health endpoint on :${port}`);
});
return server;
}
// ---------------------------------------------------------------------------
// Detect @all-bots content
// ---------------------------------------------------------------------------
const ALL_BOTS_RE = /@\*\*(?:all-bots|All Bots)\*\*/i;
function isAllBotsMention(content: string): boolean {
return ALL_BOTS_RE.test(content);
}
// ---------------------------------------------------------------------------
// Pi subprocess invocation
// ---------------------------------------------------------------------------
interface PiResult {
content: string;
error?: string;
}
function invokePi(config: ZulipConfig, message: string): PiResult {
const command = config.pi_command ?? "pi";
// Build a system prompt prefix that describes the incoming Zulip message
const fullInput = `[Zulip DM] ${message}`;
const result = spawnSync(command, ["-p", "--mode", "json", fullInput], {
encoding: "utf-8",
timeout: 60_000, // 60 second timeout
maxBuffer: 512 * 1024, // 512KB max output
env: {
...process.env,
PI_MODE: "extension",
},
});
if (result.error) {
return { content: "", error: result.error.message };
}
if (result.status !== 0) {
const stderr = result.stderr?.trim() ?? "";
const stdout = result.stdout?.trim() ?? "";
return {
content: stderr || stdout,
error: `pi exited with status ${result.status}`,
};
}
// Try to parse JSON output from pi --mode json
const stdout = result.stdout?.trim() ?? "";
try {
const parsed = JSON.parse(stdout);
// pi --mode json returns { content: "..." } or { message: "..." }
const content = parsed.content ?? parsed.message ?? parsed.response ?? stdout;
return { content: String(content) };
} catch {
// Not JSON — return raw stdout
return { content: stdout || "[pi returned no output]" };
}
}
// ---------------------------------------------------------------------------
// Main extension
// ---------------------------------------------------------------------------
export default async function (pi: ExtensionAPI) {
const config = loadConfig();
let queue: Awaited<ReturnType<typeof createZulipQueue>> | null = null;
let pollTimer: ReturnType<typeof setInterval> | null = null;
let retryCount = 0;
let connected = false;
let lastError: string | null = null;
let processedCount = 0;
let healthServer: ReturnType<typeof startHealthServer> | null = null;
// Track state for health endpoint
function getState() {
return {
connected,
email: config.zulip.email,
site: config.zulip.site,
agent: config.agent.name, agent: config.agent.name,
uptime_seconds: Math.floor((Date.now() - startTime) / 1000), is_owner: config.agent.owner_email,
zulip_connected: false, queue_id: queue?.queueId ?? null,
last_message_time: null, last_event_id: queue?.lastEventId ?? null,
timestamp: new Date().toISOString(), messages_processed: processedCount,
}) last_error: lastError,
); retry_count: retryCount,
}) };
.listen(port, "127.0.0.1", () => { }
console.log(`[zulip-extension] Health endpoint on :${port}`);
/**
* Process a single Zulip event.
*/
async function processEvent(event: ZulipEvent): Promise<void> {
const msg = event.message;
// Ignore non-message events
if (event.type !== "message") return;
// Ignore own messages (sent by this bot)
if (msg.sender_email === config.zulip.email) return;
// DM-first architecture (ADR-001, ADR-002): process private messages
if (msg.type === "private") {
processedCount++;
// Extract sender info
const senderName = msg.sender_full_name || msg.sender_email;
const senderId = msg.sender_id;
console.log(`[zulip-ext] DM from ${senderName} (id=${senderId}): ${msg.content.slice(0, 80)}...`);
// Send typing indicator
try {
await queue!.sendTypingNotification([senderId], "start");
} catch {
// Non-critical
}
// Process via pi subprocess
const formattedMessage = `New Zulip DM from @${senderName}: ${msg.content}`;
const result = invokePi(config, formattedMessage);
// Stop typing
try {
await queue!.sendTypingNotification([senderId], "stop");
} catch {
// Non-critical
}
if (result.error) {
console.error(`[zulip-ext] pi processing failed: ${result.error}`);
lastError = result.error;
await queue!.sendMessage({
type: "private",
to: senderId,
content: `:warning: Sorry, I couldn't process your message (error: ${result.error}). Please try again.`,
});
return;
}
if (!result.content) {
await queue!.sendMessage({
type: "private",
to: senderId,
content: `:grey_question: I received your message but produced no output.`,
});
return;
}
// Send response back to Zulip
const response = result.content.length > 1500
? result.content.slice(0, 1500) + "\n\n[...response truncated, see your agent for full output]"
: result.content;
await queue!.sendMessage({
type: "private",
to: senderId,
content: response,
});
console.log(`[zulip-ext] Replied to ${senderName} (${response.length} chars)`);
return;
}
// Stream messages: only respond to @all-bots (ADR-006)
if (msg.type === "stream") {
if (isAllBotsMention(msg.content)) {
processedCount++;
const streamName = typeof msg.display_recipient === "string"
? msg.display_recipient
: "unknown";
const topic = msg.subject || "general";
console.log(`[zulip-ext] @all-bots in #${streamName} > ${topic}`);
const result = invokePi(config, `[Zulip @all-bots from ${msg.sender_full_name} in #${streamName} > ${topic}]: ${msg.content.replace(/@\*\*(?:all-bots|All Bots)\*\*/gi, "").trim()}`);
if (result.error) {
console.error(`[zulip-ext] @all-bots processing failed: ${result.error}`);
return;
}
await queue!.sendMessage({
type: "stream",
to: msg.stream_id ?? streamName,
subject: topic,
content: `${result.content.slice(0, 1500)}`,
});
}
}
}
/**
* Polling loop.
*/
async function startPolling(): Promise<void> {
console.log(`[zulip-ext] Connecting to ${config.zulip.site} as ${config.zulip.email}...`);
try {
queue = await createZulipQueue(config);
connected = true;
retryCount = 0;
lastError = null;
console.log(`[zulip-ext] Connected, queue: ${queue.queueId} (last_event_id=${queue.lastEventId})`);
// Start polling
pollTimer = setInterval(async () => {
if (!queue) return;
try {
const events = await queue.poll();
for (const event of events) {
await processEvent(event);
}
} catch (err) {
const errMsg = err instanceof Error ? err.message : String(err);
// Check if queue was deregistered (need to reconnect)
if (errMsg.includes("BAD_EVENT_QUEUE_ID") || errMsg.includes("queue_id")) {
console.log(`[zulip-ext] Queue expired, reconnecting...`);
connected = false;
// Clear timer and restart connection
if (pollTimer) clearInterval(pollTimer);
pollTimer = null;
// Reconnect after delay
setTimeout(() => startPolling(), config.retry_delay_ms ?? 5000);
return;
}
// Network errors — retry on next poll interval
lastError = errMsg;
console.error(`[zulip-ext] Poll error: ${errMsg}`);
}
}, config.poll_interval_ms ?? 3000);
} catch (err) {
const errMsg = err instanceof Error ? err.message : String(err);
lastError = errMsg;
console.error(`[zulip-ext] Connection failed: ${errMsg}`);
if (retryCount < (config.max_retries ?? 3)) {
retryCount++;
const delay = (config.retry_delay_ms ?? 5000) * retryCount;
console.log(`[zulip-ext] Retrying in ${delay / 1000}s (attempt ${retryCount})...`);
setTimeout(() => startPolling(), delay);
} else {
console.error(`[zulip-ext] Max retries (${config.max_retries}) reached. Giving up.`);
lastError = `Connection failed after ${config.max_retries} retries: ${errMsg}`;
}
}
}
// -------------------------------------------------------------------------
// Extension lifecycle
// -------------------------------------------------------------------------
pi.on("session_start", async (_event, _ctx: ExtensionContext) => {
console.log(`[zulip-ext] Starting Zulip extension for ${config.agent.name} (${config.zulip.email})`);
// Start health endpoint
if (!healthServer) {
healthServer = startHealthServer(config.health_port ?? 9200, getState);
}
// Begin Zulip event polling
startPolling();
});
pi.on("session_shutdown", async (_event, _ctx: ExtensionContext) => {
console.log(`[zulip-ext] Shutting down...`);
// Clean up poller
if (pollTimer) {
clearInterval(pollTimer);
pollTimer = null;
}
// Clean up health server
if (healthServer) {
healthServer.close();
healthServer = null;
}
connected = false;
queue = null;
});
// Register /zulip-status command for diagnostics
pi.registerCommand("zulip-status", {
description: "Show Zulip extension status and stats",
handler: async (_args: unknown, ctx: ExtensionContext) => {
const state = getState();
const lines: string[] = [
"=== Zulip Extension Status ===",
"",
`Agent: ${state.agent}`,
`Email: ${state.email}`,
`Server: ${state.site}`,
`Owner: ${state.is_owner}`,
`Connected: ${state.connected ? "✅ Yes" : "❌ No"}`,
`Queue ID: ${state.queue_id ?? "N/A"}`,
`Last Event ID: ${state.last_event_id ?? "N/A"}`,
`Messages: ${state.messages_processed} processed`,
`Retries: ${state.retry_count}`,
`Health Port: :${config.health_port ?? 9200}`,
];
if (state.last_error) {
lines.push("", `Last Error: ${state.last_error}`);
}
lines.push("", "ADRs followed: DM-first (ADR-001), DM-routing (ADR-002), @all-bots content (ADR-006)");
ctx.ui.notify(lines.join("\n"), "info");
},
});
// Also register a /zulip-command-js tool for ad-hoc Zulip API calls
// Register zulip-send-test command
// Usage: /zulip-send-test "targetEmail" "message body"
pi.registerCommand("zulip-send-test", {
description: "Send a test message to Zulip user: /zulip-send-test <email> <message>",
handler: async (args: string, ctx: ExtensionContext) => {
const parts = args.trim().match(/^([^\s]+)\s+(.+)$/);
const targetEmail = parts?.[1] ?? config.agent.owner_email;
const message = parts?.[2] ?? "Test message from pi Zulip extension";
if (!queue) {
ctx.ui.notify("Zulip queue not connected. Use /zulip-status to check.", "error");
return;
}
try {
await queue.sendMessage({
type: "private",
to: targetEmail,
content: message,
});
ctx.ui.notify(`Test message sent to ${targetEmail}`, "info");
} catch (err) {
const errMsg = err instanceof Error ? err.message : String(err);
ctx.ui.notify(`Failed to send: ${errMsg}`, "error");
}
},
}); });
} }
pi-zulip-extension/src/types.d.ts
Vendored
+34
View File
@@ -0,0 +1,34 @@
// Type declarations for zulip-js (no @types available)
declare module "zulip-js" {
interface ZulipClient {
queues: {
register(params: {
event_types: string[];
narrow?: Array<Array<string | number>>;
}): Promise<{ queue_id: string; last_event_id: number }>;
};
users: {
me: {
get(): Promise<any>;
};
};
messages: {
store: {
send(rawContent: string): Promise<any>;
};
};
events: {
retrieve(params: {
queue_id: string;
last_event_id: number;
dont_block?: boolean;
}): Promise<{ events: any[]; result?: string; msg?: string }>;
};
}
export function createClient(config: {
username: string;
apiKey: string;
realm: string;
}): Promise<ZulipClient>;
}
scripts/deploy.sh
+35 -37
View File
@@ -1,5 +1,5 @@
#!/usr/bin/env bash #!/usr/bin/env bash
# deploy.sh GitOps deployment for zulip-platform-plugins # deploy.sh — GitOps deployment for zulip-platform-plugins
# Usage: ./deploy.sh <tag|branch> [--ct=<agent>] [--dry-run] # Usage: ./deploy.sh <tag|branch> [--ct=<agent>] [--dry-run]
# ./deploy.sh v1.0.0 # Deploy to all 6 CTs # ./deploy.sh v1.0.0 # Deploy to all 6 CTs
# ./deploy.sh main # Deploy latest (staging only!) # ./deploy.sh main # Deploy latest (staging only!)
@@ -36,28 +36,28 @@ declare -A AGENTS=(
["abiba"]="amdpve CT 100" ["abiba"]="amdpve CT 100"
) )
# Platform paths per agent type # Platform classification for each agent
declare -A PLUGIN_PATHS=( declare -A PLATFORM=(
["tanko"]="/opt/hermes-zulip-plugin" ["tanko"]="hermes"
["mumuni"]="/opt/hermes-zulip-plugin" ["mumuni"]="hermes"
["koonimo"]="/opt/hermes-zulip-plugin" ["koonimo"]="hermes"
["koby\"]="/opt/hermes-zulip-plugin" ["koby"]="hermes"
["kagentz\"]="/opt/agent-zero-plugin" ["kagentz"]="agent-zero"
["abiba\"]="/root/.pi/agent/extensions/zulip.ts" ["abiba"]="pi"
) )
# Correcting the backslash artifacts from the original file # Plugin paths by platform
PLUGIN_PATHS["koby"]="/opt/hermes-zulip-plugin" declare -A PLUGIN_PATHS_BY_PLATFORM=(
PLUGIN_PATHS["kagentz"]="/opt/agent-zero-plugin" ["hermes"]="/opt/hermes-zulip-plugin"
PLUGIN_PATHS["abiba"]="/root/.pi/agent/extensions/zulip.ts" ["agent-zero"]="/opt/agent-zero-plugin"
["pi"]="/root/.pi/agent/extensions/zulip.ts"
)
declare -A SERVICE_NAMES=( # Service names by platform
["tanko"]="zulip-plugin" declare -A SERVICE_NAMES_BY_PLATFORM=(
["mumuni"]="zulip-plugin" ["hermes"]="zulip-plugin"
["koonimo"]="zulip-plugin" ["agent-zero"]="zulip-plugin"
["koby"]="zulip-plugin" ["pi"]="pi" # pi reload, not systemctl
["kagentz"]="zulip-plugin"
["abiba"]="pi" # pi reload, not systemctl
) )
GITEA_REPO="https://git.sysloggh.net/SyslogSolution/zulip-platform-plugins.git" GITEA_REPO="https://git.sysloggh.net/SyslogSolution/zulip-platform-plugins.git"
@@ -72,15 +72,16 @@ log() {
deploy_agent() { deploy_agent() {
local agent="$1" local agent="$1"
local ct_info="${AGENTS[$agent]}" local ct_info="${AGENTS[$agent]}"
local plugin_path="${PLUGIN_PATHS[$agent]}" local platform="${PLATFORM[$agent]}"
local service="${SERVICE_NAMES[$agent]}" local plugin_path="${PLUGIN_PATHS_BY_PLATFORM[$platform]}"
local service="${SERVICE_NAMES_BY_PLATFORM[$platform]}"
log "=== Deploying $agent ($ct_info) @ $TAG ===" log "=== Deploying $agent ($ct_info) @ $TAG (platform: $platform) ==="
log "Path: $plugin_path" log "Path: $plugin_path | Service: $service"
# 1. Git pull & checkout tag # 1. Git pull & checkout tag
if [[ "$DRY_RUN" != "true" ]]; then if [[ "$DRY_RUN" != "true" ]]; then
cd "$plugin_path" || { log "ERROR: $agent path $plugin_path not found"; return 1; } cd "$plugin_path" || { log "ERROR: $agent: path $plugin_path not found"; return 1; }
git fetch --tags origin git fetch --tags origin
git checkout "$TAG" git checkout "$TAG"
else else
@@ -90,15 +91,12 @@ deploy_agent() {
# 2. Install dependencies (platform-specific) # 2. Install dependencies (platform-specific)
if [[ "$DRY_RUN" != "true" ]]; then if [[ "$DRY_RUN" != "true" ]]; then
case "$agent" in case "$platform" in
tanko|mumuni|koonimo|koby) hermes|agent-zero)
pip install -r requirements.txt --quiet pip install -r requirements.txt --quiet
;; ;;
kagentz) pi)
pip install -r requirements.txt --quiet log "$agent: pi extension — skipping pip install"
;;
abiba)
log "$agent: pi extension skipping pip install"
;; ;;
esac esac
else else
@@ -107,8 +105,8 @@ deploy_agent() {
# 3. Restart service # 3. Restart service
if [[ "$DRY_RUN" != "true" ]]; then if [[ "$DRY_RUN" != "true" ]]; then
case "$agent" in case "$platform" in
abiba) pi)
log "$agent: triggering /reload" log "$agent: triggering /reload"
;; ;;
*) *)
@@ -127,7 +125,7 @@ deploy_agent() {
if curl -sf "http://localhost:$HEALTH_PORT/health" > /dev/null 2>&1; then if curl -sf "http://localhost:$HEALTH_PORT/health" > /dev/null 2>&1; then
log "OK: $agent health check passed" log "OK: $agent health check passed"
else else
log "ERROR: $agent health check failed check logs" log "ERROR: $agent health check failed — check logs"
return 1 return 1
fi fi
else else
@@ -135,8 +133,8 @@ deploy_agent() {
fi fi
} }
# --- Main ---\ # --- Main ---
log "Deploy started target: $TAG (Dry Run: $DRY_RUN)" log "Deploy started — target: $TAG (Dry Run: $DRY_RUN)"
if [[ -n "$SINGLE_CT" ]]; then if [[ -n "$SINGLE_CT" ]]; then
deploy_agent "$SINGLE_CT" deploy_agent "$SINGLE_CT"