feat: RAG 知识库真实数据 + 渠道端到端接线(并行∦wire.go)+ 主会话收口
【渠道端到端·接线(agent-72)】 - 新建 internal/services/channelbridge/:桥接服务(Start 对配置渠道 Connect(10s超时/warn)+SetMessageHandler+ReadMessages; 入站 UnifiedMessage→sessions.GetOrCreate(ResolveActor 白名单结构性识别)→loop.Run→SendMessage 回执,chat 文本)。 - cmd/cyrene/channels.go:从 CHANNEL_* env 读 4 渠道凭据(telegram/qq_obv11/weixin_ilink/webhook)。 - wire.go:app 加 channelBridge 字段,wireChannels 构造适配器+bridge.Start(),close() 收尾。 - 无真凭据降级:无 CHANNEL_* → configured()false 跳过不 dial;Connect 失败 Warn 不阻启动; webhook 仅入站(回执 ErrUnsupported 记 debug)。身份识别绝不靠提示词。收发端到端待凭据验证。 【RAG 知识库(agent-71)】 - internal/runtime/knowledge/:真实数据源(doc=资料/记忆=经历 分离 design/03 §3③)。 ListDocuments 走 projectindex.ListProjects(+ListChapters)、Search 走 projectindex.Search(标题/描述+章节 命中行+上下文)+retriever.Retrieve(配 vector 真语义,命中标 semantic:true)。 - gateway/server.go:RagStore 接口+Deps.RAG 可空+3 路由(GET /admin/rag、POST /admin/rag/query、/delete)。 - admin.go:adminListRag/adminSearchRag(空 query 400)/adminDeleteRag(写 501 not_implemented,M10)。 - projectindex/store.go:ListProjects/ListChapters。 - 前端 Rag.tsx:docs 从 GET /admin/rag 拉、检索走 POST /admin/rag/query、删除调真端点(501 显示 Notice), 上传/分块/嵌入显示 M10 Notice;api.ts/types.ts 加 adminListRag/adminSearchRag/adminDeleteRag。 【主会话收口 wire.go】projStore 提到 app struct+wireMemory 赋值;knowledge.New(...) 注入 gateway.Deps.RAG。 实机验证:admin/rag 200(documents[] 空表非壳)、channels 枚举 4 渠道、health ok。 go build/vet/test 全绿 + 前端 tsc。wire.go 567 行。 遗留(记录):① 群聊/陌生人 MemWritable 记忆隔离未接 loop(AGENTS.md 红线,待修) ② webhook InboundHandler 需 gateway 挂载 ③ RAG 语义路径 persona-scoped(project 记忆其他 persona 不浮现)。
This commit is contained in:
1 parent
518bd0a230
commit
dbcf3df8a3
13 files changed
+1510
-104
No files matched your search
@@ -0,0 +1,380 @@
|
||||
// Package channelbridge wires the registered IM channel adapters into the Cyrene
|
||||
// runtime: it connects each configured adapter, installs the inbound message
|
||||
// handler, and routes every normalized inbound message through the
|
||||
// channel→session→actor→loop→reply pipeline (DevDocs/design/08 §0-§2).
|
||||
//
|
||||
// The bridge is deliberately generic over adapter.PlatformAdapter. It owns the
|
||||
// per-platform Connect lifecycle and the inbound→reply path; the concrete
|
||||
// adapters themselves (and their configuration, sourced from env) are assembled
|
||||
// in cmd/cyrene/wire.go and handed in via Add. It never reads message text to
|
||||
// decide identity — that is ResolveActor's job (actor metadata + owner whitelist,
|
||||
// a project red line).
|
||||
package channelbridge
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/channel/session"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/logger"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/loop"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/user"
|
||||
)
|
||||
|
||||
// Knob constants. Compile-time so no bare magic number appears in the run path
|
||||
// (project rule §10).
|
||||
const (
|
||||
// connectTimeout bounds a single adapter Connect so a hung dial cannot stall
|
||||
// boot (a failed connect logs a warning and is skipped, never fatal).
|
||||
connectTimeout = 10 * time.Second
|
||||
// maxConcurrent bounds how many inbound messages run the agent loop at once,
|
||||
// so a burst of channel messages cannot spawn unbounded goroutines.
|
||||
maxConcurrent = 16
|
||||
)
|
||||
|
||||
// ErrNilLoop is returned when the bridge is built without a chat runner.
|
||||
var ErrNilLoop = errors.New("channelbridge: nil loop runner")
|
||||
|
||||
// ErrNilSessions is returned when the bridge is built without a session store.
|
||||
var ErrNilSessions = errors.New("channelbridge: nil session store")
|
||||
|
||||
// Chatter runs a single agent turn. *loop.Loop is the production implementation;
|
||||
// tests inject a stub.
|
||||
type Chatter interface {
|
||||
Run(ctx context.Context, sessionID string, messages []model.Message) (*loop.Result, error)
|
||||
}
|
||||
|
||||
// SessionStore resolves the channel↔session binding for an inbound message.
|
||||
// *session.SessionManager is the production implementation.
|
||||
type SessionStore interface {
|
||||
GetOrCreate(ctx context.Context, msg adapter.UnifiedMessage, owner *user.OwnerProfile) (session.Session, error)
|
||||
}
|
||||
|
||||
// Options carries the bridge's injected dependencies. Loop and Sessions are
|
||||
// required; Owner may be nil (then no account is recognized as owner, the safe
|
||||
// default-deny posture from design/11 §0).
|
||||
type Options struct {
|
||||
Log *slog.Logger
|
||||
Loop Chatter
|
||||
Sessions SessionStore
|
||||
Owner *user.OwnerProfile
|
||||
}
|
||||
|
||||
// Bridge owns the channel adapter lifecycle and the inbound message pipeline. It
|
||||
// is safe for concurrent use.
|
||||
type Bridge struct {
|
||||
log *slog.Logger
|
||||
run Chatter
|
||||
sessions SessionStore
|
||||
owner *user.OwnerProfile
|
||||
|
||||
mu sync.RWMutex
|
||||
adapters map[string]adapter.PlatformAdapter
|
||||
connected map[string]bool
|
||||
closed bool
|
||||
|
||||
ctx context.Context
|
||||
cancel context.CancelFunc
|
||||
wg sync.WaitGroup
|
||||
|
||||
sem chan struct{}
|
||||
}
|
||||
|
||||
// New builds a Bridge from opts. It returns an error when a required dependency
|
||||
// is missing so the caller fails fast rather than silently dropping messages.
|
||||
func New(opts Options) (*Bridge, error) {
|
||||
if opts.Loop == nil {
|
||||
return nil, ErrNilLoop
|
||||
}
|
||||
if opts.Sessions == nil {
|
||||
return nil, ErrNilSessions
|
||||
}
|
||||
log := opts.Log
|
||||
if log == nil {
|
||||
log = logger.New(logger.Options{Out: nil})
|
||||
}
|
||||
return &Bridge{
|
||||
log: logger.WithSource(log, "services/channelbridge"),
|
||||
run: opts.Loop,
|
||||
sessions: opts.Sessions,
|
||||
owner: opts.Owner,
|
||||
adapters: make(map[string]adapter.PlatformAdapter),
|
||||
connected: make(map[string]bool),
|
||||
cancel: func() {},
|
||||
sem: make(chan struct{}, maxConcurrent),
|
||||
}, nil
|
||||
}
|
||||
|
||||
// Add registers an adapter by its Name. It is safe to call before Start; adding
|
||||
// after Start has no effect on already-running adapters.
|
||||
func (b *Bridge) Add(a adapter.PlatformAdapter) {
|
||||
if a == nil {
|
||||
return
|
||||
}
|
||||
b.mu.Lock()
|
||||
defer b.mu.Unlock()
|
||||
b.adapters[a.Name()] = a
|
||||
}
|
||||
|
||||
// Start connects every configured adapter, installs the inbound handler, and
|
||||
// launches the read loops. Adapters without credentials (or whose Connect fails)
|
||||
// are logged and left disconnected; they never fail startup. Start blocks only
|
||||
// long enough to connect (bounded by connectTimeout per adapter) and returns.
|
||||
func (b *Bridge) Start() {
|
||||
b.mu.Lock()
|
||||
if b.closed {
|
||||
b.mu.Unlock()
|
||||
return
|
||||
}
|
||||
b.ctx, b.cancel = context.WithCancel(context.Background())
|
||||
b.mu.Unlock()
|
||||
|
||||
for _, a := range b.snapshot() {
|
||||
if !b.configured(a) {
|
||||
b.log.Warn("channel skipped: no credentials", "channel", a.Name())
|
||||
continue
|
||||
}
|
||||
connCtx, cancel := context.WithTimeout(context.Background(), connectTimeout)
|
||||
err := a.Connect(connCtx)
|
||||
cancel()
|
||||
if err != nil {
|
||||
b.log.Warn("channel connect failed, leaving disconnected",
|
||||
"channel", a.Name(), "error", err)
|
||||
continue
|
||||
}
|
||||
b.mu.Lock()
|
||||
b.connected[a.Name()] = true
|
||||
b.mu.Unlock()
|
||||
|
||||
a.SetMessageHandler(b.onMessage)
|
||||
b.wg.Add(1)
|
||||
go b.readLoop(a)
|
||||
b.log.Info("channel connected", "channel", a.Name())
|
||||
}
|
||||
b.log.Info("channel bridge started", "connected", b.ConnectedCount(),
|
||||
"registered", len(b.snapshot()))
|
||||
}
|
||||
|
||||
// Stop cancels the read loops, disconnects every adapter and waits for in-flight
|
||||
// message processing to finish. It is idempotent.
|
||||
func (b *Bridge) Stop(ctx context.Context) {
|
||||
b.mu.Lock()
|
||||
if b.closed {
|
||||
b.mu.Unlock()
|
||||
return
|
||||
}
|
||||
b.closed = true
|
||||
cancel := b.cancel
|
||||
b.mu.Unlock()
|
||||
|
||||
cancel()
|
||||
for name, a := range b.snapshot() {
|
||||
if err := a.Disconnect(ctx); err != nil {
|
||||
b.log.Warn("channel disconnect", "channel", name, "error", err)
|
||||
}
|
||||
}
|
||||
b.wg.Wait()
|
||||
b.log.Info("channel bridge stopped")
|
||||
}
|
||||
|
||||
// IsConnected reports whether the named adapter is currently connected.
|
||||
func (b *Bridge) IsConnected(name string) bool {
|
||||
b.mu.RLock()
|
||||
defer b.mu.RUnlock()
|
||||
return b.connected[name]
|
||||
}
|
||||
|
||||
// ConnectedCount reports how many adapters are currently connected.
|
||||
func (b *Bridge) ConnectedCount() int {
|
||||
b.mu.RLock()
|
||||
defer b.mu.RUnlock()
|
||||
n := 0
|
||||
for _, ok := range b.connected {
|
||||
if ok {
|
||||
n++
|
||||
}
|
||||
}
|
||||
return n
|
||||
}
|
||||
|
||||
// snapshot returns a copy of the registered adapters for a lock-free iteration.
|
||||
func (b *Bridge) snapshot() []adapter.PlatformAdapter {
|
||||
b.mu.RLock()
|
||||
defer b.mu.RUnlock()
|
||||
out := make([]adapter.PlatformAdapter, 0, len(b.adapters))
|
||||
for _, a := range b.adapters {
|
||||
out = append(out, a)
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// configured reports whether the adapter carries enough configuration to attempt
|
||||
// a Connect. It inspects the adapter's own Config() snapshot for the credential
|
||||
// each platform needs, so a bare (empty) adapter is skipped rather than dialed.
|
||||
// This is what keeps weixin-ilink from falling into its blocking QR-login flow
|
||||
// when no bot token is present.
|
||||
func (b *Bridge) configured(a adapter.PlatformAdapter) bool {
|
||||
cfg := a.Config()
|
||||
switch a.Name() {
|
||||
case "telegram", "weixin-ilink":
|
||||
return cfgStr(cfg, "bot_token") != ""
|
||||
case "qq-obv11":
|
||||
if cfgStr(cfg, "mode") == "client" {
|
||||
return cfgStr(cfg, "remoteURL") != ""
|
||||
}
|
||||
return cfgStr(cfg, "listenAddr") != ""
|
||||
case "webhook":
|
||||
// Inbound-only: no external credential is required to hold the session.
|
||||
return true
|
||||
default:
|
||||
return true
|
||||
}
|
||||
}
|
||||
|
||||
// readLoop runs an adapter's blocking read path until the bridge stops. The emit
|
||||
// sink is wired to onMessage so poll-style adapters (Telegram getUpdates, weixin
|
||||
// long poll) deliver through the same pipeline as push-style adapters.
|
||||
func (b *Bridge) readLoop(a adapter.PlatformAdapter) {
|
||||
defer b.wg.Done()
|
||||
ctx := b.ctx
|
||||
a.ReadMessages(ctx, func(msg adapter.UnifiedMessage) {
|
||||
b.onMessage(ctx, msg)
|
||||
})
|
||||
}
|
||||
|
||||
// onMessage is the adapter-facing inbound callback. It runs the pipeline in a
|
||||
// bounded goroutine so the adapter's read loop never blocks on a slow agent turn.
|
||||
func (b *Bridge) onMessage(ctx context.Context, msg adapter.UnifiedMessage) {
|
||||
b.mu.Lock()
|
||||
if b.closed {
|
||||
b.mu.Unlock()
|
||||
return
|
||||
}
|
||||
b.mu.Unlock()
|
||||
|
||||
if err := msg.Validate(); err != nil {
|
||||
b.log.Warn("inbound message invalid, dropped", "platform", msg.Platform, "error", err)
|
||||
return
|
||||
}
|
||||
|
||||
select {
|
||||
case b.sem <- struct{}{}:
|
||||
case <-ctx.Done():
|
||||
return
|
||||
}
|
||||
|
||||
// Add the worker under the same lock that Stop uses to set closed, so a
|
||||
// concurrent Stop either sees our Add before it waits, or we observe closed
|
||||
// and bail without leaking a goroutine.
|
||||
b.mu.Lock()
|
||||
if b.closed {
|
||||
b.mu.Unlock()
|
||||
<-b.sem
|
||||
return
|
||||
}
|
||||
b.wg.Add(1)
|
||||
b.mu.Unlock()
|
||||
|
||||
go func() {
|
||||
defer b.wg.Done()
|
||||
defer func() { <-b.sem }()
|
||||
b.process(ctx, msg)
|
||||
}()
|
||||
}
|
||||
|
||||
// process runs the full inbound pipeline: resolve the session/actor, drive the
|
||||
// agent loop with the chat text, and send the loop's reply back through the
|
||||
// channel. Identity is resolved structurally by the session layer; the bridge
|
||||
// never reads message content to decide who the sender is.
|
||||
func (b *Bridge) process(ctx context.Context, msg adapter.UnifiedMessage) {
|
||||
text := strings.TrimSpace(msg.Content)
|
||||
if text == "" {
|
||||
b.log.Debug("inbound message has no chat text, skipped",
|
||||
"platform", msg.Platform, "channel", msg.ChannelID)
|
||||
return
|
||||
}
|
||||
|
||||
sess, err := b.sessions.GetOrCreate(ctx, msg, b.owner)
|
||||
if err != nil {
|
||||
b.log.Warn("session resolve failed", "platform", msg.Platform,
|
||||
"channel", msg.ChannelID, "error", err)
|
||||
return
|
||||
}
|
||||
|
||||
b.log.Debug("inbound → session",
|
||||
"platform", msg.Platform, "channel", msg.ChannelID,
|
||||
"session", sess.SessionID, "actor_role", sess.ActorRole,
|
||||
"env", sess.Env(), "mem_writable", sess.MemWritable)
|
||||
|
||||
res, err := b.run.Run(ctx, sess.SessionID, []model.Message{
|
||||
model.TextMessage(model.RoleUser, text),
|
||||
})
|
||||
if err != nil {
|
||||
b.log.Warn("agent run failed", "platform", msg.Platform,
|
||||
"session", sess.SessionID, "error", err)
|
||||
return
|
||||
}
|
||||
|
||||
reply := strings.TrimSpace(res.Text)
|
||||
if reply == "" {
|
||||
b.log.Debug("no reply produced", "platform", msg.Platform, "session", sess.SessionID)
|
||||
return
|
||||
}
|
||||
if err := b.sendReply(ctx, msg, reply); err != nil {
|
||||
if errors.Is(err, adapter.ErrUnsupported) {
|
||||
b.log.Debug("channel does not support outbound reply, skipped",
|
||||
"platform", msg.Platform, "session", sess.SessionID)
|
||||
return
|
||||
}
|
||||
b.log.Warn("send reply failed", "platform", msg.Platform, "error", err)
|
||||
}
|
||||
}
|
||||
|
||||
// sendReply delivers the loop's text back to the channel's source conversation.
|
||||
// It mirrors the inbound message's channel targeting fields so the adapter
|
||||
// resolves the same peer the message came from.
|
||||
func (b *Bridge) sendReply(ctx context.Context, in adapter.UnifiedMessage, text string) error {
|
||||
a := b.adapter(in.Platform)
|
||||
if a == nil {
|
||||
return fmt.Errorf("channelbridge: no adapter for platform %q", in.Platform)
|
||||
}
|
||||
out := adapter.UnifiedMessage{
|
||||
Platform: in.Platform,
|
||||
ChannelType: in.ChannelType,
|
||||
ChannelID: in.ChannelID,
|
||||
GroupID: in.GroupID,
|
||||
GroupName: in.GroupName,
|
||||
SenderID: in.SenderID,
|
||||
Sender: in.Sender,
|
||||
Content: text,
|
||||
ContentType: adapter.ContentText,
|
||||
Timestamp: time.Now().UTC(),
|
||||
}
|
||||
return a.SendMessage(ctx, adapter.UnifiedResponse{Messages: []adapter.UnifiedMessage{out}})
|
||||
}
|
||||
|
||||
// adapter returns the registered adapter for a platform name, or nil.
|
||||
func (b *Bridge) adapter(platform string) adapter.PlatformAdapter {
|
||||
b.mu.RLock()
|
||||
defer b.mu.RUnlock()
|
||||
return b.adapters[platform]
|
||||
}
|
||||
|
||||
// cfgStr reads a string field from an adapter config map, returning "" when
|
||||
// absent.
|
||||
func cfgStr(m map[string]any, key string) string {
|
||||
if m == nil {
|
||||
return ""
|
||||
}
|
||||
if s, ok := m[key].(string); ok {
|
||||
return s
|
||||
}
|
||||
return ""
|
||||
}
|
||||
@@ -0,0 +1,209 @@
|
||||
package channelbridge
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/channel/session"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/loop"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/user"
|
||||
)
|
||||
|
||||
// fakeAdapter is a minimal PlatformAdapter that records outbound sends so the
|
||||
// bridge's reply path can be asserted without network I/O.
|
||||
type fakeAdapter struct {
|
||||
name string
|
||||
connected bool
|
||||
sends []adapter.UnifiedMessage
|
||||
handler adapter.MessageHandler
|
||||
sendErr error
|
||||
}
|
||||
|
||||
func (f *fakeAdapter) Name() string { return f.name }
|
||||
func (f *fakeAdapter) Config() map[string]any {
|
||||
return map[string]any{"bot_token": "configured"}
|
||||
}
|
||||
func (f *fakeAdapter) Connect(context.Context) error { f.connected = true; return nil }
|
||||
func (f *fakeAdapter) Disconnect(context.Context) error {
|
||||
f.connected = false
|
||||
return nil
|
||||
}
|
||||
func (f *fakeAdapter) IsConnected() bool { return f.connected }
|
||||
func (f *fakeAdapter) HealthCheck(context.Context) error { return nil }
|
||||
func (f *fakeAdapter) ReadMessages(ctx context.Context, _ func(adapter.UnifiedMessage)) {
|
||||
<-ctx.Done()
|
||||
}
|
||||
func (f *fakeAdapter) SendMessage(_ context.Context, resp adapter.UnifiedResponse) error {
|
||||
for _, m := range resp.Messages {
|
||||
f.sends = append(f.sends, m)
|
||||
}
|
||||
return f.sendErr
|
||||
}
|
||||
func (f *fakeAdapter) SendProactive(context.Context, adapter.UnifiedMessage) error { return nil }
|
||||
func (f *fakeAdapter) SetMessageHandler(fn adapter.MessageHandler) { f.handler = fn }
|
||||
func (f *fakeAdapter) Capabilities() adapter.Capabilities { return adapter.DefaultCapabilities() }
|
||||
func (f *fakeAdapter) ListChannels(context.Context) ([]adapter.ChannelInfo, error) { return nil, nil }
|
||||
|
||||
// stubChatter records the session id and messages passed to the agent loop.
|
||||
type stubChatter struct {
|
||||
sessionID string
|
||||
messages []model.Message
|
||||
res *loop.Result
|
||||
err error
|
||||
}
|
||||
|
||||
func (s *stubChatter) Run(_ context.Context, sessionID string, msgs []model.Message) (*loop.Result, error) {
|
||||
s.sessionID = sessionID
|
||||
s.messages = msgs
|
||||
return s.res, s.err
|
||||
}
|
||||
|
||||
// stubSessions records the inbound message and returns a fixed session.
|
||||
type stubSessions struct {
|
||||
msg adapter.UnifiedMessage
|
||||
owner *user.OwnerProfile
|
||||
sess session.Session
|
||||
err error
|
||||
}
|
||||
|
||||
func (s *stubSessions) GetOrCreate(_ context.Context, msg adapter.UnifiedMessage, owner *user.OwnerProfile) (session.Session, error) {
|
||||
s.msg = msg
|
||||
s.owner = owner
|
||||
if s.err != nil {
|
||||
return session.Session{}, s.err
|
||||
}
|
||||
return s.sess, nil
|
||||
}
|
||||
|
||||
func directMessage() adapter.UnifiedMessage {
|
||||
return adapter.UnifiedMessage{
|
||||
Platform: "telegram",
|
||||
ChannelID: "100",
|
||||
ChannelType: adapter.ChannelDirect,
|
||||
SenderID: "100",
|
||||
Content: "hello",
|
||||
ContentType: adapter.ContentText,
|
||||
}
|
||||
}
|
||||
|
||||
func newTestBridge(t *testing.T) (*Bridge, *stubChatter, *stubSessions, *fakeAdapter) {
|
||||
t.Helper()
|
||||
chat := &stubChatter{res: &loop.Result{Text: "hi back"}}
|
||||
sess := &stubSessions{sess: session.Session{SessionID: "sess-1", Owner: true, MemWritable: true}}
|
||||
ad := &fakeAdapter{name: "telegram"}
|
||||
|
||||
b, err := New(Options{Loop: chat, Sessions: sess, Owner: &user.OwnerProfile{}})
|
||||
if err != nil {
|
||||
t.Fatalf("New: %v", err)
|
||||
}
|
||||
b.Add(ad)
|
||||
return b, chat, sess, ad
|
||||
}
|
||||
|
||||
func TestProcessRunsPipelineAndReplies(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
b, chat, sess, ad := newTestBridge(t)
|
||||
ctx := context.Background()
|
||||
|
||||
b.process(ctx, directMessage())
|
||||
|
||||
if sess.msg.SenderID != "100" {
|
||||
t.Errorf("resolved sender = %q, want 100", sess.msg.SenderID)
|
||||
}
|
||||
if chat.sessionID != "sess-1" {
|
||||
t.Errorf("loop session id = %q, want sess-1", chat.sessionID)
|
||||
}
|
||||
if len(chat.messages) != 1 || chat.messages[0].Role != model.RoleUser {
|
||||
t.Fatalf("loop messages = %+v, want single user message", chat.messages)
|
||||
}
|
||||
if got := chat.messages[0].Content[0].Text; got != "hello" {
|
||||
t.Errorf("loop text = %q, want hello", got)
|
||||
}
|
||||
if len(ad.sends) != 1 {
|
||||
t.Fatalf("send count = %d, want 1", len(ad.sends))
|
||||
}
|
||||
reply := ad.sends[0]
|
||||
if reply.Content != "hi back" || reply.ContentType != adapter.ContentText {
|
||||
t.Errorf("reply = %+v, want text 'hi back'", reply)
|
||||
}
|
||||
if reply.ChannelID != "100" || reply.Platform != "telegram" {
|
||||
t.Errorf("reply target = %+v, want platform telegram channel 100", reply)
|
||||
}
|
||||
}
|
||||
|
||||
func TestProcessSkipsEmptyText(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
b, chat, _, ad := newTestBridge(t)
|
||||
msg := directMessage()
|
||||
msg.Content = " "
|
||||
|
||||
b.process(context.Background(), msg)
|
||||
|
||||
if chat.sessionID != "" {
|
||||
t.Errorf("loop ran for empty content, session = %q", chat.sessionID)
|
||||
}
|
||||
if len(ad.sends) != 0 {
|
||||
t.Errorf("send count = %d, want 0 for empty content", len(ad.sends))
|
||||
}
|
||||
}
|
||||
|
||||
func TestProcessLoopErrorDoesNotSend(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
b, chat, _, ad := newTestBridge(t)
|
||||
chat.res = nil
|
||||
chat.err = context.DeadlineExceeded
|
||||
|
||||
b.process(context.Background(), directMessage())
|
||||
|
||||
if chat.sessionID != "sess-1" {
|
||||
t.Errorf("loop should still run for valid text, session = %q", chat.sessionID)
|
||||
}
|
||||
if len(ad.sends) != 0 {
|
||||
t.Errorf("send count = %d, want 0 on loop error", len(ad.sends))
|
||||
}
|
||||
}
|
||||
|
||||
func TestProcessGroupMessageKeepsGroupTarget(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
b, chat, _, ad := newTestBridge(t)
|
||||
msg := adapter.UnifiedMessage{
|
||||
Platform: "telegram",
|
||||
ChannelID: "-20",
|
||||
ChannelType: adapter.ChannelGroup,
|
||||
GroupID: "-20",
|
||||
GroupName: "Gang",
|
||||
SenderID: "9",
|
||||
Content: "hello all",
|
||||
ContentType: adapter.ContentText,
|
||||
}
|
||||
chat.res = &loop.Result{Text: "group ack"}
|
||||
|
||||
b.process(context.Background(), msg)
|
||||
|
||||
if len(ad.sends) != 1 {
|
||||
t.Fatalf("send count = %d, want 1", len(ad.sends))
|
||||
}
|
||||
reply := ad.sends[0]
|
||||
if reply.ChannelType != adapter.ChannelGroup || reply.GroupID != "-20" {
|
||||
t.Errorf("group reply target = %+v, want group -20", reply)
|
||||
}
|
||||
}
|
||||
|
||||
func TestProcessEmptyReplySendsNothing(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
b, chat, _, ad := newTestBridge(t)
|
||||
chat.res = &loop.Result{Text: " "}
|
||||
|
||||
b.process(context.Background(), directMessage())
|
||||
|
||||
if len(ad.sends) != 0 {
|
||||
t.Errorf("send count = %d, want 0 for empty reply", len(ad.sends))
|
||||
}
|
||||
}
|
||||
@@ -6,11 +6,13 @@ import (
|
||||
"errors"
|
||||
"net/http"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/auth"
|
||||
capreg "git.yeij.top/AskaEth/Cyrene_v2/internal/capability/registry"
|
||||
apperr "git.yeij.top/AskaEth/Cyrene_v2/internal/errors"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/knowledge"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/pkg/plugins/sdk"
|
||||
)
|
||||
|
||||
@@ -251,6 +253,150 @@ func (s *Server) adminListBackthink(w http.ResponseWriter, r *http.Request) {
|
||||
writeJSON(w, http.StatusOK, out)
|
||||
}
|
||||
|
||||
// AdminRagDoc is the wire shape of one knowledge-base document (project-index
|
||||
// entry) in the admin RAG list.
|
||||
type AdminRagDoc struct {
|
||||
ID string `json:"id"`
|
||||
Name string `json:"name"`
|
||||
Desc string `json:"desc,omitempty"`
|
||||
Chunks int `json:"chunks"`
|
||||
Embedded bool `json:"embedded"`
|
||||
}
|
||||
|
||||
// AdminRagHit is the wire shape of one RAG retrieval hit.
|
||||
type AdminRagHit struct {
|
||||
DocID string `json:"docId"`
|
||||
DocName string `json:"docName"`
|
||||
Chapter string `json:"chapter,omitempty"`
|
||||
Text string `json:"text"`
|
||||
Score float64 `json:"score"`
|
||||
Semantic bool `json:"semantic"`
|
||||
}
|
||||
|
||||
// AdminRagResponse is the wire shape of GET /api/v1/admin/rag. Hits is present
|
||||
// only when a query parameter was supplied, so callers distinguish "no search
|
||||
// requested" from "searched and found nothing".
|
||||
type AdminRagResponse struct {
|
||||
Documents []AdminRagDoc `json:"documents"`
|
||||
Hits []AdminRagHit `json:"hits,omitempty"`
|
||||
}
|
||||
|
||||
// ragQueryRequest is the POST /api/v1/admin/rag/query body.
|
||||
type ragQueryRequest struct {
|
||||
Query string `json:"query"`
|
||||
}
|
||||
|
||||
// ragDeleteRequest is the POST /api/v1/admin/rag/delete body.
|
||||
type ragDeleteRequest struct {
|
||||
ID string `json:"id"`
|
||||
}
|
||||
|
||||
// adminListRag lists the knowledge-base documents and, when ?query= is present,
|
||||
// returns the matching hits too. Without a wired RagStore it degrades to an empty
|
||||
// list (backward compatible — never a 500).
|
||||
func (s *Server) adminListRag(w http.ResponseWriter, r *http.Request) {
|
||||
out := AdminRagResponse{Documents: make([]AdminRagDoc, 0)}
|
||||
if s.rag == nil {
|
||||
writeJSON(w, http.StatusOK, out)
|
||||
return
|
||||
}
|
||||
docs, err := s.rag.ListDocuments(r.Context())
|
||||
if err != nil {
|
||||
s.log.Error("list rag", "err", err)
|
||||
writeError(w, apperr.Wrap(apperr.ErrInternal, "rag_failed", "failed to list knowledge documents"))
|
||||
return
|
||||
}
|
||||
out.Documents = ragDocsFrom(docs)
|
||||
if query := strings.TrimSpace(r.URL.Query().Get("query")); query != "" {
|
||||
hits, herr := s.rag.Search(r.Context(), query)
|
||||
if herr != nil {
|
||||
s.log.Error("search rag", "err", herr)
|
||||
writeError(w, apperr.Wrap(apperr.ErrInternal, "rag_search_failed", "failed to search knowledge base"))
|
||||
return
|
||||
}
|
||||
out.Hits = ragHitsFrom(hits)
|
||||
}
|
||||
writeJSON(w, http.StatusOK, out)
|
||||
}
|
||||
|
||||
// adminSearchRag runs a knowledge-base query and returns the matching hits.
|
||||
func (s *Server) adminSearchRag(w http.ResponseWriter, r *http.Request) {
|
||||
if s.rag == nil {
|
||||
s.auditClaims(r, "rag.query", "error")
|
||||
writeError(w, &apperr.AppError{Code: "not_implemented", Message: "知识库 RAG 服务未配置(未注入 RagStore)", HTTPCode: http.StatusNotImplemented})
|
||||
return
|
||||
}
|
||||
var req ragQueryRequest
|
||||
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
|
||||
writeError(w, apperr.Wrap(apperr.ErrBadRequest, "bad_request", "invalid request body"))
|
||||
return
|
||||
}
|
||||
if strings.TrimSpace(req.Query) == "" {
|
||||
writeError(w, apperr.Wrap(apperr.ErrBadRequest, "bad_request", "query is required"))
|
||||
return
|
||||
}
|
||||
hits, err := s.rag.Search(r.Context(), req.Query)
|
||||
if err != nil {
|
||||
s.log.Error("search rag", "err", err)
|
||||
writeError(w, apperr.Wrap(apperr.ErrInternal, "rag_search_failed", "failed to search knowledge base"))
|
||||
return
|
||||
}
|
||||
s.auditClaims(r, "rag.query", "ok")
|
||||
writeJSON(w, http.StatusOK, map[string]any{"query": req.Query, "hits": ragHitsFrom(hits)})
|
||||
}
|
||||
|
||||
// adminDeleteRag answers the document delete honestly: write operations
|
||||
// (upload/chunk/embed/delete) are milestone M10, so it returns a clear 501 rather
|
||||
// than a fabricated success. The route is audited as an admin write.
|
||||
func (s *Server) adminDeleteRag(w http.ResponseWriter, r *http.Request) {
|
||||
if s.rag == nil {
|
||||
s.auditClaims(r, "rag.delete", "error")
|
||||
writeError(w, &apperr.AppError{Code: "not_implemented", Message: "知识库 RAG 服务未配置(未注入 RagStore)", HTTPCode: http.StatusNotImplemented})
|
||||
return
|
||||
}
|
||||
var req ragDeleteRequest
|
||||
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
|
||||
writeError(w, apperr.Wrap(apperr.ErrBadRequest, "bad_request", "invalid request body"))
|
||||
return
|
||||
}
|
||||
if strings.TrimSpace(req.ID) == "" {
|
||||
writeError(w, apperr.Wrap(apperr.ErrBadRequest, "bad_request", "id is required"))
|
||||
return
|
||||
}
|
||||
if err := s.rag.DeleteDocument(r.Context(), req.ID); err != nil {
|
||||
s.auditClaims(r, "rag.delete", "error")
|
||||
writeError(w, &apperr.AppError{Code: "not_implemented", Message: "知识库写操作(上传/分块/嵌入/删除)由 M10 后端提供", HTTPCode: http.StatusNotImplemented})
|
||||
return
|
||||
}
|
||||
s.auditClaims(r, "rag.delete", "ok")
|
||||
writeJSON(w, http.StatusOK, map[string]any{"id": req.ID})
|
||||
}
|
||||
|
||||
// ragDocsFrom maps knowledge docs to their admin wire shape.
|
||||
func ragDocsFrom(docs []knowledge.Doc) []AdminRagDoc {
|
||||
out := make([]AdminRagDoc, 0, len(docs))
|
||||
for _, d := range docs {
|
||||
out = append(out, AdminRagDoc{ID: d.ID, Name: d.Name, Desc: d.Desc, Chunks: d.Chunks, Embedded: d.Embedded})
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// ragHitsFrom maps knowledge hits to their admin wire shape.
|
||||
func ragHitsFrom(hits []knowledge.Hit) []AdminRagHit {
|
||||
out := make([]AdminRagHit, 0, len(hits))
|
||||
for _, h := range hits {
|
||||
out = append(out, AdminRagHit{
|
||||
DocID: h.DocID,
|
||||
DocName: h.DocName,
|
||||
Chapter: h.Chapter,
|
||||
Text: h.Text,
|
||||
Score: h.Score,
|
||||
Semantic: h.Semantic,
|
||||
})
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// adminSetPolicy upserts the reachability markers for a capability. It is the
|
||||
// audited admin write example (owner-only, enforced by the route registration).
|
||||
func (s *Server) adminSetPolicy(w http.ResponseWriter, r *http.Request) {
|
||||
|
||||
@@ -22,6 +22,7 @@ import (
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/config"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/iot"
|
||||
personareg "git.yeij.top/AskaEth/Cyrene_v2/internal/persona/registry"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/knowledge"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/loop"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/memory"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model"
|
||||
@@ -188,6 +189,21 @@ type stubMemories struct {
|
||||
|
||||
func (s *stubMemories) List(ctx context.Context) ([]memory.MemoryItem, error) { return s.list, s.err }
|
||||
|
||||
// stubRag mimics RagStore, returning domain docs/hits the gateway maps to its
|
||||
// admin wire shape.
|
||||
type stubRag struct {
|
||||
docs []knowledge.Doc
|
||||
hits []knowledge.Hit
|
||||
err error
|
||||
}
|
||||
|
||||
func (s *stubRag) ListDocuments(ctx context.Context) ([]knowledge.Doc, error) { return s.docs, s.err }
|
||||
func (s *stubRag) Search(ctx context.Context, query string) ([]knowledge.Hit, error) {
|
||||
return s.hits, s.err
|
||||
}
|
||||
func (s *stubRag) IndexDocument(ctx context.Context, doc knowledge.Doc) error { return s.err }
|
||||
func (s *stubRag) DeleteDocument(ctx context.Context, id string) error { return s.err }
|
||||
|
||||
// stubIoT mimics IoTStore.
|
||||
type stubIoT struct {
|
||||
list []iot.Device
|
||||
@@ -1219,3 +1235,132 @@ func reqWithOrigin(origin string) *http.Request {
|
||||
func reqWithQuery(k, v string) *http.Request {
|
||||
return httptest.NewRequest(http.MethodGet, "/ws/chat?"+k+"="+v, nil)
|
||||
}
|
||||
|
||||
// --- RAG knowledge base ----------------------------------------------------
|
||||
|
||||
func TestAdminRagListsAndSearches(t *testing.T) {
|
||||
t.Parallel()
|
||||
docs := []knowledge.Doc{{ID: "p1", Name: "开发时间线.md", Desc: "milestones", Chunks: 3, Embedded: true}}
|
||||
hits := []knowledge.Hit{{DocID: "p1", DocName: "开发时间线.md", Chapter: "M4", Text: "RAG 检索测试", Score: 1.0}}
|
||||
srv, ts, jwt := buildEnv(t, cfgDefault(), Deps{RAG: &stubRag{docs: docs, hits: hits}})
|
||||
_ = srv
|
||||
tok := signAny(t, jwt, "acct", auth.RoleOwner)
|
||||
|
||||
// List only.
|
||||
resp, body := doReq(t, http.MethodGet, ts.URL+"/api/v1/admin/rag", nil, tok)
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
t.Fatalf("list status = %d, want 200 (body=%s)", resp.StatusCode, body)
|
||||
}
|
||||
var list struct {
|
||||
Documents []AdminRagDoc `json:"documents"`
|
||||
Hits []AdminRagHit `json:"hits"`
|
||||
}
|
||||
if err := json.Unmarshal(decodeBody(t, body)["data"], &list); err != nil {
|
||||
t.Fatalf("decode list: %v (body=%s)", err, body)
|
||||
}
|
||||
if len(list.Documents) != 1 || list.Documents[0].Name != "开发时间线.md" {
|
||||
t.Fatalf("documents = %+v, want one 开发时间线.md", list.Documents)
|
||||
}
|
||||
if list.Hits != nil {
|
||||
t.Fatalf("hits = %+v, want omitted without query", list.Hits)
|
||||
}
|
||||
|
||||
// List + search via ?query=.
|
||||
resp2, body2 := doReq(t, http.MethodGet, ts.URL+"/api/v1/admin/rag?query=rag", nil, tok)
|
||||
if resp2.StatusCode != http.StatusOK {
|
||||
t.Fatalf("search status = %d, want 200 (body=%s)", resp2.StatusCode, body2)
|
||||
}
|
||||
var search struct {
|
||||
Documents []AdminRagDoc `json:"documents"`
|
||||
Hits []AdminRagHit `json:"hits"`
|
||||
}
|
||||
if err := json.Unmarshal(decodeBody(t, body2)["data"], &search); err != nil {
|
||||
t.Fatalf("decode search: %v (body=%s)", err, body2)
|
||||
}
|
||||
if len(search.Hits) != 1 || search.Hits[0].DocName != "开发时间线.md" {
|
||||
t.Fatalf("hits = %+v, want one hit", search.Hits)
|
||||
}
|
||||
}
|
||||
|
||||
func TestAdminRagQueryPost(t *testing.T) {
|
||||
t.Parallel()
|
||||
hits := []knowledge.Hit{{DocID: "p1", DocName: "开发时间线.md", Text: "RAG 检索测试", Score: 1.0}}
|
||||
srv, ts, jwt := buildEnv(t, cfgDefault(), Deps{RAG: &stubRag{hits: hits}})
|
||||
_ = srv
|
||||
tok := signAny(t, jwt, "acct", auth.RoleOwner)
|
||||
|
||||
resp, body := doReq(t, http.MethodPost, ts.URL+"/api/v1/admin/rag/query", map[string]any{"query": "rag"}, tok)
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
t.Fatalf("status = %d, want 200 (body=%s)", resp.StatusCode, body)
|
||||
}
|
||||
var res struct {
|
||||
Query string `json:"query"`
|
||||
Hits []AdminRagHit `json:"hits"`
|
||||
}
|
||||
if err := json.Unmarshal(decodeBody(t, body)["data"], &res); err != nil {
|
||||
t.Fatalf("decode: %v (body=%s)", err, body)
|
||||
}
|
||||
if res.Query != "rag" || len(res.Hits) != 1 {
|
||||
t.Fatalf("response = %+v, want query=rag with 1 hit", res)
|
||||
}
|
||||
}
|
||||
|
||||
func TestAdminRagEmptyWhenUnwired(t *testing.T) {
|
||||
t.Parallel()
|
||||
srv, ts, jwt := buildEnv(t, cfgDefault(), Deps{})
|
||||
_ = srv
|
||||
tok := signAny(t, jwt, "acct", auth.RoleOwner)
|
||||
|
||||
// List degrades to empty (never a 500).
|
||||
resp, body := doReq(t, http.MethodGet, ts.URL+"/api/v1/admin/rag", nil, tok)
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
t.Fatalf("list status = %d, want 200 (body=%s)", resp.StatusCode, body)
|
||||
}
|
||||
var list struct {
|
||||
Documents []AdminRagDoc `json:"documents"`
|
||||
}
|
||||
if err := json.Unmarshal(decodeBody(t, body)["data"], &list); err != nil {
|
||||
t.Fatalf("decode: %v (body=%s)", err, body)
|
||||
}
|
||||
if len(list.Documents) != 0 {
|
||||
t.Fatalf("documents = %+v, want empty", list.Documents)
|
||||
}
|
||||
|
||||
// Query reports a clear 501 (backend not implemented).
|
||||
resp2, body2 := doReq(t, http.MethodPost, ts.URL+"/api/v1/admin/rag/query", map[string]any{"query": "rag"}, tok)
|
||||
if resp2.StatusCode != http.StatusNotImplemented {
|
||||
t.Fatalf("query status = %d, want 501 (body=%s)", resp2.StatusCode, body2)
|
||||
}
|
||||
wantErrorShape(t, body2)
|
||||
if !strings.Contains(string(body2), "not_implemented") {
|
||||
t.Fatalf("error body = %s, want not_implemented code", body2)
|
||||
}
|
||||
}
|
||||
|
||||
func TestAdminRagDeleteNotImplemented(t *testing.T) {
|
||||
t.Parallel()
|
||||
srv, ts, jwt := buildEnv(t, cfgDefault(), Deps{RAG: &stubRag{err: knowledge.ErrWriteUnsupported}})
|
||||
_ = srv
|
||||
tok := signAny(t, jwt, "acct", auth.RoleOwner)
|
||||
|
||||
resp, body := doReq(t, http.MethodPost, ts.URL+"/api/v1/admin/rag/delete", map[string]any{"id": "p1"}, tok)
|
||||
if resp.StatusCode != http.StatusNotImplemented {
|
||||
t.Fatalf("status = %d, want 501 (body=%s)", resp.StatusCode, body)
|
||||
}
|
||||
wantErrorShape(t, body)
|
||||
if !strings.Contains(string(body), "M10") {
|
||||
t.Fatalf("error body = %s, want M10 write message", body)
|
||||
}
|
||||
}
|
||||
|
||||
func TestAdminRagDeleteRequiresID(t *testing.T) {
|
||||
t.Parallel()
|
||||
srv, ts, jwt := buildEnv(t, cfgDefault(), Deps{RAG: &stubRag{}})
|
||||
_ = srv
|
||||
tok := signAny(t, jwt, "acct", auth.RoleOwner)
|
||||
|
||||
resp, _ := doReq(t, http.MethodPost, ts.URL+"/api/v1/admin/rag/delete", map[string]any{"id": " "}, tok)
|
||||
if resp.StatusCode != http.StatusBadRequest {
|
||||
t.Fatalf("status = %d, want 400", resp.StatusCode)
|
||||
}
|
||||
}
|
||||
@@ -35,6 +35,7 @@ import (
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/iot"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/logger"
|
||||
personareg "git.yeij.top/AskaEth/Cyrene_v2/internal/persona/registry"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/knowledge"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/loop"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/memory"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model"
|
||||
@@ -132,6 +133,23 @@ type BackthinkStore interface {
|
||||
List(ctx context.Context, limit int) ([]backthink.Record, error)
|
||||
}
|
||||
|
||||
// RagStore backs the admin knowledge-base RAG module (design/10 §3.12 module 8).
|
||||
// *runtime/knowledge.Store is the production implementation. Every method is
|
||||
// read-only for this milestone: IndexDocument/DeleteDocument return a clear
|
||||
// "write lands in M10" error rather than a fabricated success.
|
||||
type RagStore interface {
|
||||
// ListDocuments enumerates the knowledge-base documents (project-index
|
||||
// entries).
|
||||
ListDocuments(ctx context.Context) ([]knowledge.Doc, error)
|
||||
// Search answers a knowledge-base query (line-level hierarchy + optional
|
||||
// semantic/vector ranking).
|
||||
Search(ctx context.Context, query string) ([]knowledge.Hit, error)
|
||||
// IndexDocument uploads/chunks/embeds a document. M10; returns an error.
|
||||
IndexDocument(ctx context.Context, doc knowledge.Doc) error
|
||||
// DeleteDocument removes a document. M10; returns an error.
|
||||
DeleteDocument(ctx context.Context, id string) error
|
||||
}
|
||||
|
||||
// Deps carries the gateway's injected dependencies. Every field is nil-tolerant
|
||||
// except Auth, which is required (authentication backs login and every RBAC
|
||||
// decision). Registry/Loop are satisfied by the concrete *capability.Registry and
|
||||
@@ -170,6 +188,9 @@ type Deps struct {
|
||||
Users UserStore
|
||||
// Backthink lists background-think records. Optional.
|
||||
Backthink BackthinkStore
|
||||
// RAG backs the admin knowledge-base RAG module. Optional; list/search
|
||||
// return empty when unset, and write endpoints report a clear 501.
|
||||
RAG RagStore
|
||||
// Streaming marks the injected Loop as streaming-capable (its OnStream is
|
||||
// wired to publish ws.stream_chunk). When set the hub skips its own single
|
||||
// final stream_chunk emission, since deltas already stream through OnStream.
|
||||
@@ -197,6 +218,7 @@ type Server struct {
|
||||
automations AutomationStore
|
||||
users UserStore
|
||||
backthink BackthinkStore
|
||||
rag RagStore
|
||||
voice VoiceService
|
||||
streaming bool
|
||||
frontend fs.FS
|
||||
@@ -229,6 +251,7 @@ func New(cfg *config.Config, deps Deps) (*Server, error) {
|
||||
automations: deps.Automations,
|
||||
users: deps.Users,
|
||||
backthink: deps.Backthink,
|
||||
rag: deps.RAG,
|
||||
streaming: deps.Streaming,
|
||||
frontend: deps.Frontend,
|
||||
log: sourcedLog(deps.Log),
|
||||
@@ -269,6 +292,9 @@ func (s *Server) Handler() http.Handler {
|
||||
mux.Handle("GET /api/v1/admin/automations", adminAuth(http.HandlerFunc(s.adminListAutomations)))
|
||||
mux.Handle("GET /api/v1/admin/users", adminAuth(http.HandlerFunc(s.adminListUsers)))
|
||||
mux.Handle("GET /api/v1/admin/backthink", adminAuth(http.HandlerFunc(s.adminListBackthink)))
|
||||
mux.Handle("GET /api/v1/admin/rag", adminAuth(http.HandlerFunc(s.adminListRag)))
|
||||
mux.Handle("POST /api/v1/admin/rag/query", adminAuth(http.HandlerFunc(s.adminSearchRag)))
|
||||
mux.Handle("POST /api/v1/admin/rag/delete", adminAuth(http.HandlerFunc(s.adminDeleteRag)))
|
||||
mux.Handle("POST /api/v1/admin/plugins/policy", s.auth.Authorize(auth.RoleOwner)(http.HandlerFunc(s.adminSetPolicy)))
|
||||
|
||||
var h http.Handler = mux
|
||||
|
||||
Reference in new issue
Block a user