Files
Agent-System-Backend/internal/server/server.go
T
weichen 797a6eaab7 Add inbox activity cards: cardsvc, store, events, API and producers
Why: T2 of the interaction foundation — activity cards as first-class state
  objects behind the right-rail inbox (proposal D25): upload ingest cards
  (SR-A5 failures explicit, artifact retrievable), HITL projection cards
  (double-presence, ref-linked two-way sync), squad orchestration cards and
  the mock #upload demo of the full card lifecycle.

Evidence: go test ./... green across 14 packages incl. new cardsvc suite
  (lifecycle, action labels and idempotency, pin/dismiss, owner isolation,
  resolve-by-ref idempotency) and updated migration test (v6 activity_cards,
  both dialects); events flow through the session event buffer (persist, seq,
  replay); receipts appended to the conversation timeline on card actions.

Impact: additive; cards disabled (SetCards nil) degenerates to the T0
  baseline; upload cards persist at project level per the v1 inbox boundary
  (conversation-scoped inbox; project-level aggregation is follow-up).

Refs: modules/documents docs/proposals/2026-09-30-ui-inbox-command-workspace.md
2026-09-30 06:52:48 +00:00

941 lines
31 KiB
Go
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
// Package server 提供 HTTP/SSE API 与内嵌前端(go:embed)。
// API-first:前端只是第一个客户端;全部能力走公开 API。
package server
import (
"encoding/json"
"errors"
"fmt"
"io"
"io/fs"
"net/http"
"os"
"path/filepath"
"strconv"
"strings"
"time"
"agent-system-backend/internal/cardsvc"
"agent-system-backend/internal/ingest"
"agent-system-backend/internal/plugins"
"agent-system-backend/internal/profiles"
"agent-system-backend/internal/ratelimit"
"agent-system-backend/internal/runmgr"
"agent-system-backend/internal/settings"
"agent-system-backend/internal/store"
"agent-system-backend/internal/tools"
"agent-system-backend/web"
core "agent-system-core"
)
type Server struct {
st *store.Store
Auth *Authenticator // API key 认证(nil/未启用 = 默认租户;组合根注入)
Retriever core.Retriever // 混合检索(nil = store 全文;组合根注入)
Indexer core.DocumentIndexer // 检索索引入口(nil = 整文档单块;组合根注入)
Limiter ratelimit.Limiter // 租户限流(nil = 不限;组合根注入)
LimitLookup func() int // 当前限流值(次/分;<=0 不限;组合根缓存注入)
mgr *runmgr.Manager
Cards *cardsvc.Service // 收件箱任务卡(nil = 不启用;组合根注入)
reg *tools.Registry
prof *profiles.Service
set *settings.Service
plugins *plugins.Host
static fs.FS
// Subscribe 注入 bus 订阅(事件在线转发;重放走 store 游标)。
Subscribe func(conversationID string) (<-chan core.StreamEvent, func())
}
func New(st *store.Store, mgr *runmgr.Manager, reg *tools.Registry, p *profiles.Service, s *settings.Service, ph *plugins.Host) *Server {
sub, err := web.Dist()
if err != nil {
panic(err) // embed 布局错误属构建期问题
}
return &Server{st: st, mgr: mgr, reg: reg, prof: p, set: s, plugins: ph, static: sub}
}
func (s *Server) Handler() http.Handler {
mux := http.NewServeMux()
// ---- 项目 / 会话 ----
mux.HandleFunc("GET /api/projects", s.listProjects)
mux.HandleFunc("POST /api/projects", s.createProject)
mux.HandleFunc("PATCH /api/projects/{id}", s.patchProject)
mux.HandleFunc("DELETE /api/projects/{id}", s.deleteProject)
mux.HandleFunc("GET /api/projects/{id}/conversations", s.listConversations)
mux.HandleFunc("POST /api/projects/{id}/conversations", s.createConversation)
mux.HandleFunc("GET /api/conversations/{id}", s.conversationDetail)
mux.HandleFunc("PATCH /api/conversations/{id}", s.patchConversation)
mux.HandleFunc("DELETE /api/conversations/{id}", s.deleteConversation)
mux.HandleFunc("POST /api/conversations/{id}/messages", s.postMessage)
mux.HandleFunc("POST /api/runs/{id}/cancel", s.cancelRun)
mux.HandleFunc("GET /api/runs/{id}", s.getRun)
// ---- HITL ----
mux.HandleFunc("POST /api/hitl/{id}/resolve", s.resolveHITL)
// ---- 收件箱任务卡(提案 T2)----
mux.HandleFunc("GET /api/conversations/{id}/cards", s.listCards)
mux.HandleFunc("POST /api/cards/{id}/actions/{actionId}", s.cardAction)
mux.HandleFunc("POST /api/cards/{id}/pin", s.cardPin)
mux.HandleFunc("POST /api/cards/{id}/dismiss", s.cardDismiss)
// ---- 事件流(SSE)----
mux.HandleFunc("GET /api/events", s.events)
// ---- 上传与工件 ----
mux.HandleFunc("POST /api/projects/{id}/uploads", s.upload)
mux.HandleFunc("GET /api/projects/{id}/artifacts", s.listArtifacts)
mux.HandleFunc("GET /api/artifacts/{id}", s.getArtifact)
mux.HandleFunc("GET /api/artifacts/{id}/content", s.artifactContent)
mux.HandleFunc("PUT /api/artifacts/{id}/content", s.putArtifactContent)
mux.HandleFunc("DELETE /api/artifacts/{id}", s.deleteArtifact)
// ---- 画像 / 设置 / 工具 / 插件 / 注册表 ----
mux.HandleFunc("GET /api/profiles", s.listProfiles)
mux.HandleFunc("POST /api/profiles", s.saveProfile)
mux.HandleFunc("DELETE /api/profiles/{id}", s.deleteProfile)
mux.HandleFunc("GET /api/settings", s.getSettings)
mux.HandleFunc("PUT /api/settings/user", s.putSettingsUser)
mux.HandleFunc("PUT /api/settings/project/{id}", s.putSettingsProject)
mux.HandleFunc("GET /api/tools", s.listTools)
mux.HandleFunc("POST /api/tools", s.createUserTool)
mux.HandleFunc("DELETE /api/tools/{name}", s.deleteUserTool)
mux.HandleFunc("GET /api/plugins", s.listPlugins)
mux.HandleFunc("POST /api/plugins/reload", s.reloadPlugins)
mux.HandleFunc("GET /api/plugins/settings-schema", s.pluginSettingsSchema)
mux.HandleFunc("GET /api/registry", s.registry)
// ---- 计划 / 待办 / 命令 / 解析器 / 检索 ----
mux.HandleFunc("GET /api/conversations/{id}/plan", s.getPlan)
mux.HandleFunc("GET /api/conversations/{id}/todos", s.getTodos)
mux.HandleFunc("GET /api/commands", s.listCommands)
mux.HandleFunc("GET /api/parsers", s.listParsers)
mux.HandleFunc("POST /api/projects/{id}/knowledge", s.addKnowledge)
mux.HandleFunc("GET /api/knowledge/search", s.searchKnowledge)
mux.HandleFunc("GET /api/health", func(w http.ResponseWriter, r *http.Request) {
writeJSON(w, http.StatusOK, map[string]any{"ok": true, "ts": time.Now().UTC()})
})
// ---- 管理(嵌入式档在线备份)----
mux.HandleFunc("POST /api/admin/backup", s.backup)
// API 未匹配路径一律 404(不回落 SPA,避免 /api/* 返回 HTML 200)。
mux.HandleFunc("/api/", func(w http.ResponseWriter, r *http.Request) { http.NotFound(w, r) })
// 静态前端(embed);API 全链路:认证(未启用 = 默认租户)→ 租户限流。
inner := s.withRateLimit(mux)
var h http.Handler = inner
if s.Auth != nil {
h = s.Auth.Middleware(inner)
}
mux.HandleFunc("/", s.serveStatic)
return h
}
// ---- SSE:游标重放 + 在线订阅 + 终态对账由 Query 承担 ----
func (s *Server) events(w http.ResponseWriter, r *http.Request) {
conversationID := r.URL.Query().Get("conversationId")
if conversationID == "" {
http.Error(w, "conversationId required", http.StatusBadRequest)
return
}
flusher, ok := w.(http.Flusher)
if !ok {
http.Error(w, "streaming unsupported", http.StatusInternalServerError)
return
}
w.Header().Set("Content-Type", "text/event-stream")
w.Header().Set("Cache-Control", "no-cache")
w.Header().Set("Connection", "keep-alive")
w.Header().Set("X-Accel-Buffering", "no")
// 租户边界:跨租户会话不可订阅(NotFound,不泄露存在性)。
if _, err := s.st.GetConversation(s.owner(r), conversationID); err != nil {
writeStoreErr(w, err)
return
}
since := int64(0)
if ids := r.Header.Get("Last-Event-ID"); ids != "" {
since, _ = strconv.ParseInt(ids, 10, 64)
} else if q := r.URL.Query().Get("since"); q != "" {
since, _ = strconv.ParseInt(q, 10, 64)
}
// 先订阅再重放,避免重放与订阅间隙丢事件;按 seq 去重。
live, stop := s.Subscribe(conversationID)
defer stop()
send := func(e core.StreamEvent) bool {
buf, _ := json.Marshal(e)
fmt.Fprintf(w, "id: %d\nevent: %s\ndata: %s\n\n", e.Seq, e.Type, buf)
flusher.Flush()
return r.Context().Err() == nil
}
lastSent := since
replay, err := s.st.EventsAfter(conversationID, since, 1000)
if err == nil {
for _, e := range replay {
if e.Seq <= lastSent {
continue
}
if !send(e) {
stop()
return
}
lastSent = e.Seq
}
}
// 心跳 + 在线转发。
heartbeat := time.NewTicker(15 * time.Second)
defer heartbeat.Stop()
for {
select {
case <-r.Context().Done():
stop()
return
case <-heartbeat.C:
fmt.Fprint(w, ": ping\n\n")
flusher.Flush()
case e := <-live:
if e.ConversationID != conversationID || e.Seq <= lastSent {
continue
}
if !send(e) {
stop()
return
}
lastSent = e.Seq
}
}
}
func (s *Server) serveStatic(w http.ResponseWriter, r *http.Request) {
path := strings.TrimPrefix(r.URL.Path, "/")
if path == "" {
path = "index.html"
}
data, err := fs.ReadFile(s.static, path)
if err != nil {
// SPA 回落 index.html。
data, err = fs.ReadFile(s.static, "index.html")
if err != nil {
http.NotFound(w, r)
return
}
path = "index.html"
}
switch {
case strings.HasSuffix(path, ".html"):
w.Header().Set("Content-Type", "text/html; charset=utf-8")
case strings.HasSuffix(path, ".js"):
w.Header().Set("Content-Type", "text/javascript; charset=utf-8")
case strings.HasSuffix(path, ".css"):
w.Header().Set("Content-Type", "text/css; charset=utf-8")
case strings.HasSuffix(path, ".svg"):
w.Header().Set("Content-Type", "image/svg+xml")
}
w.Write(data)
}
// ---- handlers:项目与会话 ----
func (s *Server) listProjects(w http.ResponseWriter, r *http.Request) {
ps, err := s.st.ListProjects(s.owner(r))
writeJSONOr(w, ps, err)
}
func (s *Server) createProject(w http.ResponseWriter, r *http.Request) {
var req struct {
Name string `json:"name"`
}
if err := json.NewDecoder(r.Body).Decode(&req); err != nil || strings.TrimSpace(req.Name) == "" {
http.Error(w, "name required", http.StatusBadRequest)
return
}
p, err := s.st.CreateProject(s.owner(r), strings.TrimSpace(req.Name))
writeJSONOr(w, p, err)
}
func (s *Server) patchProject(w http.ResponseWriter, r *http.Request) {
id := r.PathValue("id")
var req struct {
Name *string `json:"name"`
Settings map[string]any `json:"settings"`
}
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
http.Error(w, err.Error(), http.StatusBadRequest)
return
}
if req.Name != nil {
if err := s.st.RenameProject(s.owner(r), id, *req.Name); err != nil {
writeStoreErr(w, err)
return
}
}
if req.Settings != nil {
if err := s.st.UpdateProjectSettings(s.owner(r), id, req.Settings); err != nil {
writeStoreErr(w, err)
return
}
}
p, err := s.st.GetProject(s.owner(r), id)
writeJSONOr(w, p, err)
}
func (s *Server) deleteProject(w http.ResponseWriter, r *http.Request) {
writeOKOr(w, s.st.DeleteProject(s.owner(r), r.PathValue("id")))
}
func (s *Server) listConversations(w http.ResponseWriter, r *http.Request) {
owner := s.owner(r)
if _, err := s.st.GetProject(owner, r.PathValue("id")); err != nil {
writeStoreErr(w, err)
return
}
cs, err := s.st.ListConversations(owner, r.PathValue("id"))
writeJSONOr(w, cs, err)
}
func (s *Server) createConversation(w http.ResponseWriter, r *http.Request) {
var req struct {
Title string `json:"title"`
ProfileID string `json:"profileId"`
}
json.NewDecoder(r.Body).Decode(&req)
if req.Title == "" {
req.Title = "新会话"
}
c, err := s.st.CreateConversation(s.owner(r), r.PathValue("id"), req.Title, req.ProfileID)
writeJSONOr(w, c, err)
}
// conversationDetail 终态对账权威:消息树 + runs + 活动 HITL 卡片。
func (s *Server) conversationDetail(w http.ResponseWriter, r *http.Request) {
id := r.PathValue("id")
owner := s.owner(r)
conv, err := s.st.GetConversation(owner, id)
if err != nil {
writeStoreErr(w, err)
return
}
msgs, _ := s.st.ListMessages(id)
runs, _ := s.st.ListRunsByConversation(id)
cards, _ := s.st.ActiveHITLCards(owner, id)
lastSeq, _ := s.st.LastEventSeq(id)
writeJSON(w, http.StatusOK, map[string]any{
"conversation": conv, "messages": msgs, "runs": runs,
"activeHitl": cards, "lastSeq": lastSeq,
})
}
func (s *Server) patchConversation(w http.ResponseWriter, r *http.Request) {
id := r.PathValue("id")
var req struct {
Title string `json:"title"`
ProfileID string `json:"profileId"`
}
json.NewDecoder(r.Body).Decode(&req)
writeOKOr(w, s.st.UpdateConversation(s.owner(r), id, req.Title, req.ProfileID))
}
func (s *Server) deleteConversation(w http.ResponseWriter, r *http.Request) {
writeOKOr(w, s.st.DeleteConversation(s.owner(r), r.PathValue("id")))
}
// ---- 消息发送(含命令路由与附件) ----
func (s *Server) postMessage(w http.ResponseWriter, r *http.Request) {
convID := r.PathValue("id")
owner := s.owner(r)
conv, err := s.st.GetConversation(owner, convID)
if err != nil {
writeStoreErr(w, err)
return
}
var req struct {
Text string `json:"text"`
AttachmentIDs []string `json:"attachmentIds"`
}
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
http.Error(w, err.Error(), http.StatusBadRequest)
return
}
text := strings.TrimSpace(req.Text)
if text == "" && len(req.AttachmentIDs) == 0 {
http.Error(w, "text or attachments required", http.StatusBadRequest)
return
}
// 命令路由:/ 开头走内置命令,不启动 run(4.11 输入增强)。
if strings.HasPrefix(text, "/") {
s.handleCommand(w, r, conv, text)
return
}
parts := []core.Part{}
if text != "" {
parts = append(parts, core.Part{Type: core.PartText, Text: text})
}
var parsedDocs []core.ParsedDocument
for _, artID := range req.AttachmentIDs {
art, err := s.st.GetArtifact(owner, artID)
if err != nil {
http.Error(w, "attachment "+artID+" not found", http.StatusBadRequest)
return
}
content, _ := s.st.ArtifactContent(owner, artID)
doc := ingest.Parse(art.Name, art.MIME, content)
parsedDocs = append(parsedDocs, doc)
parts = append(parts, core.Part{Type: core.PartArtifactRef, Artifact: &core.ArtifactRef{ArtifactID: art.ID, Name: art.Name, Kind: art.Kind}})
}
msg := core.Message{ConversationID: convID, Role: core.RoleUser, Parts: parts}
if err := s.st.AppendMessage(&msg); err != nil {
writeStoreErr(w, err)
return
}
run, err := s.mgr.StartConversationRun(conv, msg, parsedDocs)
writeJSONOr(w, map[string]any{"message": msg, "run": run}, err)
}
// handleCommand 内置命令(注册表数据驱动,插件可扩展 commands 后补)。
func (s *Server) handleCommand(w http.ResponseWriter, r *http.Request, conv core.Conversation, text string) {
fields := strings.Fields(text)
cmd := strings.ToLower(fields[0])
switch cmd {
case "/help":
writeJSON(w, http.StatusOK, map[string]any{"command": cmd, "reply": "可用命令:/new 新会话 · /mock on|off 演示模式 · /model <名称> 设置主模型 · /profile <id> 切换画像 · /help 帮助"})
case "/new":
c, err := s.st.CreateConversation(conv.Owner, conv.ProjectID, "新会话", conv.ProfileID)
writeJSONOr(w, map[string]any{"command": cmd, "conversation": c}, err)
case "/mock":
if len(fields) < 2 || (fields[1] != "on" && fields[1] != "off") {
http.Error(w, "usage: /mock on|off", http.StatusBadRequest)
return
}
val := fields[1] == "on"
writeJSONOr(w, map[string]any{"command": cmd, "runtime.mock": val}, s.set.MergeUser(conv.Owner, map[string]any{"runtime": map[string]any{"mock": val}}))
case "/model":
if len(fields) < 2 {
http.Error(w, "usage: /model <model-name>", http.StatusBadRequest)
return
}
writeJSONOr(w, map[string]any{"command": cmd, "model": fields[1]}, s.set.MergeUser(conv.Owner, map[string]any{"models": map[string]any{"chat": map[string]any{"model": fields[1]}}}))
case "/profile":
if len(fields) < 2 {
http.Error(w, "usage: /profile <id>", http.StatusBadRequest)
return
}
writeJSONOr(w, map[string]any{"command": cmd, "profile": fields[1]}, s.st.UpdateConversation(conv.Owner, conv.ID, "", fields[1]))
default:
http.Error(w, "未知命令 "+cmd+"(/help 查看)", http.StatusBadRequest)
}
}
func (s *Server) cancelRun(w http.ResponseWriter, r *http.Request) {
writeOKOr(w, s.mgr.CancelRun(r.PathValue("id")))
}
func (s *Server) getRun(w http.ResponseWriter, r *http.Request) {
run, err := s.st.GetRun(s.owner(r), r.PathValue("id"))
if err != nil {
writeStoreErr(w, err)
return
}
children, _ := s.st.ChildRuns(run.ID)
writeJSON(w, http.StatusOK, map[string]any{"run": run, "children": children})
}
// ---- HITL ----
func (s *Server) resolveHITL(w http.ResponseWriter, r *http.Request) {
var resp core.HITLResponse
if err := json.NewDecoder(r.Body).Decode(&resp); err != nil {
http.Error(w, err.Error(), http.StatusBadRequest)
return
}
resp.CardID = r.PathValue("id")
pivotText, pivoted, err := s.mgr.ResolveHITL(s.owner(r), string(resp.CardID), resp)
if err != nil {
writeStoreErr(w, err)
return
}
writeJSON(w, http.StatusOK, map[string]any{"resolved": true, "pivoted": pivoted, "pivotText": pivotText})
}
// ---- 收件箱任务卡 ----
func (s *Server) listCards(w http.ResponseWriter, r *http.Request) {
owner := s.owner(r)
if _, err := s.st.GetConversation(owner, r.PathValue("id")); err != nil {
writeStoreErr(w, err)
return
}
cards, err := s.Cards.List(owner, r.PathValue("id"))
writeJSONOr(w, map[string]any{"cards": cards}, err)
}
// cardAction 处理卡上按钮:resolved + 回执消息进对话时间线(确认即留痕)。
func (s *Server) cardAction(w http.ResponseWriter, r *http.Request) {
owner := s.owner(r)
var body struct {
FreeText string `json:"freeText,omitempty"`
}
_ = json.NewDecoder(r.Body).Decode(&body)
card, label, err := s.Cards.Action(owner, r.PathValue("id"), r.PathValue("actionId"), body.FreeText)
if err != nil {
writeStoreErr(w, err)
return
}
// 回执消息:确认结论写进对话时间线(含所点按钮语义),审计锚点为 card.resolved 事件。
if card.ConversationID != "" {
msg := core.Message{ID: store.NewID("msg"), ConversationID: card.ConversationID, Role: core.RoleAssistant, CreatedAt: time.Now().UTC()}
msg.Parts = []core.Part{{Type: core.PartText, Text: fmt.Sprintf("✓ 已按「%s」处理「%s」——结论已记录并留痕(card:%s)", label, card.Title, card.ID)}}
_ = s.st.AppendMessage(&msg)
}
writeJSON(w, http.StatusOK, map[string]any{"card": card})
}
func (s *Server) cardPin(w http.ResponseWriter, r *http.Request) {
var body struct {
Pinned bool `json:"pinned"`
}
_ = json.NewDecoder(r.Body).Decode(&body)
card, err := s.Cards.SetPinned(s.owner(r), r.PathValue("id"), body.Pinned)
writeJSONOr(w, map[string]any{"card": card}, err)
}
func (s *Server) cardDismiss(w http.ResponseWriter, r *http.Request) {
card, err := s.Cards.Dismiss(s.owner(r), r.PathValue("id"))
writeJSONOr(w, map[string]any{"card": card}, err)
}
// ---- 上传与工件 ----
func (s *Server) upload(w http.ResponseWriter, r *http.Request) {
projectID := r.PathValue("id")
r.ParseMultipartForm(64 << 20)
file, header, err := r.FormFile("file")
if err != nil {
http.Error(w, "file field required", http.StatusBadRequest)
return
}
defer file.Close()
content, err := io.ReadAll(io.LimitReader(file, 100<<20))
if err != nil {
http.Error(w, err.Error(), http.StatusBadRequest)
return
}
owner := s.owner(r)
if _, err := s.st.GetProject(owner, projectID); err != nil {
writeStoreErr(w, err)
return
}
mime := header.Header.Get("Content-Type")
art := core.Artifact{ProjectID: projectID, Owner: owner, Name: header.Filename, Kind: kindOf(header.Filename), MIME: mime}
if err := s.st.CreateArtifact(&art, content); err != nil {
writeStoreErr(w, err)
return
}
// 摄取卡(SR-A5 界面化):解析过程显式可见,失败不静默——原始文件已存工件可取回。
// v1 边界:上传为项目级事件,卡片按项目持久化(conversationId 空);会话内演示见 mock #upload。
ingestCard := core.ActivityCard{
ProjectID: projectID, Kind: core.CardKindProgress, Title: header.Filename,
Status: core.CardActive, Tone: "info",
Body: []core.UINode{{Type: core.UIProgress, Progress: &core.UIProgressSpec{Label: "解析中", Value: 0.3}}},
}
if s.Cards != nil {
ingestCard, _ = s.Cards.Create(owner, ingestCard, "")
}
doc := ingest.Parse(art.Name, mime, content)
if s.Cards != nil {
ingestCard.Status = core.CardDone
ingestCard.Tone = "ok"
body := []core.UINode{
{Type: core.UIKV, KV: [][2]string{{"解析器", doc.Meta.Parser}, {"内容块", fmt.Sprint(len(doc.Blocks))}}},
{Type: core.UIActions, Actions: []core.UIAction{{ID: "open", Label: "📄 打开工件", Command: ""}}},
}
if len(doc.Warnings) > 0 {
ingestCard.Status = core.CardActionRequired
ingestCard.Tone = "warn"
for _, wn := range doc.Warnings {
body = append(body, core.UINode{Type: core.UIMarkdown, Markdown: "⚠ " + wn})
}
body = append(body, core.UINode{Type: core.UIMarkdown, Markdown: "_解析不完整不静默:原始文件已永久保留,可取回或重试。_"})
}
ingestCard.Body = body
_ = s.Cards.Update(owner, ingestCard)
}
// 文本可解析的进检索索引(管线分块+全文+可选向量)。
if plain := doc.PlainText(); plain != "" && len(plain) < 200000 {
if s.Indexer != nil {
_ = s.Indexer.IndexDocument(owner, projectID, "upload", art.Name, art.ID, plain)
} else {
_ = s.st.AddKnowledge(core.KnowledgeDoc{Owner: owner, ProjectID: projectID, Source: "upload", Title: art.Name, Ref: art.ID, Chunk: plain})
}
}
writeJSON(w, http.StatusOK, core.UploadInfo{ArtifactID: art.ID, Filename: art.Name, MIME: mime, Size: art.Size, Parsed: &doc})
}
func kindOf(filename string) string {
switch {
case strings.HasSuffix(strings.ToLower(filename), ".md"), strings.HasSuffix(strings.ToLower(filename), ".txt"), strings.HasSuffix(strings.ToLower(filename), ".docx"), strings.HasSuffix(strings.ToLower(filename), ".pdf"):
return "document"
case strings.Contains(strings.ToLower(filename), ".png"), strings.Contains(strings.ToLower(filename), ".jpg"), strings.Contains(strings.ToLower(filename), ".jpeg"), strings.Contains(strings.ToLower(filename), ".webp"), strings.Contains(strings.ToLower(filename), ".gif"):
return "image"
default:
return "data"
}
}
func (s *Server) listArtifacts(w http.ResponseWriter, r *http.Request) {
owner := s.owner(r)
if _, err := s.st.GetProject(owner, r.PathValue("id")); err != nil {
writeStoreErr(w, err)
return
}
as, err := s.st.ListArtifacts(owner, r.PathValue("id"))
writeJSONOr(w, as, err)
}
func (s *Server) getArtifact(w http.ResponseWriter, r *http.Request) {
a, err := s.st.GetArtifact(s.owner(r), r.PathValue("id"))
writeJSONOr(w, a, err)
}
func (s *Server) artifactContent(w http.ResponseWriter, r *http.Request) {
a, err := s.st.GetArtifact(s.owner(r), r.PathValue("id"))
if err != nil {
writeStoreErr(w, err)
return
}
content, err := s.st.ArtifactContent(s.owner(r), a.ID)
if err != nil {
writeStoreErr(w, err)
return
}
w.Header().Set("Content-Type", contentTypeOf(a))
w.Write(content)
}
func (s *Server) putArtifactContent(w http.ResponseWriter, r *http.Request) {
body, _ := io.ReadAll(io.LimitReader(r.Body, 20<<20))
writeOKOr(w, s.st.UpdateArtifactContent(s.owner(r), r.PathValue("id"), body))
}
func (s *Server) deleteArtifact(w http.ResponseWriter, r *http.Request) {
writeOKOr(w, s.st.DeleteArtifact(s.owner(r), r.PathValue("id")))
}
// ---- 画像 ----
func (s *Server) listProfiles(w http.ResponseWriter, r *http.Request) {
ps, err := s.prof.List(s.owner(r))
writeJSONOr(w, ps, err)
}
func (s *Server) saveProfile(w http.ResponseWriter, r *http.Request) {
var p core.AgentProfile
if err := json.NewDecoder(r.Body).Decode(&p); err != nil {
http.Error(w, err.Error(), http.StatusBadRequest)
return
}
if p.ID == "" {
p.ID = "prof_" + strconv.FormatInt(time.Now().UnixNano()%100000, 10)
}
if err := s.prof.Save(s.owner(r), p); err != nil {
http.Error(w, err.Error(), http.StatusBadRequest)
return
}
writeJSON(w, http.StatusOK, p)
}
func (s *Server) deleteProfile(w http.ResponseWriter, r *http.Request) {
writeOKOr(w, s.prof.Delete(s.owner(r), r.PathValue("id")))
}
// ---- 设置 ----
func (s *Server) getSettings(w http.ResponseWriter, r *http.Request) {
owner := s.owner(r)
projectID := r.URL.Query().Get("project")
schema := append(settings.Schema(), s.plugins.SettingsSchema()...)
writeJSON(w, http.StatusOK, map[string]any{
"schema": schema,
"defaults": settings.Masked(core.EffectiveSettings(settings.SystemDefaults())),
"user": maskMap(mustUser(s.set, owner)),
"project": maskMap(mustScope(s.set, owner, projectID)),
})
}
func mustUser(s *settings.Service, owner string) map[string]any {
m, _ := s.UserScope(owner)
if m == nil {
return map[string]any{}
}
return m
}
func mustScope(s *settings.Service, owner, projectID string) map[string]any {
if projectID == "" {
return map[string]any{}
}
m, _ := s.ProjectScope(owner, projectID)
return m
}
func (s *Server) putSettingsUser(w http.ResponseWriter, r *http.Request) {
var values map[string]any
if err := json.NewDecoder(r.Body).Decode(&values); err != nil {
http.Error(w, err.Error(), http.StatusBadRequest)
return
}
writeOKOr(w, s.set.MergeUser(s.owner(r), values))
}
func (s *Server) putSettingsProject(w http.ResponseWriter, r *http.Request) {
var values map[string]any
if err := json.NewDecoder(r.Body).Decode(&values); err != nil {
http.Error(w, err.Error(), http.StatusBadRequest)
return
}
writeOKOr(w, s.set.MergeProject(s.owner(r), r.PathValue("id"), values))
}
// ---- 工具(用户声明式 HTTP 工具注册)----
func (s *Server) listTools(w http.ResponseWriter, r *http.Request) {
writeJSON(w, http.StatusOK, s.reg.List())
}
func (s *Server) createUserTool(w http.ResponseWriter, r *http.Request) {
var desc core.ToolDesc
if err := json.NewDecoder(r.Body).Decode(&desc); err != nil {
http.Error(w, err.Error(), http.StatusBadRequest)
return
}
desc.Source = core.SourceUser
if desc.HTTP == nil {
http.Error(w, "v1 用户工具需为声明式 HTTP 工具(提供 http 段)", http.StatusBadRequest)
return
}
lookup := func(key string) string { return s.set.Effective(s.owner(r), "").GetString(key, "") }
if err := s.reg.Register(desc, tools.HTTPExecutor(*desc.HTTP, lookup)); err != nil {
http.Error(w, err.Error(), http.StatusBadRequest)
return
}
writeJSON(w, http.StatusOK, desc)
}
func (s *Server) deleteUserTool(w http.ResponseWriter, r *http.Request) {
name := r.PathValue("name")
if desc, _, ok := s.reg.Get(name); ok && desc.Source != core.SourceUser {
http.Error(w, "仅可删除 user 来源工具", http.StatusForbidden)
return
}
s.reg.UnregisterBySource(core.SourceUser, name)
writeJSON(w, http.StatusOK, map[string]any{"deleted": name})
}
// ---- 插件 ----
type pluginView struct {
Manifest core.PluginManifest `json:"manifest"`
Panels []core.PluginPanel `json:"panels"`
Cards map[string][]core.UINode `json:"cards"`
Workflows []core.WorkflowDef `json:"workflows"`
Intents []json.RawMessage `json:"intents"`
Prompts map[string]string `json:"prompts"`
Settings []core.SettingDesc `json:"settings"`
}
func (s *Server) listPlugins(w http.ResponseWriter, r *http.Request) {
var out []pluginView
for _, ld := range s.plugins.List() {
out = append(out, pluginView{
Manifest: ld.Manifest, Panels: ld.Panels, Cards: ld.Cards,
Workflows: ld.Workflows, Intents: ld.Intents, Prompts: ld.Prompts, Settings: ld.PluginSettings,
})
}
writeJSON(w, http.StatusOK, orEmptySlice(out))
}
func (s *Server) reloadPlugins(w http.ResponseWriter, r *http.Request) {
results := s.plugins.LoadDir()
writeJSON(w, http.StatusOK, map[string]any{"reloaded": len(results), "errors": results})
}
func (s *Server) pluginSettingsSchema(w http.ResponseWriter, r *http.Request) {
writeJSON(w, http.StatusOK, s.plugins.SettingsSchema())
}
// registry 统一注册表视图(来源标记 + 按来源启停入口)。
func (s *Server) registry(w http.ResponseWriter, r *http.Request) {
reg := map[string]any{
"tools": s.reg.List(),
"parsers": ingest.ParsersDescription(),
"panels": flattenPanels(s.plugins.List()),
}
writeJSON(w, http.StatusOK, reg)
}
func flattenPanels(lds []*plugins.Loaded) []core.PluginPanel {
var out []core.PluginPanel
for _, ld := range lds {
out = append(out, ld.Panels...)
}
return out
}
// ---- 计划 / 待办 / 命令 / 检索 ----
func (s *Server) getPlan(w http.ResponseWriter, r *http.Request) {
p, err := s.st.GetPlan(r.PathValue("id"))
if errors.Is(err, store.ErrNotFound) {
writeJSON(w, http.StatusOK, map[string]any{})
return
}
writeJSONOr(w, p, err)
}
func (s *Server) getTodos(w http.ResponseWriter, r *http.Request) {
l, err := s.st.GetTodos(r.PathValue("id"))
if errors.Is(err, store.ErrNotFound) {
writeJSON(w, http.StatusOK, map[string]any{})
return
}
writeJSONOr(w, l, err)
}
var builtinCommands = []map[string]string{
{"name": "/help", "description": "查看可用命令"},
{"name": "/new", "description": "新建会话"},
{"name": "/mock", "description": "演示模式 on|off(不调用真实模型)"},
{"name": "/model", "description": "设置主模型名称"},
{"name": "/profile", "description": "切换会话画像"},
}
func (s *Server) listCommands(w http.ResponseWriter, r *http.Request) {
writeJSON(w, http.StatusOK, builtinCommands)
}
func (s *Server) listParsers(w http.ResponseWriter, r *http.Request) {
writeJSON(w, http.StatusOK, ingest.ParsersDescription())
}
func (s *Server) addKnowledge(w http.ResponseWriter, r *http.Request) {
var docs []core.KnowledgeDoc
if err := json.NewDecoder(r.Body).Decode(&docs); err != nil {
http.Error(w, err.Error(), http.StatusBadRequest)
return
}
owner := s.owner(r)
if _, err := s.st.GetProject(owner, r.PathValue("id")); err != nil {
writeStoreErr(w, err)
return
}
for i := range docs {
docs[i].Owner = owner
docs[i].ProjectID = r.PathValue("id")
if docs[i].Source == "" {
docs[i].Source = "manual"
}
}
writeOKOr(w, s.st.AddKnowledge(docs...))
}
func (s *Server) searchKnowledge(w http.ResponseWriter, r *http.Request) {
q := r.URL.Query().Get("q")
if q == "" {
http.Error(w, "q required", http.StatusBadRequest)
return
}
topK, _ := strconv.Atoi(r.URL.Query().Get("topK"))
query := core.RetrievalQuery{ProjectID: r.URL.Query().Get("project"), Text: q, TopK: topK}
if s.Retriever != nil {
hits, err := s.Retriever.Retrieve(r.Context(), s.owner(r), query)
writeJSONOr(w, hits, err)
return
}
hits, err := s.st.SearchKnowledge(s.owner(r), query)
writeJSONOr(w, hits, err)
}
// ---- 帮助 ----
func writeJSON(w http.ResponseWriter, status int, v any) {
w.Header().Set("Content-Type", "application/json; charset=utf-8")
w.WriteHeader(status)
json.NewEncoder(w).Encode(v)
}
func writeJSONOr(w http.ResponseWriter, v any, err error) {
if err != nil {
writeStoreErr(w, err)
return
}
writeJSON(w, http.StatusOK, v)
}
func writeOKOr(w http.ResponseWriter, err error) {
if err != nil {
writeStoreErr(w, err)
return
}
writeJSON(w, http.StatusOK, map[string]any{"ok": true})
}
func writeStoreErr(w http.ResponseWriter, err error) {
if errors.Is(err, store.ErrNotFound) {
http.Error(w, err.Error(), http.StatusNotFound)
return
}
if errors.Is(err, runmgr.ErrConversationBusy) {
http.Error(w, err.Error(), http.StatusConflict)
return
}
http.Error(w, err.Error(), http.StatusInternalServerError)
}
// backup 触发一致性在线备份(嵌入式档:VACUUM INTO;PG 档见运维指南 pg_dump)。
func (s *Server) backup(w http.ResponseWriter, r *http.Request) {
dir := filepath.Join(s.st.DataDir(), "backups")
if err := os.MkdirAll(dir, 0o755); err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
dest := filepath.Join(dir, "agent-system-"+time.Now().UTC().Format("20060102T150405Z")+".db")
if err := s.st.Backup(dest); err != nil {
http.Error(w, err.Error(), http.StatusNotImplemented)
return
}
writeJSON(w, http.StatusOK, map[string]any{"ok": true, "path": dest})
}
func contentTypeOf(a core.Artifact) string {
if a.MIME != "" {
return a.MIME
}
if a.Kind == "image" {
return "application/octet-stream"
}
return "text/plain; charset=utf-8"
}
func maskMap(m map[string]any) map[string]any {
if m == nil {
return map[string]any{}
}
return settings.Masked(m)
}
func orEmptySlice[T any](s []T) []T {
if s == nil {
return []T{}
}
return s
}