feat(plugin): event delivery and webhooks for plugins (P2 batch 6) (#3728)

* feat(plugin): event delivery and webhooks for plugins

.github/workflows/plugin-sdk.yml

@@ -12,12 +12,14 @@ on:
       - 'pluginsdk/**'
       - 'examples/plugins/notebooks/**'
       - 'examples/plugins/links/**'
+      - 'examples/plugins/activity/**'
       - '.github/workflows/plugin-sdk.yml'
   pull_request:
     paths:
       - 'pluginsdk/**'
       - 'examples/plugins/notebooks/**'
       - 'examples/plugins/links/**'
+      - 'examples/plugins/activity/**'
       - '.github/workflows/plugin-sdk.yml'
 
 defaults:
@@ -78,5 +80,6 @@ jobs:
         run: |
           (cd ../../examples/plugins/notebooks && python -m unittest -v test_main)
           (cd ../../examples/plugins/links && python -m unittest -v test_main)
+          (cd ../../examples/plugins/activity && python -m unittest -v test_main)
       - name: build the package
         run: python -m pip install build && python -m build

examples/plugins/activity/main.py

@@ -0,0 +1,87 @@
+"""A WeKnora plugin that keeps an activity feed of the workspace, written
+with the Python SDK. It shows the three ways data reaches a plugin without a
+user asking: events WeKnora sends (at least once, so they are deduplicated
+by ID), a webhook other systems call, and a page that reads the feed."""
+
+from __future__ import annotations
+
+import hmac
+from datetime import datetime, timezone
+
+from weknora_plugin import ErrorCode, Plugin, PluginError, UIResponse, WebhookResponse
+
+plugin = Plugin("weknora-examples.activity", "1.0.0")
+
+FEED_KEY = "feed"
+MAX_ENTRIES = 100
+
+
+def store(call):
+    host = call.host()
+    if host is None:
+        # Retryable: the Host API may just be unreachable for a moment.
+        raise PluginError(ErrorCode.UNAVAILABLE, "the Host API is not available to this plugin")
+    return host
+
+
+def describe(ev, include_answers: bool) -> dict | None:
+    d = ev.data or {}
+    title = d.get("title") or d.get("fileName") or d.get("knowledgeId", "")
+    if ev.type == "knowledge.ingested":
+        return {"kind": "ingested", "text": title}
+    if ev.type == "knowledge.failed":
+        return {"kind": "failed", "text": f"{title}: {d.get('error', '')}".strip(": ")}
+    if ev.type == "knowledge.deleted":
+        return {"kind": "deleted", "text": title}
+    if ev.type == "chat.answered":
+        text = (d.get("question") or "").strip()[:200]
+        if include_answers:
+            first = (d.get("answer") or "").strip().splitlines()
+            text += f" → {first[0][:200]}" if first else ""
+        return {"kind": "answered", "text": text}
+    return None  # a type added after this plugin was written
+
+
+def append(host, entry: dict) -> None:
+    feed = host.kv_get(FEED_KEY, [])
+    # Deliveries repeat after a failure; the event ID makes them harmless.
+    if any(e.get("id") == entry["id"] for e in feed):
+        return
+    host.kv_put(FEED_KEY, ([entry] + feed)[:MAX_ENTRIES])
+
+
+@plugin.on_event
+def on_event(call, ev):
+    entry = describe(ev, call.tenant.get("include_answers") is True)
+    if entry is None:
+        return
+    entry.update(id=ev.id, at=(ev.occurred_at or datetime.now(timezone.utc)).isoformat())
+    append(store(call), entry)
+
+
+@plugin.webhook("notes")
+def notes(call, req):
+    secret = str(call.tenant.get("webhook_secret") or "")
+    given = req.headers.get("X-Activity-Secret", "")
+    if not secret or not hmac.compare_digest(secret, given):
+        return WebhookResponse(status=401, body="bad or missing X-Activity-Secret")
+    try:
+        text = str((req.json() or {}).get("text") or "").strip()
+    except ValueError:
+        text = ""
+    if not text:
+        return WebhookResponse(status=400, body='send {"text": "..."}')
+    now = datetime.now(timezone.utc)
+    append(store(call), {"id": f"note-{now.timestamp()}", "kind": "note", "text": text[:500], "at": now.isoformat()})
+    return {"ok": True}
+
+
+@plugin.ui
+def ui(call, req):
+    if req.method == "GET" and req.path == "/feed":
+        return UIResponse(body=store(call).kv_get(FEED_KEY, []))
+    return UIResponse(status=404, body={"error": "not found"})
+
+
+if __name__ == "__main__":
+    plugin.serve()

examples/plugins/activity/package.sh

@@ -0,0 +1,19 @@
+#!/usr/bin/env bash
+# Packages the activity example plugin as a .wkp: plugin.yaml, schemas,
+# main.py, the page under ui/ with the bridge library, and the Python SDK
+# under vendor/.
+set -euo pipefail
+cd "$(dirname "$0")"
+version=$(sed -n 's/^version: //p' plugin.yaml)
+out=${1:-"weknora-examples-activity-${version}.wkp"}
+stage=$(mktemp -d)
+trap 'rm -rf "$stage"' EXIT
+cp plugin.yaml main.py "$stage"/
+cp -r ui schemas "$stage"/
+cp ../../../packages/plugin-ui/index.js "$stage/ui/weknora-plugin-ui.js"
+mkdir -p "$stage/vendor"
+cp -r ../../../pluginsdk/python/src/weknora_plugin "$stage/vendor/"
+find "$stage" -name __pycache__ -prune -exec rm -rf {} +
+rm -f "$out"
+(cd "$stage" && zip -qr - .) > "$out"
+echo "wrote $out"

examples/plugins/activity/plugin.yaml

@@ -0,0 +1,30 @@
+schemaVersion: 1
+id: weknora-examples.activity
+version: 1.0.0
+apiVersion: weknora.plugin/v1
+name: { en-US: Activity feed, zh-CN: 动态 }
+description:
+  en-US: A feed of what happens in the workspace (documents ingested, failed, deleted, questions answered), plus notes other systems post through a webhook. Shows events and webhooks, written in Python.
+  zh-CN: 空间动态：文档入库、失败、删除、问答完成，以及外部系统经 Webhook 推送的消息。演示事件与 Webhook，用 Python 编写。
+publisher: { id: weknora-examples, name: WeKnora examples }
+homepage: https://github.com/Tencent/WeKnora/tree/main/examples/plugins/activity
+license: MIT
+runtime: { type: host, kind: python, entry: main.py }
+config:
+  tenant: schemas/tenant.yaml
+permissions:
+  hostApi: [kv]
+  # Answers include the question and the answer text.
+  events: [knowledge.ingested, knowledge.failed, knowledge.deleted, chat.answered]
+contributes:
+  pages:
+    - id: feed
+      name: { en-US: Activity, zh-CN: 动态 }
+      description: { en-US: What happened in this workspace lately, zh-CN: 本空间最近的动态 }
+      entry: ui/index.html
+  webhooks:
+    - id: notes
+      name: { en-US: Post a note, zh-CN: 推送消息 }
+      description:
+        en-US: "POST {\"text\": \"...\"} with the header X-Activity-Secret set to the secret in this plugin's settings."
+        zh-CN: "POST {\"text\": \"...\"}，并带上请求头 X-Activity-Secret，值为本插件设置中的密钥。"

examples/plugins/activity/schemas/tenant.yaml

@@ -0,0 +1,18 @@
+type: object
+properties:
+  webhook_secret:
+    type: string
+    title: Webhook secret
+    description: Callers of the note webhook send it in the X-Activity-Secret header
+    x-secret: true
+    x-i18n:
+      title: { zh-CN: Webhook 密钥 }
+      description: { zh-CN: 调用推送消息 Webhook 时放在 X-Activity-Secret 请求头里 }
+  include_answers:
+    type: boolean
+    title: Show answer text
+    description: Keep the first line of each answer in the feed, not just the question
+    default: false
+    x-i18n:
+      title: { zh-CN: 显示回答内容 }
+      description: { zh-CN: 在动态里保留每个回答的第一行，而不只是问题 }

examples/plugins/activity/test_main.py

@@ -0,0 +1,66 @@
+import json
+import os
+import sys
+import unittest
+
+HERE = os.path.dirname(os.path.abspath(__file__))
+sys.path.insert(0, os.path.join(HERE, "..", "..", "..", "pluginsdk", "python", "src"))
+sys.path.insert(0, HERE)
+
+import main  # noqa: E402
+from weknora_plugin import EventDelivery, UIRequest, WebhookRequest  # noqa: E402
+from weknora_plugin.plugin import Call  # noqa: E402
+
+
+class FakeHost:
+    def __init__(self):
+        self.data = {}
+
+    def kv_get(self, key, default=None):
+        return self.data.get(key, default)
+
+    def kv_put(self, key, value, ttl=0):
+        self.data[key] = value
+
+
+def call(**tenant):
+    return Call({"context": {"tenantId": 1}, "config": {"tenant": tenant}})
+
+
+class ActivityTest(unittest.TestCase):
+    def setUp(self):
+        self.host = FakeHost()
+        main.store = lambda c: self.host
+
+    def feed(self):
+        return self.host.data.get("feed", [])
+
+    def test_events_are_recorded_once(self):
+        ev = EventDelivery(id="e1", type="knowledge.ingested", data={"knowledgeId": "k1", "title": "Spec"})
+        main.on_event(call(), ev)
+        main.on_event(call(), ev)  # a retry after a failure
+        main.on_event(call(), EventDelivery(id="e2", type="some.future.type", data={}))
+        self.assertEqual([(e["id"], e["kind"], e["text"]) for e in self.feed()], [("e1", "ingested", "Spec")])
+
+    def test_answers(self):
+        ev = EventDelivery(id="a", type="chat.answered", data={"question": "What is RAG?", "answer": "Retrieval.\nMore"})
+        main.on_event(call(), ev)
+        main.on_event(call(include_answers=True), EventDelivery(id="b", type="chat.answered", data=ev.data))
+        self.assertEqual([e["text"] for e in self.feed()], ["What is RAG? → Retrieval.", "What is RAG?"])
+
+    def test_note_webhook(self):
+        req = WebhookRequest(method="POST", headers={"X-Activity-Secret": "s3"}, body=json.dumps({"text": "Deploy done"}).encode())
+        self.assertEqual(main.notes(call(webhook_secret="s3"), req), {"ok": True})
+        self.assertEqual(self.feed()[0]["text"], "Deploy done")
+        self.assertEqual(main.notes(call(webhook_secret="other"), req).status, 401)
+        self.assertEqual(main.notes(call(), req).status, 401)
+        self.assertEqual(main.notes(call(webhook_secret="s3"), WebhookRequest(headers={"X-Activity-Secret": "s3"}, body=b"{}")).status, 400)
+
+    def test_page_reads_the_feed(self):
+        main.on_event(call(), EventDelivery(id="e1", type="knowledge.deleted", data={"title": "Old"}))
+        self.assertEqual(main.ui(call(), UIRequest(method="GET", path="/feed")).body[0]["kind"], "deleted")
+        self.assertEqual(main.ui(call(), UIRequest(method="GET", path="/other")).status, 404)
+
+
+if __name__ == "__main__":
+    unittest.main()

examples/plugins/activity/ui/app.css

@@ -0,0 +1,23 @@
+:root {
+  --fg: var(--wk-text-color-primary, #1f2329);
+  --muted: var(--wk-text-color-secondary, #646a73);
+  --line: var(--wk-component-border, #dee0e3);
+  --card: var(--wk-bg-color-container, #fff);
+  --brand: var(--wk-brand-color, #0052d9);
+  --danger: var(--wk-error-color, #d54941);
+  --ok: var(--wk-success-color, #2ba471);
+  --radius: var(--wk-radius-default, 6px);
+}
+:root[data-theme='dark'] { color-scheme: dark; }
+body { margin: 0; font: 14px/1.5 var(--wk-font-family, system-ui, sans-serif); color: var(--fg); background: transparent; }
+main { padding: 4px 2px 12px; }
+.bar { display: flex; justify-content: flex-end; margin-bottom: 8px; }
+button { padding: 4px 12px; font: inherit; border-radius: var(--radius); cursor: pointer; border: 1px solid var(--line); background: var(--card); color: var(--fg); }
+.empty { color: var(--muted); padding: 24px 0; text-align: center; }
+ol { list-style: none; margin: 0; padding: 0; display: grid; gap: 6px; }
+li { display: flex; gap: 10px; align-items: baseline; padding: 8px 12px; border: 1px solid var(--line); border-radius: var(--radius); background: var(--card); }
+.kind { flex: none; font-size: 12px; padding: 0 6px; border-radius: 4px; background: color-mix(in srgb, var(--brand) 12%, transparent); color: var(--brand); }
+.kind.failed, .kind.deleted { color: var(--danger); background: color-mix(in srgb, var(--danger) 12%, transparent); }
+.kind.ingested { color: var(--ok); background: color-mix(in srgb, var(--ok) 12%, transparent); }
+.text { flex: 1; min-width: 0; overflow-wrap: anywhere; }
+time { flex: none; color: var(--muted); font-size: 12px; }

examples/plugins/activity/ui/feed.js

@@ -0,0 +1,54 @@
+import { connect } from './weknora-plugin-ui.js'
+
+const LABELS = {
+  en: { ingested: 'Ingested', failed: 'Failed', deleted: 'Deleted', answered: 'Answered', note: 'Note', empty: 'Nothing yet.', refresh: 'Refresh' },
+  zh: { ingested: '已入库', failed: '失败', deleted: '已删除', answered: '已回答', note: '消息', empty: '暂无动态。', refresh: '刷新' },
+}
+
+const wk = await connect()
+const app = document.getElementById('app')
+let text = pick(wk.context.locale)
+wk.on('locale', (locale) => {
+  text = pick(locale)
+  void load()
+})
+
+function pick(locale) {
+  return String(locale || '').startsWith('zh') ? LABELS.zh : LABELS.en
+}
+
+function el(tag, props = {}, ...children) {
+  const node = Object.assign(document.createElement(tag), props)
+  node.append(...children)
+  return node
+}
+
+async function load() {
+  app.setAttribute('aria-busy', 'true')
+  try {
+    const { body } = await wk.get('/feed')
+    render(Array.isArray(body) ? body : [])
+  } catch (e) {
+    app.replaceChildren(el('p', { className: 'empty', textContent: e.message }))
+  } finally {
+    app.removeAttribute('aria-busy')
+  }
+}
+
+function render(feed) {
+  const refresh = el('button', { type: 'button', textContent: text.refresh })
+  refresh.addEventListener('click', () => void load())
+  const list = el('ol')
+  for (const entry of feed) {
+    const at = new Date(entry.at)
+    list.append(
+      el('li', {},
+        el('span', { className: `kind ${entry.kind}`, textContent: text[entry.kind] || entry.kind }),
+        el('span', { className: 'text', textContent: entry.text }),
+        el('time', { dateTime: entry.at, textContent: isNaN(at) ? '' : at.toLocaleString(wk.context.locale) })),
+    )
+  }
+  app.replaceChildren(el('div', { className: 'bar' }, refresh), feed.length ? list : el('p', { className: 'empty', textContent: text.empty }))
+}
+
+await load()

examples/plugins/activity/ui/index.html

@@ -0,0 +1,13 @@
+<!doctype html>
+<html lang="en">
+<head>
+  <meta charset="utf-8" />
+  <meta name="viewport" content="width=device-width, initial-scale=1" />
+  <title>Activity</title>
+  <link rel="stylesheet" href="app.css" />
+</head>
+<body>
+  <main id="app" aria-busy="true"></main>
+  <script type="module" src="feed.js"></script>
+</body>
+</html>

frontend/src/api/plugin.ts

@@ -17,6 +17,7 @@ export type ExtensionPoint =
   | 'pages'
   | 'settingsSections'
   | 'kbTabs'
+  | 'webhooks'
 
 export interface PluginContribution {
   id: string
@@ -137,6 +138,20 @@ export function pluginPageRequest(
   return post<{ data: PluginPageResponse }>(`/api/v1/plugins/${encodeURIComponent(pluginId)}/ui-request`, req)
 }
 
+/** A webhook of a plugin with this workspace's URL (Admin+: the URL is a secret). */
+export interface PluginWebhook {
+  id: string
+  name: LocalizedText
+  description?: LocalizedText
+  /** Path on this deployment; `url` is the full URL when the server knows its public address. */
+  path: string
+  url?: string
+}
+
+export function listPluginWebhooks(id: string) {
+  return get<{ data: PluginWebhook[] }>(`/api/v1/plugins/${encodeURIComponent(id)}/webhooks`)
+}
+
 /** The workspace's configuration of a plugin (Admin+). */
 export function getPluginConfig(id: string) {
   return get<{ data: PluginConfig }>(`/api/v1/plugins/${encodeURIComponent(id)}/config`)

frontend/src/i18n/locales/en-US.ts

@@ -2867,6 +2867,10 @@ export default {
   pluginCenter: {
     installed: 'Installed',
     configure: 'Configure',
+    webhooks: 'Webhook URLs',
+    webhooksHint: 'Set these URLs up in the third-party service so it sends its events to this plugin. They carry this workspace\'s secret; keep them private.',
+    copy: 'Copy',
+    webhookCopied: 'Webhook URL copied',
     configTitle: 'Configure {name}',
     configDescription: 'This workspace\'s settings for the plugin, such as its API key. Secrets are stored encrypted and never shown again.',
     configLoadFailed: 'Failed to load the plugin configuration',
@@ -2897,7 +2901,8 @@ export default {
       mcpServers: 'MCP servers',
       pages: 'Pages',
       settingsSections: 'Settings sections',
-      kbTabs: 'Knowledge base tabs'
+      kbTabs: 'Knowledge base tabs',
+      webhooks: 'Webhooks'
     }
   },
   schemaForm: {

frontend/src/i18n/locales/ja-JP.ts

@@ -2867,6 +2867,10 @@ export default {
   pluginCenter: {
     installed: 'インストール済み',
     configure: '設定',
+    webhooks: 'Webhook URL',
+    webhooksHint: 'サードパーティのサービスにこれらの URL を設定すると、イベントがこのプラグインに送られます。このワークスペースの秘密を含むため、公開しないでください。',
+    copy: 'コピー',
+    webhookCopied: 'Webhook URL をコピーしました',
     configTitle: '{name} の設定',
     configDescription: 'このワークスペースでプラグインを使うための設定（API キーなど）。シークレットは暗号化して保存され、再表示されません。',
     configLoadFailed: 'プラグイン設定の読み込みに失敗しました',
@@ -2897,7 +2901,8 @@ export default {
       mcpServers: 'MCP サーバー',
       pages: 'ページ',
       settingsSections: '設定セクション',
-      kbTabs: 'ナレッジベースのタブ'
+      kbTabs: 'ナレッジベースのタブ',
+      webhooks: 'Webhook'
     }
   },
   schemaForm: {

frontend/src/i18n/locales/ko-KR.ts

@@ -5385,6 +5385,10 @@ export default {
   pluginCenter: {
     installed: '설치됨',
     configure: '설정',
+    webhooks: 'Webhook URL',
+    webhooksHint: '타사 서비스에 이 URL을 설정하면 이벤트가 이 플러그인으로 전송됩니다. 이 워크스페이스의 비밀이 포함되어 있으니 공개하지 마세요.',
+    copy: '복사',
+    webhookCopied: 'Webhook URL을 복사했습니다',
     configTitle: '{name} 설정',
     configDescription: '이 워크스페이스에서 플러그인을 사용하기 위한 설정(예: API 키). 비밀 값은 암호화되어 저장되며 다시 표시되지 않습니다.',
     configLoadFailed: '플러그인 설정을 불러오지 못했습니다',
@@ -5415,7 +5419,8 @@ export default {
       mcpServers: 'MCP 서버',
       pages: '페이지',
       settingsSections: '설정 섹션',
-      kbTabs: '지식 베이스 탭'
+      kbTabs: '지식 베이스 탭',
+      webhooks: 'Webhook'
     }
   },
   schemaForm: {

frontend/src/i18n/locales/ru-RU.ts

@@ -5385,6 +5385,10 @@ export default {
   pluginCenter: {
     installed: 'Установлен',
     configure: 'Настроить',
+    webhooks: 'Адреса вебхуков',
+    webhooksHint: 'Укажите эти адреса в стороннем сервисе, чтобы он отправлял события этому плагину. Они содержат секрет рабочего пространства; не публикуйте их.',
+    copy: 'Копировать',
+    webhookCopied: 'Адрес вебхука скопирован',
     configTitle: 'Настройки: {name}',
     configDescription: 'Настройки плагина для этого рабочего пространства, например API-ключ. Секреты хранятся в зашифрованном виде и больше не показываются.',
     configLoadFailed: 'Не удалось загрузить настройки плагина',
@@ -5415,7 +5419,8 @@ export default {
       mcpServers: 'MCP-серверы',
       pages: 'Страницы',
       settingsSections: 'Разделы настроек',
-      kbTabs: 'Вкладки базы знаний'
+      kbTabs: 'Вкладки базы знаний',
+      webhooks: 'Вебхуки'
     }
   },
   schemaForm: {

frontend/src/i18n/locales/zh-CN.ts

@@ -5387,6 +5387,10 @@ export default {
   pluginCenter: {
     installed: '已安装',
     configure: '配置',
+    webhooks: 'Webhook 地址',
+    webhooksHint: '在第三方系统里配置这些地址，它们会把事件推送给本插件。地址包含本空间的密钥，请勿公开。',
+    copy: '复制',
+    webhookCopied: '已复制 Webhook 地址',
     configTitle: '配置 {name}',
     configDescription: '本空间使用该插件时的配置，例如 API 密钥。密钥加密保存，不会再次显示。',
     configLoadFailed: '加载插件配置失败',
@@ -5417,7 +5421,8 @@ export default {
       mcpServers: 'MCP 服务',
       pages: '页面',
       settingsSections: '设置页',
-      kbTabs: '知识库页签'
+      kbTabs: '知识库页签',
+      webhooks: 'Webhook'
     }
   },
   schemaForm: {

frontend/src/views/settings/PluginCenter.vue

@@ -60,7 +60,7 @@
             />
           </t-tooltip>
           <t-button
-            v-if="canManage && hasTenantConfig(p.manifest)"
+            v-if="canManage && canConfigure(p.manifest)"
             size="small"
             variant="text"
             theme="primary"
@@ -79,37 +79,66 @@
       icon="setting"
       :confirm-loading="configSaving"
       :confirm-disabled="!configSchema || configSaving"
+      :hide-footer="!configSchema"
       @confirm="saveConfig"
     >
       <div v-if="configLoading" class="plugin-center__state"><t-loading size="small" /></div>
-      <section v-else-if="configSchema" class="setting-drawer__section">
-        <SchemaForm v-model="configValues" :schema="configSchema" :errors="configErrors" />
-      </section>
+      <template v-else>
+        <section v-if="configSchema" class="setting-drawer__section">
+          <SchemaForm v-model="configValues" :schema="configSchema" :errors="configErrors" />
+        </section>
+        <section v-if="configWebhooks.length" class="setting-drawer__section">
+          <h4 class="setting-drawer__section-title">{{ t('pluginCenter.webhooks') }}</h4>
+          <p class="plugin-center__hint">{{ t('pluginCenter.webhooksHint') }}</p>
+          <div v-for="hook in configWebhooks" :key="hook.id" class="plugin-center__webhook">
+            <div class="plugin-center__webhook-name">{{ localizedText(hook.name, locale) }}</div>
+            <p v-if="hook.description" class="plugin-center__hint">{{ localizedText(hook.description, locale) }}</p>
+            <div class="plugin-center__webhook-url">
+              <code>{{ webhookUrl(hook, origin) }}</code>
+              <t-button size="small" variant="text" theme="primary"
+                @click="copyWithToast(webhookUrl(hook, origin), 'pluginCenter.webhookCopied')">
+                {{ t('pluginCenter.copy') }}
+              </t-button>
+            </div>
+          </div>
+        </section>
+      </template>
     </SettingDrawer>
   </div>
 </template>
 
 <script setup lang="ts">
-import { usePluginPagesStore } from '@/stores/pluginPages'
 import { computed, onMounted, ref } from 'vue'
 import { useI18n } from 'vue-i18n'
 import { MessagePlugin } from 'tdesign-vue-next'
 
 import {
   getPluginConfig,
+  listPluginWebhooks,
   listPlugins,
   setPluginEnabled,
   updatePluginConfig,
   type ExtensionPoint,
+  type PluginWebhook,
   type TenantPlugin,
 } from '@/api/plugin'
 import SchemaForm from '@/components/schema-form/SchemaForm.vue'
 import { validateConfig, type ConfigSchema, type ConfigValue, type FieldError } from '@/components/schema-form/schema'
 import SettingDrawer from '@/components/settings/SettingDrawer.vue'
 import { useAuthStore } from '@/stores/auth'
+import { usePluginPagesStore } from '@/stores/pluginPages'
 import { localizedText } from '@/utils/localizedText'
+import { copyWithToast } from '@/utils/clipboard'
 
-import { EXTENSION_POINTS, contributionSummary, filterPlugins, hasTenantConfig } from './pluginCenterState'
+import {
+  EXTENSION_POINTS,
+  canConfigure,
+  contributionSummary,
+  filterPlugins,
+  hasTenantConfig,
+  hasWebhooks,
+  webhookUrl,
+} from './pluginCenterState'
 
 // Lists the plugins this deployment knows — builtins included — and lets an
 // admin turn them off for the workspace. A disabled plugin's integrations drop
@@ -180,18 +209,27 @@ const configValues = ref<ConfigValue>({})
 const configErrors = ref<FieldError[]>([])
 const configLoading = ref(false)
 const configSaving = ref(false)
+const configWebhooks = ref<PluginWebhook[]>([])
+const origin = window.location.origin
 
 async function openConfig(p: TenantPlugin) {
   configPlugin.value = p
   configSchema.value = null
   configValues.value = {}
   configErrors.value = []
+  configWebhooks.value = []
   configOpen.value = true
   configLoading.value = true
   try {
-    const res = await getPluginConfig(p.manifest.id)
-    configSchema.value = res.data.schema
-    configValues.value = res.data.values ?? {}
+    const [config, hooks] = await Promise.all([
+      hasTenantConfig(p.manifest) ? getPluginConfig(p.manifest.id) : null,
+      hasWebhooks(p.manifest) ? listPluginWebhooks(p.manifest.id) : null,
+    ])
+    if (config) {
+      configSchema.value = config.data.schema
+      configValues.value = config.data.values ?? {}
+    }
+    configWebhooks.value = hooks?.data ?? []
   } catch (e: any) {
     MessagePlugin.error(e?.message || t('pluginCenter.configLoadFailed'))
   } finally {
@@ -281,6 +319,47 @@ onMounted(load)
   }
 }
 
+.plugin-center__hint {
+  margin: 0 0 8px;
+  font-size: var(--app-text-sm);
+  line-height: 1.5;
+  color: var(--td-text-color-placeholder);
+}
+
+.plugin-center__webhook {
+  display: flex;
+  flex-direction: column;
+  gap: 4px;
+
+  & + & {
+    margin-top: 12px;
+  }
+}
+
+.plugin-center__webhook-name {
+  font-size: var(--app-text-sm);
+  font-weight: 500;
+  color: var(--td-text-color-primary);
+}
+
+.plugin-center__webhook-url {
+  display: flex;
+  align-items: center;
+  gap: 8px;
+  min-width: 0;
+
+  code {
+    flex: 1;
+    min-width: 0;
+    padding: 6px 8px;
+    font-size: var(--app-text-xs);
+    word-break: break-all;
+    color: var(--td-text-color-secondary);
+    background: var(--td-bg-color-secondarycontainer);
+    border-radius: var(--app-radius-sm);
+  }
+}
+
 .plugin-center__state {
   display: flex;
   justify-content: center;

frontend/src/views/settings/pluginCenterState.test.ts

@@ -63,3 +63,14 @@ test('pluginSkillChoices lists skills of enabled plugins with their install sour
   assert.equal(zh?.name, '分诊')
   assert.equal(zh?.description, 'Sort issues')
 })
+
+test('canConfigure covers webhooks, and webhookUrl prefers the server URL', async () => {
+  const { canConfigure, hasWebhooks, webhookUrl } = await import('./pluginCenterState')
+  const bare = plugin('acme.bare', 'Bare', {}).manifest
+  assert.equal(canConfigure(bare), false)
+  const hooks = plugin('acme.hooks', 'Hooks', { webhooks: [{ id: 'inbox', name: { default: 'Inbox' } }] }).manifest
+  assert.equal(hasWebhooks(hooks), true)
+  assert.equal(canConfigure(hooks), true)
+  assert.equal(webhookUrl({ path: '/api/v1/plugin-callbacks/a/b/c' }, 'https://x.example'), 'https://x.example/api/v1/plugin-callbacks/a/b/c')
+  assert.equal(webhookUrl({ path: '/p', url: 'https://public.example/p' }, 'https://x.example'), 'https://public.example/p')
+})

frontend/src/views/settings/pluginCenterState.ts

@@ -6,6 +6,7 @@ import { localizedText } from '../../utils/localizedText'
 /** Extension points in the order the plugin center shows them. */
 export const EXTENSION_POINTS: ExtensionPoint[] = [
   'modelVendors', 'connectors', 'imChannels', 'webSearch', 'tools', 'parsers', 'skills', 'mcpServers',
+  'pages', 'settingsSections', 'kbTabs', 'webhooks',
 ]
 
 /** Whether the workspace can configure the plugin (it declares config.tenant). */
@@ -13,6 +14,21 @@ export function hasTenantConfig(m: PluginManifest): boolean {
   return !!m.config?.tenantSchema?.properties && Object.keys(m.config.tenantSchema.properties).length > 0
 }
 
+/** Whether the plugin has webhooks, whose URLs admins set up elsewhere. */
+export function hasWebhooks(m: PluginManifest): boolean {
+  return (m.contributes?.webhooks?.length ?? 0) > 0
+}
+
+/** Whether the plugin's configuration drawer has anything to show. */
+export function canConfigure(m: PluginManifest): boolean {
+  return hasTenantConfig(m) || hasWebhooks(m)
+}
+
+/** A webhook's full URL: the server's, or this page's origin with the path. */
+export function webhookUrl(hook: { path: string; url?: string }, origin: string): string {
+  return hook.url || `${origin}${hook.path}`
+}
+
 /**
  * Skills of enabled plugins, for the skill catalog's "from plugin" picker.
  * Each carries the install source the backend understands.

internal/application/service/knowledge_delete.go

@@ -10,8 +10,10 @@ import (
 	"github.com/Tencent/WeKnora/internal/application/service/retriever"
 	apperrors "github.com/Tencent/WeKnora/internal/errors"
 	"github.com/Tencent/WeKnora/internal/logger"
+	pluginevents "github.com/Tencent/WeKnora/internal/plugin/events"
 	"github.com/Tencent/WeKnora/internal/types"
 	"github.com/Tencent/WeKnora/internal/types/interfaces"
+	"github.com/Tencent/WeKnora/pluginsdk/pluginapi"
 	"github.com/hibiken/asynq"
 	"golang.org/x/sync/errgroup"
 )
@@ -604,6 +606,10 @@ func (s *knowledgeService) executeKnowledgeDelete(plan *knowledgeDeletePlan, sin
 		for _, knowledge := range knowledges {
 			knowledgeIDs = append(knowledgeIDs, knowledge.ID)
 			titles = append(titles, knowledge.Title)
+			pluginevents.Publish(ctx, tenantInfo.ID, pluginapi.EventKnowledgeDeleted, pluginapi.KnowledgeEventData{
+				KnowledgeBaseID: kbID, KnowledgeID: knowledge.ID, Title: knowledge.Title,
+				FileName: knowledge.FileName, FileType: knowledge.FileType, Source: knowledge.Source,
+			})
 		}
 		details := map[string]any{"count": len(knowledgeIDs)}
 		if len(knowledgeIDs) <= 20 {

internal/container/container.go

@@ -74,11 +74,13 @@ import (
 	"github.com/Tencent/WeKnora/internal/plugin/activate"
 	pluginbuiltin "github.com/Tencent/WeKnora/internal/plugin/builtin"
 	plugindriver "github.com/Tencent/WeKnora/internal/plugin/driver"
+	pluginevents "github.com/Tencent/WeKnora/internal/plugin/events"
 	pluginmanifest "github.com/Tencent/WeKnora/internal/plugin/manifest"
 	"github.com/Tencent/WeKnora/internal/plugin/reconcile"
 	pluginregistry "github.com/Tencent/WeKnora/internal/plugin/registry"
 	pluginremote "github.com/Tencent/WeKnora/internal/plugin/remote"
 	plugintenancy "github.com/Tencent/WeKnora/internal/plugin/tenancy"
+	pluginwebhook "github.com/Tencent/WeKnora/internal/plugin/webhook"
 	"github.com/Tencent/WeKnora/internal/router"
 	"github.com/Tencent/WeKnora/internal/sandbox"
 	"github.com/Tencent/WeKnora/internal/storageallowlist"
@@ -150,7 +152,10 @@ func BuildContainer(container *dig.Container) *dig.Container {
 	must(container.Provide(repository.NewTenantInvitationRepository))
 	must(container.Provide(repository.NewAuditLogRepository))
 	must(container.Provide(repository.NewKnowledgeBaseRepository))
-	must(container.Provide(repository.NewKnowledgeRepository))
+	// Documents reaching completed or failed raise plugin events.
+	must(container.Provide(func(db *gorm.DB) interfaces.KnowledgeRepository {
+		return pluginevents.WatchKnowledge(repository.NewKnowledgeRepository(db))
+	}))
 	must(container.Provide(repository.NewKnowledgeSpanRepository))
 	must(container.Provide(repository.NewChunkRepository))
 	must(container.Provide(repository.NewKnowledgeTagRepository))
@@ -178,6 +183,13 @@ func BuildContainer(container *dig.Container) *dig.Container {
 	must(container.Provide(activate.NewConnectors))
 	must(container.Provide(activate.NewParsers))
 	must(container.Provide(activate.NewUIPages))
+	must(container.Provide(func(
+		reg *pluginregistry.Registry, t *plugintenancy.Service, enq interfaces.TaskEnqueuer, iv *activate.Invoker,
+	) *pluginevents.Dispatcher {
+		return pluginevents.NewDispatcher(reg, t, enq, iv)
+	}))
+	must(container.Provide(func(d *pluginevents.Dispatcher) interfaces.TaskHandler { return d },
+		dig.Name("pluginEvents")))
 	must(container.Provide(newMCPServiceRepository))
 	must(container.Provide(repository.NewMCPToolApprovalRepository))
 	must(container.Provide(repository.NewMCPOAuthRepository))
@@ -613,6 +625,8 @@ func BuildContainer(container *dig.Container) *dig.Container {
 	must(container.Provide(handler.NewPluginHandler))
 	must(container.Provide(handler.NewPluginAdminHandler))
 	must(container.Provide(handler.NewPluginUIHandler))
+	must(container.Provide(pluginwebhook.NewTokensFromEnv))
+	must(container.Provide(handler.NewPluginWebhookHandler))
 	logger.Debugf(ctx, "[Container] HTTP handlers registered")
 
 	// Wire the chat package's local image resolver so multimodal chat can read

internal/container/plugin_runtime.go

@@ -17,6 +17,7 @@ import (
 	"github.com/Tencent/WeKnora/internal/handler"
 	"github.com/Tencent/WeKnora/internal/logger"
 	"github.com/Tencent/WeKnora/internal/plugin/activate"
+	pluginevents "github.com/Tencent/WeKnora/internal/plugin/events"
 	"github.com/Tencent/WeKnora/internal/plugin/host"
 	"github.com/Tencent/WeKnora/internal/plugin/hostapi"
 	"github.com/Tencent/WeKnora/internal/plugin/hostpool"
@@ -26,6 +27,7 @@ import (
 	pluginregistry "github.com/Tencent/WeKnora/internal/plugin/registry"
 	"github.com/Tencent/WeKnora/internal/plugin/remote"
 	plugintenancy "github.com/Tencent/WeKnora/internal/plugin/tenancy"
+	"github.com/Tencent/WeKnora/internal/plugin/webhook"
 	"github.com/Tencent/WeKnora/internal/types/interfaces"
 	"github.com/Tencent/WeKnora/pluginsdk/client"
 )
@@ -71,7 +73,9 @@ func (a pluginActivators) list() []reconcile.Activator {
 // (remote, on a plugin host) only through WEKNORA_PLUGIN_HOST_API_URL.
 func newPluginHostAPI(
 	cfg *config.Config, iv *activate.Invoker, repo interfaces.PluginKVRepository, cleaner interfaces.ResourceCleaner,
+	hooks *webhook.Tokens,
 ) *hostapi.Handler {
+	iv.SetWebhooks(hooks, webhook.PublicBase())
 	issuer := hostapi.NewIssuerFromEnv()
 	// Plugins on this node always use loopback; the public address is for
 	// plugins elsewhere (remote, on a plugin host), which reach this node
@@ -211,8 +215,9 @@ func newPluginInstaller(
 // traffic, then keeps this node in step with the others.
 func startPluginReconciler(
 	r *reconcile.Reconciler, hostManager *host.Manager, remoteManager *remote.Manager,
-	cleaner interfaces.ResourceCleaner,
+	events *pluginevents.Dispatcher, cleaner interfaces.ResourceCleaner,
 ) {
+	pluginevents.SetDefault(events)
 	if kinds := hostManager.Kinds(); len(kinds) > 0 {
 		logger.Infof(context.Background(), "[plugin] this node runs %s host plugins", strings.Join(kinds, ", "))
 	}

internal/handler/plugin_webhook.go

@@ -0,0 +1,193 @@
+package handler
+
+import (
+	"context"
+	"io"
+	"net/http"
+	"strings"
+	"sync"
+
+	"github.com/gin-gonic/gin"
+	"golang.org/x/time/rate"
+
+	"github.com/Tencent/WeKnora/internal/errors"
+	"github.com/Tencent/WeKnora/internal/logger"
+	"github.com/Tencent/WeKnora/internal/plugin/activate"
+	"github.com/Tencent/WeKnora/internal/plugin/manifest"
+	"github.com/Tencent/WeKnora/internal/plugin/registry"
+	"github.com/Tencent/WeKnora/internal/plugin/tenancy"
+	"github.com/Tencent/WeKnora/internal/plugin/webhook"
+	"github.com/Tencent/WeKnora/internal/types"
+	"github.com/Tencent/WeKnora/pluginsdk/pluginapi"
+)
+
+// maxWebhookBody bounds one inbound call.
+const maxWebhookBody = 1 << 20
+
+// Per-URL rate: a burst, then a steady rate.
+const (
+	webhookBurst  = 40
+	webhookPerSec = 20
+)
+
+// PluginWebhookHandler receives third parties' calls to plugin webhooks and
+// relays them to the plugin, and shows workspace admins their URLs.
+type PluginWebhookHandler struct {
+	registry *registry.Registry
+	tenancy  *tenancy.Service
+	invoker  *activate.Invoker
+	tokens   *webhook.Tokens
+
+	mu       sync.Mutex
+	limiters map[string]*rate.Limiter
+}
+
+// NewPluginWebhookHandler creates the handler.
+func NewPluginWebhookHandler(
+	reg *registry.Registry, t *tenancy.Service, iv *activate.Invoker, tokens *webhook.Tokens,
+) *PluginWebhookHandler {
+	return &PluginWebhookHandler{
+		registry: reg,
+		tenancy:  t,
+		invoker:  iv,
+		tokens:   tokens,
+		limiters: map[string]*rate.Limiter{},
+	}
+}
+
+func (h *PluginWebhookHandler) allow(token string) bool {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	l, ok := h.limiters[token]
+	if !ok {
+		if len(h.limiters) > 10000 {
+			h.limiters = map[string]*rate.Limiter{} // bounded memory; limits restart
+		}
+		l = rate.NewLimiter(webhookPerSec, webhookBurst)
+		h.limiters[token] = l
+	}
+	return l.Allow()
+}
+
+// webhookOf finds webhook id among a loaded plugin's contributions.
+func (h *PluginWebhookHandler) webhookOf(pluginID, id string) (*manifest.Manifest, *manifest.Contribution) {
+	m, ok := h.registry.Plugin(pluginID)
+	if !ok || (m.Runtime.Type != manifest.RuntimeHost && m.Runtime.Type != manifest.RuntimeRemote) {
+		return nil, nil
+	}
+	for i, c := range m.Contributes[manifest.PointWebhooks] {
+		if c.ID == id {
+			return m, &m.Contributes[manifest.PointWebhooks][i]
+		}
+	}
+	return nil, nil
+}
+
+// Receive relays one inbound call. It needs no login: the URL's token names
+// the workspace and proves WeKnora issued the URL. Everything that is not
+// a live webhook of a plugin the workspace has on answers 404.
+func (h *PluginWebhookHandler) Receive(c *gin.Context) {
+	pluginID, hookID, token := c.Param("id"), c.Param("hook"), c.Param("token")
+	tenantID, ok := h.tokens.Verify(pluginID, hookID, token)
+	if !ok {
+		c.Status(http.StatusNotFound)
+		return
+	}
+	m, _ := h.webhookOf(pluginID, hookID)
+	if m == nil {
+		c.Status(http.StatusNotFound)
+		return
+	}
+	ctx := c.Request.Context()
+	if on, err := h.tenancy.PluginEnabled(ctx, tenantID, pluginID); err != nil || !on {
+		c.Status(http.StatusNotFound)
+		return
+	}
+	if !h.allow(token) {
+		c.Status(http.StatusTooManyRequests)
+		return
+	}
+	body, err := io.ReadAll(io.LimitReader(c.Request.Body, maxWebhookBody+1))
+	if err != nil || len(body) > maxWebhookBody {
+		c.Status(http.StatusRequestEntityTooLarge)
+		return
+	}
+	headers := map[string]string{}
+	for k, v := range c.Request.Header {
+		if len(v) > 0 && !strings.EqualFold(k, "Cookie") {
+			headers[k] = v[0]
+		}
+	}
+	path := c.Param("path")
+	if path == "" {
+		path = "/"
+	}
+	in := pluginapi.WebhookRequest{
+		Method: c.Request.Method, Path: path, Query: c.Request.URL.RawQuery, Headers: headers, Body: body,
+	}
+	var out pluginapi.WebhookResponse
+	ctx = context.WithValue(ctx, types.TenantIDContextKey, tenantID)
+	if err := h.invoker.Call(ctx, m, pluginapi.WebhookPath(hookID), nil, in, &out); err != nil {
+		logger.Warnf(ctx, "[plugin] webhook %s/%s for tenant %d: %v", pluginID, hookID, tenantID, err)
+		status := http.StatusBadGateway
+		if pe, ok := pluginapi.AsError(err); ok && pe.Retryable {
+			status = http.StatusServiceUnavailable
+		}
+		c.Status(status)
+		return
+	}
+	status := out.Status
+	if status < 100 || status > 599 {
+		status = http.StatusOK
+	}
+	ctype := out.ContentType
+	if ctype == "" {
+		ctype = "application/octet-stream"
+	}
+	c.Data(status, ctype, out.Body)
+}
+
+// PluginWebhookDTO is one webhook of a plugin, with the workspace's URL.
+type PluginWebhookDTO struct {
+	ID          string                 `json:"id"`
+	Name        manifest.LocalizedText `json:"name"`
+	Description manifest.LocalizedText `json:"description,omitzero"`
+	// Path is the URL path; URL is the full URL when WeKnora knows its
+	// public address (APP_EXTERNAL_URL).
+	Path string `json:"path"`
+	URL  string `json:"url,omitempty"`
+}
+
+// List godoc
+// @Summary      插件 Webhook 地址
+// @Description  返回本空间该插件各个 Webhook 的地址，供在第三方系统中配置；地址含密钥，仅管理员可见
+// @Tags         Plugins
+// @Produce      json
+// @Param        id   path      string  true  "插件 ID"
+// @Success      200  {object}  map[string]interface{}
+// @Security     Bearer
+// @Router       /plugins/{id}/webhooks [get]
+func (h *PluginWebhookHandler) List(c *gin.Context) {
+	pluginID := c.Param("id")
+	m, ok := h.registry.Plugin(pluginID)
+	if !ok {
+		_ = c.Error(errors.NewNotFoundError("plugin not found"))
+		return
+	}
+	tenantID := c.GetUint64(types.TenantIDContextKey.String())
+	base := webhook.PublicBase()
+	out := []PluginWebhookDTO{}
+	for _, w := range m.Contributes[manifest.PointWebhooks] {
+		dto := PluginWebhookDTO{
+			ID:          w.ID,
+			Name:        w.Name,
+			Description: w.Description,
+			Path:        h.tokens.Path(m.ID, w.ID, tenantID),
+		}
+		if base != "" {
+			dto.URL = base + dto.Path
+		}
+		out = append(out, dto)
+	}
+	c.JSON(http.StatusOK, gin.H{"success": true, "data": out})
+}

internal/handler/plugin_webhook_test.go

@@ -0,0 +1,140 @@
+package handler
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+
+	"github.com/gin-gonic/gin"
+
+	"github.com/Tencent/WeKnora/internal/plugin/activate"
+	"github.com/Tencent/WeKnora/internal/plugin/manifest"
+	"github.com/Tencent/WeKnora/internal/plugin/plugintest"
+	"github.com/Tencent/WeKnora/internal/plugin/registry"
+	"github.com/Tencent/WeKnora/internal/plugin/tenancy"
+	"github.com/Tencent/WeKnora/internal/plugin/webhook"
+	"github.com/Tencent/WeKnora/internal/types"
+	"github.com/Tencent/WeKnora/pluginsdk"
+	"github.com/Tencent/WeKnora/pluginsdk/client"
+	"github.com/Tencent/WeKnora/pluginsdk/pluginapi"
+)
+
+func webhookEngine(t *testing.T) (*gin.Engine, *webhook.Tokens, *plugintest.MemTenantSettings) {
+	t.Helper()
+	gin.SetMode(gin.TestMode)
+	reg := registry.New()
+	m := &manifest.Manifest{
+		SchemaVersion: manifest.SchemaVersion, ID: "acme.hooks", Version: "1.0.0", APIVersion: pluginapi.APIVersion,
+		Name: manifest.Text("Hooks", nil), Publisher: manifest.Publisher{ID: "acme"},
+		Runtime: manifest.Runtime{Type: manifest.RuntimeHost, Kind: "binary", Entry: "bin/x"},
+		Contributes: manifest.Contributions{
+			manifest.PointWebhooks: {{ID: "inbox", Name: manifest.Text("Inbox", nil)}},
+		},
+	}
+	if err := reg.Register(m); err != nil {
+		t.Fatal(err)
+	}
+	settings := &plugintest.MemTenantSettings{}
+	_ = settings.Upsert(
+		context.Background(),
+		&types.PluginTenantSetting{TenantID: 7, PluginID: "acme.hooks", Enabled: true},
+	)
+
+	p := pluginsdk.New(pluginsdk.Info{ID: "acme.hooks", Version: "1.0.0"})
+	p.Webhook(
+		"inbox",
+		func(
+			_ context.Context, call *pluginsdk.Call, req pluginapi.WebhookRequest,
+		) (*pluginapi.WebhookResponse, error) {
+			if req.Headers["X-Signature"] != "ok" {
+				return &pluginapi.WebhookResponse{Status: 401}, nil
+			}
+			body, _ := json.Marshal(map[string]any{
+				"tenant": call.TenantID, "method": req.Method, "path": req.Path, "query": req.Query,
+				"body": string(req.Body), "cookie": req.Headers["Cookie"],
+			})
+			return &pluginapi.WebhookResponse{Status: 202, ContentType: "application/json", Body: body}, nil
+		},
+	)
+	srv := httptest.NewServer(p.Handler())
+	t.Cleanup(srv.Close)
+	tokens := webhook.NewTokens([]byte("k"))
+	h := NewPluginWebhookHandler(reg, tenancy.NewService(reg, settings),
+		activate.NewInvoker(oneClient{client.New(srv.URL, nil, nil)}), tokens)
+	r := gin.New()
+	r.Any(webhook.PathPrefix+"/:id/:hook/:token", h.Receive)
+	r.Any(webhook.PathPrefix+"/:id/:hook/:token/*path", h.Receive)
+	r.GET("/plugins/:id/webhooks", func(c *gin.Context) {
+		c.Set(types.TenantIDContextKey.String(), uint64(7))
+		h.List(c)
+	})
+	return r, tokens, settings
+}
+
+func call(r *gin.Engine, method, path string, body string, headers map[string]string) *httptest.ResponseRecorder {
+	w := httptest.NewRecorder()
+	req := httptest.NewRequest(method, path, bytes.NewBufferString(body))
+	for k, v := range headers {
+		req.Header.Set(k, v)
+	}
+	r.ServeHTTP(w, req)
+	return w
+}
+
+func TestPluginWebhooksRelayCalls(t *testing.T) {
+	r, tokens, settings := webhookEngine(t)
+	url := tokens.Path("acme.hooks", "inbox", 7)
+	signed := map[string]string{"X-Signature": "ok", "Cookie": "session=secret"}
+
+	w := call(r, http.MethodPost, url+"/issues?since=1", `{"id":1}`, signed)
+	if w.Code != 202 || w.Header().Get("Content-Type") != "application/json" {
+		t.Fatalf("relay = %d %s", w.Code, w.Body)
+	}
+	var got map[string]any
+	_ = json.Unmarshal(w.Body.Bytes(), &got)
+	if got["tenant"] != float64(7) || got["path"] != "/issues" || got["query"] != "since=1" ||
+		got["body"] != `{"id":1}` || got["cookie"] != "" {
+		t.Fatalf("plugin saw %v", got)
+	}
+	if w := call(r, http.MethodPost, url, `{}`, nil); w.Code != 401 {
+		t.Fatalf("the plugin's answer is relayed: %d", w.Code)
+	}
+
+	for _, bad := range []string{
+		tokens.Path("acme.hooks", "inbox", 8) + "x",
+		strings.Replace(url, "/inbox/", "/other/", 1),
+		webhook.PathPrefix + "/acme.hooks/inbox/7.nope",
+		tokens.Path("acme.nope", "inbox", 7),
+	} {
+		if w := call(r, http.MethodPost, bad, `{}`, signed); w.Code != http.StatusNotFound {
+			t.Errorf("%s = %d", bad, w.Code)
+		}
+	}
+	_ = settings.Upsert(
+		context.Background(),
+		&types.PluginTenantSetting{TenantID: 7, PluginID: "acme.hooks", Enabled: false},
+	)
+	if w := call(r, http.MethodPost, url, `{}`, signed); w.Code != http.StatusNotFound {
+		t.Fatalf("a switched-off plugin's webhook = %d", w.Code)
+	}
+}
+
+func TestPluginWebhookList(t *testing.T) {
+	r, tokens, _ := webhookEngine(t)
+	t.Setenv("APP_EXTERNAL_URL", "https://weknora.example.com/")
+	w := call(r, http.MethodGet, "/plugins/acme.hooks/webhooks", "", nil)
+	var resp struct {
+		Data []PluginWebhookDTO `json:"data"`
+	}
+	if json.Unmarshal(w.Body.Bytes(), &resp) != nil || len(resp.Data) != 1 {
+		t.Fatalf("list = %s", w.Body)
+	}
+	want := tokens.Path("acme.hooks", "inbox", 7)
+	if resp.Data[0].Path != want || resp.Data[0].URL != "https://weknora.example.com"+want {
+		t.Fatalf("webhook = %+v", resp.Data[0])
+	}
+}

internal/handler/session/qa.go

@@ -20,6 +20,7 @@ import (
 	"github.com/Tencent/WeKnora/internal/event"
 	"github.com/Tencent/WeKnora/internal/logger"
 	"github.com/Tencent/WeKnora/internal/models/api"
+	pluginevents "github.com/Tencent/WeKnora/internal/plugin/events"
 	"github.com/Tencent/WeKnora/internal/sandbox"
 	"github.com/Tencent/WeKnora/internal/storageurl"
 	"github.com/Tencent/WeKnora/internal/stream"
@@ -27,6 +28,7 @@ import (
 	"github.com/Tencent/WeKnora/internal/types"
 	"github.com/Tencent/WeKnora/internal/types/interfaces"
 	secutils "github.com/Tencent/WeKnora/internal/utils"
+	"github.com/Tencent/WeKnora/pluginsdk/pluginapi"
 	"github.com/gin-gonic/gin"
 	"github.com/google/uuid"
 )
@@ -1925,6 +1927,16 @@ func (h *Handler) completeAssistantMessage(
 		logger.Errorf(ctx, "Failed to persist assistant message %s: %v", assistantMessage.ID, err)
 		return err
 	}
+	// The stop button completes the message a first time without the
+	// question; the run's own completion follows with it.
+	if userQuery != "" {
+		tenantID, _ := types.TenantIDFromContext(ctx)
+		userID, _ := types.UserIDFromContext(ctx)
+		pluginevents.Publish(ctx, tenantID, pluginapi.EventChatAnswered, pluginapi.ChatEventData{
+			SessionID: assistantMessage.SessionID, MessageID: assistantMessage.ID, AgentID: assistantMessage.AgentID,
+			UserID: userID, Question: userQuery, Answer: assistantMessage.Content,
+		})
+	}
 
 	// Asynchronously index the Q&A pair into the chat history knowledge base for vector search.
 	// Use WithoutCancel so the goroutine survives after the HTTP request context is done.

internal/plugin/activate/invoke.go

@@ -8,6 +8,7 @@ import (
 	"github.com/Tencent/WeKnora/internal/plugin/install"
 	"github.com/Tencent/WeKnora/internal/plugin/manifest"
 	"github.com/Tencent/WeKnora/internal/plugin/tenancy"
+	"github.com/Tencent/WeKnora/internal/plugin/webhook"
 	"github.com/Tencent/WeKnora/internal/types"
 	"github.com/Tencent/WeKnora/internal/types/interfaces"
 	"github.com/Tencent/WeKnora/pluginsdk/client"
@@ -37,6 +38,17 @@ type Invoker struct {
 	tokens        TokenIssuer
 	hostURL       string
 	publicHostURL string
+	// webhooks and publicBase give plugins their workspace's webhook URLs.
+	webhooks   *webhook.Tokens
+	publicBase string
+}
+
+// SetWebhooks lets calls carry the workspace's webhook URLs, when WeKnora's
+// public address (publicBase) is known.
+func (iv *Invoker) SetWebhooks(tokens *webhook.Tokens, publicBase string) {
+	iv.mu.Lock()
+	iv.webhooks, iv.publicBase = tokens, publicBase
+	iv.mu.Unlock()
 }
 
 // TokenIssuer signs the Host API token of one call.
@@ -111,7 +123,16 @@ func (iv *Invoker) Envelope(
 	if !iv.clients.OnThisNode(m.ID) {
 		hostURL = iv.publicHostURL
 	}
+	hooks, publicBase := iv.webhooks, iv.publicBase
 	iv.mu.RUnlock()
+	if hooks != nil && publicBase != "" && env.Context.TenantID != 0 {
+		for _, c := range m.Contributes[manifest.PointWebhooks] {
+			if env.Context.Webhooks == nil {
+				env.Context.Webhooks = map[string]string{}
+			}
+			env.Context.Webhooks[c.ID] = publicBase + hooks.Path(m.ID, c.ID, env.Context.TenantID)
+		}
+	}
 	if tokens != nil && hostURL != "" && env.Context.TenantID != 0 && len(m.Permissions.HostAPI) > 0 {
 		token, _, err := tokens.Issue(m.ID, m.Version, env.Context.TenantID, m.Permissions.HostAPI)
 		if err != nil {

internal/plugin/activate/invoke_test.go

@@ -6,6 +6,7 @@ import (
 
 	"github.com/Tencent/WeKnora/internal/plugin/hostapi"
 	"github.com/Tencent/WeKnora/internal/plugin/manifest"
+	"github.com/Tencent/WeKnora/internal/plugin/webhook"
 	"github.com/Tencent/WeKnora/internal/types"
 )
 
@@ -46,3 +47,21 @@ func TestEnvelopeCarriesHostAccessOnlyForGrantedPlugins(t *testing.T) {
 		t.Fatalf("remote envelope = %+v", env.Context)
 	}
 }
+
+func TestEnvelopeCarriesWebhookURLs(t *testing.T) {
+	iv := NewInvoker(fakeClients{})
+	m := &manifest.Manifest{ID: "acme.hooks", Contributes: manifest.Contributions{
+		manifest.PointWebhooks: {{ID: "inbox"}},
+	}}
+	ctx := context.WithValue(context.Background(), types.TenantIDContextKey, uint64(7))
+	if env, _ := iv.Envelope(ctx, m, nil); env.Context.Webhooks != nil {
+		t.Fatal("no URLs without a public address")
+	}
+	tokens := webhook.NewTokens([]byte("k"))
+	iv.SetWebhooks(tokens, "https://weknora.example.com")
+	env, err := iv.Envelope(ctx, m, nil)
+	want := "https://weknora.example.com" + tokens.Path("acme.hooks", "inbox", 7)
+	if err != nil || env.Context.Webhooks["inbox"] != want {
+		t.Fatalf("webhooks = %v, %v", env.Context.Webhooks, err)
+	}
+}

internal/plugin/events/events.go

@@ -0,0 +1,151 @@
+// Package events delivers WeKnora's business events (a document ingested,
+// an answer given) to the plugins that subscribed to them in
+// permissions.events. Publishing enqueues one task per subscribed plugin
+// that the workspace has switched on; the task calls the plugin's
+// /v1/events and is retried with backoff while the plugin fails with a
+// retryable error, keeping the event ID (at least once).
+package events
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"slices"
+	"sync/atomic"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/hibiken/asynq"
+
+	"github.com/Tencent/WeKnora/internal/logger"
+	"github.com/Tencent/WeKnora/internal/plugin/manifest"
+	"github.com/Tencent/WeKnora/internal/plugin/registry"
+	"github.com/Tencent/WeKnora/internal/plugin/tenancy"
+	"github.com/Tencent/WeKnora/internal/types"
+	"github.com/Tencent/WeKnora/internal/types/interfaces"
+	"github.com/Tencent/WeKnora/pluginsdk/pluginapi"
+)
+
+// TaskType is the task that delivers one event to one plugin.
+const TaskType = types.TypePluginEvent
+
+// Delivery bounds: retries back off (asynq's default, or the Lite
+// executor's), each attempt within its own timeout.
+const (
+	maxRetry        = 10
+	deliveryTimeout = 2 * time.Minute
+)
+
+// payload is one queued delivery.
+type payload struct {
+	PluginID string                  `json:"pluginId"`
+	TenantID uint64                  `json:"tenantId"`
+	Event    pluginapi.EventDelivery `json:"event"`
+}
+
+// Caller makes a call to a plugin (activate.Invoker); an interface so
+// services that publish events do not import the plugin runtime.
+type Caller interface {
+	Call(ctx context.Context, m *manifest.Manifest, path string, instance map[string]any, input, out any) error
+}
+
+// Dispatcher fans events out to subscribed plugins and delivers them.
+type Dispatcher struct {
+	registry *registry.Registry
+	tenancy  *tenancy.Service
+	enqueue  interfaces.TaskEnqueuer
+	invoker  Caller
+	now      func() time.Time
+}
+
+// NewDispatcher creates a Dispatcher.
+func NewDispatcher(
+	reg *registry.Registry, t *tenancy.Service, enq interfaces.TaskEnqueuer, iv Caller,
+) *Dispatcher {
+	return &Dispatcher{registry: reg, tenancy: t, enqueue: enq, invoker: iv, now: time.Now}
+}
+
+var current atomic.Pointer[Dispatcher]
+
+// SetDefault makes d the dispatcher Publish uses.
+func SetDefault(d *Dispatcher) { current.Store(d) }
+
+// Publish sends an event of a workspace to the plugins subscribed to it. It
+// does nothing until the plugin runtime starts, and never fails the caller:
+// a delivery it cannot queue is logged.
+func Publish(ctx context.Context, tenantID uint64, typ string, data any) {
+	if d := current.Load(); d != nil && tenantID != 0 {
+		d.Publish(ctx, tenantID, typ, data)
+	}
+}
+
+// subscribed reports whether a loaded plugin wants an event and can take
+// it (it has code).
+func subscribed(m *manifest.Manifest, typ string) bool {
+	rt := m.Runtime.Type
+	return (rt == manifest.RuntimeHost || rt == manifest.RuntimeRemote) && slices.Contains(m.Permissions.Events, typ)
+}
+
+// Publish enqueues a delivery for every subscribed plugin the workspace has
+// switched on.
+func (d *Dispatcher) Publish(ctx context.Context, tenantID uint64, typ string, data any) {
+	raw, err := json.Marshal(data)
+	if err != nil {
+		logger.Warnf(ctx, "[plugin] event %s: %v", typ, err)
+		return
+	}
+	ev := pluginapi.EventDelivery{ID: uuid.NewString(), Type: typ, OccurredAt: d.now().UTC(), Data: raw}
+	for _, m := range d.registry.Plugins() {
+		if !subscribed(m, typ) {
+			continue
+		}
+		if on, err := d.tenancy.PluginEnabled(ctx, tenantID, m.ID); err != nil || !on {
+			if err != nil {
+				logger.Warnf(ctx, "[plugin] event %s to %s: read switch: %v", typ, m.ID, err)
+			}
+			continue
+		}
+		body, _ := json.Marshal(payload{PluginID: m.ID, TenantID: tenantID, Event: ev})
+		task := asynq.NewTask(TaskType, body)
+		_, err := d.enqueue.Enqueue(task, asynq.Queue(types.QueueMaintenance), asynq.MaxRetry(maxRetry),
+			asynq.Timeout(deliveryTimeout))
+		if err != nil {
+			logger.Warnf(ctx, "[plugin] event %s to %s: enqueue: %v", typ, m.ID, err)
+		}
+	}
+}
+
+// Handle delivers one event (the TaskType handler). An event is dropped when
+// the plugin is gone, no longer subscribed or switched off, or refuses it
+// for good; a retryable failure is returned so the task runs again.
+func (d *Dispatcher) Handle(ctx context.Context, task *asynq.Task) error {
+	var p payload
+	if err := json.Unmarshal(task.Payload(), &p); err != nil {
+		return fmt.Errorf("decode plugin event: %w: %w", err, asynq.SkipRetry)
+	}
+	m, ok := d.registry.Plugin(p.PluginID)
+	if !ok || !subscribed(m, p.Event.Type) {
+		return nil
+	}
+	if on, err := d.tenancy.PluginEnabled(ctx, p.TenantID, p.PluginID); err != nil {
+		return err
+	} else if !on {
+		return nil
+	}
+	ev := p.Event
+	ev.Attempt = 1
+	if n, ok := asynq.GetRetryCount(ctx); ok {
+		ev.Attempt = n + 1
+	}
+	ctx = context.WithValue(ctx, types.TenantIDContextKey, p.TenantID)
+	err := d.invoker.Call(ctx, m, pluginapi.EventsPath, nil, ev, nil)
+	if err == nil {
+		return nil
+	}
+	if pe, ok := pluginapi.AsError(err); ok && !pe.Retryable {
+		logger.Warnf(ctx, "[plugin] %s refused event %s (%s): %v", p.PluginID, ev.ID, ev.Type, err)
+		return nil
+	}
+	logger.Infof(ctx, "[plugin] event %s to %s failed (attempt %d), will retry: %v", ev.ID, p.PluginID, ev.Attempt, err)
+	return err
+}

internal/plugin/events/events_test.go

@@ -0,0 +1,208 @@
+package events
+
+import (
+	"context"
+	"encoding/json"
+	"net/http/httptest"
+	"sync"
+	"testing"
+
+	"github.com/hibiken/asynq"
+
+	"github.com/Tencent/WeKnora/internal/plugin/activate"
+	"github.com/Tencent/WeKnora/internal/plugin/manifest"
+	"github.com/Tencent/WeKnora/internal/plugin/plugintest"
+	"github.com/Tencent/WeKnora/internal/plugin/registry"
+	"github.com/Tencent/WeKnora/internal/plugin/tenancy"
+	"github.com/Tencent/WeKnora/internal/types"
+	"github.com/Tencent/WeKnora/internal/types/interfaces"
+	"github.com/Tencent/WeKnora/pluginsdk"
+	"github.com/Tencent/WeKnora/pluginsdk/client"
+	"github.com/Tencent/WeKnora/pluginsdk/pluginapi"
+)
+
+type queue struct {
+	mu    sync.Mutex
+	tasks []*asynq.Task
+}
+
+func (q *queue) Enqueue(task *asynq.Task, _ ...asynq.Option) (*asynq.TaskInfo, error) {
+	q.mu.Lock()
+	defer q.mu.Unlock()
+	q.tasks = append(q.tasks, task)
+	return &asynq.TaskInfo{}, nil
+}
+
+func (q *queue) take() []*asynq.Task {
+	q.mu.Lock()
+	defer q.mu.Unlock()
+	out := q.tasks
+	q.tasks = nil
+	return out
+}
+
+type oneClient struct{ c *client.Client }
+
+func (o oneClient) Client(context.Context, *manifest.Manifest) (*client.Client, error) {
+	return o.c, nil
+}
+func (o oneClient) OnThisNode(string) bool { return true }
+
+type received struct {
+	mu     sync.Mutex
+	events []pluginapi.EventDelivery
+	tenant []uint64
+}
+
+func setup(t *testing.T) (*Dispatcher, *queue, *received, *plugintest.MemTenantSettings) {
+	t.Helper()
+	reg := registry.New()
+	plugin := func(id string, events ...string) *manifest.Manifest {
+		return &manifest.Manifest{
+			SchemaVersion: manifest.SchemaVersion, ID: id, Version: "1.0.0", APIVersion: pluginapi.APIVersion,
+			Name: manifest.Text(id, nil), Publisher: manifest.Publisher{ID: "acme"},
+			Runtime:     manifest.Runtime{Type: manifest.RuntimeHost, Kind: "binary", Entry: "bin/x"},
+			Permissions: manifest.Permissions{Events: events},
+		}
+	}
+	listener := plugin("acme.audit", pluginapi.EventKnowledgeIngested, pluginapi.EventKnowledgeFailed)
+	other := plugin("acme.other", pluginapi.EventChatAnswered)
+	for _, m := range []*manifest.Manifest{listener, other} {
+		if err := reg.Register(m); err != nil {
+			t.Fatal(err)
+		}
+	}
+	settings := &plugintest.MemTenantSettings{}
+	ctx := context.Background()
+	for _, id := range []string{"acme.audit", "acme.other"} {
+		_ = settings.Upsert(ctx, &types.PluginTenantSetting{TenantID: 7, PluginID: id, Enabled: true})
+	}
+
+	got := &received{}
+	p := pluginsdk.New(pluginsdk.Info{ID: "acme.audit", Version: "1.0.0"})
+	p.OnEvent(func(_ context.Context, call *pluginsdk.Call, ev pluginapi.EventDelivery) error {
+		got.mu.Lock()
+		got.events = append(got.events, ev)
+		got.tenant = append(got.tenant, call.TenantID)
+		got.mu.Unlock()
+		var data pluginapi.KnowledgeEventData
+		_ = json.Unmarshal(ev.Data, &data)
+		switch data.KnowledgeID {
+		case "down":
+			return pluginapi.Errorf(pluginapi.CodeUnavailable, "try later")
+		case "bad":
+			return pluginapi.Errorf(pluginapi.CodeBadRequest, "never")
+		}
+		return nil
+	})
+	srv := httptest.NewServer(p.Handler())
+	t.Cleanup(srv.Close)
+	q := &queue{}
+	d := NewDispatcher(reg, tenancy.NewService(reg, settings), q,
+		activate.NewInvoker(oneClient{client.New(srv.URL, nil, nil)}))
+	return d, q, got, settings
+}
+
+func TestPublishFansOutToSubscribedEnabledPlugins(t *testing.T) {
+	ctx := context.Background()
+	d, q, got, settings := setup(t)
+
+	d.Publish(ctx, 7, pluginapi.EventKnowledgeIngested, pluginapi.KnowledgeEventData{KnowledgeID: "k1"})
+	tasks := q.take()
+	if len(tasks) != 1 || tasks[0].Type() != TaskType {
+		t.Fatalf("tasks = %d", len(tasks))
+	}
+	if err := d.Handle(ctx, tasks[0]); err != nil {
+		t.Fatal(err)
+	}
+	if len(got.events) != 1 || got.events[0].Type != pluginapi.EventKnowledgeIngested || got.tenant[0] != 7 ||
+		got.events[0].Attempt != 1 || got.events[0].ID == "" {
+		t.Fatalf("delivered = %+v tenants %v", got.events, got.tenant)
+	}
+
+	// Another workspace never switched the plugins on.
+	d.Publish(ctx, 8, pluginapi.EventKnowledgeIngested, pluginapi.KnowledgeEventData{KnowledgeID: "k1"})
+	if n := len(q.take()); n != 0 {
+		t.Fatalf("an off plugin got %d deliveries", n)
+	}
+
+	// A plugin switched off after the event was queued does not get it.
+	d.Publish(ctx, 7, pluginapi.EventKnowledgeIngested, pluginapi.KnowledgeEventData{KnowledgeID: "k2"})
+	queued := q.take()
+	_ = settings.Upsert(ctx, &types.PluginTenantSetting{TenantID: 7, PluginID: "acme.audit", Enabled: false})
+	if err := d.Handle(ctx, queued[0]); err != nil || len(got.events) != 1 {
+		t.Fatalf("delivered to a switched-off plugin: %v", err)
+	}
+}
+
+func TestHandleRetriesOnlyRetryableFailures(t *testing.T) {
+	ctx := context.Background()
+	d, q, got, _ := setup(t)
+	d.Publish(ctx, 7, pluginapi.EventKnowledgeFailed, pluginapi.KnowledgeEventData{KnowledgeID: "down"})
+	d.Publish(ctx, 7, pluginapi.EventKnowledgeFailed, pluginapi.KnowledgeEventData{KnowledgeID: "bad"})
+	tasks := q.take()
+	if err := d.Handle(ctx, tasks[0]); err == nil {
+		t.Fatal("a retryable failure must be retried")
+	}
+	if err := d.Handle(ctx, tasks[0]); err == nil || got.events[0].ID != got.events[1].ID {
+		t.Fatal("a retry keeps the event ID")
+	}
+	if err := d.Handle(ctx, tasks[1]); err != nil {
+		t.Fatalf("a refused event is dropped, got %v", err)
+	}
+}
+
+// fakeKnowledge is the part of the repository the watch uses.
+type fakeKnowledge struct {
+	interfaces.KnowledgeRepository
+	rows map[string]*types.Knowledge
+}
+
+func (f *fakeKnowledge) GetKnowledgeByIDOnly(_ context.Context, id string) (*types.Knowledge, error) {
+	k := *f.rows[id]
+	return &k, nil
+}
+
+func (f *fakeKnowledge) UpdateKnowledge(_ context.Context, k *types.Knowledge) error {
+	c := *k
+	f.rows[k.ID] = &c
+	return nil
+}
+
+func (f *fakeKnowledge) FinalizeSubtask(_ context.Context, id string) (int, bool, error) {
+	f.rows[id].ParseStatus = types.ParseStatusCompleted
+	return 0, true, nil
+}
+
+func TestWatchKnowledgePublishesTransitionsOnly(t *testing.T) {
+	ctx := context.Background()
+	d, q, _, _ := setup(t)
+	SetDefault(d)
+	t.Cleanup(func() { SetDefault(nil) })
+	base := &fakeKnowledge{rows: map[string]*types.Knowledge{
+		"k1": {ID: "k1", TenantID: 7, KnowledgeBaseID: "kb", ParseStatus: types.ParseStatusProcessing},
+		"k2": {ID: "k2", TenantID: 7, KnowledgeBaseID: "kb", ParseStatus: types.ParseStatusFinalizing},
+	}}
+	repo := WatchKnowledge(base)
+
+	k := *base.rows["k1"]
+	k.ParseStatus, k.ErrorMessage = types.ParseStatusFailed, "boom"
+	_ = repo.UpdateKnowledge(ctx, &k)
+	_ = repo.UpdateKnowledge(ctx, &k) // already failed: no second event
+	k.ParseStatus = types.ParseStatusProcessing
+	_ = repo.UpdateKnowledge(ctx, &k) // not terminal
+	_, _, _ = repo.FinalizeSubtask(ctx, "k2")
+
+	var seen []string
+	for _, task := range q.take() {
+		var p payload
+		_ = json.Unmarshal(task.Payload(), &p)
+		var data pluginapi.KnowledgeEventData
+		_ = json.Unmarshal(p.Event.Data, &data)
+		seen = append(seen, p.Event.Type+":"+data.KnowledgeID+":"+data.Error)
+	}
+	want := []string{"knowledge.failed:k1:boom", "knowledge.ingested:k2:"}
+	if len(seen) != 2 || seen[0] != want[0] || seen[1] != want[1] {
+		t.Fatalf("published %v, want %v", seen, want)
+	}
+}

internal/plugin/events/knowledge.go

@@ -0,0 +1,122 @@
+package events
+
+import (
+	"context"
+
+	"github.com/Tencent/WeKnora/internal/types"
+	"github.com/Tencent/WeKnora/internal/types/interfaces"
+	"github.com/Tencent/WeKnora/pluginsdk/pluginapi"
+)
+
+// WatchKnowledge wraps the knowledge repository to publish
+// knowledge.ingested and knowledge.failed. A document reaches either state
+// through many code paths (enrichment finishing last, no enrichment to do,
+// a dozen failure branches), but only through these few writes, so the
+// events are raised here, on the transition only.
+func WatchKnowledge(repo interfaces.KnowledgeRepository) interfaces.KnowledgeRepository {
+	return &knowledgeWatch{KnowledgeRepository: repo}
+}
+
+type knowledgeWatch struct {
+	interfaces.KnowledgeRepository
+}
+
+func terminal(status string) bool {
+	return status == types.ParseStatusCompleted || status == types.ParseStatusFailed
+}
+
+// publishKnowledge raises the event for a row that just reached status.
+func publishKnowledge(ctx context.Context, k *types.Knowledge) {
+	if k == nil {
+		return
+	}
+	data := pluginapi.KnowledgeEventData{
+		KnowledgeBaseID: k.KnowledgeBaseID, KnowledgeID: k.ID, Title: k.Title,
+		FileName: k.FileName, FileType: k.FileType, Source: k.Source,
+	}
+	switch k.ParseStatus {
+	case types.ParseStatusCompleted:
+		Publish(ctx, k.TenantID, pluginapi.EventKnowledgeIngested, data)
+	case types.ParseStatusFailed:
+		data.Error = k.ErrorMessage
+		Publish(ctx, k.TenantID, pluginapi.EventKnowledgeFailed, data)
+	}
+}
+
+// previous reads a row's status before a write that may change it, only
+// when events are on (the read is not free).
+func (w *knowledgeWatch) previous(ctx context.Context, id string) string {
+	if current.Load() == nil {
+		return ""
+	}
+	before, err := w.GetKnowledgeByIDOnly(ctx, id)
+	if err != nil || before == nil {
+		return ""
+	}
+	return before.ParseStatus
+}
+
+func (w *knowledgeWatch) UpdateKnowledge(ctx context.Context, k *types.Knowledge) error {
+	if k == nil || !terminal(k.ParseStatus) {
+		return w.KnowledgeRepository.UpdateKnowledge(ctx, k)
+	}
+	was := w.previous(ctx, k.ID)
+	if err := w.KnowledgeRepository.UpdateKnowledge(ctx, k); err != nil {
+		return err
+	}
+	if was != k.ParseStatus {
+		publishKnowledge(ctx, k)
+	}
+	return nil
+}
+
+func (w *knowledgeWatch) UpdateKnowledgeForTransfer(ctx context.Context, before, after *types.Knowledge) error {
+	if err := w.KnowledgeRepository.UpdateKnowledgeForTransfer(ctx, before, after); err != nil {
+		return err
+	}
+	if after != nil && terminal(after.ParseStatus) && (before == nil || before.ParseStatus != after.ParseStatus) {
+		publishKnowledge(ctx, after)
+	}
+	return nil
+}
+
+func (w *knowledgeWatch) UpdateKnowledgeColumns(ctx context.Context, id string, values map[string]interface{}) error {
+	status, _ := values["parse_status"].(string)
+	if !terminal(status) {
+		return w.KnowledgeRepository.UpdateKnowledgeColumns(ctx, id, values)
+	}
+	was := w.previous(ctx, id)
+	if err := w.KnowledgeRepository.UpdateKnowledgeColumns(ctx, id, values); err != nil {
+		return err
+	}
+	if was != status {
+		w.publishByID(ctx, id)
+	}
+	return nil
+}
+
+func (w *knowledgeWatch) FinalizeSubtask(ctx context.Context, id string) (int, bool, error) {
+	n, promoted, err := w.KnowledgeRepository.FinalizeSubtask(ctx, id)
+	if err == nil && promoted {
+		w.publishByID(ctx, id)
+	}
+	return n, promoted, err
+}
+
+func (w *knowledgeWatch) CompleteProcessingWithoutSubtasks(ctx context.Context, id string) (bool, error) {
+	completed, err := w.KnowledgeRepository.CompleteProcessingWithoutSubtasks(ctx, id)
+	if err == nil && completed {
+		w.publishByID(ctx, id)
+	}
+	return completed, err
+}
+
+func (w *knowledgeWatch) publishByID(ctx context.Context, id string) {
+	if current.Load() == nil {
+		return
+	}
+	k, err := w.GetKnowledgeByIDOnly(ctx, id)
+	if err == nil {
+		publishKnowledge(ctx, k)
+	}
+}

internal/plugin/host/process.go

@@ -341,6 +341,9 @@ func CheckServedManifest(want *manifest.Manifest, got *pluginapi.Manifest) error
 	if got.APIVersion != pluginapi.APIVersion {
 		return fmt.Errorf("plugin speaks %q, this WeKnora speaks %q", got.APIVersion, pluginapi.APIVersion)
 	}
+	if len(want.Permissions.Events) > 0 && len(got.Contributes["events"]) == 0 {
+		return errors.New("plugin.yaml subscribes to events but the plugin handles none")
+	}
 	for point, contribs := range want.Contributes {
 		if info, ok := manifest.LookupPoint(point); !ok || info.Declarative {
 			continue

internal/plugin/manifest/manifest.go

@@ -15,6 +15,8 @@ import (
 
 	"golang.org/x/mod/semver"
 	"gopkg.in/yaml.v3"
+
+	"github.com/Tencent/WeKnora/pluginsdk/pluginapi"
 )
 
 // SchemaVersion is the manifest format version this build understands.
@@ -329,6 +331,14 @@ func (m *Manifest) validatePermissions(add func(string, ...any)) {
 			)
 		}
 	}
+	for _, e := range m.Permissions.Events {
+		if !slices.Contains(pluginapi.EventTypes, e) {
+			add("permissions.events %q is not an event type (known: %s)", e, strings.Join(pluginapi.EventTypes, ", "))
+		}
+	}
+	if len(m.Permissions.Events) > 0 && m.Runtime.Type == RuntimeDeclarative {
+		add("permissions.events needs code; declarative plugins cannot receive events")
+	}
 }
 
 // storedTypeIDLimit is how long a qualified contribution ID may be where
@@ -337,7 +347,8 @@ func (m *Manifest) validatePermissions(add func(string, ...any)) {
 var storedTypeIDLimit = map[Point]int{PointConnectors: 50, PointWebSearch: 50}
 
 func (m *Manifest) validateContributions(add func(string, ...any)) {
-	if len(m.Contributes) == 0 {
+	// A plugin that only listens to events contributes nothing else.
+	if len(m.Contributes) == 0 && len(m.Permissions.Events) == 0 {
 		add("contributes must declare at least one contribution")
 	}
 	for point := range m.Contributes {

internal/plugin/manifest/point.go

@@ -23,6 +23,9 @@ const (
 	PointSettingsSections Point = "settingsSections"
 	// PointKBTabs contributes tabs of the knowledge base page.
 	PointKBTabs Point = "kbTabs"
+	// PointWebhooks contributes inbound URLs third parties call, one per
+	// workspace; WeKnora relays the calls to the plugin.
+	PointWebhooks Point = "webhooks"
 )
 
 // UIRoot is the package directory plugin pages live in. Only files under it
@@ -64,6 +67,7 @@ var points = []PointInfo{
 	{Point: PointPages, ThirdParty: true, Declarative: true},
 	{Point: PointSettingsSections, ThirdParty: true, Declarative: true},
 	{Point: PointKBTabs, ThirdParty: true, Declarative: true},
+	{Point: PointWebhooks, ThirdParty: true},
 }
 
 // Points returns every known extension point in display order.

internal/plugin/tenancy/tenancy.go

@@ -64,6 +64,27 @@ func (s *Service) List(ctx context.Context, tenantID uint64) ([]TenantPlugin, er
 	return out, nil
 }
 
+// PluginEnabled reports whether a tenant has a plugin switched on. Unlike
+// EnabledFilter it fails closed: it decides whether workspace data goes to
+// the plugin.
+func (s *Service) PluginEnabled(ctx context.Context, tenantID uint64, pluginID string) (bool, error) {
+	m, ok := s.registry.Plugin(pluginID)
+	if !ok {
+		return false, nil
+	}
+	if m.Required {
+		return true, nil
+	}
+	row, err := s.repo.Get(ctx, tenantID, pluginID)
+	if err != nil {
+		return false, err
+	}
+	if row == nil {
+		return enabledByDefault(m), nil
+	}
+	return row.Enabled, nil
+}
+
 // SetEnabled turns a plugin on or off for a tenant.
 func (s *Service) SetEnabled(
 	ctx context.Context, tenantID uint64, pluginID string, enabled bool, updatedBy string,

internal/plugin/webhook/webhook.go

@@ -0,0 +1,87 @@
+// Package webhook gives each workspace a secret URL per plugin webhook.
+// Third parties call WeKnora there; the URL alone names the workspace, so
+// it carries the tenant ID and a MAC that only WeKnora can compute.
+package webhook
+
+import (
+	"crypto/hmac"
+	"crypto/rand"
+	"crypto/sha256"
+	"encoding/base64"
+	"os"
+	"strconv"
+	"strings"
+
+	"github.com/Tencent/WeKnora/internal/utils"
+)
+
+// PathPrefix is where webhooks are received:
+// {prefix}/{pluginId}/{webhookId}/{token}[/...].
+const PathPrefix = "/api/v1/plugin-callbacks"
+
+// macBytes is the length of a token's MAC.
+const macBytes = 16
+
+// Tokens signs and checks webhook URLs.
+type Tokens struct{ key []byte }
+
+// NewTokens signs with key.
+func NewTokens(key []byte) *Tokens { return &Tokens{key: key} }
+
+// NewTokensFromEnv derives the key from SYSTEM_AES_KEY or JWT_SECRET, the
+// same on every node. Without either, URLs change on restart.
+func NewTokensFromEnv() *Tokens {
+	var secret []byte
+	switch {
+	case utils.GetAESKey() != nil:
+		secret = utils.GetAESKey()
+	case strings.TrimSpace(os.Getenv("JWT_SECRET")) != "":
+		secret = []byte(strings.TrimSpace(os.Getenv("JWT_SECRET")))
+	default:
+		secret = make([]byte, 32)
+		_, _ = rand.Read(secret)
+	}
+	mac := hmac.New(sha256.New, secret)
+	_, _ = mac.Write([]byte("weknora plugin webhooks"))
+	return &Tokens{key: mac.Sum(nil)}
+}
+
+func (t *Tokens) mac(pluginID, webhookID string, tenantID uint64) []byte {
+	m := hmac.New(sha256.New, t.key)
+	_, _ = m.Write([]byte(pluginID + "\x00" + webhookID + "\x00" + strconv.FormatUint(tenantID, 10)))
+	return m.Sum(nil)[:macBytes]
+}
+
+// Token is the URL segment naming a workspace: "<tenant>.<mac>".
+func (t *Tokens) Token(pluginID, webhookID string, tenantID uint64) string {
+	return strconv.FormatUint(tenantID, 36) + "." +
+		base64.RawURLEncoding.EncodeToString(t.mac(pluginID, webhookID, tenantID))
+}
+
+// Verify returns the workspace a token names, if it is genuine.
+func (t *Tokens) Verify(pluginID, webhookID, token string) (uint64, bool) {
+	tenant, sig, ok := strings.Cut(token, ".")
+	if !ok {
+		return 0, false
+	}
+	tenantID, err := strconv.ParseUint(tenant, 36, 64)
+	if err != nil || tenantID == 0 {
+		return 0, false
+	}
+	got, err := base64.RawURLEncoding.DecodeString(sig)
+	if err != nil || !hmac.Equal(got, t.mac(pluginID, webhookID, tenantID)) {
+		return 0, false
+	}
+	return tenantID, true
+}
+
+// Path is the URL path of a webhook for a workspace.
+func (t *Tokens) Path(pluginID, webhookID string, tenantID uint64) string {
+	return PathPrefix + "/" + pluginID + "/" + webhookID + "/" + t.Token(pluginID, webhookID, tenantID)
+}
+
+// PublicBase is WeKnora's public address (APP_EXTERNAL_URL), without a
+// trailing slash; empty when not configured.
+func PublicBase() string {
+	return strings.TrimSuffix(strings.TrimSpace(os.Getenv("APP_EXTERNAL_URL")), "/")
+}

internal/plugin/webhook/webhook_test.go

@@ -0,0 +1,31 @@
+package webhook
+
+import (
+	"strings"
+	"testing"
+)
+
+func TestTokens(t *testing.T) {
+	tok := NewTokens([]byte("k"))
+	token := tok.Token("acme.jira", "events", 42)
+	if id, ok := tok.Verify("acme.jira", "events", token); !ok || id != 42 {
+		t.Fatalf("verify = %d, %v", id, ok)
+	}
+	for _, bad := range []struct{ plugin, hook, token string }{
+		{"acme.other", "events", token},
+		{"acme.jira", "other", token},
+		{"acme.jira", "events", strings.Replace(token, "16.", "17.", 1)},
+		{"acme.jira", "events", "16"},
+		{"acme.jira", "events", "0." + strings.SplitN(token, ".", 2)[1]},
+	} {
+		if _, ok := tok.Verify(bad.plugin, bad.hook, bad.token); ok {
+			t.Errorf("%+v verified", bad)
+		}
+	}
+	if _, ok := NewTokens([]byte("other")).Verify("acme.jira", "events", token); ok {
+		t.Fatal("another key verified the token")
+	}
+	if p := tok.Path("acme.jira", "events", 42); !strings.HasPrefix(p, PathPrefix+"/acme.jira/events/16.") {
+		t.Fatalf("path = %s", p)
+	}
+}

internal/router/router.go

@@ -21,6 +21,7 @@ import (
 	"github.com/Tencent/WeKnora/internal/mcpserver"
 	"github.com/Tencent/WeKnora/internal/middleware"
 	"github.com/Tencent/WeKnora/internal/plugin/hostapi"
+	"github.com/Tencent/WeKnora/internal/plugin/webhook"
 	"github.com/Tencent/WeKnora/internal/tracing/langfuse"
 	"github.com/Tencent/WeKnora/internal/types/interfaces"
 
@@ -95,6 +96,7 @@ type RouterParams struct {
 	PluginAdminHandler           *handler.PluginAdminHandler
 	PluginHostAPI                *hostapi.Handler
 	PluginUIHandler              *handler.PluginUIHandler
+	PluginWebhookHandler         *handler.PluginWebhookHandler
 	WeKnoraCloudHandler          *handler.WeKnoraCloudHandler
 	WikiPageHandler              *handler.WikiPageHandler
 	MemoryHandler                *handler.MemoryHandler
@@ -219,6 +221,13 @@ func NewRouter(params RouterParams) *gin.Engine {
 	if params.PluginUIHandler != nil {
 		r.GET(handler.PluginUIAssetsPrefix+"/:id/:version/*path", params.PluginUIHandler.ServeAsset)
 	}
+	// Plugin webhooks: third parties call a secret per-workspace URL; the
+	// handler checks it, so no login applies.
+	if params.PluginWebhookHandler != nil {
+		hooks := r.Group(webhook.PathPrefix)
+		hooks.Any("/:id/:hook/:token", params.PluginWebhookHandler.Receive)
+		hooks.Any("/:id/:hook/:token/*path", params.PluginWebhookHandler.Receive)
+	}
 
 	// 认证中间件
 	r.Use(middleware.Auth(params.TenantService, params.UserService, params.TenantMemberService, params.TenantAPIKeyService, params.Config))
@@ -339,7 +348,7 @@ func NewRouter(params RouterParams) *gin.Engine {
 		RegisterEmbedChannelRoutes(v1, params.EmbedChannelHandler, rbacGuards)
 		RegisterMCPEndpointRoutes(v1, params.MCPEndpointHandler, rbacGuards)
 		RegisterDataSourceRoutes(v1, params.DataSourceHandler, params.DataSourceCredentialsHandler, rbacGuards)
-		RegisterPluginRoutes(v1, params.PluginHandler, params.PluginUIHandler, rbacGuards)
+		RegisterPluginRoutes(v1, params.PluginHandler, params.PluginUIHandler, params.PluginWebhookHandler, rbacGuards)
 		RegisterPluginAdminRoutes(v1, params.PluginAdminHandler, rbacGuards)
 		RegisterWeKnoraCloudRoutes(v1, params.WeKnoraCloudHandler, rbacGuards)
 		RegisterWikiPageRoutes(v1, params.WikiPageHandler, rbacGuards)

internal/router/routes_plugin.go

@@ -12,7 +12,10 @@ import (
 // plugin switches. Any member may read the catalog; only admins change the
 // switches. Like the other integration catalogs it stays closed to scoped
 // API keys.
-func RegisterPluginRoutes(r *gin.RouterGroup, h *handler.PluginHandler, ui *handler.PluginUIHandler, g *rbacGuards) {
+func RegisterPluginRoutes(
+	r *gin.RouterGroup, h *handler.PluginHandler, ui *handler.PluginUIHandler, hooks *handler.PluginWebhookHandler,
+	g *rbacGuards,
+) {
 	plugins := g.apiKeyGroup(r.Group("/plugins"), apiKeyFullAccess())
 	{
 		plugins.GET("", g.Viewer(), h.ListPlugins)
@@ -30,6 +33,10 @@ func RegisterPluginRoutes(r *gin.RouterGroup, h *handler.PluginHandler, ui *hand
 			ui.SetRoleGuard(func(need types.TenantRole) gin.HandlerFunc { return middleware.RequireRole(need, g.cfg) })
 			plugins.POST("/:id/ui-request", g.Viewer(), ui.Request)
 		}
+		// Webhook URLs carry their secret — Admin+.
+		if hooks != nil {
+			plugins.GET("/:id/webhooks", g.Admin(), hooks.List)
+		}
 	}
 }
 

internal/router/sync_task.go

@@ -233,8 +233,10 @@ type SyncTaskParams struct {
 	KnowledgeAutoTag     interfaces.TaskHandler `name:"knowledgeAutoTag"`
 	KnowledgeBaseProfile interfaces.TaskHandler `name:"knowledgeBaseProfile"`
 	WikiIngest           interfaces.TaskHandler `name:"wikiIngest"`
-	TemporaryDocument    interfaces.TemporaryDocumentService
-	MemoryService        interfaces.MemoryService
+	// PluginEvents delivers events to plugins (plugin:event).
+	PluginEvents      interfaces.TaskHandler `name:"pluginEvents"`
+	TemporaryDocument interfaces.TemporaryDocumentService
+	MemoryService     interfaces.MemoryService
 }
 
 // RegisterSyncHandlers registers all task handlers on the SyncTaskExecutor.
@@ -266,5 +268,6 @@ func RegisterSyncHandlers(params SyncTaskParams) {
 	params.Executor.RegisterHandler(types.TypeWikiIngest, params.WikiIngest.Handle)
 	params.Executor.RegisterHandler(types.TypeWikiFinalize, params.WikiIngest.Handle)
 	params.Executor.RegisterHandler(types.TypeMemoryExtract, params.MemoryService.Handle)
+	params.Executor.RegisterHandler(types.TypePluginEvent, params.PluginEvents.Handle)
 	logger.Infof(context.Background(), "[SyncTask] All task handlers registered (Lite mode, no Redis)")
 }

internal/router/task.go

@@ -45,10 +45,12 @@ type AsynqTaskParams struct {
 	KnowledgeAutoTag     interfaces.TaskHandler `name:"knowledgeAutoTag"`
 	KnowledgeBaseProfile interfaces.TaskHandler `name:"knowledgeBaseProfile"`
 	WikiIngest           interfaces.TaskHandler `name:"wikiIngest"`
-	TemporaryDocument    interfaces.TemporaryDocumentService
-	MemoryService        interfaces.MemoryService
-	DeadLetterRepo       interfaces.TaskDeadLetterRepository
-	SpanTracker          service.SpanTracker
+	// PluginEvents delivers events to plugins (plugin:event).
+	PluginEvents      interfaces.TaskHandler `name:"pluginEvents"`
+	TemporaryDocument interfaces.TemporaryDocumentService
+	MemoryService     interfaces.MemoryService
+	DeadLetterRepo    interfaces.TaskDeadLetterRepository
+	SpanTracker       service.SpanTracker
 }
 
 // defaultRedisOpTimeout is the previous hard-coded read timeout. The 100ms
@@ -263,6 +265,7 @@ func RunAsynqServer(params AsynqTaskParams) *asynq.ServeMux {
 
 	// Register extract handlers - router will dispatch to appropriate handler
 	mux.HandleFunc(types.TypeChunkExtract, params.ChunkExtractor.Handle)
+	mux.HandleFunc(types.TypePluginEvent, params.PluginEvents.Handle)
 	mux.HandleFunc(types.TypeDataTableSummary, params.DataTableSummary.Handle)
 
 	// Register document processing handler

internal/types/task.go

@@ -234,6 +234,7 @@ type WorkerServerStat struct {
 const (
 	TypeChunkExtract             = "chunk:extract"
 	TypeDocumentProcess          = "document:process"           // 文档处理任务
+	TypePluginEvent              = "plugin:event"               // 向插件投递业务事件
 	TypeFAQImport                = "faq:import"                 // FAQ导入任务（包含dry run模式）
 	TypeQuestionGeneration       = "question:generation"        // 问题生成任务
 	TypeSummaryGeneration        = "summary:generation"         // 摘要生成任务

pluginsdk/README.md

@@ -104,6 +104,53 @@ p.UI(func(ctx context.Context, call *pluginsdk.Call, req pluginapi.UIRequest) (*
 WeKnora refuses a caller below the page's `minRole` before the call.
 The handler can check `req.Role` for anything finer.
 
+## Events
+
+List the events a plugin wants in `permissions.events`. The administrator
+sees them at install.
+
+| Event | When | Data |
+| --- | --- | --- |
+| `knowledge.ingested` | A document finished processing | `KnowledgeEventData` |
+| `knowledge.failed` | A document failed for good | `KnowledgeEventData` with `error` |
+| `knowledge.deleted` | A document was deleted | `KnowledgeEventData` |
+| `chat.answered` | An answer completed | `ChatEventData`, including the question and answer |
+
+```go
+p.OnEvent(func(ctx context.Context, call *pluginsdk.Call, ev pluginapi.EventDelivery) error {
+	// ev.ID stays the same across retries: deduplicate on it.
+	return nil
+})
+```
+
+Delivery is at least once, in the background:
+- A workspace receives events only while it has the plugin switched on.
+- A retryable error (`unavailable`, `rate_limited`) or a timeout brings
+  the event back later with the same `ID` and a higher `Attempt`.
+- Any other error drops the event.
+- Ignore types you do not know: the list only grows.
+
+## Webhooks
+
+`contributes.webhooks` gives each workspace a secret URL per webhook,
+under `/api/v1/plugin-callbacks/`. Workspace admins copy it from the plugin
+center, and it is in `call.Context.Webhooks` when WeKnora knows its
+public address.
+
+```go
+p.Webhook("events", func(ctx context.Context, call *pluginsdk.Call, req pluginapi.WebhookRequest) (*pluginapi.WebhookResponse, error) {
+	// call is the workspace the URL belongs to; verify the sender with its configuration.
+	return &pluginapi.WebhookResponse{Status: 204}, nil
+})
+```
+
+WeKnora answers 404 unless all of these hold:
+- the URL is genuine;
+- the webhook exists;
+- the workspace has the plugin on.
+
+Bodies are limited to 1 MB, with 20 calls a second per URL.
+
 ## Calling back into WeKnora (Host API)
 
 A plugin that declares `permissions.hostApi` gets a short-lived token with
@@ -164,6 +211,7 @@ Complete plugins with their `package.sh`:
 - `examples/plugins/subtitles`: a parser using the Host API.
 - `examples/plugins/notebooks`: a parser written in Python.
 - `examples/plugins/links`: pages (toolbox, settings, knowledge base tab), in Python.
+- `examples/plugins/activity`: events, a webhook and a page, in Python.
 
 ## Testing
 

pluginsdk/conformance/conformance.go

@@ -149,6 +149,23 @@ func Run(ctx context.Context, t Target) Report {
 			return protocolAnswer(err)
 		})
 	}
+	if len(m.Contributes["events"]) > 0 {
+		check("events accepts a delivery", func(ctx context.Context) error {
+			// A type the plugin does not know: it must answer, not crash.
+			ev := pluginapi.EventDelivery{
+				ID: "conformance", Type: "weknora.conformance", OccurredAt: time.Now(), Attempt: 1,
+				Data: json.RawMessage(`{}`),
+			}
+			return protocolAnswer(t.Client.Call(ctx, pluginapi.EventsPath, envelope(), ev, nil))
+		})
+	}
+	for _, id := range m.Contributes["webhooks"] {
+		check("webhooks/"+id+" answers", func(ctx context.Context) error {
+			var out pluginapi.WebhookResponse
+			in := pluginapi.WebhookRequest{Method: http.MethodPost, Path: "/", Body: []byte(`{}`)}
+			return protocolAnswer(t.Client.Call(ctx, pluginapi.WebhookPath(id), envelope(), in, &out))
+		})
+	}
 	if len(m.Contributes["ui"]) > 0 {
 		check("ui/request answers", func(ctx context.Context) error {
 			var out pluginapi.UIResponse

pluginsdk/events.go

@@ -0,0 +1,69 @@
+package pluginsdk
+
+import (
+	"context"
+	"encoding/json"
+	"net/http"
+
+	"github.com/Tencent/WeKnora/pluginsdk/pluginapi"
+)
+
+// EventHandler handles one event delivery. Deliveries repeat (at least
+// once): make it idempotent on ev.ID. Return a retryable error (unavailable,
+// rate_limited) to be called again later; any other error drops the event.
+// Ignore event types you do not know.
+type EventHandler func(ctx context.Context, call *Call, ev pluginapi.EventDelivery) error
+
+// OnEvent registers the handler of the events the plugin subscribed to in
+// permissions.events.
+func (p *Plugin) OnEvent(h EventHandler) { p.events = h }
+
+// WebhookHandler answers one inbound call to a webhook.
+type WebhookHandler func(
+	ctx context.Context, call *Call, req pluginapi.WebhookRequest,
+) (*pluginapi.WebhookResponse, error)
+
+// Webhook registers webhook id (contributes.webhooks[].id). call carries the
+// workspace the URL belongs to and its configuration, so the handler can
+// verify the sender's signature with the workspace's secret.
+func (p *Plugin) Webhook(id string, h WebhookHandler) { p.webhooks[id] = h }
+
+func (p *Plugin) routeEvents(mux *http.ServeMux) {
+	mux.HandleFunc("POST /v1/events", func(w http.ResponseWriter, r *http.Request) {
+		if p.events == nil {
+			writeError(w, pluginapi.Errorf(pluginapi.CodeNotFound, "this plugin handles no events"))
+			return
+		}
+		p.unary(func(ctx context.Context, call *Call, raw json.RawMessage) (any, error) {
+			ev, err := decodeInput[pluginapi.EventDelivery](raw)
+			if err != nil {
+				return nil, err
+			}
+			return struct{}{}, p.events(ctx, call, ev)
+		})(w, r)
+	})
+	mux.HandleFunc("POST /v1/webhooks/{id}", func(w http.ResponseWriter, r *http.Request) {
+		h, ok := p.webhooks[r.PathValue("id")]
+		if !ok {
+			writeError(w, pluginapi.Errorf(pluginapi.CodeNotFound, "no webhook %q", r.PathValue("id")))
+			return
+		}
+		p.unary(func(ctx context.Context, call *Call, raw json.RawMessage) (any, error) {
+			in, err := decodeInput[pluginapi.WebhookRequest](raw)
+			if err != nil {
+				return nil, err
+			}
+			out, err := h(ctx, call, in)
+			if err != nil {
+				return nil, err
+			}
+			if out == nil {
+				out = &pluginapi.WebhookResponse{}
+			}
+			if out.Status == 0 {
+				out.Status = http.StatusOK
+			}
+			return out, nil
+		})(w, r)
+	})
+}

pluginsdk/openapi_test.go

@@ -25,7 +25,7 @@ func TestOpenAPIMatchesRoutes(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	for _, f := range []string{"websearch.go", "connector.go", "parser.go", "ui.go"} {
+	for _, f := range []string{"websearch.go", "connector.go", "parser.go", "ui.go", "events.go"} {
 		b, err := os.ReadFile(f)
 		if err != nil {
 			t.Fatal(err)

pluginsdk/plugin.go

@@ -64,6 +64,8 @@ type Plugin struct {
 	connectors map[string]Connector
 	parsers    map[string]Parser
 	ui         UIHandler
+	events     EventHandler
+	webhooks   map[string]WebhookHandler
 	validate   ConfigValidator
 	logger     *slog.Logger
 	// ShutdownTimeout bounds how long Serve waits for calls in flight after
@@ -78,6 +80,7 @@ func New(info Info) *Plugin {
 		webSearch:       map[string]WebSearcher{},
 		connectors:      map[string]Connector{},
 		parsers:         map[string]Parser{},
+		webhooks:        map[string]WebhookHandler{},
 		logger:          slog.New(slog.NewTextHandler(os.Stderr, nil)),
 		ShutdownTimeout: 60 * time.Second,
 	}
@@ -104,6 +107,10 @@ func (p *Plugin) Manifest() pluginapi.Manifest {
 	add("webSearch", keys(p.webSearch))
 	add("connectors", keys(p.connectors))
 	add("parsers", keys(p.parsers))
+	add("webhooks", keys(p.webhooks))
+	if p.events != nil {
+		m.Contributes["events"] = []string{"handler"}
+	}
 	if p.ui != nil {
 		m.Contributes["ui"] = []string{"request"}
 	}
@@ -143,6 +150,7 @@ func (p *Plugin) Handler() http.Handler {
 	p.routeConnectors(mux)
 	p.routeParsers(mux)
 	p.routeUI(mux)
+	p.routeEvents(mux)
 	mux.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
 		writeError(w, pluginapi.Errorf(pluginapi.CodeNotFound, "no endpoint %s %s", r.Method, r.URL.Path))
 	})

pluginsdk/plugin_test.go

@@ -292,3 +292,55 @@ func TestUIRequests(t *testing.T) {
 		t.Fatalf("status = %d, %v", out.Status, err)
 	}
 }
+
+func TestEventsAndWebhooks(t *testing.T) {
+	ctx := context.Background()
+	p := New(Info{ID: "acme.hooks", Version: "1.0.0"})
+	var seen []string
+	p.OnEvent(func(_ context.Context, call *Call, ev pluginapi.EventDelivery) error {
+		if ev.Type == pluginapi.EventKnowledgeFailed {
+			return pluginapi.Errorf(pluginapi.CodeUnavailable, "try later")
+		}
+		seen = append(seen, fmt.Sprintf("%s#%d@%d", ev.Type, ev.Attempt, call.TenantID))
+		return nil
+	})
+	p.Webhook("jira", func(
+		_ context.Context, _ *Call, req pluginapi.WebhookRequest,
+	) (*pluginapi.WebhookResponse, error) {
+		if req.Headers["X-Signature"] != "ok" {
+			return &pluginapi.WebhookResponse{Status: 401}, nil
+		}
+		body := append([]byte(req.Path+" "), req.Body...)
+		return &pluginapi.WebhookResponse{ContentType: "text/plain", Body: body}, nil
+	})
+	m := p.Manifest()
+	if len(m.Contributes["events"]) != 1 || m.Contributes["webhooks"][0] != "jira" {
+		t.Fatalf("manifest = %+v", m)
+	}
+	srv := httptest.NewServer(p.Handler())
+	defer srv.Close()
+	c := client.New(srv.URL, nil, nil)
+	env := pluginapi.Envelope{Context: pluginapi.Context{TenantID: 9}}
+
+	ev := pluginapi.EventDelivery{ID: "e1", Type: pluginapi.EventKnowledgeIngested, Attempt: 2}
+	if err := c.Call(ctx, pluginapi.EventsPath, env, ev, nil); err != nil || seen[0] != "knowledge.ingested#2@9" {
+		t.Fatalf("event = %v, %v", seen, err)
+	}
+	ev.Type = pluginapi.EventKnowledgeFailed
+	err := c.Call(ctx, pluginapi.EventsPath, env, ev, nil)
+	if pe, ok := pluginapi.AsError(err); !ok || !pe.Retryable {
+		t.Fatalf("retryable failure = %v", err)
+	}
+
+	var out pluginapi.WebhookResponse
+	req := pluginapi.WebhookRequest{
+		Method: "POST", Path: "/issue", Headers: map[string]string{"X-Signature": "ok"}, Body: []byte("hi"),
+	}
+	if err := c.Call(ctx, pluginapi.WebhookPath("jira"), env, req, &out); err != nil || out.Status != 200 ||
+		string(out.Body) != "/issue hi" || out.ContentType != "text/plain" {
+		t.Fatalf("webhook = %+v, %v", out, err)
+	}
+	if err := c.Call(ctx, pluginapi.WebhookPath("nope"), env, req, &out); !isCode(err, pluginapi.CodeNotFound) {
+		t.Fatalf("unknown webhook = %v", err)
+	}
+}

pluginsdk/pluginapi/envelope.go

@@ -26,6 +26,10 @@ type Context struct {
 	// Host lets the plugin call back into WeKnora (Host API); absent when
 	// the plugin was granted no scopes.
 	Host *HostAccess `json:"host,omitempty"`
+	// Webhooks are this workspace's URLs of the plugin's webhooks, by
+	// webhook ID, for registering them with a third party. Present only
+	// when WeKnora knows its public address.
+	Webhooks map[string]string `json:"webhooks,omitempty"`
 }
 
 // HostAccess is where and how a plugin calls the WeKnora Host API.

pluginsdk/pluginapi/events.go

@@ -0,0 +1,94 @@
+package pluginapi
+
+import (
+	"encoding/json"
+	"time"
+)
+
+// EventsPath receives the events a plugin subscribed to in
+// permissions.events. Delivery is at least once: a delivery that fails with
+// a retryable error (or times out) is sent again later with the same ID, so
+// handlers must be idempotent on ID. A non-retryable error drops it.
+const EventsPath = "/v1/events"
+
+// Event types a plugin can subscribe to. The set only grows.
+const (
+	// EventKnowledgeIngested: a document finished processing and is
+	// searchable. Data is KnowledgeEventData.
+	EventKnowledgeIngested = "knowledge.ingested"
+	// EventKnowledgeFailed: a document failed processing for good.
+	// Data is KnowledgeEventData with Error.
+	EventKnowledgeFailed = "knowledge.failed"
+	// EventKnowledgeDeleted: a document was deleted. Data is
+	// KnowledgeEventData.
+	EventKnowledgeDeleted = "knowledge.deleted"
+	// EventChatAnswered: an assistant answer was completed. Data is
+	// ChatEventData, question and answer included.
+	EventChatAnswered = "chat.answered"
+)
+
+// EventTypes lists every event type, for manifest validation.
+var EventTypes = []string{EventKnowledgeIngested, EventKnowledgeFailed, EventKnowledgeDeleted, EventChatAnswered}
+
+// EventDelivery is one event sent to a plugin. The tenant is the envelope's.
+type EventDelivery struct {
+	// ID identifies the event; retries keep it (idempotency key).
+	ID         string    `json:"id"`
+	Type       string    `json:"type"`
+	OccurredAt time.Time `json:"occurredAt"`
+	// Attempt counts deliveries of this event, from 1.
+	Attempt int             `json:"attempt"`
+	Data    json.RawMessage `json:"data,omitempty"`
+}
+
+// KnowledgeEventData describes a document of a knowledge base.
+type KnowledgeEventData struct {
+	KnowledgeBaseID string `json:"knowledgeBaseId"`
+	KnowledgeID     string `json:"knowledgeId"`
+	Title           string `json:"title,omitempty"`
+	FileName        string `json:"fileName,omitempty"`
+	FileType        string `json:"fileType,omitempty"`
+	// Source is the URL a URL or data-source document came from.
+	Source string `json:"source,omitempty"`
+	// Error is why processing failed (knowledge.failed).
+	Error string `json:"error,omitempty"`
+}
+
+// ChatEventData describes an answered question.
+type ChatEventData struct {
+	SessionID string `json:"sessionId"`
+	// MessageID is the assistant message.
+	MessageID string `json:"messageId"`
+	AgentID   string `json:"agentId,omitempty"`
+	UserID    string `json:"userId,omitempty"`
+	Question  string `json:"question,omitempty"`
+	Answer    string `json:"answer"`
+}
+
+// WebhookPath answers calls to webhook id (contributes.webhooks). Third
+// parties call WeKnora at a secret per-workspace URL; WeKnora checks the
+// URL, finds the workspace and relays the call here with the workspace's
+// envelope. Verifying the sender (a signature header) is the plugin's job.
+func WebhookPath(id string) string { return "/v1/webhooks/" + id }
+
+// WebhookRequest is one inbound call.
+type WebhookRequest struct {
+	Method string `json:"method"`
+	// Path is what followed the webhook URL, starting with "/" ("/" if
+	// nothing did).
+	Path  string `json:"path"`
+	Query string `json:"query,omitempty"`
+	// Headers are the request headers, one value each, cookies removed.
+	Headers map[string]string `json:"headers,omitempty"`
+	// Body is the raw body; JSON carries it as base64.
+	Body []byte `json:"body,omitempty"`
+}
+
+// WebhookResponse is what the caller receives.
+type WebhookResponse struct {
+	// Status defaults to 200.
+	Status int `json:"status,omitempty"`
+	// ContentType of Body, if any.
+	ContentType string `json:"contentType,omitempty"`
+	Body        []byte `json:"body,omitempty"`
+}

pluginsdk/pluginapi/openapi.yaml

@@ -218,6 +218,56 @@ paths:
                 required: [output]
                 properties: { output: { $ref: "#/components/schemas/UIResponse" } }
         default: { $ref: "#/components/responses/Error" }
+  /v1/events:
+    post:
+      summary: Receive an event the plugin subscribed to (permissions.events)
+      description: |
+        At least once: a retryable error or a timeout brings the same event
+        back later with the same id and a higher attempt, so handle it
+        idempotently. A non-retryable error drops it. Ignore unknown types.
+        The manifest lists "events": ["handler"] under contributes when the
+        plugin handles events.
+      requestBody:
+        required: true
+        content:
+          application/json:
+            schema:
+              allOf:
+                - $ref: "#/components/schemas/Envelope"
+                - properties: { input: { $ref: "#/components/schemas/EventDelivery" } }
+      responses:
+        "200":
+          description: Handled
+          content:
+            application/json:
+              schema: { type: object, required: [output], properties: { output: { type: object } } }
+        default: { $ref: "#/components/responses/Error" }
+  /v1/webhooks/{id}:
+    post:
+      summary: Answer an inbound call to webhook id (contributes.webhooks)
+      description: |
+        Third parties call a secret per-workspace URL of WeKnora; WeKnora
+        relays the call here with that workspace's envelope. Verify the
+        sender (e.g. a signature header) with the workspace configuration.
+      parameters: [{ $ref: "#/components/parameters/ContributionID" }]
+      requestBody:
+        required: true
+        content:
+          application/json:
+            schema:
+              allOf:
+                - $ref: "#/components/schemas/Envelope"
+                - properties: { input: { $ref: "#/components/schemas/WebhookRequest" } }
+      responses:
+        "200":
+          description: The answer to the caller
+          content:
+            application/json:
+              schema:
+                type: object
+                required: [output]
+                properties: { output: { $ref: "#/components/schemas/WebhookResponse" } }
+        default: { $ref: "#/components/responses/Error" }
 components:
   securitySchemes:
     hostToken: { type: http, scheme: bearer }
@@ -269,6 +319,10 @@ components:
               properties:
                 url: { type: string }
                 token: { type: string }
+            webhooks:
+              type: object
+              description: This workspace's webhook URLs by webhook ID, when WeKnora knows its public address
+              additionalProperties: { type: string }
         config:
           type: object
           description: Configuration scopes, secrets decrypted, never merged
@@ -375,6 +429,54 @@ components:
               mimeType: { type: string }
               data: { type: string, contentEncoding: base64 }
         metadata: { type: object, additionalProperties: { type: string } }
+    EventDelivery:
+      type: object
+      required: [id, type, occurredAt, attempt]
+      properties:
+        id: { type: string, description: Idempotency key; kept across retries }
+        type: { enum: [knowledge.ingested, knowledge.failed, knowledge.deleted, chat.answered] }
+        occurredAt: { type: string, format: date-time }
+        attempt: { type: integer, minimum: 1 }
+        data:
+          oneOf:
+            - $ref: "#/components/schemas/KnowledgeEventData"
+            - $ref: "#/components/schemas/ChatEventData"
+    KnowledgeEventData:
+      type: object
+      required: [knowledgeBaseId, knowledgeId]
+      properties:
+        knowledgeBaseId: { type: string }
+        knowledgeId: { type: string }
+        title: { type: string }
+        fileName: { type: string }
+        fileType: { type: string }
+        source: { type: string }
+        error: { type: string }
+    ChatEventData:
+      type: object
+      required: [sessionId, messageId, answer]
+      properties:
+        sessionId: { type: string }
+        messageId: { type: string }
+        agentId: { type: string }
+        userId: { type: string }
+        question: { type: string }
+        answer: { type: string }
+    WebhookRequest:
+      type: object
+      required: [method, path]
+      properties:
+        method: { type: string }
+        path: { type: string }
+        query: { type: string }
+        headers: { type: object, additionalProperties: { type: string } }
+        body: { type: string, contentEncoding: base64 }
+    WebhookResponse:
+      type: object
+      properties:
+        status: { type: integer }
+        contentType: { type: string }
+        body: { type: string, contentEncoding: base64 }
     UIRequest:
       type: object
       required: [mount, method, path]

pluginsdk/python/README.md

@@ -77,6 +77,18 @@ def ui(call, req: UIRequest):
 The pages themselves are HTML under `ui/` that use
 [`@weknora/plugin-ui`](../../packages/plugin-ui).
 
+**Events and webhooks.**
+
+```python
+@plugin.on_event
+def on_event(call, ev: EventDelivery):   # permissions.events; dedupe on ev.id
+    ...
+
+@plugin.webhook("notes")                 # contributes.webhooks
+def notes(call, req: WebhookRequest):    # req.headers, req.body, req.json()
+    return WebhookResponse(status=204)
+```
+
 ## Errors
 
 Raise `PluginError(ErrorCode.X, "message")` to choose what WeKnora sees:

pluginsdk/python/src/weknora_plugin/__init__.py

@@ -23,8 +23,13 @@ from .protocol import API_VERSION, PROTOCOL_VERSION, ErrorCode, PluginError, inv
 from .types import (
     FETCH_FULL,
     FETCH_INCREMENTAL,
+    EVENT_CHAT_ANSWERED,
+    EVENT_KNOWLEDGE_DELETED,
+    EVENT_KNOWLEDGE_FAILED,
+    EVENT_KNOWLEDGE_INGESTED,
     ConnectorConfig,
     Cursor,
+    EventDelivery,
     FetchedItem,
     FetchInput,
     KVEntry,
@@ -37,6 +42,8 @@ from .types import (
     SearchResult,
     UIRequest,
     UIResponse,
+    WebhookRequest,
+    WebhookResponse,
 )
 
 __version__ = "0.1.0"
@@ -49,6 +56,11 @@ __all__ = [
     "Call",
     "ConnectorConfig",
     "Cursor",
+    "EVENT_CHAT_ANSWERED",
+    "EVENT_KNOWLEDGE_DELETED",
+    "EVENT_KNOWLEDGE_FAILED",
+    "EVENT_KNOWLEDGE_INGESTED",
+    "EventDelivery",
     "ErrorCode",
     "FetchInput",
     "FetchedItem",
@@ -67,5 +79,7 @@ __all__ = [
     "StreamClosed",
     "UIRequest",
     "UIResponse",
+    "WebhookRequest",
+    "WebhookResponse",
     "invalid_config",
 ]

pluginsdk/python/src/weknora_plugin/plugin.py

@@ -31,8 +31,11 @@ from .types import (
     ParseOutput,
     SearchInput,
     SearchResult,
+    EventDelivery,
     UIRequest,
     UIResponse,
+    WebhookRequest,
+    WebhookResponse,
     from_wire,
     parse_time,
     to_wire,
@@ -53,6 +56,9 @@ class Call:
         #: When WeKnora stops waiting; give up by then.
         self.deadline: Optional[datetime] = parse_time(ctx.get("deadline"))
         self._host = ctx.get("host") or {}
+        #: This workspace's webhook URLs by webhook ID, when WeKnora knows
+        #: its public address.
+        self.webhooks: Dict[str, str] = ctx.get("webhooks") or {}
         #: Platform-wide configuration (config.system).
         self.system: Dict[str, Any] = cfg.get("system") or {}
         #: Workspace configuration (config.tenant).
@@ -159,6 +165,8 @@ WebSearchFunc = Callable[[Call, SearchInput], List[SearchResult]]
 ParserFunc = Callable[[Call, ParseInput], ParseOutput]
 ConfigValidator = Callable[[Call], None]
 UIHandler = Callable[[Call, UIRequest], Any]
+EventHandler = Callable[[Call, EventDelivery], None]
+WebhookHandler = Callable[[Call, WebhookRequest], Any]
 
 _ROUTE = re.compile(r"^/v1/(websearch|connectors|parsers)/([^/]+)/([a-z-]+)$")
 
@@ -176,6 +184,8 @@ class Plugin:
         self._parsers: Dict[str, ParserFunc] = {}
         self._validate: Optional[ConfigValidator] = None
         self._ui: Optional[UIHandler] = None
+        self._events: Optional[EventHandler] = None
+        self._webhooks: Dict[str, WebhookHandler] = {}
         #: Seconds serve() waits for calls in flight after SIGTERM.
         self.shutdown_timeout = 60.0
         if logger is None:
@@ -216,6 +226,21 @@ class Plugin:
 
         return register
 
+    def on_event(self, fn: EventHandler) -> EventHandler:
+        """Registers the handler of the events subscribed to in
+        permissions.events: fn(call, EventDelivery). Deliveries repeat, so
+        be idempotent on the event's id. Raise a retryable PluginError
+        (unavailable, rate_limited) to be called again later; any other
+        error drops the event. Ignore types you do not know."""
+        self._events = fn
+        return fn
+
+    def webhook(self, id: str, fn: Optional[WebhookHandler] = None) -> Any:
+        """Registers webhook id (contributes.webhooks): fn(call,
+        WebhookRequest) returns a WebhookResponse, or any JSON value for a
+        200. call is the workspace the URL belongs to."""
+        return self._register(self._webhooks, id, fn)
+
     def ui(self, fn: UIHandler) -> UIHandler:
         """Registers the handler behind the plugin's pages: fn(call,
         UIRequest) returns a UIResponse, or any JSON value for a 200."""
@@ -247,11 +272,14 @@ class Plugin:
             ("webSearch", self._web_search),
             ("connectors", self._connectors),
             ("parsers", self._parsers),
+            ("webhooks", self._webhooks),
         ):
             if table:
                 contributes[point] = sorted(table)
         if self._ui is not None:
             contributes["ui"] = ["request"]
+        if self._events is not None:
+            contributes["events"] = ["handler"]
         return {"id": self.id, "version": self.version, "apiVersion": p.API_VERSION, "contributes": contributes}
 
     # Dispatch.
@@ -272,6 +300,20 @@ class Plugin:
             else:
                 self._unary(h, body, lambda call, raw: _ui_output(self._ui(call, from_wire(UIRequest, raw))))
             return
+        if method == "POST" and path == "/v1/events":
+            if self._events is None:
+                h._send_error(PluginError(ErrorCode.NOT_FOUND, "this plugin handles no events"))
+            else:
+                events = self._events
+                self._unary(h, body, lambda call, raw: events(call, from_wire(EventDelivery, raw)), empty=True)
+            return
+        if method == "POST" and path.startswith("/v1/webhooks/"):
+            hook = self._webhooks.get(path[len("/v1/webhooks/"):])
+            if hook is None:
+                h._send_error(PluginError(ErrorCode.NOT_FOUND, f"no webhook {path.rsplit('/', 1)[-1]!r}"))
+            else:
+                self._unary(h, body, lambda call, raw: _webhook_output(hook(call, from_wire(WebhookRequest, raw))))
+            return
         m = _ROUTE.match(path)
         if method == "POST" and m:
             point, cid, action = m.groups()
@@ -432,6 +474,19 @@ def _connector_call(c: Any, action: str, call: Call, raw: Any) -> Any:
     return {"ancestors": list(resolve(call, cfg, list(raw.get("resourceIds") or [])) or [])}
 
 
+def _webhook_output(out: Any) -> Any:
+    if not isinstance(out, WebhookResponse):
+        import json as _json
+
+        out = WebhookResponse(content_type="application/json", body=_json.dumps(out).encode())
+    wire: dict = {"status": out.status or 200}
+    if out.content_type:
+        wire["contentType"] = out.content_type
+    if out.body:
+        wire["body"] = to_wire(out.body)
+    return wire
+
+
 def _ui_output(out: Any) -> Any:
     if not isinstance(out, UIResponse):
         out = UIResponse(body=out)

pluginsdk/python/src/weknora_plugin/types.py

@@ -263,3 +263,52 @@ class UIResponse:
 
     body: Any = None
     status: int = 200
+
+
+# Event types a plugin can subscribe to in permissions.events.
+EVENT_KNOWLEDGE_INGESTED = "knowledge.ingested"
+EVENT_KNOWLEDGE_FAILED = "knowledge.failed"
+EVENT_KNOWLEDGE_DELETED = "knowledge.deleted"
+EVENT_CHAT_ANSWERED = "chat.answered"
+
+
+@dataclass
+class EventDelivery:
+    """One event. Deliveries repeat (at least once): be idempotent on id.
+    data is the event's payload as a dict (see the protocol for each type)."""
+
+    id: str = ""
+    type: str = ""
+    occurred_at: Optional[datetime] = None
+    attempt: int = 1
+    data: Any = None
+
+
+@dataclass
+class WebhookRequest:
+    """An inbound call to one of the plugin's webhooks. body is the raw
+    bytes; headers carry one value each."""
+
+    method: str = ""
+    path: str = "/"
+    query: str = ""
+    headers: Dict[str, str] = field(default_factory=dict)
+    body: bytes = b""
+
+    def json(self) -> Any:
+        import json as _json
+
+        return _json.loads(self.body or b"null")
+
+
+@dataclass
+class WebhookResponse:
+    """What the caller receives. A str body is sent as UTF-8."""
+
+    status: int = 200
+    content_type: str = ""
+    body: Optional[Union[bytes, str]] = None
+
+    def __post_init__(self) -> None:
+        if isinstance(self.body, str):
+            self.body = self.body.encode()

pluginsdk/python/tests/fixture.py

@@ -20,9 +20,12 @@ from weknora_plugin import (  # noqa: E402
     Resource,
     SearchResult,
     UIResponse,
+    WebhookResponse,
     invalid_config,
 )
 
+seen_events = []
+
 plugin = Plugin("acme.fixture", "1.0.0")
 
 
@@ -75,6 +78,20 @@ class Notes:
         return Cursor(state={"after": start + 3})
 
 
+@plugin.on_event
+def on_event(call, ev):
+    if ev.type == "knowledge.failed":
+        raise PluginError(ErrorCode.UNAVAILABLE, "try later")
+    seen_events.append((ev.id, ev.type, ev.attempt, call.tenant_id, (ev.data or {}).get("knowledgeId")))
+
+
+@plugin.webhook("inbox")
+def inbox(call, req):
+    if req.headers.get("X-Signature") != "ok":
+        return WebhookResponse(status=401, body="bad signature")
+    return {"path": req.path, "got": req.json(), "tenant": call.tenant_id}
+
+
 @plugin.ui
 def ui(call, req):
     if req.path == "/missing":

pluginsdk/python/tests/test_plugin.py

@@ -80,7 +80,9 @@ class PluginTest(unittest.TestCase):
                     "webSearch": ["echo"],
                     "connectors": ["notes"],
                     "parsers": ["upper"],
+                    "webhooks": ["inbox"],
                     "ui": ["request"],
+                    "events": ["handler"],
                 },
             },
         )
@@ -122,6 +124,26 @@ class PluginTest(unittest.TestCase):
         self.assertEqual(out["images"], [{"originalRef": "img/dot.png", "data": base64.b64encode(b"\x89PNG").decode(), "mimeType": "image/png"}])
         self.assertEqual(out["metadata"], {"fileType": "txt"})
 
+    def test_events(self):
+        from fixture import seen_events
+
+        ev = {"id": "e1", "type": "knowledge.ingested", "occurredAt": "2026-09-26T10:51:18.5Z", "attempt": 2, "data": {"knowledgeId": "k1"}}
+        self.assertEqual(self.c.call("/v1/events", ev), (200, {"output": {}}))
+        self.assertIn(("e1", "knowledge.ingested", 2, 7, "k1"), seen_events)
+        status, body = self.c.call("/v1/events", dict(ev, type="knowledge.failed"))
+        self.assertEqual((status, body["error"]["retryable"]), (503, True))
+
+    def test_webhooks(self):
+        body = base64.b64encode(b'{"a": 1}').decode()
+        req = {"method": "POST", "path": "/issue", "headers": {"X-Signature": "ok"}, "body": body}
+        status, out = self.c.call("/v1/webhooks/inbox", req)
+        self.assertEqual(status, 200)
+        self.assertEqual(out["output"]["contentType"], "application/json")
+        self.assertEqual(json.loads(base64.b64decode(out["output"]["body"])), {"path": "/issue", "got": {"a": 1}, "tenant": 7})
+        status, out = self.c.call("/v1/webhooks/inbox", dict(req, headers={}))
+        self.assertEqual((out["output"]["status"], base64.b64decode(out["output"]["body"])), (401, b"bad signature"))
+        self.assertEqual(self.c.call("/v1/webhooks/nope", req)[0], 404)
+
     def test_ui_requests(self):
         req = {"mount": "pages/links", "method": "PUT", "path": "/links", "body": {"a": 1}, "role": "admin"}
         status, body = self.c.call("/v1/ui/request", req)

website-docs/03-features/25-plugins.md

@@ -104,6 +104,28 @@ Helm 设置 `pluginHost.enabled=true` 即可。使用本地存储（`STORAGE_TYP
 - **通信**：页面只能经 [`@weknora/plugin-ui`](https://github.com/Tencent/WeKnora/tree/main/packages/plugin-ui) 桥与 WeKnora 通信，由 WeKnora 代发请求给插件后端，或者弹提示、确认框、跳转页面。
 - **鉴权**：每次请求，WeKnora 都校验空间已启用该插件、用户满足页面的最低角色，并把用户角色一并交给插件后端。
 
+## 事件与 Webhook
+
+**事件**：插件在 `permissions.events` 中声明要订阅的事件，安装时由系统管理员审阅。
+
+| 事件 | 时机 |
+| --- | --- |
+| `knowledge.ingested` | 文档处理完成、可被检索 |
+| `knowledge.failed` | 文档处理最终失败 |
+| `knowledge.deleted` | 文档被删除 |
+| `chat.answered` | 一次回答完成，含问题与回答内容 |
+
+- 只有启用了该插件的空间才会向它投递事件。
+- 事件经后台任务队列（有 Redis 时为 asynq）异步投递，至少一次。
+- 插件返回可重试错误时，同一事件会以相同 ID 重新投递，最多 10 次。
+
+**Webhook**：插件在 `contributes.webhooks` 中声明入站地址。
+
+- 每个空间得到各自的秘密地址 `/api/v1/plugin-callbacks/...`，空间管理员可在「设置 → 插件」的「配置」中复制。
+- 第三方系统调用该地址时，WeKnora 先校验地址、确认空间已启用插件，再把请求转给插件。
+- 插件自行用空间配置里的密钥校验调用方。
+- 设置 `APP_EXTERNAL_URL` 后，插件还能拿到完整地址，自动向第三方注册。
+
 ## 开发插件
 
 - **Go**：[pluginsdk](https://github.com/Tencent/WeKnora/tree/main/pluginsdk)，含协议定义、SDK、客户端和一致性测试工具 `weknora-plugin-conformance`。
@@ -112,6 +134,7 @@ Helm 设置 `pluginHost.enabled=true` 即可。使用本地存储（`STORAGE_TYP
   - `rss`：数据源连接器，Go；
   - `subtitles`：文档解析器，Go，使用 Host API；
   - `notebooks`：Jupyter 笔记本解析器，Python；
-  - `links`：带三种页面的团队链接插件，Python。
+  - `links`：带三种页面的团队链接插件，Python；
+  - `activity`：订阅事件、接收 Webhook 的空间动态插件，Python。
 
 同一个插件既可以打包成 `host` 插件由 WeKnora 运行，也可以作为 `remote` 服务独立部署，代码不用改。