feat(plugin): Host API with a key-value store, and document parsers from plugins (#3721)

* feat(plugin): Host API with plugin tokens and a key-value store
* feat(plugin): document parsers from code plugins
* docs(pluginsdk): parsers and the Host API
This commit is contained in:
lyingbug
2026-09-26 14:45:39 +08:00
committed by GitHub
parent 4328ab26a9
commit ff49f9491d
44 changed files with 1950 additions and 23 deletions
+130
View File
@@ -0,0 +1,130 @@
// Command subtitles is an example parser plugin: it turns SRT and WebVTT
// subtitle files into a Markdown transcript, one paragraph per cue with its
// start time. It also keeps a per-workspace count of parsed files through
// the Host API key-value store, to show how a plugin keeps state without a
// database.
//
// Build a package with ./package.sh.
package main
import (
"bufio"
"bytes"
"context"
"fmt"
"log"
"regexp"
"strings"
"time"
"github.com/Tencent/WeKnora/pluginsdk"
"github.com/Tencent/WeKnora/pluginsdk/pluginapi"
)
// Version must match plugin.yaml.
const Version = "1.0.0"
func main() {
p := pluginsdk.New(pluginsdk.Info{ID: "weknora-examples.subtitles", Version: Version})
p.Parser("subtitles", pluginsdk.ParserFunc(parse))
if err := p.Serve(); err != nil {
log.Fatal(err)
}
}
// timing matches a cue timing line: 00:01:02,500 --> 00:01:04,000 (SRT) or
// 01:02.500 --> 01:04.000 (WebVTT, hours optional).
var timing = regexp.MustCompile(`^\s*((?:\d+:)?\d{2}:\d{2}[.,]\d{3})\s*-->\s*`)
// tags strips WebVTT / SRT inline markup such as <i> or <v Speaker>.
var tags = regexp.MustCompile(`<[^>]*>`)
type cue struct {
start string
text []string
}
func parse(ctx context.Context, call *pluginsdk.Call, in pluginapi.ParseInput) (*pluginapi.ParseOutput, error) {
if len(in.Content) == 0 {
return nil, pluginapi.Errorf(pluginapi.CodeInvalidConfig, "the subtitle file is empty")
}
cues := parseCues(in.Content)
if len(cues) == 0 {
return nil, pluginapi.Errorf(
pluginapi.CodeInvalidConfig,
"no subtitle cues found; is this an .srt or .vtt file?",
)
}
title := strings.TrimSpace(in.Title)
if title == "" {
title = strings.TrimSuffix(in.FileName, "."+in.FileType)
}
var md strings.Builder
fmt.Fprintf(&md, "# %s\n\n", title)
for _, c := range cues {
fmt.Fprintf(&md, "**[%s]** %s\n\n", shortTime(c.start), strings.Join(c.text, " "))
}
countParsed(ctx, call)
return &pluginapi.ParseOutput{
Markdown: strings.TrimSpace(md.String()) + "\n",
Metadata: map[string]string{"cues": fmt.Sprint(len(cues))},
}, nil
}
func parseCues(data []byte) []cue {
data = bytes.TrimPrefix(data, []byte("\xef\xbb\xbf"))
var out []cue
var cur *cue
sc := bufio.NewScanner(bytes.NewReader(data))
sc.Buffer(make([]byte, 0, 64<<10), 1<<20)
for sc.Scan() {
line := strings.TrimRight(sc.Text(), "\r")
if m := timing.FindStringSubmatch(line); m != nil {
out = append(out, cue{start: m[1]})
cur = &out[len(out)-1]
continue
}
if strings.TrimSpace(line) == "" {
cur = nil
continue
}
if cur == nil {
continue // cue numbers, WEBVTT header, NOTE blocks
}
if text := strings.TrimSpace(tags.ReplaceAllString(line, "")); text != "" {
cur.text = append(cur.text, text)
}
}
kept := out[:0]
for _, c := range out {
if len(c.text) > 0 {
kept = append(kept, c)
}
}
return kept
}
// shortTime drops milliseconds and a zero hour: 00:01:02,500 → 01:02.
func shortTime(t string) string {
t = strings.NewReplacer(",", ".").Replace(t)
if i := strings.IndexByte(t, '.'); i >= 0 {
t = t[:i]
}
return strings.TrimPrefix(t, "00:")
}
// countParsed bumps the workspace's parsed-file counter. It is best effort:
// a missing grant or an unreachable host must not fail the parse.
func countParsed(ctx context.Context, call *pluginsdk.Call) {
host := call.Host()
if host == nil {
return
}
ctx, cancel := context.WithTimeout(ctx, 5*time.Second)
defer cancel()
var n int
if _, err := host.KVGet(ctx, "stats/parsed", &n); err != nil {
return
}
_ = host.KVPut(ctx, "stats/parsed", n+1, 0)
}
+146
View File
@@ -0,0 +1,146 @@
package main
import (
"bytes"
"context"
"go/parser"
"go/token"
"net/http"
"net/http/httptest"
"os"
"path/filepath"
"strings"
"testing"
"github.com/gin-gonic/gin"
"github.com/google/uuid"
"gorm.io/driver/sqlite"
"gorm.io/gorm"
"github.com/Tencent/WeKnora/internal/application/repository"
"github.com/Tencent/WeKnora/internal/plugin/hostapi"
"github.com/Tencent/WeKnora/internal/types"
"github.com/Tencent/WeKnora/pluginsdk"
"github.com/Tencent/WeKnora/pluginsdk/client"
"github.com/Tencent/WeKnora/pluginsdk/conformance"
"github.com/Tencent/WeKnora/pluginsdk/pluginapi"
)
const srt = "\ufeff1\r\n00:00:01,000 --> 00:00:03,500\r\nHello <i>world</i>.\r\n\r\n" +
"2\r\n00:01:02,250 --> 00:01:04,000\r\nSecond line\r\ncontinues here.\r\n"
const vtt = `WEBVTT
NOTE a comment
01:02.500 --> 01:04.000
<v Ada>Plugins run out of process.
1:00:00.000 --> 1:00:02.000 align:start
Bye.
`
func TestParseSRTAndVTT(t *testing.T) {
ctx := context.Background()
out, err := parse(
ctx,
&pluginsdk.Call{},
pluginapi.ParseInput{FileName: "talk.srt", FileType: "srt", Content: []byte(srt)},
)
if err != nil {
t.Fatal(err)
}
want := "# talk\n\n**[00:01]** Hello world.\n\n**[01:02]** Second line continues here.\n"
if out.Markdown != want || out.Metadata["cues"] != "2" {
t.Fatalf("srt markdown = %q", out.Markdown)
}
out, err = parse(
ctx,
&pluginsdk.Call{},
pluginapi.ParseInput{FileName: "t.vtt", FileType: "vtt", Title: "Talk", Content: []byte(vtt)},
)
if err != nil {
t.Fatal(err)
}
if !strings.Contains(out.Markdown, "# Talk") ||
!strings.Contains(out.Markdown, "**[01:02]** Plugins run out of process.") ||
!strings.Contains(out.Markdown, "**[1:00:00]** Bye.") {
t.Fatalf("vtt markdown = %q", out.Markdown)
}
for _, bad := range []string{"", "just some text\nwith no cues\n"} {
_, err := parse(
ctx,
&pluginsdk.Call{},
pluginapi.ParseInput{FileName: "x.srt", FileType: "srt", Content: []byte(bad)},
)
if e, ok := pluginapi.AsError(err); !ok || e.Code != pluginapi.CodeInvalidConfig || e.Retryable {
t.Errorf("%q: want a permanent invalid_config error, got %v", bad, err)
}
}
}
func TestParseCountsThroughTheHostAPI(t *testing.T) {
gin.SetMode(gin.TestMode)
db, err := gorm.Open(sqlite.Open("file:"+uuid.NewString()+"?mode=memory&cache=shared"), &gorm.Config{})
if err != nil {
t.Fatal(err)
}
if err := db.AutoMigrate(&types.PluginKV{}); err != nil {
t.Fatal(err)
}
iss := hostapi.NewIssuer([]byte("k"))
r := gin.New()
hostapi.NewHandler(iss, hostapi.NewKV(repository.NewPluginKVRepository(db))).Register(r)
srv := httptest.NewServer(r)
defer srv.Close()
tok, _, _ := iss.Issue("weknora-examples.subtitles", Version, 3, []string{"kv"})
call := &pluginsdk.Call{
Context: pluginapi.Context{TenantID: 3, Host: &pluginapi.HostAccess{URL: srv.URL, Token: tok}},
}
in := pluginapi.ParseInput{FileName: "a.srt", FileType: "srt", Content: []byte(srt)}
for range 2 {
if _, err := parse(context.Background(), call, in); err != nil {
t.Fatal(err)
}
}
var n int
if ok, err := call.Host().KVGet(context.Background(), "stats/parsed", &n); !ok || err != nil || n != 2 {
t.Fatalf("parsed count = %d (%v, %v)", n, ok, err)
}
}
func TestConformsAndStaysExternal(t *testing.T) {
p := pluginsdk.New(pluginsdk.Info{ID: "weknora-examples.subtitles", Version: Version})
p.Parser("subtitles", pluginsdk.ParserFunc(parse))
srv := httptest.NewServer(p.Handler())
defer srv.Close()
rep := conformance.Run(context.Background(), conformance.Target{
Client: client.New(srv.URL, nil, nil),
Raw: func(ctx context.Context, path string, body []byte) (*http.Response, error) {
req, _ := http.NewRequestWithContext(ctx, http.MethodPost, srv.URL+path, bytes.NewReader(body))
return http.DefaultClient.Do(req)
},
})
for _, res := range rep.Results {
if !res.Passed {
t.Errorf("%s: %s", res.Name, res.Detail)
}
}
files, _ := filepath.Glob("*.go")
for _, f := range files {
if strings.HasSuffix(f, "_test.go") {
continue
}
src, _ := os.ReadFile(f)
parsed, err := parser.ParseFile(token.NewFileSet(), f, src, parser.ImportsOnly)
if err != nil {
t.Fatal(err)
}
for _, imp := range parsed.Imports {
if strings.Contains(imp.Path.Value, "WeKnora/internal") {
t.Errorf("%s imports %s", f, imp.Path.Value)
}
}
}
}
+20
View File
@@ -0,0 +1,20 @@
#!/usr/bin/env bash
# Builds the subtitles example plugin for the platforms WeKnora runs on and zips it
# into a .wkp package: plugin.yaml and bin/<os>-<arch>/subtitles.
set -euo pipefail
cd "$(dirname "$0")"
version=$(sed -n 's/^version: //p' plugin.yaml)
out=${1:-"weknora-examples-subtitles-${version}.wkp"}
stage=$(mktemp -d)
trap 'rm -rf "$stage"' EXIT
cp plugin.yaml "$stage"/
for target in linux/amd64 linux/arm64 darwin/amd64 darwin/arm64 windows/amd64; do
os=${target%/*}
arch=${target#*/}
bin="$stage/bin/${os}-${arch}/subtitles"
[ "$os" = windows ] && bin="$bin.exe"
CGO_ENABLED=0 GOOS=$os GOARCH=$arch go build -trimpath -ldflags="-s -w" -o "$bin" .
done
rm -f "$out"
(cd "$stage" && zip -qr - .) > "$out"
echo "wrote $out"
+23
View File
@@ -0,0 +1,23 @@
schemaVersion: 1
id: weknora-examples.subtitles
version: 1.0.0
apiVersion: weknora.plugin/v1
name: { en-US: Subtitles, zh-CN: 字幕解析 }
description:
en-US: Parse SRT and WebVTT subtitles into a searchable transcript.
zh-CN: 把 SRT、WebVTT 字幕解析为可检索的文字稿。
publisher: { id: weknora-examples, name: WeKnora examples }
homepage: https://github.com/Tencent/WeKnora/tree/main/examples/plugins/subtitles
license: MIT
runtime: { type: host, kind: binary, entry: "bin/{os}-{arch}/subtitles" }
permissions:
# Counts parsed files per workspace in the Host API key-value store.
hostApi: [kv]
contributes:
parsers:
- id: subtitles
name: { en-US: Subtitles, zh-CN: 字幕解析 }
description:
en-US: SRT and WebVTT subtitles as a transcript with cue times
zh-CN: 带时间点的字幕文字稿
fileTypes: [srt, vtt]
+4
View File
@@ -107,6 +107,10 @@ export interface ParserEngineInfo {
FileTypes: string[]
Available?: boolean
UnavailableReason?: string
/** Set on engines an installed plugin provides (Name is then acme.ocr/ocr). */
PluginID?: string
/** Localized names of plugin engines, keyed by locale; "default" as fallback. */
DisplayNames?: Record<string, string>
}
/** 解析引擎配置(引擎连接参数存空间;聊天附件解析策略在智能体中配置) */
@@ -66,6 +66,7 @@
</template>
<script setup lang="ts">
import { pickLocale } from '@/utils/localizedText'
import { ref, watch, computed, onMounted, onUnmounted } from 'vue'
import { useI18n } from 'vue-i18n'
import { type ParserEngineInfo } from '@/api/system'
@@ -73,10 +74,13 @@ import { useEditorResourcesStore } from '@/stores/editorResources'
import { useUIStore } from '@/stores/ui'
import { storeToRefs } from 'pinia'
const { t } = useI18n()
const { t, locale } = useI18n()
const editorResources = useEditorResourcesStore()
function getEngineDisplayName(engineName: string): string {
// Plugin engines name themselves.
const names = parserEngines.value.find((e) => e.Name === engineName)?.DisplayNames
if (names) return pickLocale(names, locale.value) || names.default || engineName
const key = `kbSettings.parser.engines.${engineName}.name`
const translated = t(key)
return translated !== key ? translated : engineName
@@ -409,6 +409,7 @@
</template>
<script setup lang="ts">
import { pickLocale } from '@/utils/localizedText'
import { ref, computed, onMounted, nextTick } from 'vue'
import { useI18n } from 'vue-i18n'
import { useUIStore } from '@/stores/ui'
@@ -425,7 +426,7 @@ import {
} from '@/api/system'
import { getWeKnoraCloudStatus } from '@/api/model'
const { t } = useI18n()
const { t, locale } = useI18n()
const uiStore = useUIStore()
const authStore = useAuthStore()
@@ -546,6 +547,9 @@ function engineInitial(engineName: string): string {
}
function getEngineDisplayName(engineName: string): string {
// Plugin engines name themselves.
const names = engines.value.find((e) => e.Name === engineName)?.DisplayNames
if (names) return pickLocale(names, locale.value) || names.default || engineName
const key = `kbSettings.parser.engines.${engineName}.name`
const translated = t(key)
return translated !== key ? translated : engineName
@@ -0,0 +1,90 @@
package repository
import (
"context"
"strings"
"time"
"gorm.io/gorm"
"gorm.io/gorm/clause"
"github.com/Tencent/WeKnora/internal/types"
"github.com/Tencent/WeKnora/internal/types/interfaces"
)
// pluginKVRepository stores plugins' Host API key-value data (plugin_kv,
// migration 000117).
type pluginKVRepository struct {
db *gorm.DB
}
// NewPluginKVRepository wires the repository into the container.
func NewPluginKVRepository(db *gorm.DB) interfaces.PluginKVRepository {
return &pluginKVRepository{db: db}
}
// live restricts a query to entries that have not expired.
func live(q *gorm.DB, now time.Time) *gorm.DB {
return q.Where("expires_at IS NULL OR expires_at > ?", now)
}
func (r *pluginKVRepository) scope(ctx context.Context, pluginID string, tenantID uint64) *gorm.DB {
return r.db.WithContext(ctx).Model(&types.PluginKV{}).
Where("plugin_id = ? AND tenant_id = ?", pluginID, tenantID)
}
func (r *pluginKVRepository) Get(
ctx context.Context, pluginID string, tenantID uint64, key string,
) (*types.PluginKV, error) {
var rows []types.PluginKV
err := live(r.scope(ctx, pluginID, tenantID).Where("key = ?", key), time.Now()).Limit(1).Find(&rows).Error
if err != nil || len(rows) == 0 {
return nil, err
}
return &rows[0], nil
}
func (r *pluginKVRepository) Put(ctx context.Context, e *types.PluginKV) error {
return r.db.WithContext(ctx).Clauses(clause.OnConflict{
Columns: []clause.Column{{Name: "plugin_id"}, {Name: "tenant_id"}, {Name: "key"}},
DoUpdates: clause.AssignmentColumns([]string{"value", "expires_at", "updated_at"}),
}).Create(e).Error
}
func (r *pluginKVRepository) Delete(ctx context.Context, pluginID string, tenantID uint64, key string) (bool, error) {
res := r.db.WithContext(ctx).
Where("plugin_id = ? AND tenant_id = ? AND key = ?", pluginID, tenantID, key).
Delete(&types.PluginKV{})
return res.RowsAffected > 0, res.Error
}
// escapeLike makes a prefix match literally in LIKE.
func escapeLike(s string) string {
return strings.NewReplacer(`\`, `\\`, `%`, `\%`, `_`, `\_`).Replace(s)
}
func (r *pluginKVRepository) List(
ctx context.Context, pluginID string, tenantID uint64, prefix, after string, limit int,
) ([]types.PluginKV, error) {
q := live(r.scope(ctx, pluginID, tenantID), time.Now())
if prefix != "" {
q = q.Where(`key LIKE ? ESCAPE '\'`, escapeLike(prefix)+"%")
}
if after != "" {
q = q.Where("key > ?", after)
}
var rows []types.PluginKV
err := q.Order("key ASC").Limit(limit).Find(&rows).Error
return rows, err
}
func (r *pluginKVRepository) Count(ctx context.Context, pluginID string, tenantID uint64) (int64, error) {
var n int64
err := live(r.scope(ctx, pluginID, tenantID), time.Now()).Count(&n).Error
return n, err
}
func (r *pluginKVRepository) DeleteExpired(ctx context.Context, now time.Time) (int64, error) {
res := r.db.WithContext(ctx).Where("expires_at IS NOT NULL AND expires_at <= ?", now).Delete(&types.PluginKV{})
return res.RowsAffected, res.Error
}
@@ -93,3 +93,52 @@ func TestPluginTenantSettingUpsertColumns(t *testing.T) {
require.False(t, got.Enabled)
require.JSONEq(t, `{"region":"eu"}`, string(got.Config), "flipping the switch must keep the config")
}
func TestPluginKVRepository(t *testing.T) {
db, err := gorm.Open(sqlite.Open("file:"+uuid.NewString()+"?mode=memory&cache=shared"), &gorm.Config{})
require.NoError(t, err)
require.NoError(t, db.AutoMigrate(&types.PluginKV{}))
repo := NewPluginKVRepository(db)
ctx := context.Background()
put := func(tenant uint64, key, value string, expires *time.Time) {
t.Helper()
require.NoError(t, repo.Put(ctx, &types.PluginKV{
PluginID: "acme.x", TenantID: tenant, Key: key, Value: types.JSON(value),
ExpiresAt: expires, UpdatedAt: time.Now(),
}))
}
past := time.Now().Add(-time.Minute)
put(1, "cursor:a", `1`, nil)
put(1, "cursor:b", `2`, nil)
put(1, "cursor_x", `3`, nil) // "_" must not act as a LIKE wildcard
put(1, "old", `4`, &past)
put(2, "cursor:a", `99`, nil)
put(1, "cursor:a", `10`, nil) // upsert
got, err := repo.Get(ctx, "acme.x", 1, "cursor:a")
require.NoError(t, err)
require.JSONEq(t, `10`, string(got.Value))
got, err = repo.Get(ctx, "acme.x", 1, "old")
require.NoError(t, err)
require.Nil(t, got, "expired entries read as missing")
list, err := repo.List(ctx, "acme.x", 1, "cursor:", "", 10)
require.NoError(t, err)
require.Len(t, list, 2)
list, err = repo.List(ctx, "acme.x", 1, "cursor:", "cursor:a", 10)
require.NoError(t, err)
require.Len(t, list, 1)
require.Equal(t, "cursor:b", list[0].Key)
n, err := repo.Count(ctx, "acme.x", 1)
require.NoError(t, err)
require.EqualValues(t, 3, n)
removed, err := repo.DeleteExpired(ctx, time.Now())
require.NoError(t, err)
require.EqualValues(t, 1, removed)
ok, err := repo.Delete(ctx, "acme.x", 1, "cursor:b")
require.NoError(t, err)
require.True(t, ok)
got, _ = repo.Get(ctx, "acme.x", 2, "cursor:a")
require.JSONEq(t, `99`, string(got.Value), "tenants are separate partitions")
}
@@ -10,6 +10,7 @@ import (
"mime/multipart"
"net/http"
"os"
"slices"
"strings"
"time"
@@ -56,8 +57,11 @@ func isSupportedImportExtension(ext string) bool {
if ext == "" || ext == unknownFileType {
return false
}
_, ok := supportedImportFileExtensions[ext]
return ok
if _, ok := supportedImportFileExtensions[ext]; ok {
return true
}
// Installed plugins' parser engines bring their own file types.
return slices.Contains(docparser.PluginFileTypes(), ext)
}
// isValidFileType checks if a filename's extension is supported for import.
+6 -1
View File
@@ -169,8 +169,11 @@ func BuildContainer(container *dig.Container) *dig.Container {
must(container.Provide(activate.NewModelVendors))
must(container.Provide(pluginhost.NewManager))
must(container.Provide(newPluginInvoker))
must(container.Provide(repository.NewPluginKVRepository))
must(container.Provide(newPluginHostAPI))
must(container.Provide(activate.NewWebSearch))
must(container.Provide(activate.NewConnectors))
must(container.Provide(activate.NewParsers))
must(container.Provide(newMCPServiceRepository))
must(container.Provide(repository.NewMCPToolApprovalRepository))
must(container.Provide(repository.NewMCPOAuthRepository))
@@ -1854,8 +1857,10 @@ func installPluginGate(
webSearch *handler.WebSearchProviderHandler,
dataSources *handler.DataSourceHandler,
imHandler *handler.IMHandler,
system *handler.SystemHandler,
) {
for _, h := range []interface{ SetPluginGate(interfaces.PluginGate) }{models, webSearch, dataSources, imHandler} {
gated := []interface{ SetPluginGate(interfaces.PluginGate) }{models, webSearch, dataSources, imHandler, system}
for _, h := range gated {
h.SetPluginGate(gate)
}
}
+29 -1
View File
@@ -2,6 +2,10 @@ package container
import (
"context"
"fmt"
"os"
"strings"
"time"
"github.com/redis/go-redis/v9"
"go.uber.org/dig"
@@ -13,6 +17,7 @@ import (
"github.com/Tencent/WeKnora/internal/handler"
"github.com/Tencent/WeKnora/internal/plugin/activate"
"github.com/Tencent/WeKnora/internal/plugin/host"
"github.com/Tencent/WeKnora/internal/plugin/hostapi"
"github.com/Tencent/WeKnora/internal/plugin/install"
"github.com/Tencent/WeKnora/internal/plugin/reconcile"
pluginregistry "github.com/Tencent/WeKnora/internal/plugin/registry"
@@ -38,6 +43,7 @@ type pluginActivators struct {
Host *host.Manager
WebSearch *activate.WebSearch
Connectors *activate.Connectors
Parsers *activate.Parsers
MCP *activate.MCPServers
Skills *activate.Skills
Vendors *activate.ModelVendors
@@ -47,7 +53,29 @@ type pluginActivators struct {
// list orders the activators: the host first, so a code plugin's process is
// running before anything routes calls to it.
func (a pluginActivators) list() []reconcile.Activator {
return []reconcile.Activator{a.Host, a.WebSearch, a.Connectors, a.Vendors, a.MCP, a.Skills}
return []reconcile.Activator{a.Host, a.WebSearch, a.Connectors, a.Parsers, a.Vendors, a.MCP, a.Skills}
}
// newPluginHostAPI serves the Host API and gives calls a way back to it: the
// embedded host's plugins reach this node on loopback.
func newPluginHostAPI(
cfg *config.Config, iv *activate.Invoker, repo interfaces.PluginKVRepository, cleaner interfaces.ResourceCleaner,
) *hostapi.Handler {
issuer := hostapi.NewIssuerFromEnv()
url := strings.TrimSpace(os.Getenv("WEKNORA_PLUGIN_HOST_API_URL"))
if url == "" {
port := 8080
if cfg != nil && cfg.Server != nil && cfg.Server.Port > 0 {
port = cfg.Server.Port
}
url = fmt.Sprintf("http://127.0.0.1:%d", port)
}
iv.SetHostAPI(issuer, url)
kv := hostapi.NewKV(repo)
ctx, cancel := context.WithCancel(context.Background())
kv.StartSweeper(ctx, 10*time.Minute)
cleaner.RegisterWithName("PluginKVSweeper", func() error { cancel(); return nil })
return hostapi.NewHandler(issuer, kv)
}
// bindPluginActivators hands the activators what they need once the plugin
@@ -38,6 +38,7 @@ var versionedSQLiteTables = []string{
"plugin_tenant_settings",
"plugins",
"plugin_versions",
"plugin_kv",
}
// versionedSQLiteColumns maps each existing table to the columns that the
@@ -72,7 +73,7 @@ var versionedSQLiteColumns = map[string][]string{
}, // 000028
}
const expectedSQLiteMigrationVersion = 35
const expectedSQLiteMigrationVersion = 36
func TestSQLiteMigrationsCreateVersionedSchema(t *testing.T) {
repoRoot := sqliteRepoRoot(t)
+14 -1
View File
@@ -24,6 +24,7 @@ import (
"github.com/Tencent/WeKnora/internal/infrastructure/docparser"
"github.com/Tencent/WeKnora/internal/logger"
modellimiter "github.com/Tencent/WeKnora/internal/models/limiter"
"github.com/Tencent/WeKnora/internal/plugin/manifest"
"github.com/Tencent/WeKnora/internal/runtime"
"github.com/Tencent/WeKnora/internal/types"
"github.com/Tencent/WeKnora/internal/types/interfaces"
@@ -38,6 +39,7 @@ type runtimeKnowledgeCanceller interface {
// SystemHandler handles system-related requests
type SystemHandler struct {
pluginGated
cfg *config.Config
neo4jDriver neo4j.Driver
documentReader interfaces.DocumentReader
@@ -420,7 +422,18 @@ func (h *SystemHandler) ListParserEngines(c *gin.Context) {
connected := reader != nil && reader.IsConnected()
remoteEngines := h.fetchRemoteEngines(c.Request.Context(), reader, overrides)
engines := docparser.ListAllEngines(connected, overrides, remoteEngines)
c.JSON(200, gin.H{"code": 0, "msg": "success", "data": engines, "docreader_addr": docreaderAddr, "docreader_transport": docreaderTransport, "connected": connected})
// Engines of plugins the workspace has turned off leave the list.
enabled := h.pluginFilter(c)
visible := engines[:0]
for _, e := range engines {
if enabled(manifest.PointParsers, e.Name) {
visible = append(visible, e)
}
}
c.JSON(200, gin.H{
"code": 0, "msg": "success", "data": visible,
"docreader_addr": docreaderAddr, "docreader_transport": docreaderTransport, "connected": connected,
})
}
// ReconnectDocReader reconnects the document converter to a new (or same) DocReader address.
@@ -3,6 +3,9 @@ package docparser
import (
"context"
"fmt"
"sort"
"strings"
"sync"
"github.com/Tencent/WeKnora/internal/types"
"github.com/Tencent/WeKnora/internal/types/interfaces"
@@ -41,32 +44,102 @@ type ReaderDeps struct {
WeKnoraCloudCredentials func(ctx context.Context) *types.WeKnoraCloudCredentials
}
// enginesMu guards the registries: plugin engines come and go while the
// server runs.
var enginesMu sync.RWMutex
// localEngines holds all locally registered parser engines, in registration
// order — which is also the order the engine list is shown in.
var localEngines []EngineRegistration
// pluginEngines holds engines installed plugins provide, by qualified name
// ("acme.ocr/ocr"). They are kept apart from localEngines so the builtin
// plugin catalog never describes them as builtins.
var pluginEngines = map[string]EngineRegistration{}
// PluginEngineInfo is what a plugin engine adds to the engine list.
type PluginEngineInfo interface {
PluginID() string
// DisplayNames localizes the engine's name, keyed by locale.
DisplayNames() map[string]string
}
// RegisterEngine adds an engine to the local registry. Called from init().
func RegisterEngine(e EngineRegistration) {
enginesMu.Lock()
defer enginesMu.Unlock()
localEngines = append(localEngines, e)
}
// Engines returns the locally registered engines in registration order.
// RegisterPluginEngine adds or replaces a plugin's engine. It refuses a name
// a builtin engine uses.
func RegisterPluginEngine(e EngineRegistration) error {
enginesMu.Lock()
defer enginesMu.Unlock()
for _, builtin := range localEngines {
if builtin.Name() == e.Name() {
return fmt.Errorf("parser engine %s already exists", e.Name())
}
}
pluginEngines[e.Name()] = e
return nil
}
// UnregisterPluginEngine removes a plugin's engine.
func UnregisterPluginEngine(name string) {
enginesMu.Lock()
defer enginesMu.Unlock()
delete(pluginEngines, name)
}
// Engines returns the builtin engines in registration order.
func Engines() []EngineRegistration {
enginesMu.RLock()
defer enginesMu.RUnlock()
out := make([]EngineRegistration, len(localEngines))
copy(out, localEngines)
return out
}
// lookupEngine returns the locally registered engine with this name.
// pluginEngineList returns the plugin engines sorted by name.
func pluginEngineList() []EngineRegistration {
enginesMu.RLock()
defer enginesMu.RUnlock()
out := make([]EngineRegistration, 0, len(pluginEngines))
for _, e := range pluginEngines {
out = append(out, e)
}
sort.Slice(out, func(i, j int) bool { return out[i].Name() < out[j].Name() })
return out
}
// PluginFileTypes returns the file types plugin engines parse, so uploads
// of those types are accepted.
func PluginFileTypes() []string {
var out []string
for _, e := range pluginEngineList() {
out = append(out, e.FileTypes(true)...)
}
return out
}
// lookupEngine returns the registered engine with this name.
func lookupEngine(name string) (EngineRegistration, bool) {
enginesMu.RLock()
defer enginesMu.RUnlock()
for _, engine := range localEngines {
if engine.Name() == name {
return engine, true
}
}
return nil, false
e, ok := pluginEngines[name]
return e, ok
}
// isPluginEngineName reports whether a name is a qualified plugin
// contribution ID, which no builtin or docreader engine uses.
func isPluginEngineName(name string) bool { return strings.Contains(name, "/") }
// NewReader builds the reader for an engine.
//
// An empty engine name means "no explicit choice": simple formats are handled
@@ -79,6 +152,11 @@ func NewReader(
if registration, ok := lookupEngine(engine); ok {
return registration.NewReader(ctx, deps)
}
if isPluginEngineName(engine) {
// Its plugin was uninstalled or disabled: the Python docreader does
// not know it either, so say so instead of sending it there.
return nil, errEngineUnavailable(engine, "its plugin is not installed or not running")
}
if engine == "" && !isURL && IsSimpleFormat(fileType) {
return &SimpleFormatReader{}, nil
}
@@ -112,10 +190,11 @@ func ListAllEngines(
remoteMap[re.Name] = re
}
seen := make(map[string]bool, len(localEngines))
result := make([]types.ParserEngineInfo, 0, len(localEngines)+len(remoteEngines))
locals := Engines()
seen := make(map[string]bool, len(locals))
result := make([]types.ParserEngineInfo, 0, len(locals)+len(remoteEngines))
for _, e := range localEngines {
for _, e := range locals {
name := e.Name()
seen[name] = true
@@ -148,6 +227,18 @@ func ListAllEngines(
result = append(result, re)
}
for _, e := range pluginEngineList() {
available, reason := e.CheckAvailable(docreaderConnected, overrides)
info := types.ParserEngineInfo{
Name: e.Name(), Description: e.Description(), FileTypes: e.FileTypes(docreaderConnected),
Available: available, UnavailableReason: reason,
}
if meta, ok := e.(PluginEngineInfo); ok {
info.PluginID, info.DisplayNames = meta.PluginID(), meta.DisplayNames()
}
result = append(result, info)
}
return result
}
+24 -1
View File
@@ -28,6 +28,22 @@ type Invoker struct {
mu sync.RWMutex
tenancy *tenancy.Service
plugins interfaces.PluginRepository
// tokens and hostURL give plugins granted Host API scopes a way back.
tokens TokenIssuer
hostURL string
}
// TokenIssuer signs the Host API token of one call.
type TokenIssuer interface {
Issue(pluginID, version string, tenantID uint64, scopes []string) (string, time.Time, error)
}
// SetHostAPI lets calls carry Host API access: where plugins reach WeKnora
// and how their tokens are signed.
func (iv *Invoker) SetHostAPI(tokens TokenIssuer, url string) {
iv.mu.Lock()
iv.tokens, iv.hostURL = tokens, url
iv.mu.Unlock()
}
// NewInvoker creates an Invoker; Bind completes it.
@@ -77,8 +93,15 @@ func (iv *Invoker) Envelope(
env.Context.Deadline = &d
}
iv.mu.RLock()
t, plugins := iv.tenancy, iv.plugins
t, plugins, tokens, hostURL := iv.tenancy, iv.plugins, iv.tokens, iv.hostURL
iv.mu.RUnlock()
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 {
return env, err
}
env.Context.Host = &pluginapi.HostAccess{URL: hostURL, Token: token}
}
if plugins != nil && len(m.Config.SystemSchema) > 0 {
sys, _, err := install.OpenSystemConfig(ctx, plugins, m)
if err != nil {
+37
View File
@@ -0,0 +1,37 @@
package activate
import (
"context"
"testing"
"github.com/Tencent/WeKnora/internal/plugin/hostapi"
"github.com/Tencent/WeKnora/internal/plugin/manifest"
"github.com/Tencent/WeKnora/internal/types"
)
func TestEnvelopeCarriesHostAccessOnlyForGrantedPlugins(t *testing.T) {
iss := hostapi.NewIssuer([]byte("k"))
iv := NewInvoker(fakeClients{})
iv.SetHostAPI(iss, "http://127.0.0.1:9")
ctx := context.WithValue(context.Background(), types.TenantIDContextKey, uint64(7))
granted := &manifest.Manifest{
ID: "acme.x", Version: "1.0.0", Permissions: manifest.Permissions{HostAPI: []string{"kv"}},
}
env, err := iv.Envelope(ctx, granted, nil)
if err != nil || env.Context.Host == nil || env.Context.Host.URL != "http://127.0.0.1:9" {
t.Fatalf("envelope = %+v, %v", env.Context, err)
}
claims, err := iss.Verify(env.Context.Host.Token)
if err != nil || claims.PluginID != "acme.x" || claims.TenantID != 7 || !claims.Has("kv") {
t.Fatalf("token claims = %+v, %v", claims, err)
}
plain := &manifest.Manifest{ID: "acme.y", Version: "1.0.0"}
if env, _ := iv.Envelope(ctx, plain, nil); env.Context.Host != nil {
t.Fatal("a plugin without Host API scopes must get no token")
}
if env, _ := iv.Envelope(context.Background(), granted, nil); env.Context.Host != nil {
t.Fatal("a call without a tenant must get no token")
}
}
+152
View File
@@ -0,0 +1,152 @@
package activate
import (
"context"
"errors"
"fmt"
"path"
"strings"
"sync"
"time"
"github.com/Tencent/WeKnora/internal/infrastructure/docparser"
"github.com/Tencent/WeKnora/internal/plugin/manifest"
"github.com/Tencent/WeKnora/internal/plugin/reconcile"
"github.com/Tencent/WeKnora/internal/types"
"github.com/Tencent/WeKnora/internal/types/interfaces"
"github.com/Tencent/WeKnora/pluginsdk/pluginapi"
)
// parseTimeout bounds one plugin parse when the caller set no deadline.
const parseTimeout = 10 * time.Minute
// Parsers registers code plugins' document parsers as parser engines under
// their qualified IDs, next to the builtin engines and the docreader's.
type Parsers struct {
iv *Invoker
mu sync.Mutex
registered map[string][]string
}
// NewParsers creates the parser activator.
func NewParsers(iv *Invoker) *Parsers {
return &Parsers{iv: iv, registered: map[string][]string{}}
}
// Name implements reconcile.Activator.
func (a *Parsers) Name() string { return "parsers" }
// Activate implements reconcile.Activator: all of a plugin's parsers or none.
func (a *Parsers) Activate(_ context.Context, l *reconcile.Loaded) error {
var ids []string
for _, c := range l.Manifest.Contributes[manifest.PointParsers] {
e := &pluginEngine{
iv: a.iv, m: l.Manifest, local: c.ID, name: manifest.QualifiedID(l.Manifest.ID, c.ID),
description: c.Description.Default, names: displayNames(c.Name), fileTypes: c.FileTypes,
}
if err := docparser.RegisterPluginEngine(e); err != nil {
for _, done := range ids {
docparser.UnregisterPluginEngine(done)
}
return fmt.Errorf("parser %s: %w", c.ID, err)
}
ids = append(ids, e.name)
}
a.mu.Lock()
a.registered[l.Manifest.ID] = ids
a.mu.Unlock()
return nil
}
// Deactivate implements reconcile.Activator.
func (a *Parsers) Deactivate(_ context.Context, pluginID string) error {
a.mu.Lock()
ids := a.registered[pluginID]
delete(a.registered, pluginID)
a.mu.Unlock()
for _, id := range ids {
docparser.UnregisterPluginEngine(id)
}
return nil
}
func displayNames(t manifest.LocalizedText) map[string]string {
out := map[string]string{}
for k, v := range t.Locales {
out[k] = v
}
if t.Default != "" {
out["default"] = t.Default
}
return out
}
// pluginEngine is a plugin parser as a docparser engine.
type pluginEngine struct {
iv *Invoker
m *manifest.Manifest
local, name string
description string
names map[string]string
fileTypes []string
}
var (
_ docparser.EngineRegistration = (*pluginEngine)(nil)
_ docparser.PluginEngineInfo = (*pluginEngine)(nil)
)
func (e *pluginEngine) Name() string { return e.name }
func (e *pluginEngine) Description() string { return e.description }
func (e *pluginEngine) FileTypes(bool) []string { return e.fileTypes }
func (e *pluginEngine) PluginID() string { return e.m.ID }
func (e *pluginEngine) DisplayNames() map[string]string { return e.names }
func (e *pluginEngine) CheckAvailable(bool, map[string]string) (bool, string) {
if _, err := e.iv.clients.Client(e.m.ID); err != nil {
return false, err.Error()
}
return true, ""
}
// NewReader implements docparser.EngineRegistration. deps.Overrides carry
// other engines' credentials and are deliberately not passed on; the
// plugin's own settings travel in its system and tenant configuration.
func (e *pluginEngine) NewReader(context.Context, docparser.ReaderDeps) (interfaces.DocReader, error) {
return &remoteParser{engine: e}, nil
}
// remoteParser reads one document through the plugin.
type remoteParser struct{ engine *pluginEngine }
// Read implements interfaces.DocReader. A retryable plugin failure (the
// plugin is down, rate limited) is returned as an error so the task is
// retried; any other failure is final and reported in the result.
func (r *remoteParser) Read(ctx context.Context, req *types.ReadRequest) (*types.ReadResult, error) {
ctx, cancel := withDefaultTimeout(ctx, parseTimeout)
defer cancel()
in := pluginapi.ParseInput{
FileName: req.FileName, FileType: strings.ToLower(strings.TrimPrefix(req.FileType, ".")),
Content: req.FileContent, URL: req.URL, Title: req.Title,
}
var out pluginapi.ParseOutput
err := r.engine.iv.Call(ctx, r.engine.m, pluginapi.ParsePath(r.engine.local), nil, in, &out)
if err != nil {
var pe *pluginapi.Error
if errors.As(err, &pe) && !pe.Retryable {
return &types.ReadResult{Error: err.Error()}, nil
}
return nil, err
}
res := &types.ReadResult{MarkdownContent: out.Markdown, Metadata: out.Metadata}
for _, img := range out.Images {
if img.OriginalRef == "" || len(img.Data) == 0 {
continue
}
res.ImageRefs = append(res.ImageRefs, types.ImageRef{
Filename: path.Base(img.OriginalRef), OriginalRef: img.OriginalRef,
MimeType: img.MimeType, ImageData: img.Data,
})
}
return res, nil
}
+120
View File
@@ -0,0 +1,120 @@
package activate
import (
"context"
"net/http/httptest"
"slices"
"strings"
"testing"
"github.com/Tencent/WeKnora/internal/infrastructure/docparser"
"github.com/Tencent/WeKnora/internal/plugin/pkg"
"github.com/Tencent/WeKnora/internal/plugin/plugintest"
"github.com/Tencent/WeKnora/internal/plugin/reconcile"
"github.com/Tencent/WeKnora/internal/types"
"github.com/Tencent/WeKnora/pluginsdk"
"github.com/Tencent/WeKnora/pluginsdk/client"
"github.com/Tencent/WeKnora/pluginsdk/pluginapi"
)
const ocrManifest = `schemaVersion: 1
id: acme.ocr
version: 1.0.0
apiVersion: weknora.plugin/v1
name: { en-US: ACME OCR }
publisher: { id: acme }
runtime: { type: host, kind: binary, entry: bin/ocr }
contributes:
parsers:
- id: ocr
name: { en-US: ACME OCR, zh-CN: ACME 识别 }
fileTypes: [tiff, pdf]
`
func TestPluginParser(t *testing.T) {
ctx := context.Background()
plugin := pluginsdk.New(pluginsdk.Info{ID: "acme.ocr", Version: "1.0.0"})
plugin.Parser("ocr", pluginsdk.ParserFunc(
func(_ context.Context, _ *pluginsdk.Call, in pluginapi.ParseInput) (*pluginapi.ParseOutput, error) {
switch string(in.Content) {
case "busy":
return nil, pluginapi.Errorf(pluginapi.CodeRateLimited, "slow down")
case "broken":
return nil, pluginapi.Errorf(pluginapi.CodeInvalidConfig, "not a scanned document")
}
return &pluginapi.ParseOutput{
Markdown: "# " + in.FileName + "\n\n![scan](page-1.png)",
Images: []pluginapi.ParsedImage{
{OriginalRef: "page-1.png", MimeType: "image/png", Data: []byte{1, 2}},
},
Metadata: map[string]string{"pages": "1"},
}, nil
}))
srv := httptest.NewServer(plugin.Handler())
defer srv.Close()
p, err := pkg.Open(plugintest.Zip(t, map[string]string{"plugin.yaml": ocrManifest, "bin/ocr": "x"}))
if err != nil {
t.Fatal(err)
}
a := NewParsers(NewInvoker(fakeClients{client.New(srv.URL, nil, nil)}))
if err := a.Activate(ctx, &reconcile.Loaded{Manifest: p.Manifest, Package: p}); err != nil {
t.Fatalf("Activate: %v", err)
}
defer func() { _ = a.Deactivate(ctx, "acme.ocr") }()
var listed *types.ParserEngineInfo
for _, e := range docparser.ListAllEngines(false, nil, nil) {
if e.Name == "acme.ocr/ocr" {
listed = &e
}
}
if listed == nil || !listed.Available || listed.PluginID != "acme.ocr" ||
listed.DisplayNames["zh-CN"] != "ACME 识别" ||
!slices.Equal(listed.FileTypes, []string{"tiff", "pdf"}) {
t.Fatalf("listed = %+v", listed)
}
if !slices.Contains(docparser.PluginFileTypes(), "tiff") {
t.Fatal("plugin file types must be importable")
}
reader, err := docparser.NewReader(ctx, "acme.ocr/ocr", "tiff", false, docparser.ReaderDeps{
Overrides: map[string]string{"mineru_api_key": "must-not-leak"},
})
if err != nil {
t.Fatal(err)
}
res, err := reader.Read(
ctx,
&types.ReadRequest{FileName: "scan.tiff", FileType: ".TIFF", FileContent: []byte("img")},
)
if err != nil || !strings.HasPrefix(res.MarkdownContent, "# scan.tiff") || res.Metadata["pages"] != "1" ||
len(
res.ImageRefs,
) != 1 || res.ImageRefs[0].OriginalRef != "page-1.png" || res.ImageRefs[0].Filename != "page-1.png" {
t.Fatalf("Read = %+v, %v", res, err)
}
if _, err := reader.Read(ctx, &types.ReadRequest{FileContent: []byte("busy")}); err == nil {
t.Fatal("a retryable plugin error must come back as an error, so the task retries")
}
res, err = reader.Read(ctx, &types.ReadRequest{FileContent: []byte("broken")})
if err != nil || res.Error != "not a scanned document" {
t.Fatalf("a permanent plugin error must fail the document: %+v, %v", res, err)
}
if err := a.Deactivate(ctx, "acme.ocr"); err != nil {
t.Fatal(err)
}
if _, err := docparser.NewReader(ctx, "acme.ocr/ocr", "tiff", false, docparser.ReaderDeps{}); err == nil ||
!strings.Contains(err.Error(), "not installed or not running") {
t.Fatalf("a removed plugin engine must not fall through to the docreader: %v", err)
}
}
func TestPluginParserCannotShadowABuiltin(t *testing.T) {
e := &pluginEngine{name: "simple"}
if err := docparser.RegisterPluginEngine(e); err == nil {
t.Fatal("a plugin engine must not take a builtin engine's name")
}
}
+4 -2
View File
@@ -167,8 +167,10 @@ func (p *process) childEnv(socket, token string) []string {
"HTTPS_PROXY=" + p.proxy.URL(),
"http_proxy=" + p.proxy.URL(),
"https_proxy=" + p.proxy.URL(),
"NO_PROXY=",
"no_proxy=",
// The Host API is on this node's loopback; the proxy is for the
// outside world.
"NO_PROXY=127.0.0.1,localhost,::1",
"no_proxy=127.0.0.1,localhost,::1",
}
for _, k := range []string{"PATH", "LANG", "LC_ALL", "TZ", "SYSTEMROOT", "WINDIR"} {
if v, ok := os.LookupEnv(k); ok {
+111
View File
@@ -0,0 +1,111 @@
package hostapi
import (
"context"
"net/http/httptest"
"strings"
"testing"
"time"
"github.com/gin-gonic/gin"
"github.com/google/uuid"
"github.com/stretchr/testify/require"
"gorm.io/driver/sqlite"
"gorm.io/gorm"
"github.com/Tencent/WeKnora/internal/application/repository"
"github.com/Tencent/WeKnora/internal/types"
"github.com/Tencent/WeKnora/pluginsdk"
"github.com/Tencent/WeKnora/pluginsdk/pluginapi"
)
func TestTokens(t *testing.T) {
iss := NewIssuer([]byte("k"))
tok, exp, err := iss.Issue("acme.x", "1.0.0", 7, []string{"kv"})
require.NoError(t, err)
require.WithinDuration(t, time.Now().Add(TokenTTL), exp, time.Second)
c, err := iss.Verify(tok)
require.NoError(t, err)
require.Equal(t, uint64(7), c.TenantID)
require.True(t, c.Has("kv"))
_, err = NewIssuer([]byte("other")).Verify(tok)
require.ErrorIs(t, err, ErrInvalidToken, "another key must not verify")
later := NewIssuer([]byte("k"))
later.now = func() time.Time { return time.Now().Add(TokenTTL + time.Minute) }
_, err = later.Verify(tok)
require.ErrorIs(t, err, ErrInvalidToken, "expired tokens must not verify")
_, err = iss.Verify(tok[:len(tok)-2] + "xx")
require.ErrorIs(t, err, ErrInvalidToken)
}
// hostCall is what a plugin's call would carry, pointing at a test server.
func hostCall(t *testing.T, url string, iss *Issuer, tenant uint64, scopes ...string) *pluginsdk.Call {
t.Helper()
tok, _, err := iss.Issue("acme.x", "1.0.0", tenant, scopes)
require.NoError(t, err)
return &pluginsdk.Call{
Context: pluginapi.Context{TenantID: tenant, Host: &pluginapi.HostAccess{URL: url, Token: tok}},
}
}
func TestKVThroughTheSDK(t *testing.T) {
gin.SetMode(gin.TestMode)
db, err := gorm.Open(sqlite.Open("file:"+uuid.NewString()+"?mode=memory&cache=shared"), &gorm.Config{})
require.NoError(t, err)
require.NoError(t, db.AutoMigrate(&types.PluginKV{}))
iss := NewIssuer([]byte("k"))
h := NewHandler(iss, NewKV(repository.NewPluginKVRepository(db)))
r := gin.New()
h.Register(r)
srv := httptest.NewServer(r)
defer srv.Close()
ctx := context.Background()
host := hostCall(t, srv.URL, iss, 1, "kv").Host()
require.NotNil(t, host)
type cursor struct{ Page int }
ok, err := host.KVGet(ctx, "cursor/a", &cursor{})
require.NoError(t, err)
require.False(t, ok)
require.NoError(t, host.KVPut(ctx, "cursor/a", cursor{Page: 3}, 0))
require.NoError(t, host.KVPut(ctx, "cursor/b", cursor{Page: 4}, time.Hour))
var got cursor
ok, err = host.KVGet(ctx, "cursor/a", &got)
require.NoError(t, err)
require.True(t, ok)
require.Equal(t, 3, got.Page)
page, err := host.KVList(ctx, "cursor/", "", 1)
require.NoError(t, err)
require.Len(t, page.Entries, 1)
require.Equal(t, "cursor/a", page.Next)
page, err = host.KVList(ctx, "cursor/", page.Next, 1)
require.NoError(t, err)
require.Equal(t, "cursor/b", page.Entries[0].Key)
require.NotNil(t, page.Entries[0].ExpiresAt)
// Another tenant sees nothing of tenant 1.
other := hostCall(t, srv.URL, iss, 2, "kv").Host()
ok, err = other.KVGet(ctx, "cursor/a", &got)
require.NoError(t, err)
require.False(t, ok)
require.NoError(t, host.KVDelete(ctx, "cursor/a"))
ok, _ = host.KVGet(ctx, "cursor/a", &got)
require.False(t, ok)
// Without the scope, or with a forged token, the call is refused.
noScope := hostCall(t, srv.URL, iss, 1).Host()
_, err = noScope.KVGet(ctx, "cursor/b", &got)
require.ErrorContains(t, err, `not granted the "kv" scope`)
forged := hostCall(t, srv.URL, NewIssuer([]byte("wrong")), 1, "kv").Host()
_, err = forged.KVGet(ctx, "cursor/b", &got)
require.ErrorContains(t, err, "invalid or expired")
// Bad input is a bad request, not a crash.
err = host.KVPut(ctx, strings.Repeat("k", 300), 1, 0)
e, ok := pluginapi.AsError(err)
require.True(t, ok)
require.Equal(t, pluginapi.CodeBadRequest, e.Code)
}
+125
View File
@@ -0,0 +1,125 @@
package hostapi
import (
"encoding/json"
"errors"
"io"
"net/http"
"strconv"
"strings"
"github.com/gin-gonic/gin"
"github.com/Tencent/WeKnora/internal/logger"
"github.com/Tencent/WeKnora/pluginsdk/pluginapi"
)
// PathPrefix is where the Host API is served. The global auth middleware
// lets it through; Handler authenticates plugin tokens itself.
const PathPrefix = "/api/v1/plugin-host/"
const claimsKey = "pluginHostClaims"
// Handler serves the Host API.
type Handler struct {
issuer *Issuer
kv *KV
}
// NewHandler creates the Host API handler.
func NewHandler(issuer *Issuer, kv *KV) *Handler { return &Handler{issuer: issuer, kv: kv} }
func writeErr(c *gin.Context, err error) {
var e *pluginapi.Error
if !errors.As(err, &e) {
logger.Errorf(c.Request.Context(), "[plugin] host api %s: %v", c.Request.URL.Path, err)
e = &pluginapi.Error{Code: pluginapi.CodeInternal, Message: "host api failed"}
}
c.AbortWithStatusJSON(e.Code.HTTPStatus(), pluginapi.ErrorBody{Error: *e})
}
// authenticate verifies the plugin token and the scope the route needs.
func (h *Handler) authenticate(scope string) gin.HandlerFunc {
return func(c *gin.Context) {
token, ok := strings.CutPrefix(c.GetHeader("Authorization"), "Bearer ")
if !ok || token == "" {
writeErr(c, pluginapi.Errorf(pluginapi.CodeUnauthorized, "missing plugin token"))
return
}
claims, err := h.issuer.Verify(token)
if err != nil {
writeErr(c, pluginapi.Errorf(pluginapi.CodeUnauthorized, "invalid or expired plugin token"))
return
}
if !claims.Has(scope) {
writeErr(
c,
pluginapi.Errorf(pluginapi.CodeUnauthorized, "the plugin was not granted the %q scope", scope),
)
return
}
c.Set(claimsKey, claims)
c.Next()
}
}
func claimsOf(c *gin.Context) *Claims {
v, _ := c.Get(claimsKey)
claims, _ := v.(*Claims)
return claims
}
// Register mounts the Host API on the engine root.
func (h *Handler) Register(r gin.IRouter) {
g := r.Group(strings.TrimSuffix(PathPrefix, "/"))
kv := g.Group("/kv", h.authenticate(ScopeKV))
kv.GET("", h.kvGet)
kv.PUT("", h.kvPut)
kv.DELETE("", h.kvDelete)
kv.GET("/list", h.kvList)
}
func (h *Handler) kvGet(c *gin.Context) {
e, err := h.kv.Get(c.Request.Context(), claimsOf(c), c.Query("key"))
if err != nil {
writeErr(c, err)
return
}
c.JSON(http.StatusOK, e)
}
func (h *Handler) kvPut(c *gin.Context) {
var in pluginapi.KVPut
body, err := io.ReadAll(io.LimitReader(c.Request.Body, pluginapi.KVMaxValueBytes+pluginapi.KVMaxKeyBytes+1024))
if err == nil {
err = json.Unmarshal(body, &in)
}
if err != nil {
writeErr(c, pluginapi.Errorf(pluginapi.CodeBadRequest, "body must be {key, value, ttlSeconds}"))
return
}
e, err := h.kv.Put(c.Request.Context(), claimsOf(c), in)
if err != nil {
writeErr(c, err)
return
}
c.JSON(http.StatusOK, e)
}
func (h *Handler) kvDelete(c *gin.Context) {
if err := h.kv.Delete(c.Request.Context(), claimsOf(c), c.Query("key")); err != nil {
writeErr(c, err)
return
}
c.Status(http.StatusNoContent)
}
func (h *Handler) kvList(c *gin.Context) {
limit, _ := strconv.Atoi(c.Query("limit"))
out, err := h.kv.List(c.Request.Context(), claimsOf(c), c.Query("prefix"), c.Query("after"), limit)
if err != nil {
writeErr(c, err)
return
}
c.JSON(http.StatusOK, out)
}
+162
View File
@@ -0,0 +1,162 @@
package hostapi
import (
"context"
"encoding/json"
"strings"
"time"
"unicode"
"unicode/utf8"
"github.com/Tencent/WeKnora/internal/logger"
"github.com/Tencent/WeKnora/internal/types"
"github.com/Tencent/WeKnora/internal/types/interfaces"
"github.com/Tencent/WeKnora/pluginsdk/pluginapi"
)
// ScopeKV grants a plugin its own key-value store in the calling tenant.
const ScopeKV = "kv"
// KV is the plugins' key-value store. Every operation is confined to the
// plugin and tenant of the token, so a plugin never sees another plugin's
// or another workspace's data.
type KV struct {
repo interfaces.PluginKVRepository
now func() time.Time
}
// NewKV creates the store.
func NewKV(repo interfaces.PluginKVRepository) *KV { return &KV{repo: repo, now: time.Now} }
func checkKey(key string) error {
switch {
case key == "":
return pluginapi.Errorf(pluginapi.CodeBadRequest, "key is required")
case len(key) > pluginapi.KVMaxKeyBytes:
return pluginapi.Errorf(pluginapi.CodeBadRequest, "key is over %d bytes", pluginapi.KVMaxKeyBytes)
case !utf8.ValidString(key) || strings.ContainsFunc(key, unicode.IsControl):
return pluginapi.Errorf(pluginapi.CodeBadRequest, "key must be printable UTF-8")
}
return nil
}
func toEntry(e *types.PluginKV) pluginapi.KVEntry {
return pluginapi.KVEntry{
Key: e.Key,
Value: json.RawMessage(e.Value),
ExpiresAt: e.ExpiresAt,
UpdatedAt: e.UpdatedAt,
}
}
// Get reads a key.
func (s *KV) Get(ctx context.Context, c *Claims, key string) (*pluginapi.KVEntry, error) {
if err := checkKey(key); err != nil {
return nil, err
}
e, err := s.repo.Get(ctx, c.PluginID, c.TenantID, key)
if err != nil {
return nil, err
}
if e == nil {
return nil, pluginapi.Errorf(pluginapi.CodeNotFound, "no key %q", key)
}
out := toEntry(e)
return &out, nil
}
// Put writes a key, refusing new keys past the per-tenant quota.
func (s *KV) Put(ctx context.Context, c *Claims, in pluginapi.KVPut) (*pluginapi.KVEntry, error) {
if err := checkKey(in.Key); err != nil {
return nil, err
}
if len(in.Value) == 0 || !json.Valid(in.Value) {
return nil, pluginapi.Errorf(pluginapi.CodeBadRequest, "value must be JSON")
}
if len(in.Value) > pluginapi.KVMaxValueBytes {
return nil, pluginapi.Errorf(pluginapi.CodeBadRequest, "value is over %d bytes", pluginapi.KVMaxValueBytes)
}
if in.TTLSeconds < 0 {
return nil, pluginapi.Errorf(pluginapi.CodeBadRequest, "ttlSeconds must not be negative")
}
existing, err := s.repo.Get(ctx, c.PluginID, c.TenantID, in.Key)
if err != nil {
return nil, err
}
if existing == nil {
n, err := s.repo.Count(ctx, c.PluginID, c.TenantID)
if err != nil {
return nil, err
}
if n >= pluginapi.KVMaxKeys {
return nil, pluginapi.Errorf(
pluginapi.CodeRateLimited,
"the store holds its maximum of %d keys",
pluginapi.KVMaxKeys,
)
}
}
now := s.now()
e := &types.PluginKV{
PluginID: c.PluginID, TenantID: c.TenantID, Key: in.Key, Value: types.JSON(in.Value), UpdatedAt: now,
}
if in.TTLSeconds > 0 {
exp := now.Add(time.Duration(in.TTLSeconds) * time.Second)
e.ExpiresAt = &exp
}
if err := s.repo.Put(ctx, e); err != nil {
return nil, err
}
out := toEntry(e)
return &out, nil
}
// Delete removes a key; deleting a missing key is not an error.
func (s *KV) Delete(ctx context.Context, c *Claims, key string) error {
if err := checkKey(key); err != nil {
return err
}
_, err := s.repo.Delete(ctx, c.PluginID, c.TenantID, key)
return err
}
// List pages through keys with a prefix.
func (s *KV) List(ctx context.Context, c *Claims, prefix, after string, limit int) (*pluginapi.KVList, error) {
if limit <= 0 {
limit = 100
}
limit = min(limit, pluginapi.KVMaxListLimit)
rows, err := s.repo.List(ctx, c.PluginID, c.TenantID, prefix, after, limit+1)
if err != nil {
return nil, err
}
out := &pluginapi.KVList{Entries: []pluginapi.KVEntry{}}
for i := range rows {
if i == limit {
out.Next = rows[i-1].Key
break
}
out.Entries = append(out.Entries, toEntry(&rows[i]))
}
return out, nil
}
// StartSweeper deletes expired entries every interval until ctx ends.
func (s *KV) StartSweeper(ctx context.Context, interval time.Duration) {
go func() {
t := time.NewTicker(interval)
defer t.Stop()
for {
select {
case <-ctx.Done():
return
case <-t.C:
if n, err := s.repo.DeleteExpired(ctx, s.now()); err != nil {
logger.Warnf(ctx, "[plugin] sweep expired kv: %v", err)
} else if n > 0 {
logger.Infof(ctx, "[plugin] swept %d expired kv entries", n)
}
}
}
}()
}
+110
View File
@@ -0,0 +1,110 @@
// Package hostapi is the Host API: how a code plugin calls back into
// WeKnora. Every call to a plugin that was granted Host API scopes carries a
// short-lived token bound to the plugin, the tenant and those scopes; the
// plugin presents it to /api/v1/plugin-host/*.
package hostapi
import (
"crypto/hmac"
"crypto/rand"
"crypto/sha256"
"errors"
"fmt"
"os"
"slices"
"strings"
"time"
"github.com/golang-jwt/jwt/v5"
"github.com/Tencent/WeKnora/internal/utils"
)
// Audience of Host API tokens; they are refused anywhere else.
const Audience = "weknora-host-api"
// TokenTTL is how long a token lives: one call and its follow-ups.
const TokenTTL = 5 * time.Minute
// Claims are what a Host API token asserts.
type Claims struct {
PluginID string `json:"plugin"`
Version string `json:"ver"`
TenantID uint64 `json:"tenant"`
Scopes []string `json:"scopes"`
jwt.RegisteredClaims
}
// Has reports whether the token grants a scope.
func (c *Claims) Has(scope string) bool { return slices.Contains(c.Scopes, scope) }
// Issuer signs and verifies Host API tokens.
type Issuer struct {
key []byte
now func() time.Time
}
// NewIssuer signs with key. Nodes must share it for a token issued on one
// node to verify on another.
func NewIssuer(key []byte) *Issuer { return &Issuer{key: key, now: time.Now} }
// NewIssuerFromEnv derives the signing key from the cluster-wide secrets:
// SYSTEM_AES_KEY, else JWT_SECRET. Without either it uses a random key, which
// only works while plugins call back the node that called them (the
// embedded host always does).
func NewIssuerFromEnv() *Issuer {
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 host api tokens"))
return NewIssuer(mac.Sum(nil))
}
// Issue signs a token for one call.
func (i *Issuer) Issue(pluginID, version string, tenantID uint64, scopes []string) (string, time.Time, error) {
now := i.now()
exp := now.Add(TokenTTL)
claims := Claims{
PluginID: pluginID, Version: version, TenantID: tenantID, Scopes: scopes,
RegisteredClaims: jwt.RegisteredClaims{
Subject: "plugin:" + pluginID,
Audience: jwt.ClaimStrings{Audience},
IssuedAt: jwt.NewNumericDate(now),
NotBefore: jwt.NewNumericDate(now.Add(-30 * time.Second)),
ExpiresAt: jwt.NewNumericDate(exp),
},
}
tok, err := jwt.NewWithClaims(jwt.SigningMethodHS256, claims).SignedString(i.key)
return tok, exp, err
}
// ErrInvalidToken is any token that does not verify.
var ErrInvalidToken = errors.New("invalid plugin token")
// Verify checks a token's signature, audience and lifetime.
func (i *Issuer) Verify(token string) (*Claims, error) {
var c Claims
parsed, err := jwt.ParseWithClaims(token, &c, func(*jwt.Token) (any, error) {
return i.key, nil
},
jwt.WithValidMethods([]string{jwt.SigningMethodHS256.Alg()}),
jwt.WithAudience(Audience),
jwt.WithExpirationRequired(),
jwt.WithTimeFunc(i.now),
)
if err != nil || !parsed.Valid {
return nil, fmt.Errorf("%w: %v", ErrInvalidToken, err)
}
if c.PluginID == "" || c.TenantID == 0 || c.Subject != "plugin:"+c.PluginID {
return nil, fmt.Errorf("%w: incomplete claims", ErrInvalidToken)
}
return &c, nil
}
+32
View File
@@ -10,6 +10,7 @@ import (
"errors"
"fmt"
"regexp"
"slices"
"strings"
"golang.org/x/mod/semver"
@@ -150,6 +151,8 @@ type Contribution struct {
Path string `json:"path,omitempty" yaml:"path"`
// MCP describes a remote MCP server (mcpServers).
MCP *MCPServer `json:"mcp,omitempty" yaml:"mcp"`
// FileTypes are the lower-case extensions a parser handles ("pdf").
FileTypes []string `json:"fileTypes,omitempty" yaml:"fileTypes"`
// Extra carries point-specific metadata the generic fields do not cover.
Extra map[string]any `json:"extra,omitempty" yaml:"extra"`
}
@@ -287,7 +290,19 @@ var egressPattern = regexp.MustCompile(`^(\*\.)?([a-z0-9]([a-z0-9-]*[a-z0-9])?)(
// follows links anywhere). Private addresses stay unreachable either way.
const EgressAnyHost = "*"
// HostAPIScopes are the Host API scopes a plugin may ask for in
// permissions.hostApi.
var HostAPIScopes = []string{
// kv: the plugin's own key-value store in each tenant.
"kv",
}
func (m *Manifest) validatePermissions(add func(string, ...any)) {
for _, s := range m.Permissions.HostAPI {
if !slices.Contains(HostAPIScopes, s) {
add("permissions.hostApi %q is not a Host API scope (known: %s)", s, strings.Join(HostAPIScopes, ", "))
}
}
for _, e := range m.Permissions.Egress {
if e != EgressAnyHost && !egressPattern.MatchString(strings.ToLower(e)) {
add(
@@ -345,6 +360,9 @@ func (m *Manifest) validateContributions(add func(string, ...any)) {
add("%s.aliases may only be declared by builtin plugins", where)
}
validateDeclarative(point, c, m.Builtin, where, add)
if point == PointParsers && !m.Builtin {
validateFileTypes(c.FileTypes, where, add)
}
for _, alias := range c.Aliases {
// Aliases share the ID alphabet, which has no '/', so an
// alias can never shadow a qualified ID.
@@ -358,6 +376,20 @@ func (m *Manifest) validateContributions(add func(string, ...any)) {
// validateDeclarative checks the fields a declarative contribution needs.
// Builtins describe their implementation in code instead.
// fileTypePattern is a lower-case file extension without the dot.
var fileTypePattern = regexp.MustCompile(`^[a-z0-9]{1,16}$`)
func validateFileTypes(types []string, where string, add func(string, ...any)) {
if len(types) == 0 {
add("%s.fileTypes is required: the extensions the parser handles, such as [pdf]", where)
}
for _, t := range types {
if !fileTypePattern.MatchString(t) {
add("%s.fileTypes %q must be a lower-case extension without the dot", where, t)
}
}
}
func validateDeclarative(point Point, c Contribution, builtin bool, where string, add func(string, ...any)) {
if builtin {
return
+1 -1
View File
@@ -40,7 +40,7 @@ var points = []PointInfo{
{Point: PointIMChannels},
{Point: PointWebSearch, ThirdParty: true},
{Point: PointTools},
{Point: PointParsers},
{Point: PointParsers, ThirdParty: true},
{Point: PointSkills, ThirdParty: true, Declarative: true},
{Point: PointMCPServers, ThirdParty: true, Declarative: true},
}
+8
View File
@@ -20,6 +20,7 @@ import (
"github.com/Tencent/WeKnora/internal/logger"
"github.com/Tencent/WeKnora/internal/mcpserver"
"github.com/Tencent/WeKnora/internal/middleware"
"github.com/Tencent/WeKnora/internal/plugin/hostapi"
"github.com/Tencent/WeKnora/internal/tracing/langfuse"
"github.com/Tencent/WeKnora/internal/types/interfaces"
@@ -92,6 +93,7 @@ type RouterParams struct {
DataSourceCredentialsHandler *handler.DataSourceCredentialsHandler
PluginHandler *handler.PluginHandler
PluginAdminHandler *handler.PluginAdminHandler
PluginHostAPI *hostapi.Handler
WeKnoraCloudHandler *handler.WeKnoraCloudHandler
WikiPageHandler *handler.WikiPageHandler
MemoryHandler *handler.MemoryHandler
@@ -206,6 +208,12 @@ func NewRouter(params RouterParams) *gin.Engine {
r.POST("/api/v1/local-browser/extension/authorize", params.SessionHandler.BrowserSkillAuthorize)
r.POST("/api/v1/local-browser/internal", params.SessionHandler.BrowserSkillInternal)
// Host API: plugins call back with their own short-lived tokens, which
// the handler verifies; user and API key auth do not apply.
if params.PluginHostAPI != nil {
params.PluginHostAPI.Register(r)
}
// 认证中间件
r.Use(middleware.Auth(params.TenantService, params.UserService, params.TenantMemberService, params.TenantAPIKeyService, params.Config))
+4
View File
@@ -51,6 +51,10 @@ type ParserEngineInfo struct {
FileTypes []string
Available bool
UnavailableReason string
// PluginID and DisplayNames describe an engine an installed plugin
// provides; DisplayNames is keyed by locale.
PluginID string `json:",omitempty"`
DisplayNames map[string]string `json:",omitempty"`
}
// --- Internal types used by chunking pipeline ---
+24
View File
@@ -2,6 +2,7 @@ package interfaces
import (
"context"
"time"
"github.com/Tencent/WeKnora/internal/plugin/manifest"
"github.com/Tencent/WeKnora/internal/types"
@@ -41,3 +42,26 @@ type PluginRepository interface {
GetVersion(ctx context.Context, pluginID, version string) (*types.PluginVersion, error)
SaveVersion(ctx context.Context, v *types.PluginVersion) error
}
// PluginKVRepository stores plugins' Host API key-value data. Expired
// entries read as missing.
type PluginKVRepository interface {
// Get returns (nil, nil) for a missing or expired key.
Get(ctx context.Context, pluginID string, tenantID uint64, key string) (*types.PluginKV, error)
Put(ctx context.Context, e *types.PluginKV) error
// Delete reports whether the key existed.
Delete(ctx context.Context, pluginID string, tenantID uint64, key string) (bool, error)
// List returns live keys starting with prefix, after the given key, in
// key order.
List(
ctx context.Context,
pluginID string,
tenantID uint64,
prefix, after string,
limit int,
) ([]types.PluginKV, error)
// Count counts live keys of a plugin in a tenant.
Count(ctx context.Context, pluginID string, tenantID uint64) (int64, error)
// DeleteExpired removes entries that expired before now.
DeleteExpired(ctx context.Context, now time.Time) (int64, error)
}
+14
View File
@@ -48,3 +48,17 @@ type PluginVersion struct {
// TableName pins the table so GORM's pluralizer cannot drift.
func (PluginVersion) TableName() string { return "plugin_versions" }
// PluginKV is one entry of a plugin's key-value store (Host API kv),
// partitioned by plugin and tenant.
type PluginKV struct {
PluginID string `json:"-" gorm:"type:varchar(128);primaryKey"`
TenantID uint64 `json:"-" gorm:"primaryKey"`
Key string `json:"key" gorm:"type:varchar(256);primaryKey"`
Value JSON `json:"value" gorm:"type:json"`
ExpiresAt *time.Time `json:"expiresAt,omitempty"`
UpdatedAt time.Time `json:"updatedAt"`
}
// TableName pins the table so GORM's pluralizer cannot drift.
func (PluginKV) TableName() string { return "plugin_kv" }
@@ -0,0 +1 @@
DROP TABLE IF EXISTS plugin_kv;
+12
View File
@@ -0,0 +1,12 @@
-- Host API key-value store of plugins (versioned 000117).
CREATE TABLE IF NOT EXISTS plugin_kv (
plugin_id VARCHAR(128) NOT NULL,
tenant_id BIGINT NOT NULL,
key VARCHAR(256) NOT NULL,
value TEXT NOT NULL,
expires_at DATETIME,
updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
PRIMARY KEY (plugin_id, tenant_id, key)
);
CREATE INDEX IF NOT EXISTS idx_plugin_kv_expires_at ON plugin_kv (expires_at);
@@ -0,0 +1 @@
DROP TABLE IF EXISTS plugin_kv;
@@ -0,0 +1,18 @@
-- Migration 000117: the key-value store plugins use through the Host API.
-- Plugins keep no database of their own; their state lives here, partitioned
-- by plugin and tenant so a plugin cannot read another workspace's data.
DO $$ BEGIN RAISE NOTICE '[Migration 000117] Creating plugin_kv'; END $$;
CREATE TABLE IF NOT EXISTS plugin_kv (
plugin_id VARCHAR(128) NOT NULL,
tenant_id BIGINT NOT NULL,
key VARCHAR(256) NOT NULL,
value JSONB NOT NULL,
expires_at TIMESTAMPTZ,
updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
PRIMARY KEY (plugin_id, tenant_id, key)
);
CREATE INDEX IF NOT EXISTS idx_plugin_kv_expires_at ON plugin_kv (expires_at) WHERE expires_at IS NOT NULL;
COMMENT ON TABLE plugin_kv IS 'Host API key-value store of plugins, per plugin and tenant.';
+29 -3
View File
@@ -1,7 +1,7 @@
# WeKnora plugin SDK (Go)
Build code plugins for WeKnora: web search providers and data source
connectors that run as their own process. The module has no dependencies.
Build code plugins for WeKnora: web search providers, data source
connectors and document parsers that run as their own process. The module has no dependencies.
| Package | What it is |
| --- | --- |
@@ -69,6 +69,31 @@ decrypted, in three separate scopes:
Keep plugins stateless and never cache a tenant's credentials across calls.
## Parsers
`p.Parser(id, ...)` turns a document (`in.Content`, base64 on the wire) into
Markdown. WeKnora chunks the text and stores the images the Markdown
references: return them in `Images` with an `OriginalRef` matching the
`![](...)` target. Declare the extensions it handles in `plugin.yaml`
(`fileTypes: [srt, vtt]`). Return a retryable error (`unavailable`,
`rate_limited`) to have the document retried later; any other error fails
it for good.
## Calling back into WeKnora (Host API)
A plugin that declares `permissions.hostApi` gets a short-lived token with
each call; `call.Host()` returns a client for it, or nil without a grant.
Scopes:
- `kv`: a key-value store of the plugin, separate per workspace, so the
plugin needs no database.
- `KVGet`, `KVPut` (optional TTL), `KVDelete` and `KVList`.
- Keys up to 256 bytes, JSON values up to 64 KB, 10,000 keys per
workspace.
The token is valid for a few minutes: use `call.Host()` within the call and
don't keep it.
## Packaging
A package is a zip (`.wkp`) with `plugin.yaml` at its root:
@@ -103,7 +128,8 @@ A few rules the manifest and host enforce:
(`acme.notes/notes`); for connectors and web search that must stay within
50 characters.
See `examples/plugins/rss` for a complete plugin and its `package.sh`.
See `examples/plugins/rss` (a connector) and `examples/plugins/subtitles` (a
parser using the Host API) for complete plugins with their `package.sh`.
## Testing
+16 -1
View File
@@ -140,6 +140,15 @@ func Run(ctx context.Context, t Target) Report {
return protocolAnswer(err)
})
}
for _, id := range m.Contributes["parsers"] {
check("parsers/"+id+" parse answers", func(ctx context.Context) error {
var out pluginapi.ParseOutput
err := t.Client.Call(ctx, pluginapi.ParsePath(id), envelope(), pluginapi.ParseInput{
FileName: "conformance.txt", FileType: "txt", Content: []byte("WeKnora conformance check\n"),
}, &out)
return protocolAnswer(err)
})
}
for _, id := range m.Contributes["connectors"] {
check("connectors/"+id+" validate answers", func(ctx context.Context) error {
return protocolAnswer(t.Client.Call(ctx, pluginapi.ConnectorValidatePath(id), envelope(), nil, nil))
@@ -227,7 +236,13 @@ func wantErrorBody(resp *http.Response, code pluginapi.ErrorCode) error {
return fmt.Errorf("HTTP %d without a protocol error body: %s", resp.StatusCode, b)
}
if resp.StatusCode != code.HTTPStatus() || body.Error.Code != code {
return fmt.Errorf("want HTTP %d %s, got HTTP %d %s", code.HTTPStatus(), code, resp.StatusCode, body.Error.Code)
return fmt.Errorf(
"want HTTP %d %s, got HTTP %d %s",
code.HTTPStatus(),
code,
resp.StatusCode,
body.Error.Code,
)
}
return nil
}
+13
View File
@@ -65,12 +65,25 @@ func TestSDKPluginConforms(t *testing.T) {
},
),
)
p.Parser(
"text",
pluginsdk.ParserFunc(
func(_ context.Context, _ *pluginsdk.Call, in pluginapi.ParseInput) (*pluginapi.ParseOutput, error) {
return &pluginapi.ParseOutput{Markdown: string(in.Content)}, nil
},
),
)
rep := conformance.Run(context.Background(), target(t, p.Handler()))
names := map[string]bool{}
for _, r := range rep.Results {
names[r.Name] = true
if !r.Passed {
t.Errorf("%s: %s", r.Name, r.Detail)
}
}
if !names["parsers/text parse answers"] {
t.Error("parsers must be checked")
}
if !rep.Passed() || rep.Plugin != "acme.strict@1.0.0" || len(rep.Results) < 10 {
t.Fatalf("report = %+v", rep)
}
+114
View File
@@ -0,0 +1,114 @@
package pluginsdk
import (
"bytes"
"context"
"encoding/json"
"io"
"net/http"
"net/url"
"strconv"
"strings"
"time"
"github.com/Tencent/WeKnora/pluginsdk/pluginapi"
)
// Host calls back into WeKnora on behalf of one call: the tenant and scopes
// are those the call's token carries. It is only valid while the token lives
// (a few minutes), so do not keep it beyond the call.
type Host struct {
base string
token string
http *http.Client
}
// Host returns the Host API client of this call, or nil when the plugin was
// granted no Host API scopes.
func (c *Call) Host() *Host {
if c.Context.Host == nil || c.Context.Host.URL == "" || c.Context.Host.Token == "" {
return nil
}
return &Host{
base: strings.TrimSuffix(c.Context.Host.URL, "/"), token: c.Context.Host.Token,
http: &http.Client{Timeout: 30 * time.Second},
}
}
func (h *Host) do(ctx context.Context, method, path string, query url.Values, body, out any) error {
var rd io.Reader
if body != nil {
b, err := json.Marshal(body)
if err != nil {
return err
}
rd = bytes.NewReader(b)
}
u := h.base + path
if len(query) > 0 {
u += "?" + query.Encode()
}
req, err := http.NewRequestWithContext(ctx, method, u, rd)
if err != nil {
return err
}
req.Header.Set("Authorization", "Bearer "+h.token)
if body != nil {
req.Header.Set("Content-Type", "application/json")
}
resp, err := h.http.Do(req)
if err != nil {
return pluginapi.Errorf(pluginapi.CodeUnavailable, "host api: %v", err)
}
defer func() { _ = resp.Body.Close() }()
if resp.StatusCode/100 != 2 {
var eb pluginapi.ErrorBody
b, _ := io.ReadAll(io.LimitReader(resp.Body, 64<<10))
if json.Unmarshal(b, &eb) == nil && eb.Error.Code != "" {
return &eb.Error
}
return pluginapi.Errorf(pluginapi.CodeInternal, "host api answered HTTP %d", resp.StatusCode)
}
if out == nil || resp.StatusCode == http.StatusNoContent {
return nil
}
return json.NewDecoder(resp.Body).Decode(out)
}
// KVGet reads key into v. ok is false when the key does not exist.
func (h *Host) KVGet(ctx context.Context, key string, v any) (ok bool, err error) {
var e pluginapi.KVEntry
err = h.do(ctx, http.MethodGet, pluginapi.HostKVPath, url.Values{"key": {key}}, nil, &e)
if pe, isPE := pluginapi.AsError(err); isPE && pe.Code == pluginapi.CodeNotFound {
return false, nil
}
if err != nil {
return false, err
}
return true, json.Unmarshal(e.Value, v)
}
// KVPut stores v (as JSON) under key; ttl 0 keeps it until deleted.
func (h *Host) KVPut(ctx context.Context, key string, v any, ttl time.Duration) error {
raw, err := json.Marshal(v)
if err != nil {
return err
}
in := pluginapi.KVPut{Key: key, Value: raw, TTLSeconds: int(ttl / time.Second)}
return h.do(ctx, http.MethodPut, pluginapi.HostKVPath, nil, in, nil)
}
// KVDelete removes key; a missing key is not an error.
func (h *Host) KVDelete(ctx context.Context, key string) error {
return h.do(ctx, http.MethodDelete, pluginapi.HostKVPath, url.Values{"key": {key}}, nil, nil)
}
// KVList returns a page of keys with prefix after the given key.
func (h *Host) KVList(ctx context.Context, prefix, after string, limit int) (*pluginapi.KVList, error) {
q := url.Values{"prefix": {prefix}, "after": {after}}
if limit > 0 {
q.Set("limit", strconv.Itoa(limit))
}
var out pluginapi.KVList
return &out, h.do(ctx, http.MethodGet, pluginapi.HostKVListPath, q, nil, &out)
}
+1 -1
View File
@@ -25,7 +25,7 @@ func TestOpenAPIMatchesRoutes(t *testing.T) {
if err != nil {
t.Fatal(err)
}
for _, f := range []string{"websearch.go", "connector.go"} {
for _, f := range []string{"websearch.go", "connector.go", "parser.go"} {
b, err := os.ReadFile(f)
if err != nil {
t.Fatal(err)
+55
View File
@@ -0,0 +1,55 @@
package pluginsdk
import (
"context"
"encoding/json"
"net/http"
"github.com/Tencent/WeKnora/pluginsdk/pluginapi"
)
// Parser turns a document into Markdown. Return a non-retryable error (such
// as pluginapi.CodeInvalidConfig or CodeNotFound) for a document that will
// never parse; retryable ones (CodeUnavailable, CodeRateLimited) make
// WeKnora try again later.
type Parser interface {
Parse(ctx context.Context, call *Call, in pluginapi.ParseInput) (*pluginapi.ParseOutput, error)
}
// ParserFunc adapts a function to Parser.
type ParserFunc func(ctx context.Context, call *Call, in pluginapi.ParseInput) (*pluginapi.ParseOutput, error)
// Parse implements Parser.
func (f ParserFunc) Parse(ctx context.Context, call *Call, in pluginapi.ParseInput) (*pluginapi.ParseOutput, error) {
return f(ctx, call, in)
}
// Parser registers parser id (contributes.parsers[].id).
func (p *Plugin) Parser(id string, parser Parser) { p.parsers[id] = parser }
func (p *Plugin) routeParsers(mux *http.ServeMux) {
mux.HandleFunc("POST /v1/parsers/{id}/parse", func(w http.ResponseWriter, r *http.Request) {
parser, ok := p.parsers[r.PathValue("id")]
if !ok {
writeError(w, pluginapi.Errorf(pluginapi.CodeNotFound, "no parser %q", r.PathValue("id")))
return
}
p.unary(func(ctx context.Context, call *Call, raw json.RawMessage) (any, error) {
in, err := decodeInput[pluginapi.ParseInput](raw)
if err != nil {
return nil, err
}
if len(in.Content) == 0 && in.URL == "" {
return nil, pluginapi.Errorf(pluginapi.CodeBadRequest, "content or url is required")
}
out, err := parser.Parse(ctx, call, in)
if err != nil {
return nil, err
}
if out == nil {
out = &pluginapi.ParseOutput{}
}
return out, nil
})(w, r)
})
}
+4
View File
@@ -62,6 +62,7 @@ type Plugin struct {
info Info
webSearch map[string]WebSearcher
connectors map[string]Connector
parsers map[string]Parser
validate ConfigValidator
logger *slog.Logger
// ShutdownTimeout bounds how long Serve waits for calls in flight after
@@ -75,6 +76,7 @@ func New(info Info) *Plugin {
info: info,
webSearch: map[string]WebSearcher{},
connectors: map[string]Connector{},
parsers: map[string]Parser{},
logger: slog.New(slog.NewTextHandler(os.Stderr, nil)),
ShutdownTimeout: 60 * time.Second,
}
@@ -100,6 +102,7 @@ func (p *Plugin) Manifest() pluginapi.Manifest {
}
add("webSearch", keys(p.webSearch))
add("connectors", keys(p.connectors))
add("parsers", keys(p.parsers))
return m
}
@@ -134,6 +137,7 @@ func (p *Plugin) Handler() http.Handler {
)
p.routeWebSearch(mux)
p.routeConnectors(mux)
p.routeParsers(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))
})
+48
View File
@@ -0,0 +1,48 @@
package pluginapi
import (
"encoding/json"
"time"
)
// Host API: how a plugin calls back into WeKnora. The URL and a short-lived
// bearer token arrive in Context.Host of each call; the token is bound to the
// tenant of that call and the scopes the plugin was granted (manifest
// permissions.hostApi). Errors use ErrorBody.
// HostKVPath is the key-value endpoint (scope "kv"): GET and DELETE take
// ?key=, PUT takes a KVPut body.
const HostKVPath = "/api/v1/plugin-host/kv"
// HostKVListPath lists keys: ?prefix=&after=&limit=.
const HostKVListPath = "/api/v1/plugin-host/kv/list"
// KVEntry is one stored value.
type KVEntry struct {
Key string `json:"key"`
Value json.RawMessage `json:"value"`
ExpiresAt *time.Time `json:"expiresAt,omitempty"`
UpdatedAt time.Time `json:"updatedAt"`
}
// KVPut stores a value; TTLSeconds 0 keeps it until deleted.
type KVPut struct {
Key string `json:"key"`
Value json.RawMessage `json:"value"`
TTLSeconds int `json:"ttlSeconds,omitempty"`
}
// KVList is a page of keys in key order; pass Next as after for the next
// page (empty when there is none).
type KVList struct {
Entries []KVEntry `json:"entries"`
Next string `json:"next,omitempty"`
}
// Limits of the key-value store.
const (
KVMaxKeyBytes = 256
KVMaxValueBytes = 64 << 10
KVMaxKeys = 10000 // per plugin and tenant
KVMaxListLimit = 1000
)
+50
View File
@@ -162,6 +162,33 @@ paths:
application/x-ndjson:
schema: { $ref: "#/components/schemas/Event" }
default: { $ref: "#/components/responses/Error" }
/v1/parsers/{id}/parse:
post:
summary: Parse a document into Markdown (DocReader.Read)
description: |
The file arrives as base64 content (or a URL). Answer Markdown and
the images it references; WeKnora stores the images and chunks the
text. A non-retryable error fails the document for good; a
retryable one (unavailable, rate_limited) is retried later.
parameters: [{ $ref: "#/components/parameters/ContributionID" }]
requestBody:
required: true
content:
application/json:
schema:
allOf:
- $ref: "#/components/schemas/Envelope"
- properties: { input: { $ref: "#/components/schemas/ParseInput" } }
responses:
"200":
description: The parsed document
content:
application/json:
schema:
type: object
required: [output]
properties: { output: { $ref: "#/components/schemas/ParseOutput" } }
default: { $ref: "#/components/responses/Error" }
components:
securitySchemes:
hostToken: { type: http, scheme: bearer }
@@ -296,6 +323,29 @@ components:
metadata: { type: object, additionalProperties: { type: string } }
isDeleted: { type: boolean }
sourceResourceId: { type: string }
ParseInput:
type: object
properties:
fileName: { type: string }
fileType: { type: string, description: "Lower-case extension without the dot" }
content: { type: string, contentEncoding: base64 }
url: { type: string }
title: { type: string }
ParseOutput:
type: object
required: [markdown]
properties:
markdown: { type: string }
images:
type: array
items:
type: object
required: [originalRef, data]
properties:
originalRef: { type: string }
mimeType: { type: string }
data: { type: string, contentEncoding: base64 }
metadata: { type: object, additionalProperties: { type: string } }
Event:
type: object
required: [type]
+34
View File
@@ -0,0 +1,34 @@
package pluginapi
// ParsePath is the parse endpoint of parser id.
func ParsePath(id string) string { return "/v1/parsers/" + id + "/parse" }
// ParseInput is one document to parse: its bytes (file mode) or a URL (URL
// mode). The parser's own settings arrive in Config.System and Config.Tenant.
type ParseInput struct {
FileName string `json:"fileName,omitempty"`
// FileType is the lower-case extension without the dot ("pdf").
FileType string `json:"fileType,omitempty"`
// Content is the file; JSON carries it as base64.
Content []byte `json:"content,omitempty"`
URL string `json:"url,omitempty"`
Title string `json:"title,omitempty"`
}
// ParseOutput is the document as Markdown. WeKnora chunks it; parsers do not.
type ParseOutput struct {
Markdown string `json:"markdown"`
// Images are the pictures the Markdown references: an image whose
// OriginalRef equals a ![](target) in Markdown is stored and the link
// rewritten. Inline data: URIs in the Markdown work too.
Images []ParsedImage `json:"images,omitempty"`
// Metadata such as "title" (URL mode) or "pages".
Metadata map[string]string `json:"metadata,omitempty"`
}
// ParsedImage is one image of a parsed document.
type ParsedImage struct {
OriginalRef string `json:"originalRef"`
MimeType string `json:"mimeType,omitempty"`
Data []byte `json:"data"`
}