sundynix/sundynix-agentix
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(report): 报告源/产物落 MinIO,去掉 gateway↔mcp-go 本地盘耦合

Browse Source 
多机/容器化第一时间断裂的隐患(ARCHITECTURE_REVIEW §7 #1):此前报告
源(reportStore)与 .docx 产物(reportExport/reportRender)都写 mcp-go
本地 SUNDYNIX_REPORTS_DIR,gateway 再按同一路径 c.File 回流——mcp-go
与 gateway 不共享磁盘即断。

- blob 包从 gateway/internal 提到 sundynix-shared/blob,gateway 与 mcp-go
  共用同一 MinIO;新增 PutBytes/GetBytes 走二进制(.docx)。
- mcp-go 注入 blob:report 源/产物优先 Put 到对象存储(键 reports/<id>.{json,docx}),
  导出结果返回 minio://<key>;MinIO 未就绪回退本地盘(单机降级,getSource 兼容旧本地报告)。
- gateway ExportReport 识别 minio:// → GetBytes 流式下载,否则 c.File 本地降级。
- 测试:mcp 本地回落往返单测 + shared/blob 真 MinIO 二进制往返(BLOB_TEST_ENDPOINT 门控)。
  四模块 build 绿,受影响测试全过。
- 附带:go mod tidy 引入 genproto 拆包冲突,四模块统一钉单体 genproto 新版解决
  (注:勿 go work sync,会剪掉 indirect require 回落旧版重现冲突)。

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Blizzard
2026-07-18 13:29:31 +08:00
parent 1c7e8e1ea9
commit ec9431bfa4
20 changed files with 484 additions and 182 deletions
Download Patch File Download Diff File Expand all files Collapse all files
sundynix-mcp-go/internal/mcp/gateway.go
+58 -21
View File
@@ -14,6 +14,7 @@ import (
"sync"
"time"
"github.com/sundynix/sundynix-shared/blob"
sharedbus "github.com/sundynix/sundynix-shared/bus"
"github.com/sundynix/sundynix-shared/contract"
@@ -29,6 +30,7 @@ type Gateway struct {
memory *memory.Store
history *history.Store
rag *rag.Engine
blob *blob.Store // 报告源/产物落对象存储（跨进程/多机共享；nil 或未连回退本地盘）
tools map[string]toolDef // 工具注册表：唯一事实源，dispatch 与 list_tools 共用，杜绝漂移
pgDSN string // 平台 PG DSN（sql_query 兜底库；SQL_QUERY_DSN 未设时用它）
@@ -58,8 +60,8 @@ type toolDef struct {
handler func(context.Context, *contract.ToolCall) *contract.ToolResult
}
func NewGateway(b *sharedbus.Bus, m *memory.Store, h *history.Store, r *rag.Engine, pgDSN string) *Gateway {
g := &Gateway{bus: b, memory: m, history: h, rag: r, pgDSN: pgDSN}
func NewGateway(b *sharedbus.Bus, m *memory.Store, h *history.Store, r *rag.Engine, bs *blob.Store, pgDSN string) *Gateway {
g := &Gateway{bus: b, memory: m, history: h, rag: r, blob: bs, pgDSN: pgDSN}
// 记忆 Relevance 复用 RAG 的 embedding（同一控制面下发的模型）：召回时对 query 算语义相关性。
// rag.Engine 满足 memory.Embedder；未配置 embedding 时 memory 优雅回落两项打分。
if m != nil && r != nil {
@@ -396,15 +398,54 @@ func (g *Gateway) reportRender(ctx context.Context, call *contract.ToolCall) *co
if err != nil {
return &contract.ToolResult{OK: false, Error: "report_render: " + err.Error()}
}
ref, err := g.putDocx(ctx, id, data)
if err != nil {
return &contract.ToolResult{OK: false, Error: "report_render: " + err.Error()}
}
log.Printf("[mcp_go] report_render 已生成 %s (%d 字节, %d 章节)", ref, len(data), len(secs))
return &contract.ToolResult{OK: true, Content: ref}
}
// putDocx 落地渲染出的 .docx：MinIO 可用则上传并返回 minio://<key>（跨机可取），
// 否则回退本地盘返回绝对路径（单机/共享卷降级）。
func (g *Gateway) putDocx(ctx context.Context, id string, data []byte) (string, error) {
if g.blob.Ready() {
key := contract.ReportObjectKey(id)
if err := g.blob.PutBytes(ctx, key, data, "application/vnd.openxmlformats-officedocument.wordprocessingml.document"); err != nil {
return "", fmt.Errorf("put docx: %w", err)
}
return contract.BlobScheme + key, nil
}
path := contract.ReportPath(id)
if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil {
return &contract.ToolResult{OK: false, Error: "report_render: mkdir " + err.Error()}
return "", fmt.Errorf("mkdir: %w", err)
}
if err := os.WriteFile(path, data, 0o644); err != nil {
return &contract.ToolResult{OK: false, Error: "report_render: write " + err.Error()}
return "", fmt.Errorf("write: %w", err)
}
log.Printf("[mcp_go] report_render 已生成 %s (%d 字节, %d 章节)", path, len(data), len(secs))
return &contract.ToolResult{OK: true, Content: path}
return path, nil
}
// putSource / getSource 存取报告源 JSON：优先 MinIO，回退本地盘。
func (g *Gateway) putSource(ctx context.Context, id string, data []byte) error {
if g.blob.Ready() {
return g.blob.PutBytes(ctx, contract.ReportSourceObjectKey(id), data, "application/json")
}
path := contract.ReportSourcePath(id)
if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil {
return err
}
return os.WriteFile(path, data, 0o644)
}
func (g *Gateway) getSource(ctx context.Context, id string) ([]byte, error) {
if g.blob.Ready() {
if b, err := g.blob.GetBytes(ctx, contract.ReportSourceObjectKey(id)); err == nil {
return b, nil
}
// MinIO 已启用但取不到：可能是 MinIO 启用前生成的旧报告，兜底看本地盘。
}
return os.ReadFile(contract.ReportSourcePath(id))
}
// reportSource 是报告的可序列化源数据（标题 + 章节），导出时据此渲染各格式。
@@ -415,7 +456,7 @@ type reportSource struct {
// reportStore 把报告源数据（title + sections）落盘为 JSON，供导出时按需渲染 Word/PDF/Markdown。
// 生成阶段只存源、不渲染（"导出时再处理"）。
func (g *Gateway) reportStore(_ context.Context, call *contract.ToolCall) *contract.ToolResult {
func (g *Gateway) reportStore(ctx context.Context, call *contract.ToolCall) *contract.ToolResult {
id, _ := call.Args["task_id"].(string)
if id == "" {
id = call.TaskID
@@ -429,15 +470,11 @@ func (g *Gateway) reportStore(_ context.Context, call *contract.ToolCall) *contr
_ = json.Unmarshal(raw, &secs)
}
data, _ := json.Marshal(reportSource{Title: title, Sections: secs})
path := contract.ReportSourcePath(id)
if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil {
return &contract.ToolResult{OK: false, Error: "report_store: mkdir " + err.Error()}
if err := g.putSource(ctx, id, data); err != nil {
return &contract.ToolResult{OK: false, Error: "report_store: " + err.Error()}
}
if err := os.WriteFile(path, data, 0o644); err != nil {
return &contract.ToolResult{OK: false, Error: "report_store: write " + err.Error()}
}
log.Printf("[mcp_go] report_store 已存源 %s (%d 章节)", path, len(secs))
return &contract.ToolResult{OK: true, Content: path}
log.Printf("[mcp_go] report_store 已存源 id=%s (%d 章节)", id, len(secs))
return &contract.ToolResult{OK: true, Content: id}
}
// reportExport 按需把已存报告源渲染为指定格式：
@@ -451,7 +488,7 @@ func (g *Gateway) reportExport(ctx context.Context, call *contract.ToolCall) *co
return &contract.ToolResult{OK: false, Error: "report_export: task_id 必填"}
}
format, _ := call.Args["format"].(string)
raw, err := os.ReadFile(contract.ReportSourcePath(id))
raw, err := g.getSource(ctx, id)
if err != nil {
return &contract.ToolResult{OK: false, Error: "report_export: 报告尚未生成或已过期"}
}
@@ -467,12 +504,12 @@ func (g *Gateway) reportExport(ctx context.Context, call *contract.ToolCall) *co
if rerr != nil {
return &contract.ToolResult{OK: false, Error: "report_export: " + rerr.Error()}
}
path := contract.ReportPath(id)
if err := os.WriteFile(path, data, 0o644); err != nil {
return &contract.ToolResult{OK: false, Error: "report_export: write " + err.Error()}
ref, perr := g.putDocx(ctx, id, data)
if perr != nil {
return &contract.ToolResult{OK: false, Error: "report_export: " + perr.Error()}
}
log.Printf("[mcp_go] report_export 已渲染 docx %s (%d 字节)", path, len(data))
return &contract.ToolResult{OK: true, Content: path}
log.Printf("[mcp_go] report_export 已渲染 docx %s (%d 字节)", ref, len(data))
return &contract.ToolResult{OK: true, Content: ref}
}
}