feat(gateway): chat.documents 解析 + 修两处静默吞掉(design/24 S3-B)
- REST(同步 + stream:true)与 WS 两条入口同形解析 documents:只收服务端存储名(形态 + 真实存在 + 嗅探 MIME + 本地 URL + 大小),客户端谎报 mime/url/size 一律被服务端覆盖 ⇒ 不重开 SSRF 面(有「真实可命中 URL,命中次数 0」的用例钉住)。 - 拒绝口径对齐 S1:超过 5 条 → 400 too_many_documents(且先于任何 FS 访问);形态非法/穿越/不存在 → 404 document_not_found。 - 挂载:一回合一条 RoleUser 消息,文档部件追加其后(纯附件回合不塞空文本块,空文本会被 provider 拒)。 - 修 S2 移交的两处静默吞掉:session_transcript 用 history.DecodeDocument 把 h.Document 还原为部件(坏列带来源 WARN 不静默、也不卡会话);session_tail 的 messagesEmpty 计入 ContentTypeDocument(纯附件不再被判空回合)。 - 拆文件/拆函数(AGENTS:文件≤500 行、函数≤80 行):new chat_documents.go(rest.go 原 499/500 行塞不下)、handleChat 尾段 54 行原样搬成 submitChatTurn(handleChat 90→40 行),ws_hub 增 documents 依赖(与既有 envResolver 同接法)。 - 主会话独立验收(隔离 worktree = 当前 HEAD + 仅 B 的文件 ⇒ 正是 A+B 合流态):build/vet 绿、全树 59 包 0 FAIL、13 条新用例 PASS、linux/darwin 交叉编译绿;变异独立复现(摘掉条数上限判定 ⇒ 红)。 - 主会话补的测试钉子:TestMaxChatDocumentsMatchesDesign —— 原条数用例是用常量自身算输入条数的,改常量它照样绿(实测 5→1000 全包仍绿);补钉后改 10 时新钉子红、旧用例仍绿。
This commit is contained in:
1 parent
a8851dc0a2
commit
a4789ec13e
8 files changed
+859
-87
No files matched your search
@@ -0,0 +1,157 @@
|
|||||||
|
package gateway
|
||||||
|
|
||||||
|
// 本文件归属 `chat` 请求里的文档引用(design/24 §4.2 路线① / §6.1):请求形态、
|
||||||
|
// 服务端校验(条数上限、存储名形态、文件真实存在)与该回合用户消息的装配。
|
||||||
|
//
|
||||||
|
// 它刻意**不改** S1 已落库的 `documents.go`:解析只复用那边的
|
||||||
|
// `validDocumentName` / `documentReadMIME` / `documentFilePrefix` /
|
||||||
|
// `documentNotFoundError` / `documentsLog`,读侧行为零变动。
|
||||||
|
//
|
||||||
|
// 两条硬规则:
|
||||||
|
// - 只接受**服务端签发的存储名**(`POST /api/v1/documents` 响应里的 name),
|
||||||
|
// 绝不接受任意 http(s) URL —— 不重开 F3 那种「服务端代取 URL」的 SSRF 面
|
||||||
|
// (design/24 §5);
|
||||||
|
// - 文档部件只挂在 RoleUser 回合上(design/24 §2.3-③、design/11 §10.2)。
|
||||||
|
//
|
||||||
|
// 失败一律可见(design/24 §4.4):条数超限、名字形态非法、文件不存在都让整个
|
||||||
|
// 请求被拒;绝不做「丢掉这条附件但照样起回合」的静默处理。
|
||||||
|
|
||||||
|
import (
|
||||||
|
"os"
|
||||||
|
"path/filepath"
|
||||||
|
"strings"
|
||||||
|
|
||||||
|
apperr "git.yeij.top/AskaEth/Cyrene_v2/internal/errors"
|
||||||
|
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model"
|
||||||
|
)
|
||||||
|
|
||||||
|
// maxChatDocuments 是**单回合**附件数上限(design/24 §6.1、§9.4-3 的定稿值 5)。
|
||||||
|
const maxChatDocuments = 5
|
||||||
|
|
||||||
|
// tooManyDocumentsCode 是附件数超限的机器可读错误码(design/24 §6.2 表格)。
|
||||||
|
// 它与 image 侧的 image_too_large 同一风格:客户端只看 code 就能分开处理。
|
||||||
|
const tooManyDocumentsCode = "too_many_documents"
|
||||||
|
|
||||||
|
// documentsUnavailableCode 报告本进程没有接文档解析器(未接线的构建 / 单元测试
|
||||||
|
// 构造的裸 hub)。它是**内部接线**问题,绝不是客户端输入问题:此时带了附件的
|
||||||
|
// 回合被明确拒绝,而不是把附件静默丢掉。
|
||||||
|
const documentsUnavailableCode = "chat_documents_unavailable"
|
||||||
|
|
||||||
|
// chatDocumentRef 是 `chat` 请求里的一条文档引用。它只有 `name` 一个字段:
|
||||||
|
// 客户端能给的就是 `POST /api/v1/documents` 响应里的存储名,MIME / URL / 大小
|
||||||
|
// 一律由服务端自己解析(design/24 §3.2:字节嗅探是唯一裁决,客户端声明可被
|
||||||
|
// 否决)。REST 与 WS 上行帧**同形**(design/24 §6.1)。
|
||||||
|
type chatDocumentRef struct {
|
||||||
|
Name string `json:"name"`
|
||||||
|
}
|
||||||
|
|
||||||
|
// documentResolver 把一组文档引用解析为服务端文档部件。REST 与 WS 两条入口
|
||||||
|
// 共用同一个实现(`Server.resolveChatDocuments`),因此两边不可能分叉。
|
||||||
|
type documentResolver func(refs []chatDocumentRef) ([]model.Document, error)
|
||||||
|
|
||||||
|
// resolveChatDocuments 校验并解析一个回合携带的全部文档引用。条数上限先于任何
|
||||||
|
// 文件系统访问被判定(用最小的代价拒最早),随后逐条解析;任一条非法或不存在
|
||||||
|
// 都让整个请求被拒 —— 静默丢掉一条附件正是 design/24 §4.4 禁止的
|
||||||
|
// 「传上去了但双方都不知道」。
|
||||||
|
func (s *Server) resolveChatDocuments(refs []chatDocumentRef) ([]model.Document, error) {
|
||||||
|
if len(refs) > maxChatDocuments {
|
||||||
|
return nil, tooManyDocumentsError(len(refs))
|
||||||
|
}
|
||||||
|
out := make([]model.Document, 0, len(refs))
|
||||||
|
for _, ref := range refs {
|
||||||
|
doc, err := s.resolveChatDocument(ref)
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
out = append(out, doc)
|
||||||
|
}
|
||||||
|
return out, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// resolveChatDocument 把一条引用解析为文档部件。名字形态必须与 S1 的
|
||||||
|
// `validDocumentName` 同规则(32 位小写十六进制主名 + 白名单扩展名),且文件
|
||||||
|
// 必须**确实存在**于 `<DataDir>/uploads/documents/`。形态非法、路径穿越、
|
||||||
|
// 文件不存在一律收敛到同一个 404 document_not_found —— 与读端点同一口径,
|
||||||
|
// 既不泄露「是否存在」以外的信息,也不可能被当成取任意 URL 的入口。
|
||||||
|
//
|
||||||
|
// MIME 由读侧的字节嗅探给出(`documentReadMIME`)、大小取自文件本身:落盘内容
|
||||||
|
// 就是唯一真源,不问数据库、也不信客户端。
|
||||||
|
func (s *Server) resolveChatDocument(ref chatDocumentRef) (model.Document, error) {
|
||||||
|
if !validDocumentName(ref.Name) {
|
||||||
|
return model.Document{}, documentNotFoundError()
|
||||||
|
}
|
||||||
|
dir, err := s.documentsDir()
|
||||||
|
if err != nil {
|
||||||
|
return model.Document{}, documentNotFoundError()
|
||||||
|
}
|
||||||
|
path := filepath.Join(dir, ref.Name)
|
||||||
|
if filepath.Dir(path) != filepath.Clean(dir) {
|
||||||
|
return model.Document{}, documentNotFoundError()
|
||||||
|
}
|
||||||
|
f, err := os.Open(path)
|
||||||
|
if err != nil {
|
||||||
|
return model.Document{}, documentNotFoundError()
|
||||||
|
}
|
||||||
|
defer func() { _ = f.Close() }()
|
||||||
|
info, err := f.Stat()
|
||||||
|
if err != nil || info.IsDir() {
|
||||||
|
return model.Document{}, documentNotFoundError()
|
||||||
|
}
|
||||||
|
return model.Document{
|
||||||
|
Name: ref.Name,
|
||||||
|
MIME: documentReadMIME(f),
|
||||||
|
URL: documentFilePrefix + ref.Name,
|
||||||
|
Size: info.Size(),
|
||||||
|
}, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// tooManyDocumentsError 构造附件数超限错误:400 + too_many_documents,带上限与
|
||||||
|
// 本次条数,使客户端不必猜是哪条限额挡下的。
|
||||||
|
func tooManyDocumentsError(n int) error {
|
||||||
|
return apperr.Wrap(apperr.ErrBadRequest, tooManyDocumentsCode,
|
||||||
|
"单回合附件最多 %d 条,本次 %d 条", maxChatDocuments, n)
|
||||||
|
}
|
||||||
|
|
||||||
|
// documentsUnavailableError 构造「未接文档解析器」错误:内部接线问题,因此走
|
||||||
|
// 500 + 显式错误码,绝不假装这回合只有文本。
|
||||||
|
func documentsUnavailableError() error {
|
||||||
|
return apperr.Wrap(apperr.ErrInternal, documentsUnavailableCode, "文档解析器未接线")
|
||||||
|
}
|
||||||
|
|
||||||
|
// chatDocumentErrorCode 取出错误里可对客户端暴露的机器可读码(WS 错误帧用)。
|
||||||
|
// 取不到时回退 bad_request —— 内部细节绝不回给客户端。
|
||||||
|
func chatDocumentErrorCode(err error) string {
|
||||||
|
if appErr, ok := apperr.As(err); ok && appErr.Code != "" {
|
||||||
|
return appErr.Code
|
||||||
|
}
|
||||||
|
return wsBadRequestCode
|
||||||
|
}
|
||||||
|
|
||||||
|
// chatTurnMessages 为一个回合构建用户消息:文本、图片与文档部件挂在**同一条**
|
||||||
|
// RoleUser 消息上。文档部件由 model.DocumentMessage 构造(design/24 §2.2),
|
||||||
|
// 且只可能落在 RoleUser 上(§2.3-③)。
|
||||||
|
//
|
||||||
|
// 纯附件回合(无文本、无图片)同样得到一条消息,且文档部件让它不被判成空回合
|
||||||
|
// (见 session_tail.go 的 messagesEmpty)。这一分支**不塞空文本部件**:空文本块
|
||||||
|
// 会被 provider 拒,而文档部件本身就是这条消息的内容。
|
||||||
|
func chatTurnMessages(text, image string, docs []model.Document) []model.Message {
|
||||||
|
msg := baseUserMessage(text, image)
|
||||||
|
for _, doc := range docs {
|
||||||
|
msg.Content = append(msg.Content, model.DocumentMessage(model.RoleUser, "", doc).Content...)
|
||||||
|
}
|
||||||
|
return []model.Message{msg}
|
||||||
|
}
|
||||||
|
|
||||||
|
// baseUserMessage 返回回合的基底用户消息:有图片时是可带文本的图片消息,否则是
|
||||||
|
// 文本消息。它就是引入文档之前 `userChatMessages` 的行为,逐字不变(唯一的
|
||||||
|
// 差别是文本为空时给出不带文本部件的空基底,供纯附件回合使用)。
|
||||||
|
func baseUserMessage(text, image string) model.Message {
|
||||||
|
switch {
|
||||||
|
case strings.TrimSpace(image) != "":
|
||||||
|
return model.ImageMessage(model.RoleUser, text, image)
|
||||||
|
case text == "":
|
||||||
|
return model.Message{Role: model.RoleUser}
|
||||||
|
default:
|
||||||
|
return model.TextMessage(model.RoleUser, text)
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -0,0 +1,510 @@
|
|||||||
|
package gateway
|
||||||
|
|
||||||
|
// 本文件锁定切片 S3 的 B 路(网关侧,design/24 §4.2 / §6.1 / §9.5):
|
||||||
|
// - `chat.documents` 解析为**用户回合**的文档部件(红线:只允许 RoleUser);
|
||||||
|
// - 单回合条数上限(too_many_documents)与存储名形态 / 存在性校验
|
||||||
|
// (document_not_found),且**绝不接受 URL**(不重开 SSRF 面);
|
||||||
|
// - 历史重建带回文档引用、纯附件回合不被判空(S2 移交的两处「静默吞掉」)。
|
||||||
|
//
|
||||||
|
// 断言一律走哨兵错误 + errors.Is / errors.As(机器可读的 code),不按错误文本
|
||||||
|
// 子串断言 —— 文案会变,契约不变。文档落盘用临时目录,绝不触碰 data/cyrene.db。
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"crypto/sha256"
|
||||||
|
"encoding/hex"
|
||||||
|
"errors"
|
||||||
|
"fmt"
|
||||||
|
"net/http"
|
||||||
|
"net/http/httptest"
|
||||||
|
"os"
|
||||||
|
"path/filepath"
|
||||||
|
"strconv"
|
||||||
|
"sync"
|
||||||
|
"sync/atomic"
|
||||||
|
"testing"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"git.yeij.top/AskaEth/Cyrene_v2/internal/auth"
|
||||||
|
"git.yeij.top/AskaEth/Cyrene_v2/internal/config"
|
||||||
|
apperr "git.yeij.top/AskaEth/Cyrene_v2/internal/errors"
|
||||||
|
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history"
|
||||||
|
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/loop"
|
||||||
|
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model"
|
||||||
|
)
|
||||||
|
|
||||||
|
// chatDocRunner 记录每次 Run 看到的完整消息(含文档部件)。流式回合跑在后台
|
||||||
|
// goroutine 上,因此用互斥锁守住记录。
|
||||||
|
type chatDocRunner struct {
|
||||||
|
mu sync.Mutex
|
||||||
|
calls [][]model.Message
|
||||||
|
}
|
||||||
|
|
||||||
|
func (r *chatDocRunner) Run(_ context.Context, _ string, msgs []model.Message) (*loop.Result, error) {
|
||||||
|
r.mu.Lock()
|
||||||
|
r.calls = append(r.calls, msgs)
|
||||||
|
r.mu.Unlock()
|
||||||
|
return &loop.Result{Text: "收到了", Steps: 1, StopReason: "stop"}, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (r *chatDocRunner) snapshot() [][]model.Message {
|
||||||
|
r.mu.Lock()
|
||||||
|
defer r.mu.Unlock()
|
||||||
|
return append([][]model.Message(nil), r.calls...)
|
||||||
|
}
|
||||||
|
|
||||||
|
// waitCalls 等到至少 n 次 Run 被记录(后台流式回合),超时即失败。
|
||||||
|
func (r *chatDocRunner) waitCalls(t *testing.T, n int) [][]model.Message {
|
||||||
|
t.Helper()
|
||||||
|
deadline := time.Now().Add(2 * time.Second)
|
||||||
|
for time.Now().Before(deadline) {
|
||||||
|
if calls := r.snapshot(); len(calls) >= n {
|
||||||
|
return calls
|
||||||
|
}
|
||||||
|
time.Sleep(time.Millisecond)
|
||||||
|
}
|
||||||
|
t.Fatalf("等待 %d 次回合超时(实际 %d 次)", n, len(r.snapshot()))
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// chatDocumentsEnv 打包文档聊天测试所需的依赖:真实 JWT、临时数据目录(文档
|
||||||
|
// 落盘在它下面)、接线的录制 runner。
|
||||||
|
type chatDocumentsEnv struct {
|
||||||
|
srv *Server
|
||||||
|
ts *httptest.Server
|
||||||
|
dataDir string
|
||||||
|
auth *auth.JWTService
|
||||||
|
runner *chatDocRunner
|
||||||
|
}
|
||||||
|
|
||||||
|
func newChatDocumentsEnv(t *testing.T) *chatDocumentsEnv {
|
||||||
|
t.Helper()
|
||||||
|
dataDir := t.TempDir()
|
||||||
|
jwt := auth.New(config.AuthConfig{JWTSecret: testSecret})
|
||||||
|
cfg := &config.Config{Auth: config.AuthConfig{AdminUser: "admin", AdminPass: "secret", JWTSecret: testSecret}}
|
||||||
|
runner := &chatDocRunner{}
|
||||||
|
srv, err := New(cfg, Deps{Auth: jwt, DataDir: dataDir, Loop: runner})
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("构造网关失败: %v", err)
|
||||||
|
}
|
||||||
|
ts := httptest.NewServer(srv.Handler())
|
||||||
|
t.Cleanup(ts.Close)
|
||||||
|
t.Cleanup(srv.Close)
|
||||||
|
return &chatDocumentsEnv{srv: srv, ts: ts, dataDir: dataDir, auth: jwt, runner: runner}
|
||||||
|
}
|
||||||
|
|
||||||
|
func (e *chatDocumentsEnv) token(t *testing.T) string {
|
||||||
|
t.Helper()
|
||||||
|
tok, err := e.auth.Sign("acct", auth.RoleOwner, time.Hour)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("签发 token 失败: %v", err)
|
||||||
|
}
|
||||||
|
return tok
|
||||||
|
}
|
||||||
|
|
||||||
|
// storeDocumentFile 按 S1 的内容寻址规则(sha256 前 32 位 + 扩展名)在
|
||||||
|
// `<DataDir>/uploads/documents` 下落一份文档,并返回它的存储名。
|
||||||
|
func (e *chatDocumentsEnv) storeDocumentFile(t *testing.T, content, ext string) string {
|
||||||
|
t.Helper()
|
||||||
|
sum := sha256.Sum256([]byte(content))
|
||||||
|
name := hex.EncodeToString(sum[:])[:documentHashHexLen] + ext
|
||||||
|
if !validDocumentName(name) {
|
||||||
|
t.Fatalf("构造的存储名 %q 不是合法形态(与 S1 口径不一致)", name)
|
||||||
|
}
|
||||||
|
dir := filepath.Join(e.dataDir, "uploads", documentSubDir)
|
||||||
|
if err := os.MkdirAll(dir, 0o700); err != nil {
|
||||||
|
t.Fatalf("创建文档目录失败: %v", err)
|
||||||
|
}
|
||||||
|
if err := os.WriteFile(filepath.Join(dir, name), []byte(content), 0o600); err != nil {
|
||||||
|
t.Fatalf("写入文档失败: %v", err)
|
||||||
|
}
|
||||||
|
return name
|
||||||
|
}
|
||||||
|
|
||||||
|
// validFormDocumentName 返回一个**形态合法但不存在**的存储名(用于条数上限与
|
||||||
|
// 存在性两条判定的分辨:先撞上限还是先撞不存在)。
|
||||||
|
func validFormDocumentName(seed string) string {
|
||||||
|
sum := sha256.Sum256([]byte(seed))
|
||||||
|
return hex.EncodeToString(sum[:])[:documentHashHexLen] + ".txt"
|
||||||
|
}
|
||||||
|
|
||||||
|
// chatDocumentsOf 取出这些消息里的全部文档部件。
|
||||||
|
func chatDocumentsOf(msgs []model.Message) []model.Content {
|
||||||
|
out := make([]model.Content, 0, len(msgs))
|
||||||
|
for _, m := range msgs {
|
||||||
|
for _, c := range m.Content {
|
||||||
|
if c.Type == model.ContentTypeDocument {
|
||||||
|
out = append(out, c)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return out
|
||||||
|
}
|
||||||
|
|
||||||
|
// assertOnlyUserCarriesDocuments 断言文档部件只挂在用户回合上(design/24 §2.3-③
|
||||||
|
// 的红线:编码器会对非 user 的文档块 fail-loud,因此这一条必须在这里就成立)。
|
||||||
|
func assertOnlyUserCarriesDocuments(t *testing.T, msgs []model.Message) {
|
||||||
|
t.Helper()
|
||||||
|
for _, m := range msgs {
|
||||||
|
for _, c := range m.Content {
|
||||||
|
if c.Type == model.ContentTypeDocument && m.Role != model.RoleUser {
|
||||||
|
t.Fatalf("文档部件挂在了 %s 回合上(只允许 user)", m.Role)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestChatDocumentsAttachAsUserParts 钉住主路径:REST chat 带 documents →
|
||||||
|
// 该回合的用户消息上多出一个文档部件,且它的 name / MIME / URL / 大小全部由
|
||||||
|
// **服务端**解析(客户端只给了 name)。
|
||||||
|
func TestChatDocumentsAttachAsUserParts(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
env := newChatDocumentsEnv(t)
|
||||||
|
const content = "行1\n行2\n"
|
||||||
|
name := env.storeDocumentFile(t, content, ".txt")
|
||||||
|
|
||||||
|
resp, body := doReq(t, http.MethodPost, env.ts.URL+"/api/v1/chat", map[string]any{
|
||||||
|
"sessionId": "web",
|
||||||
|
"text": "读一下这份文件",
|
||||||
|
// 客户端只给 name;MIME / URL / 大小就算谎报也必须被服务端自己的解析覆盖
|
||||||
|
// (design/24 §3.2:字节嗅探是唯一裁决)。
|
||||||
|
"documents": []map[string]any{{
|
||||||
|
"name": name,
|
||||||
|
"mime": "text/html",
|
||||||
|
"url": "https://evil.example/steal.txt",
|
||||||
|
"size": 999999,
|
||||||
|
}},
|
||||||
|
}, env.token(t))
|
||||||
|
if resp.StatusCode != http.StatusOK {
|
||||||
|
t.Fatalf("status = %d, want 200 (body=%s)", resp.StatusCode, body)
|
||||||
|
}
|
||||||
|
calls := env.runner.snapshot()
|
||||||
|
if len(calls) != 1 {
|
||||||
|
t.Fatalf("驱动了 %d 个回合,want 1", len(calls))
|
||||||
|
}
|
||||||
|
msgs := calls[0]
|
||||||
|
assertOnlyUserCarriesDocuments(t, msgs)
|
||||||
|
docs := chatDocumentsOf(msgs)
|
||||||
|
if len(docs) != 1 {
|
||||||
|
t.Fatalf("文档部件数 = %d, want 1 (msgs=%+v)", len(docs), msgs)
|
||||||
|
}
|
||||||
|
got := docs[0]
|
||||||
|
if got.DocumentName != name {
|
||||||
|
t.Fatalf("DocumentName = %q, want %q", got.DocumentName, name)
|
||||||
|
}
|
||||||
|
if got.DocumentMIME != documentSniffOKMIME {
|
||||||
|
t.Fatalf("DocumentMIME = %q, want %q(嗅探所得,绝不是客户端声明)", got.DocumentMIME, documentSniffOKMIME)
|
||||||
|
}
|
||||||
|
if want := documentFilePrefix + name; got.DocumentURL != want {
|
||||||
|
t.Fatalf("DocumentURL = %q, want %q(服务端本地引用)", got.DocumentURL, want)
|
||||||
|
}
|
||||||
|
if got.DocumentSize != int64(len(content)) {
|
||||||
|
t.Fatalf("DocumentSize = %d, want %d", got.DocumentSize, len(content))
|
||||||
|
}
|
||||||
|
// 文本仍在同一条用户消息上,顺序是「文本在前、附件在后」。
|
||||||
|
last := msgs[len(msgs)-1]
|
||||||
|
if last.Role != model.RoleUser || last.Content[0].Type != model.ContentTypeText {
|
||||||
|
t.Fatalf("末条消息 = %+v, want 用户文本消息在前", last)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestChatDocumentsPureAttachmentTurn 钉住「只带附件」的回合:既不是空回合,
|
||||||
|
// 也不会塞一个空文本块(空文本块会被 provider 拒)。
|
||||||
|
func TestChatDocumentsPureAttachmentTurn(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
env := newChatDocumentsEnv(t)
|
||||||
|
name := env.storeDocumentFile(t, "只有附件", ".md")
|
||||||
|
|
||||||
|
resp, body := doReq(t, http.MethodPost, env.ts.URL+"/api/v1/chat", map[string]any{
|
||||||
|
"sessionId": "web",
|
||||||
|
"text": "",
|
||||||
|
"documents": []chatDocumentRef{{Name: name}},
|
||||||
|
}, env.token(t))
|
||||||
|
if resp.StatusCode != http.StatusOK {
|
||||||
|
t.Fatalf("纯附件回合被拒:status = %d (body=%s)", resp.StatusCode, body)
|
||||||
|
}
|
||||||
|
calls := env.runner.snapshot()
|
||||||
|
if len(calls) != 1 {
|
||||||
|
t.Fatalf("驱动了 %d 个回合,want 1(纯附件回合是真实回合)", len(calls))
|
||||||
|
}
|
||||||
|
msgs := calls[0]
|
||||||
|
if len(chatDocumentsOf(msgs)) != 1 {
|
||||||
|
t.Fatalf("文档部件丢失:%+v", msgs)
|
||||||
|
}
|
||||||
|
for _, m := range msgs {
|
||||||
|
for _, c := range m.Content {
|
||||||
|
if c.Type == model.ContentTypeText && c.Text == "" {
|
||||||
|
t.Fatalf("纯附件回合塞了空文本部件:%+v", m)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestChatDocumentsTooManyRejected 钉住单回合条数上限:恰好 5 条通过,第 6 条
|
||||||
|
// 起 400 + too_many_documents,且条数判定**先于**任何文件系统访问(6 条全不存在
|
||||||
|
// 也报条数超限,而不是 document_not_found)。回合一次都不许起。
|
||||||
|
func TestChatDocumentsTooManyRejected(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
env := newChatDocumentsEnv(t)
|
||||||
|
refs := make([]chatDocumentRef, 0, maxChatDocuments+1)
|
||||||
|
for i := 0; i < maxChatDocuments; i++ {
|
||||||
|
refs = append(refs, chatDocumentRef{Name: env.storeDocumentFile(t, "doc-"+strconv.Itoa(i), ".txt")})
|
||||||
|
}
|
||||||
|
if docs, err := env.srv.resolveChatDocuments(refs); err != nil || len(docs) != maxChatDocuments {
|
||||||
|
t.Fatalf("恰好 %d 条必须通过:docs=%d err=%v", maxChatDocuments, len(docs), err)
|
||||||
|
}
|
||||||
|
over := append(append([]chatDocumentRef(nil), refs...), chatDocumentRef{Name: validFormDocumentName("sixth")})
|
||||||
|
if _, err := env.srv.resolveChatDocuments(over); !errors.Is(err, apperr.ErrBadRequest) {
|
||||||
|
t.Fatalf("第 %d 条应被拒为 ErrBadRequest,得到 %v", maxChatDocuments+1, err)
|
||||||
|
}
|
||||||
|
var appErr *apperr.AppError
|
||||||
|
if _, err := env.srv.resolveChatDocuments(over); !errors.As(err, &appErr) || appErr.Code != tooManyDocumentsCode {
|
||||||
|
t.Fatalf("error.code = %q, want %q", appErr.Code, tooManyDocumentsCode)
|
||||||
|
}
|
||||||
|
|
||||||
|
tok := env.token(t)
|
||||||
|
resp, body := doReq(t, http.MethodPost, env.ts.URL+"/api/v1/chat", map[string]any{
|
||||||
|
"sessionId": "web",
|
||||||
|
"text": "太多了",
|
||||||
|
"documents": over,
|
||||||
|
}, tok)
|
||||||
|
if resp.StatusCode != http.StatusBadRequest {
|
||||||
|
t.Fatalf("status = %d, want 400 (body=%s)", resp.StatusCode, body)
|
||||||
|
}
|
||||||
|
if code := decodeDocumentEnvelope(t, body).Error.Code; code != tooManyDocumentsCode {
|
||||||
|
t.Fatalf("error.code = %q, want %q", code, tooManyDocumentsCode)
|
||||||
|
}
|
||||||
|
if len(env.runner.snapshot()) != 0 {
|
||||||
|
t.Fatalf("被拒的请求起了回合 —— 附件条数必须在起回合之前判定")
|
||||||
|
}
|
||||||
|
// 恰好 5 条(都存在)照常放行:上限不得误伤边界。
|
||||||
|
resp, body = doReq(t, http.MethodPost, env.ts.URL+"/api/v1/chat", map[string]any{
|
||||||
|
"sessionId": "web",
|
||||||
|
"text": "这五份",
|
||||||
|
"documents": refs,
|
||||||
|
}, tok)
|
||||||
|
if resp.StatusCode != http.StatusOK {
|
||||||
|
t.Fatalf("恰好 %d 条必须通过:status = %d (body=%s)", maxChatDocuments, resp.StatusCode, body)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestChatDocumentsRejectMalformedOrMissingName 钉住存储名校验:形态非法(路径
|
||||||
|
// 穿越、非 32 位十六进制、白名单外扩展名、空名)与形态合法但不存在,一律
|
||||||
|
// 404 document_not_found(与读端点同一口径),且都不起回合。
|
||||||
|
func TestChatDocumentsRejectMalformedOrMissingName(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
env := newChatDocumentsEnv(t)
|
||||||
|
// 形态校验必须真的在把关:目录里真有一份**非内容寻址**的同名文件时也不
|
||||||
|
// 能因为「它存在」就被放行 —— 客户端能引用的只有 S1 签发过的存储名。
|
||||||
|
dir := filepath.Join(env.dataDir, "uploads", documentSubDir)
|
||||||
|
if err := os.MkdirAll(dir, 0o700); err != nil {
|
||||||
|
t.Fatalf("创建文档目录失败: %v", err)
|
||||||
|
}
|
||||||
|
if err := os.WriteFile(filepath.Join(dir, "notes.txt"), []byte("不是内容寻址名"), 0o600); err != nil {
|
||||||
|
t.Fatalf("写入非内容寻址文件失败: %v", err)
|
||||||
|
}
|
||||||
|
cases := []struct{ label, name string }{
|
||||||
|
{"路径穿越", "../../../etc/passwd"},
|
||||||
|
{"反斜杠穿越", `..\..\cyrene.exe`},
|
||||||
|
{"主名不是 32 位十六进制", "deadbeef.txt"},
|
||||||
|
{"扩展名不在白名单", fmt.Sprintf("%032d.html", 1)},
|
||||||
|
{"空名", ""},
|
||||||
|
{"形态合法但不存在", validFormDocumentName("missing")},
|
||||||
|
{"非内容寻址但确实存在", "notes.txt"},
|
||||||
|
}
|
||||||
|
tok := env.token(t)
|
||||||
|
for _, tc := range cases {
|
||||||
|
_, err := env.srv.resolveChatDocuments([]chatDocumentRef{{Name: tc.name}})
|
||||||
|
if !errors.Is(err, apperr.ErrNotFound) {
|
||||||
|
t.Fatalf("%s: err = %v, want ErrNotFound", tc.label, err)
|
||||||
|
}
|
||||||
|
var appErr *apperr.AppError
|
||||||
|
if _, err := env.srv.resolveChatDocuments([]chatDocumentRef{{Name: tc.name}}); !errors.As(err, &appErr) ||
|
||||||
|
appErr.Code != documentNotFoundCode {
|
||||||
|
t.Fatalf("%s: error.code = %q, want %q", tc.label, appErr.Code, documentNotFoundCode)
|
||||||
|
}
|
||||||
|
resp, body := doReq(t, http.MethodPost, env.ts.URL+"/api/v1/chat", map[string]any{
|
||||||
|
"sessionId": "web",
|
||||||
|
"text": "读一下",
|
||||||
|
"documents": []chatDocumentRef{{Name: tc.name}},
|
||||||
|
}, tok)
|
||||||
|
if resp.StatusCode != http.StatusNotFound {
|
||||||
|
t.Fatalf("%s: status = %d, want 404 (body=%s)", tc.label, resp.StatusCode, body)
|
||||||
|
}
|
||||||
|
if code := decodeDocumentEnvelope(t, body).Error.Code; code != documentNotFoundCode {
|
||||||
|
t.Fatalf("%s: error.code = %q, want %q", tc.label, code, documentNotFoundCode)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if len(env.runner.snapshot()) != 0 {
|
||||||
|
t.Fatalf("被拒的请求起了回合 —— 存储名校验必须在起回合之前完成")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestChatDocumentsNeverFetchURLs 是 SSRF 红线的正向反证(design/24 §5):把
|
||||||
|
// 一个真实可命中的 URL 当作 name 发进来,网关**一次都不许去取它** —— 文档只认
|
||||||
|
// 服务端签发的存储名。
|
||||||
|
func TestChatDocumentsNeverFetchURLs(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
env := newChatDocumentsEnv(t)
|
||||||
|
var hits atomic.Int64
|
||||||
|
evil := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||||
|
hits.Add(1)
|
||||||
|
w.WriteHeader(http.StatusOK)
|
||||||
|
}))
|
||||||
|
defer evil.Close()
|
||||||
|
|
||||||
|
tok := env.token(t)
|
||||||
|
names := []string{evil.URL + "/report.txt", "http://127.0.0.1:8080/api/v1/documents/whatever.txt"}
|
||||||
|
for _, name := range names {
|
||||||
|
resp, body := doReq(t, http.MethodPost, env.ts.URL+"/api/v1/chat", map[string]any{
|
||||||
|
"sessionId": "web",
|
||||||
|
"text": "读这个链接",
|
||||||
|
"documents": []chatDocumentRef{{Name: name}},
|
||||||
|
}, tok)
|
||||||
|
if resp.StatusCode != http.StatusNotFound {
|
||||||
|
t.Fatalf("URL 形态的 name 必须被拒:status = %d (body=%s)", resp.StatusCode, body)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if got := hits.Load(); got != 0 {
|
||||||
|
t.Fatalf("网关替客户端取了 URL %d 次 —— SSRF 面被重开", got)
|
||||||
|
}
|
||||||
|
if len(env.runner.snapshot()) != 0 {
|
||||||
|
t.Fatalf("被拒的请求起了回合")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestChatDocumentsStreamPathCarriesDocuments 钉住 `stream:true` 这条转发路径:
|
||||||
|
// 载荷必须把 documents 一并带到 hub(漏抄一个字段就是一次静默丢附件),因此
|
||||||
|
// 后台回合里同样能看到文档部件。
|
||||||
|
func TestChatDocumentsStreamPathCarriesDocuments(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
env := newChatDocumentsEnv(t)
|
||||||
|
name := env.storeDocumentFile(t, "流式回合的附件", ".txt")
|
||||||
|
|
||||||
|
resp, body := doReq(t, http.MethodPost, env.ts.URL+"/api/v1/chat", map[string]any{
|
||||||
|
"sessionId": "web",
|
||||||
|
"text": "",
|
||||||
|
"documents": []chatDocumentRef{{Name: name}},
|
||||||
|
"stream": true,
|
||||||
|
}, env.token(t))
|
||||||
|
if resp.StatusCode != http.StatusOK {
|
||||||
|
t.Fatalf("stream:true 未被受理:status = %d (body=%s)", resp.StatusCode, body)
|
||||||
|
}
|
||||||
|
calls := env.runner.waitCalls(t, 1)
|
||||||
|
if docs := chatDocumentsOf(calls[0]); len(docs) != 1 || docs[0].DocumentName != name {
|
||||||
|
t.Fatalf("流式路径丢掉了文档部件:%+v", calls[0])
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestHubStreamChatResolvesDocuments 钉住 WS 帧(`chatStreamRequest` 与 REST
|
||||||
|
// 同形):带了 documents 就按同一条解析路径装配;未接解析器时**明确回一条
|
||||||
|
// error 帧**并中止本回合,绝不静默吞掉附件。
|
||||||
|
func TestHubStreamChatResolvesDocuments(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
env := newChatDocumentsEnv(t)
|
||||||
|
name := env.storeDocumentFile(t, "ws 帧的附件", ".txt")
|
||||||
|
payload := map[string]any{"sessionId": "web", "text": "看这个", "documents": []chatDocumentRef{{Name: name}}}
|
||||||
|
|
||||||
|
hb := newTestHub(env.runner, nil)
|
||||||
|
hb.documents = env.srv.resolveChatDocuments
|
||||||
|
hb.streamChat(context.Background(), payload)
|
||||||
|
calls := env.runner.snapshot()
|
||||||
|
if len(calls) != 1 || len(chatDocumentsOf(calls[0])) != 1 {
|
||||||
|
t.Fatalf("WS 帧路径丢掉了文档部件:%+v", calls)
|
||||||
|
}
|
||||||
|
|
||||||
|
// 未接解析器(裸 hub):文档必须可见地失败,而不是被丢掉。
|
||||||
|
bare := newTestHub(env.runner, nil)
|
||||||
|
c := newOwnerConn(bare, 8)
|
||||||
|
bare.register(c)
|
||||||
|
bare.streamChat(context.Background(), payload)
|
||||||
|
ev := readFakeFrame(t, c)
|
||||||
|
if ev.Type != "error" {
|
||||||
|
t.Fatalf("帧类型 = %q, want error(附件解析失败必须可见)", ev.Type)
|
||||||
|
}
|
||||||
|
data, ok := ev.Data.(map[string]any)
|
||||||
|
if !ok || data["code"] != documentsUnavailableCode {
|
||||||
|
t.Fatalf("error data = %+v, want code=%q", ev.Data, documentsUnavailableCode)
|
||||||
|
}
|
||||||
|
if len(env.runner.snapshot()) != 1 {
|
||||||
|
t.Fatalf("解析失败的回合仍然起了 loop —— 必须在起回合之前拦住")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestTranscriptMessageKeepsDocumentReference 是 design/24 §9.5 第一处的回归锁:
|
||||||
|
// 从历史重建模型消息时必须带回文档引用(否则后续回合里附件凭空消失)。
|
||||||
|
func TestTranscriptMessageKeepsDocumentReference(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
const name = "0123456789abcdef0123456789abcdef.txt"
|
||||||
|
raw, err := history.EncodeDocument(history.DocumentRef{
|
||||||
|
Name: name, MIME: documentSniffOKMIME, Size: 12345, Status: history.DocumentStatusExtracted,
|
||||||
|
})
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("encode document: %v", err)
|
||||||
|
}
|
||||||
|
msg := transcriptMessage(history.Message{ID: "m1", Role: history.RoleUser, Content: "读一下", Document: raw})
|
||||||
|
docs := chatDocumentsOf([]model.Message{msg})
|
||||||
|
if len(docs) != 1 {
|
||||||
|
t.Fatalf("文档引用没能重建回来:%+v", msg)
|
||||||
|
}
|
||||||
|
if docs[0].DocumentName != name || docs[0].DocumentSize != 12345 ||
|
||||||
|
docs[0].DocumentURL != documentFilePrefix+name || docs[0].DocumentMIME != documentSniffOKMIME {
|
||||||
|
t.Fatalf("重建出的文档部件 = %+v, want name/mime/url/size 与落库引用一致", docs[0])
|
||||||
|
}
|
||||||
|
// 没有附件的消息照旧没有文档部件;坏列不 panic 也不产生部件(打 WARN)。
|
||||||
|
if got := chatDocumentsOf([]model.Message{
|
||||||
|
transcriptMessage(history.Message{ID: "m2", Role: history.RoleUser, Content: "无附件"}),
|
||||||
|
}); len(got) != 0 {
|
||||||
|
t.Fatalf("无附件消息产生了文档部件:%+v", got)
|
||||||
|
}
|
||||||
|
if got := chatDocumentsOf([]model.Message{
|
||||||
|
transcriptMessage(history.Message{ID: "m3", Role: history.RoleUser, Content: "坏列", Document: "{not json"}),
|
||||||
|
}); len(got) != 0 {
|
||||||
|
t.Fatalf("坏列产生了文档部件:%+v", got)
|
||||||
|
}
|
||||||
|
// 助手回合的说话者标注不受本次改动影响;附件不会落到非用户回合上。
|
||||||
|
assistant := transcriptMessage(history.Message{
|
||||||
|
ID: "m4", Role: history.RoleAssistant, Content: "答复", PersonaID: "cyrene", Document: raw,
|
||||||
|
})
|
||||||
|
if assistant.SenderName != "cyrene" {
|
||||||
|
t.Fatalf("助手消息的 SenderName = %q, want cyrene", assistant.SenderName)
|
||||||
|
}
|
||||||
|
if got := chatDocumentsOf([]model.Message{assistant}); len(got) != 0 {
|
||||||
|
t.Fatalf("文档部件落到了助手回合上:%+v", got)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestMessagesEmptyTreatsDocumentAsContent 是 design/24 §9.5 第二处的回归锁:
|
||||||
|
// 「只带附件」的回合必须算非空,否则纯文档消息会被判成空洞回合而整轮丢弃。
|
||||||
|
func TestMessagesEmptyTreatsDocumentAsContent(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
o := NewOrchestrator(OrchestratorOptions{
|
||||||
|
Run: func(context.Context, string, []model.Message, loop.SessionEnv) (*loop.Result, error) {
|
||||||
|
return &loop.Result{Text: "ok", StopReason: "stop"}, nil
|
||||||
|
},
|
||||||
|
})
|
||||||
|
docOnly := chatTurnMessages("", "", []model.Document{{Name: "0123456789abcdef0123456789abcdef.txt"}})
|
||||||
|
if o.messagesEmpty(docOnly) {
|
||||||
|
t.Fatalf("纯附件回合被判成空回合:%+v", docOnly)
|
||||||
|
}
|
||||||
|
if sr := o.Submit(context.Background(), "web", docOnly, loop.SessionEnv{}); sr.Empty || !sr.Accepted {
|
||||||
|
t.Fatalf("纯附件回合的判定 = %+v, want 被接收且非 Empty", sr)
|
||||||
|
}
|
||||||
|
// 反向:真·空回合仍然被判空(本次改动不得把空判定一起放宽)。
|
||||||
|
if !o.messagesEmpty([]model.Message{model.TextMessage(model.RoleUser, "")}) {
|
||||||
|
t.Fatal("空回合未被判空")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestMaxChatDocumentsMatchesDesign 单独把「单回合附件数上限的值」钉在设计定稿上。
|
||||||
|
//
|
||||||
|
// 为什么必须单独钉一次:TestChatDocumentsTooManyRejected 是**用本常量算输入条数**的
|
||||||
|
// (`make(..., maxChatDocuments+1)`、`for i < maxChatDocuments`),所以把常量改大它照样
|
||||||
|
// 绿 —— 实测把 5 改成 1000 时整个包仍全绿。也就是说「上限被悄悄改宽」这件事,除了这里
|
||||||
|
// 没有任何用例会发现。design/24 §9.4-3 的定稿值是 5。
|
||||||
|
func TestMaxChatDocumentsMatchesDesign(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
if maxChatDocuments != 5 {
|
||||||
|
t.Fatalf("单回合附件上限 = %d, want 5(design/24 §9.4-3 的定稿值)", maxChatDocuments)
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -7,9 +7,12 @@ import (
|
|||||||
"time"
|
"time"
|
||||||
|
|
||||||
apperr "git.yeij.top/AskaEth/Cyrene_v2/internal/errors"
|
apperr "git.yeij.top/AskaEth/Cyrene_v2/internal/errors"
|
||||||
|
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/loop"
|
||||||
|
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model"
|
||||||
)
|
)
|
||||||
|
|
||||||
// 本文件归属 REST `POST /api/v1/chat` 的 `stream:true` 派发(Android 协同 §G2)。
|
// 本文件归属 REST `POST /api/v1/chat` 的派发:`stream:true`(Android 协同 §G2)
|
||||||
|
// 与默认的同步回合(submitChatTurn)两条路径。
|
||||||
//
|
//
|
||||||
// 修复前的链路是 publishChatStream → bus.Publish → hub 订阅 → streamChat,三处
|
// 修复前的链路是 publishChatStream → bus.Publish → hub 订阅 → streamChat,三处
|
||||||
// 叠加起来让「客户端拿到 200、却什么都没发生」成为可能:
|
// 叠加起来让「客户端拿到 200、却什么都没发生」成为可能:
|
||||||
@@ -49,7 +52,7 @@ func (s *Server) handleChatStream(w http.ResponseWriter, r *http.Request, req ch
|
|||||||
if sid == "" {
|
if sid == "" {
|
||||||
sid = defaultSessionID
|
sid = defaultSessionID
|
||||||
}
|
}
|
||||||
sr, err := s.submitChatStream(detachedTurnContext(r.Context()), sid, req.Text, req.Image, req.DeviceID, req.Mode, req.ClientMsgID)
|
sr, err := s.submitChatStream(detachedTurnContext(r.Context()), sid, req)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
// 派发没成功:明确报 503,而不是让客户端以为这一轮在跑(协同 G2)。
|
// 派发没成功:明确报 503,而不是让客户端以为这一轮在跑(协同 G2)。
|
||||||
s.log.Error("chat stream dispatch", "err", err, "session", sid)
|
s.log.Error("chat stream dispatch", "err", err, "session", sid)
|
||||||
@@ -62,14 +65,18 @@ func (s *Server) handleChatStream(w http.ResponseWriter, r *http.Request, req ch
|
|||||||
// submitChatStream 把一次 stream:true 的对话交给编排器,并把它对本次提交的判定
|
// submitChatStream 把一次 stream:true 的对话交给编排器,并把它对本次提交的判定
|
||||||
// 交回 HTTP 层(协同 G2)。error 非 nil 表示这一轮**没有**被接收,此时调用方
|
// 交回 HTTP 层(协同 G2)。error 非 nil 表示这一轮**没有**被接收,此时调用方
|
||||||
// 绝不能回 2xx。
|
// 绝不能回 2xx。
|
||||||
func (s *Server) submitChatStream(ctx context.Context, sessionID, text, image, deviceID, mode, clientMsgID string) (SubmitResult, error) {
|
//
|
||||||
|
// 它接收整个 chatRequest(而不是逐个字段),使新增的可选字段(documents,
|
||||||
|
// design/24 §6.1)自动随载荷下传 —— 少抄一个字段就是一次静默丢附件。
|
||||||
|
func (s *Server) submitChatStream(ctx context.Context, sessionID string, req chatRequest) (SubmitResult, error) {
|
||||||
payload := map[string]any{
|
payload := map[string]any{
|
||||||
"sessionId": sessionID,
|
"sessionId": sessionID,
|
||||||
"text": text,
|
"text": req.Text,
|
||||||
"image": image,
|
"image": req.Image,
|
||||||
"deviceId": deviceID,
|
"deviceId": req.DeviceID,
|
||||||
"mode": mode,
|
"mode": req.Mode,
|
||||||
"clientMsgId": clientMsgID,
|
"clientMsgId": req.ClientMsgID,
|
||||||
|
"documents": req.Documents,
|
||||||
}
|
}
|
||||||
if s.hub == nil || s.hub.orchestrator == nil {
|
if s.hub == nil || s.hub.orchestrator == nil {
|
||||||
// 未接编排器:回退到既有的总线触发路径(未接线的 hub 保持完全相同的
|
// 未接编排器:回退到既有的总线触发路径(未接线的 hub 保持完全相同的
|
||||||
@@ -96,6 +103,69 @@ func (s *Server) publishChatStream(ctx context.Context, payload map[string]any)
|
|||||||
return SubmitResult{Accepted: true, Immediate: true}, nil
|
return SubmitResult{Accepted: true, Immediate: true}, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// submitChatTurn 是 `POST /api/v1/chat` 默认(非 stream)路径的派发:把一个已
|
||||||
|
// 校验装配好的回合交给编排器跑完,并把它对本次提交的判定写成响应。它是
|
||||||
|
// submitChatStream 的同步孪生 —— 两者只差「等回合跑完再回」与「立刻回提交判定」,
|
||||||
|
// 因此都住在同一个派发文件里,且共用同一条「接了编排器就直达、否则回退」的口径。
|
||||||
|
//
|
||||||
|
// 它由 handleChat 的尾段原样搬来(零逻辑改动,只把装配好的文档一起接过来),
|
||||||
|
// 使 handleChat 回到「解码 → 校验 → 装配 → 派发」的薄壳形态(函数长度上限)。
|
||||||
|
func (s *Server) submitChatTurn(w http.ResponseWriter, r *http.Request, req chatRequest, docs []model.Document) {
|
||||||
|
if s.runner == nil {
|
||||||
|
writeError(w, apperr.Wrap(apperr.ErrInternal, "chat_unavailable", "chat service is not configured"))
|
||||||
|
return
|
||||||
|
}
|
||||||
|
if req.SessionID == "" {
|
||||||
|
req.SessionID = "web"
|
||||||
|
}
|
||||||
|
// Web 主会话是主人的私密对话(JWT = owner),因此它绑定
|
||||||
|
// 全访问 env:记忆读+写,行为不变。角色
|
||||||
|
// 来自会话的绑定(design/22 §4.5)。
|
||||||
|
msgs := chatTurnMessages(req.Text, req.Image, docs)
|
||||||
|
env := s.envResolver.owner(r.Context(), req.SessionID)
|
||||||
|
// 客户端的 deviceID、行为模式与可选幂等键(协同 G4)都经 ctx 串进编排器。
|
||||||
|
ctx := withClientMsgHint(withMode(withDeviceID(r.Context(), req.DeviceID), req.Mode), clientMsgHint{ClientMsgID: req.ClientMsgID})
|
||||||
|
if s.hub.orchestrator != nil {
|
||||||
|
sr, res := s.hub.orchestrator.SubmitSync(ctx, req.SessionID, msgs, env)
|
||||||
|
// 重复到达(同一个 clientMsgId):明确告知既有消息 id,而不是再跑一遍
|
||||||
|
// 回合、再落一条用户消息(协同 G4)。
|
||||||
|
if sr.Duplicate {
|
||||||
|
writeJSON(w, http.StatusOK, map[string]any{"duplicate": true, "messageId": sr.MessageID, "sessionId": req.SessionID})
|
||||||
|
return
|
||||||
|
}
|
||||||
|
if !sr.Accepted {
|
||||||
|
if sr.PendingFull {
|
||||||
|
writeError(w, apperr.Wrap(apperr.ErrConflict, "chat_busy", "会话正忙,待处理消息已满"))
|
||||||
|
} else {
|
||||||
|
writeError(w, apperr.Wrap(apperr.ErrBadRequest, "bad_request", "text、image 或 documents 至少需要一项"))
|
||||||
|
}
|
||||||
|
return
|
||||||
|
}
|
||||||
|
// 忙碌会话返回非阻塞 ack(design/15 §2.1),而不是
|
||||||
|
// 让请求阻塞在一个可能跑一阵子的回合上。
|
||||||
|
if sr.Pending {
|
||||||
|
writeJSON(w, http.StatusAccepted, map[string]any{"queued": true, "queueDepth": sr.QueueDepth})
|
||||||
|
return
|
||||||
|
}
|
||||||
|
if sr.Err != nil {
|
||||||
|
s.log.Error("chat run", "err", sr.Err, "session", req.SessionID)
|
||||||
|
writeError(w, apperr.Wrap(apperr.ErrInternal, "chat_failed", "chat failed"))
|
||||||
|
return
|
||||||
|
}
|
||||||
|
writeJSON(w, http.StatusOK, chatResponse{Reply: res.Text, Steps: res.Steps, StopReason: res.StopReason})
|
||||||
|
return
|
||||||
|
}
|
||||||
|
// 未接线编排器时的旧同步路径(未接线的 hub
|
||||||
|
// 保持完全与 design/15 之前相同的回复契约)。
|
||||||
|
res, err := s.runner.Run(loop.WithSessionEnv(r.Context(), env), req.SessionID, msgs)
|
||||||
|
if err != nil {
|
||||||
|
s.log.Error("chat run", "err", err, "session", req.SessionID)
|
||||||
|
writeError(w, apperr.Wrap(apperr.ErrInternal, "chat_failed", "chat failed"))
|
||||||
|
return
|
||||||
|
}
|
||||||
|
writeJSON(w, http.StatusOK, chatResponse{Reply: res.Text, Steps: res.Steps, StopReason: res.StopReason})
|
||||||
|
}
|
||||||
|
|
||||||
// detachedTurnContext 返回一个与请求作用域脱钩、带上限的回合上下文(见
|
// detachedTurnContext 返回一个与请求作用域脱钩、带上限的回合上下文(见
|
||||||
// chatStreamTurnBudget)。它刻意**不**把 cancel 交出去:回合活过这次 HTTP 请求,
|
// chatStreamTurnBudget)。它刻意**不**把 cancel 交出去:回合活过这次 HTTP 请求,
|
||||||
// 在这里取消父 ctx 会连坐取消在途回合 —— cancel 只是「提前释放计时器」的优化,
|
// 在这里取消父 ctx 会连坐取消在途回合 —— cancel 只是「提前释放计时器」的优化,
|
||||||
|
|||||||
@@ -10,8 +10,6 @@ import (
|
|||||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/channel/session"
|
"git.yeij.top/AskaEth/Cyrene_v2/internal/channel/session"
|
||||||
apperr "git.yeij.top/AskaEth/Cyrene_v2/internal/errors"
|
apperr "git.yeij.top/AskaEth/Cyrene_v2/internal/errors"
|
||||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/persona/loader"
|
"git.yeij.top/AskaEth/Cyrene_v2/internal/persona/loader"
|
||||||
"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/runtime/multimodal/vision"
|
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/multimodal/vision"
|
||||||
)
|
)
|
||||||
|
|
||||||
@@ -124,7 +122,11 @@ type chatRequest struct {
|
|||||||
DeviceID string `json:"deviceId,omitempty"`
|
DeviceID string `json:"deviceId,omitempty"`
|
||||||
Mode string `json:"mode,omitempty"`
|
Mode string `json:"mode,omitempty"`
|
||||||
ClientMsgID string `json:"clientMsgId,omitempty"`
|
ClientMsgID string `json:"clientMsgId,omitempty"`
|
||||||
Stream bool `json:"stream"`
|
// Documents 是本回合附带的文档引用(design/24 §6.1):最多 maxChatDocuments
|
||||||
|
// 条,每条只给**服务端存储名**(绝不接受任意 URL)。解析与校验见
|
||||||
|
// chat_documents.go。
|
||||||
|
Documents []chatDocumentRef `json:"documents,omitempty"`
|
||||||
|
Stream bool `json:"stream"`
|
||||||
}
|
}
|
||||||
|
|
||||||
// chatResponse 是同步聊天回复的载荷。
|
// chatResponse 是同步聊天回复的载荷。
|
||||||
@@ -272,8 +274,15 @@ func (s *Server) handleChat(w http.ResponseWriter, r *http.Request) {
|
|||||||
writeError(w, err)
|
writeError(w, err)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
if strings.TrimSpace(req.Text) == "" && strings.TrimSpace(req.Image) == "" {
|
// 文档引用只认服务端存储名且必须真实存在,非法 / 不存在一律当场拒(design/24 §4.4)。
|
||||||
writeError(w, apperr.Wrap(apperr.ErrBadRequest, "bad_request", "text or image is required"))
|
docs, err := s.resolveChatDocuments(req.Documents)
|
||||||
|
if err != nil {
|
||||||
|
s.documentsLog().Warn("chat documents rejected", "session", req.SessionID, "count", len(req.Documents))
|
||||||
|
writeError(w, err)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
if strings.TrimSpace(req.Text) == "" && strings.TrimSpace(req.Image) == "" && len(req.Documents) == 0 {
|
||||||
|
writeError(w, apperr.Wrap(apperr.ErrBadRequest, "bad_request", "text、image 或 documents 至少需要一项"))
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
if req.Stream {
|
if req.Stream {
|
||||||
@@ -282,59 +291,8 @@ func (s *Server) handleChat(w http.ResponseWriter, r *http.Request) {
|
|||||||
s.handleChatStream(w, r, req)
|
s.handleChatStream(w, r, req)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
if s.runner == nil {
|
// 剩下的交给派发:把回合交给编排器(或未接线时的旧同步路径)并写响应。
|
||||||
writeError(w, apperr.Wrap(apperr.ErrInternal, "chat_unavailable", "chat service is not configured"))
|
s.submitChatTurn(w, r, req, docs)
|
||||||
return
|
|
||||||
}
|
|
||||||
if req.SessionID == "" {
|
|
||||||
req.SessionID = "web"
|
|
||||||
}
|
|
||||||
// Web 主会话是主人的私密对话(JWT = owner),因此它绑定
|
|
||||||
// 全访问 env:记忆读+写,行为不变。角色
|
|
||||||
// 来自会话的绑定(design/22 §4.5)。
|
|
||||||
msgs := userChatMessages(req.Text, req.Image)
|
|
||||||
env := s.envResolver.owner(r.Context(), req.SessionID)
|
|
||||||
// 客户端的 deviceID、行为模式与可选幂等键(协同 G4)都经 ctx 串进编排器。
|
|
||||||
ctx := withClientMsgHint(withMode(withDeviceID(r.Context(), req.DeviceID), req.Mode), clientMsgHint{ClientMsgID: req.ClientMsgID})
|
|
||||||
if s.hub.orchestrator != nil {
|
|
||||||
sr, res := s.hub.orchestrator.SubmitSync(ctx, req.SessionID, msgs, env)
|
|
||||||
// 重复到达(同一个 clientMsgId):明确告知既有消息 id,而不是再跑一遍
|
|
||||||
// 回合、再落一条用户消息(协同 G4)。
|
|
||||||
if sr.Duplicate {
|
|
||||||
writeJSON(w, http.StatusOK, map[string]any{"duplicate": true, "messageId": sr.MessageID, "sessionId": req.SessionID})
|
|
||||||
return
|
|
||||||
}
|
|
||||||
if !sr.Accepted {
|
|
||||||
if sr.PendingFull {
|
|
||||||
writeError(w, apperr.Wrap(apperr.ErrConflict, "chat_busy", "会话正忙,待处理消息已满"))
|
|
||||||
} else {
|
|
||||||
writeError(w, apperr.Wrap(apperr.ErrBadRequest, "bad_request", "text or image is required"))
|
|
||||||
}
|
|
||||||
return
|
|
||||||
}
|
|
||||||
// 忙碌会话返回非阻塞 ack(design/15 §2.1),而不是
|
|
||||||
// 让请求阻塞在一个可能跑一阵子的回合上。
|
|
||||||
if sr.Pending {
|
|
||||||
writeJSON(w, http.StatusAccepted, map[string]any{"queued": true, "queueDepth": sr.QueueDepth})
|
|
||||||
return
|
|
||||||
}
|
|
||||||
if sr.Err != nil {
|
|
||||||
s.log.Error("chat run", "err", sr.Err, "session", req.SessionID)
|
|
||||||
writeError(w, apperr.Wrap(apperr.ErrInternal, "chat_failed", "chat failed"))
|
|
||||||
return
|
|
||||||
}
|
|
||||||
writeJSON(w, http.StatusOK, chatResponse{Reply: res.Text, Steps: res.Steps, StopReason: res.StopReason})
|
|
||||||
return
|
|
||||||
}
|
|
||||||
// 未接线编排器时的旧同步路径(未接线的 hub
|
|
||||||
// 保持完全与 design/15 之前相同的回复契约)。
|
|
||||||
res, err := s.runner.Run(loop.WithSessionEnv(r.Context(), env), req.SessionID, msgs)
|
|
||||||
if err != nil {
|
|
||||||
s.log.Error("chat run", "err", err, "session", req.SessionID)
|
|
||||||
writeError(w, apperr.Wrap(apperr.ErrInternal, "chat_failed", "chat failed"))
|
|
||||||
return
|
|
||||||
}
|
|
||||||
writeJSON(w, http.StatusOK, chatResponse{Reply: res.Text, Steps: res.Steps, StopReason: res.StopReason})
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// handleListSessions 在接线了 history 存储时返回持久化聊天历史会话
|
// handleListSessions 在接线了 history 存储时返回持久化聊天历史会话
|
||||||
@@ -460,16 +418,6 @@ func firstMode(modes []string) string {
|
|||||||
return modes[0]
|
return modes[0]
|
||||||
}
|
}
|
||||||
|
|
||||||
// userChatMessages 为 chat / stream 请求构建用户回合。image
|
|
||||||
// 存在时该回合成为图片消息(可带文本);否则是
|
|
||||||
// 纯文本消息。空文本加图片得到仅图片的回合。
|
|
||||||
func userChatMessages(text, image string) []model.Message {
|
|
||||||
if strings.TrimSpace(image) != "" {
|
|
||||||
return []model.Message{model.ImageMessage(model.RoleUser, text, image)}
|
|
||||||
}
|
|
||||||
return []model.Message{model.TextMessage(model.RoleUser, text)}
|
|
||||||
}
|
|
||||||
|
|
||||||
// summaryOfSession 把 session.Session 映射为其线上结构。
|
// summaryOfSession 把 session.Session 映射为其线上结构。
|
||||||
func summaryOfSession(sess session.Session) SessionSummary {
|
func summaryOfSession(sess session.Session) SessionSummary {
|
||||||
return SessionSummary{
|
return SessionSummary{
|
||||||
|
|||||||
@@ -20,8 +20,9 @@ func (o *Orchestrator) boundedTail(msgs []model.Message) []model.Message {
|
|||||||
return out
|
return out
|
||||||
}
|
}
|
||||||
|
|
||||||
// messagesEmpty 报告回合是否不携带可用文本或图片,使
|
// messagesEmpty 报告回合是否不携带可用文本、图片或文档,使
|
||||||
// 编排器不会因空洞消息把会话标记为忙碌。
|
// 编排器不会因空洞消息把会话标记为忙碌。文档部件同样算内容:
|
||||||
|
// 「只带附件」的回合是真实回合,不该被判空(design/24 §9.5)。
|
||||||
func (o *Orchestrator) messagesEmpty(msgs []model.Message) bool {
|
func (o *Orchestrator) messagesEmpty(msgs []model.Message) bool {
|
||||||
for _, m := range msgs {
|
for _, m := range msgs {
|
||||||
for _, c := range m.Content {
|
for _, c := range m.Content {
|
||||||
@@ -31,6 +32,9 @@ func (o *Orchestrator) messagesEmpty(msgs []model.Message) bool {
|
|||||||
if c.Type == model.ContentTypeImage && c.ImageURL != "" {
|
if c.Type == model.ContentTypeImage && c.ImageURL != "" {
|
||||||
return false
|
return false
|
||||||
}
|
}
|
||||||
|
if c.Type == model.ContentTypeDocument && c.DocumentName != "" {
|
||||||
|
return false
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
return true
|
return true
|
||||||
|
|||||||
@@ -2,7 +2,9 @@ package gateway
|
|||||||
|
|
||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
|
"log/slog"
|
||||||
|
|
||||||
|
"git.yeij.top/AskaEth/Cyrene_v2/internal/logger"
|
||||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history"
|
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history"
|
||||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model"
|
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model"
|
||||||
)
|
)
|
||||||
@@ -86,7 +88,8 @@ func (o *Orchestrator) sessionTranscript(ctx context.Context, sessionID string)
|
|||||||
// persona_id)作为其说话者标签,因此即使会话被切换过,
|
// persona_id)作为其说话者标签,因此即使会话被切换过,
|
||||||
// 上下文也能把它归属到那个角色;非助手消息保持其
|
// 上下文也能把它归属到那个角色;非助手消息保持其
|
||||||
// 已存发送者标签(design/16 §4.6 的第三方显示名),绝不
|
// 已存发送者标签(design/16 §4.6 的第三方显示名),绝不
|
||||||
// 被重新标为说话角色。
|
// 被重新标为说话角色。文档引用一并还原(design/24 §9.5):
|
||||||
|
// 少了它,附件会在发起它的那一回合之后的所有回合里消失。
|
||||||
func transcriptMessage(h history.Message) model.Message {
|
func transcriptMessage(h history.Message) model.Message {
|
||||||
role := historyRoleToModel(h.Role)
|
role := historyRoleToModel(h.Role)
|
||||||
msg := model.Message{
|
msg := model.Message{
|
||||||
@@ -96,6 +99,7 @@ func transcriptMessage(h history.Message) model.Message {
|
|||||||
if h.Image != "" {
|
if h.Image != "" {
|
||||||
msg.Content = append(msg.Content, model.Content{Type: model.ContentTypeImage, ImageURL: h.Image})
|
msg.Content = append(msg.Content, model.Content{Type: model.ContentTypeImage, ImageURL: h.Image})
|
||||||
}
|
}
|
||||||
|
msg.Content = append(msg.Content, transcriptDocumentParts(role, h)...)
|
||||||
if role == model.RoleAssistant {
|
if role == model.RoleAssistant {
|
||||||
msg.SenderName = h.PersonaID
|
msg.SenderName = h.PersonaID
|
||||||
return msg
|
return msg
|
||||||
@@ -104,6 +108,39 @@ func transcriptMessage(h history.Message) model.Message {
|
|||||||
return msg
|
return msg
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// transcriptDocumentParts 把已存 document 列还原为该回合的文档部件(design/24
|
||||||
|
// §9.5)。部件只交给 RoleUser 回合(§2.3-③):文档在任何别的角色上都会让编码器
|
||||||
|
// fail-loud,而一行坏数据不该把整个会话卡死。
|
||||||
|
//
|
||||||
|
// 列被写坏(或附件落在非用户行上)时**不静默**:打一条带来源的 WARN 并省掉
|
||||||
|
// 该部件,回合照常继续 —— 附件凭空消失正是这条清单要修的缺陷,因此必须留下
|
||||||
|
// 痕迹。日志只带消息 id 与错误,绝不带正文。
|
||||||
|
func transcriptDocumentParts(role model.Role, h history.Message) []model.Content {
|
||||||
|
if h.Document == "" {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
if role != model.RoleUser {
|
||||||
|
transcriptDocumentLog().Warn("transcript document on non-user row", "message", h.ID, "role", h.Role)
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
ref, err := history.DecodeDocument(h.Document)
|
||||||
|
if err != nil {
|
||||||
|
transcriptDocumentLog().Warn("transcript document decode", "message", h.ID, "err", err)
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
if ref == nil || ref.Name == "" {
|
||||||
|
transcriptDocumentLog().Warn("transcript document without name", "message", h.ID)
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
doc := model.Document{
|
||||||
|
Name: ref.Name,
|
||||||
|
MIME: ref.MIME,
|
||||||
|
URL: documentFilePrefix + ref.Name,
|
||||||
|
Size: ref.Size,
|
||||||
|
}
|
||||||
|
return model.DocumentMessage(model.RoleUser, "", doc).Content
|
||||||
|
}
|
||||||
|
|
||||||
// historyRoleToModel 把已存聊天角色映射为模型的角色词汇,
|
// historyRoleToModel 把已存聊天角色映射为模型的角色词汇,
|
||||||
// 未知角色默认归到用户通道。
|
// 未知角色默认归到用户通道。
|
||||||
func historyRoleToModel(role string) model.Role {
|
func historyRoleToModel(role string) model.Role {
|
||||||
@@ -127,3 +164,15 @@ func assistantTailMessage(text, personaID string) model.Message {
|
|||||||
msg.SenderName = personaID
|
msg.SenderName = personaID
|
||||||
return msg
|
return msg
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// transcriptLogSource 是转录重建的日志来源标签。
|
||||||
|
const transcriptLogSource = "gateway/transcript"
|
||||||
|
|
||||||
|
// transcriptDocumentLog 返回标注来源的子日志器。转录重建是纯函数(编排器的
|
||||||
|
// 播种与转向两条路径都只拿得到 history.Message,没有 logger 在作用域内),
|
||||||
|
// 因此以进程默认 logger 为基底再盖上来源标签,使坏数据的 WARN 既可见又可
|
||||||
|
// 追溯(与 internal/auth 的 apikey/jwt 同一手法)。它只用于「数据被写坏」
|
||||||
|
// 这类冷路径,不在每回合的正常路径上打任何日志。
|
||||||
|
func transcriptDocumentLog() *slog.Logger {
|
||||||
|
return logger.WithSource(slog.Default(), transcriptLogSource)
|
||||||
|
}
|
||||||
@@ -62,12 +62,16 @@ type hub struct {
|
|||||||
// envResolver 解析每个回合的角色(design/22 §4.5);它与 server 的解析器
|
// envResolver 解析每个回合的角色(design/22 §4.5);它与 server 的解析器
|
||||||
// 同源,使 WS 路径与 REST 路径不会分叉。
|
// 同源,使 WS 路径与 REST 路径不会分叉。
|
||||||
envResolver turnEnvResolver
|
envResolver turnEnvResolver
|
||||||
log *slog.Logger
|
// documents 解析回合携带的文档引用(design/24 §6.1);同样与 server 的
|
||||||
upgrader websocket.Upgrader
|
// 解析器同源,使 WS 帧与 REST 的文档校验口径一致。为 nil 时带附件的回合
|
||||||
origins map[string]bool
|
// 被明确拒绝(见 resolveDocuments),绝不静默丢弃。
|
||||||
allowAll bool
|
documents documentResolver
|
||||||
cancel context.CancelFunc
|
log *slog.Logger
|
||||||
streamSeq atomic.Uint64
|
upgrader websocket.Upgrader
|
||||||
|
origins map[string]bool
|
||||||
|
allowAll bool
|
||||||
|
cancel context.CancelFunc
|
||||||
|
streamSeq atomic.Uint64
|
||||||
}
|
}
|
||||||
|
|
||||||
// newHub 构建一个绑定到 server 依赖的 hub。
|
// newHub 构建一个绑定到 server 依赖的 hub。
|
||||||
@@ -83,6 +87,7 @@ func newHub(s *Server) *hub {
|
|||||||
streaming: s.streaming,
|
streaming: s.streaming,
|
||||||
history: s.history,
|
history: s.history,
|
||||||
envResolver: s.envResolver,
|
envResolver: s.envResolver,
|
||||||
|
documents: s.resolveChatDocuments,
|
||||||
log: s.log,
|
log: s.log,
|
||||||
origins: origins,
|
origins: origins,
|
||||||
allowAll: allowAll,
|
allowAll: allowAll,
|
||||||
|
|||||||
@@ -10,6 +10,7 @@ import (
|
|||||||
"time"
|
"time"
|
||||||
|
|
||||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/bus"
|
"git.yeij.top/AskaEth/Cyrene_v2/internal/bus"
|
||||||
|
"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/loop"
|
||||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model"
|
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model"
|
||||||
)
|
)
|
||||||
@@ -55,18 +56,28 @@ func (h *hub) streamChat(ctx context.Context, payload any) SubmitResult {
|
|||||||
if b, err := json.Marshal(payload); err == nil {
|
if b, err := json.Marshal(payload); err == nil {
|
||||||
_ = json.Unmarshal(b, &req)
|
_ = json.Unmarshal(b, &req)
|
||||||
}
|
}
|
||||||
if h.runner == nil || (strings.TrimSpace(req.Text) == "" && strings.TrimSpace(req.Image) == "") {
|
emptyTurn := strings.TrimSpace(req.Text) == "" && strings.TrimSpace(req.Image) == "" && len(req.Documents) == 0
|
||||||
|
if h.runner == nil || emptyTurn {
|
||||||
h.broadcast(wsEvent{Type: "stream_end", TurnID: newTurnID(), Seq: int(h.streamSeq.Add(1)), SessionID: req.SessionID, Data: map[string]any{"stopReason": "idle"}})
|
h.broadcast(wsEvent{Type: "stream_end", TurnID: newTurnID(), Seq: int(h.streamSeq.Add(1)), SessionID: req.SessionID, Data: map[string]any{"stopReason": "idle"}})
|
||||||
return SubmitResult{}
|
return SubmitResult{}
|
||||||
}
|
}
|
||||||
if req.SessionID == "" {
|
if req.SessionID == "" {
|
||||||
req.SessionID = "web"
|
req.SessionID = "web"
|
||||||
}
|
}
|
||||||
|
// 文档引用在本进程内解析(只认服务端存储名 + 确实存在),失败**可见**:
|
||||||
|
// 回一条机器可读的 error 帧并中止本回合,绝不静默丢掉附件(design/24 §4.4)。
|
||||||
|
docs, derr := h.resolveDocuments(req.Documents)
|
||||||
|
if derr != nil {
|
||||||
|
logger.WithSource(h.log, documentsLogSource).Warn("chat documents rejected",
|
||||||
|
"session", req.SessionID, "count", len(req.Documents), "err", derr)
|
||||||
|
h.broadcast(wsEvent{Type: "error", SessionID: req.SessionID, Data: map[string]any{"code": chatDocumentErrorCode(derr)}})
|
||||||
|
return SubmitResult{}
|
||||||
|
}
|
||||||
// Web 主会话是 owner 的私有对话(JWT = owner),因此它绑定全权限
|
// Web 主会话是 owner 的私有对话(JWT = owner),因此它绑定全权限
|
||||||
// env:记忆读写,行为不变。角色来自会话的绑定(design/22 §4.5):
|
// env:记忆读写,行为不变。角色来自会话的绑定(design/22 §4.5):
|
||||||
// 切到角色 Y 的会话在下一轮以 Y 应答,而未绑定的/既有的会话继续
|
// 切到角色 Y 的会话在下一轮以 Y 应答,而未绑定的/既有的会话继续
|
||||||
// 解析为全局默认角色。
|
// 解析为全局默认角色。
|
||||||
msgs := userChatMessages(req.Text, req.Image)
|
msgs := chatTurnMessages(req.Text, req.Image, docs)
|
||||||
// 通过 ctx 把客户端的 deviceID、行为模式与可选幂等键串进
|
// 通过 ctx 把客户端的 deviceID、行为模式与可选幂等键串进
|
||||||
// orchestrator,使持久化的用户轮次与助手回复都带上标记,并让重复到达的
|
// orchestrator,使持久化的用户轮次与助手回复都带上标记,并让重复到达的
|
||||||
// clientMsgId 被判重(design/16 §4.5、§5.1;协同 G4)。空 clientMsgId 是
|
// clientMsgId 被判重(design/16 §4.5、§5.1;协同 G4)。空 clientMsgId 是
|
||||||
@@ -206,6 +217,24 @@ type chatStreamRequest struct {
|
|||||||
// `chat_duplicate`(data 带既有 messageId)。它可选,省略时行为与加入该
|
// `chat_duplicate`(data 带既有 messageId)。它可选,省略时行为与加入该
|
||||||
// 字段之前逐字相同。
|
// 字段之前逐字相同。
|
||||||
ClientMsgID string `json:"clientMsgId,omitempty"`
|
ClientMsgID string `json:"clientMsgId,omitempty"`
|
||||||
|
// Documents 是本回合附带的文档引用(design/24 §6.1),与 REST 的
|
||||||
|
// `chatRequest.documents` **同形**:只给服务端存储名,绝不接受 URL。
|
||||||
|
// 上行帧仍有 4096 字节硬上限 ⇒ 文档实际上走 REST;这里保持同形解析,
|
||||||
|
// 使两条入口不会分叉(解析见 resolveDocuments)。
|
||||||
|
Documents []chatDocumentRef `json:"documents,omitempty"`
|
||||||
|
}
|
||||||
|
|
||||||
|
// resolveDocuments 解析本回合携带的文档引用(design/24 §4.4/§6.1)。一条都不带
|
||||||
|
// 是空操作;未接解析器(未接线的构建 / 单元测试构造的裸 hub)时**明确报错**,
|
||||||
|
// 绝不把附件静默丢掉。
|
||||||
|
func (h *hub) resolveDocuments(refs []chatDocumentRef) ([]model.Document, error) {
|
||||||
|
if len(refs) == 0 {
|
||||||
|
return nil, nil
|
||||||
|
}
|
||||||
|
if h.documents == nil {
|
||||||
|
return nil, documentsUnavailableError()
|
||||||
|
}
|
||||||
|
return h.documents(refs)
|
||||||
}
|
}
|
||||||
|
|
||||||
// newTurnID 返回新的随机十六进制标识,用于把一串流式 WS 事件关联到
|
// newTurnID 返回新的随机十六进制标识,用于把一串流式 WS 事件关联到
|
||||||
|
|||||||
Reference in new issue
Block a user