From 0c45b86d678429159347b5d92ba5827b83eccd70 Mon Sep 17 00:00:00 2001 From: gozeros Date: Tue, 3 Nov 2026 00:13:46 +0800 Subject: [PATCH] =?UTF-8?q?chore:=20=E5=88=9D=E5=A7=8B=E5=8C=96=20eightbf-?= =?UTF-8?q?analysis-go=EF=BC=88=E5=AF=B9=E9=BD=90=20Java=20=E4=BC=9A?= =?UTF-8?q?=E5=95=86=E4=BD=9C=E4=B8=9A=E6=9C=8D=E5=8A=A1=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Go 实现 /api/v1/analysis 契约、Demo/Live Worker、DeerFlow SSE；含设计文档与冒烟脚本。 Co-authored-by: Cursor --- .gitignore | 10 + README.md | 74 ++++ cmd/server/main.go | 78 ++++ docs/env.sample | 24 ++ docs/对齐说明.md | 28 ++ docs/服务端开发设计要求.md | 379 +++++++++++++++++++ go.mod | 7 + go.sum | 4 + internal/api/handler.go | 189 ++++++++++ internal/api/response.go | 31 ++ internal/config/config.go | 144 ++++++++ internal/deerflow/auth.go | 72 ++++ internal/deerflow/client.go | 503 ++++++++++++++++++++++++++ internal/deerflow/types.go | 18 + internal/domain/job.go | 56 +++ internal/domain/status.go | 21 ++ internal/repo/job_repo.go | 298 +++++++++++++++ internal/repo/schema.go | 47 +++ internal/service/access.go | 217 +++++++++++ internal/service/access_test.go | 50 +++ internal/service/demo_worker.go | 277 ++++++++++++++ internal/service/errors.go | 23 ++ internal/service/job_service.go | 352 ++++++++++++++++++ internal/worker/live_runner.go | 623 ++++++++++++++++++++++++++++++++ internal/worker/live_worker.go | 64 ++++ internal/worker/prompts.go | 119 ++++++ internal/worker/prompts_test.go | 25 ++ schema.sql | 35 ++ scripts/smoke-demo.sh | 21 ++ 29 files changed, 3789 insertions(+) create mode 100644 .gitignore create mode 100644 README.md create mode 100644 cmd/server/main.go create mode 100644 docs/env.sample create mode 100644 docs/对齐说明.md create mode 100644 docs/服务端开发设计要求.md create mode 100644 go.mod create mode 100644 go.sum create mode 100644 internal/api/handler.go create mode 100644 internal/api/response.go create mode 100644 internal/config/config.go create mode 100644 internal/deerflow/auth.go create mode 100644 internal/deerflow/client.go create mode 100644 internal/deerflow/types.go create mode 100644 internal/domain/job.go create mode 100644 internal/domain/status.go create mode 100644 internal/repo/job_repo.go create mode 100644 internal/repo/schema.go create mode 100644 internal/service/access.go create mode 100644 internal/service/access_test.go create mode 100644 internal/service/demo_worker.go create mode 100644 internal/service/errors.go create mode 100644 internal/service/job_service.go create mode 100644 internal/worker/live_runner.go create mode 100644 internal/worker/live_worker.go create mode 100644 internal/worker/prompts.go create mode 100644 internal/worker/prompts_test.go create mode 100644 schema.sql create mode 100755 scripts/smoke-demo.sh diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..00b8e1c --- /dev/null +++ b/.gitignore @@ -0,0 +1,10 @@ +/bin/ +*.exe +*.test +*.out +/data/ +*.log +.idea/ +.vscode/ +*.swp +.DS_Store diff --git a/README.md b/README.md new file mode 100644 index 0000000..166451c --- /dev/null +++ b/README.md @@ -0,0 +1,74 @@ +# eightbf-analysis-go + +8BF「多岗位会商」后台服务的 **Go 实现**，行为对齐 Java 仓 `eightbf-analysis` 与 +[docs/服务端开发设计要求.md](./docs/服务端开发设计要求.md)。 + +- 同一套 HTTP：`/api/v1/analysis/...` +- 同一张表：`eba_analysis_job`（`schema.sql`） +- 同一套环境变量：`EBA_DB_*` / `EBA_DF_*`（Go 另支持 `EBA_DB_DSN`） +- 前端可经 Nginx `/analysis-api` 指向本进程（默认仍 `:18097`，勿与 Java 实例同时占端口） + +## 与 Java 的关系 + +| | Java | Go（本仓） | +|--|------|------------| +| 定位 | 现网/本机主用 | 同构移植，便于无 JVM 环境部署 | +| API | 权威契约 | **兼容**同一契约 | +| 表结构 | `schema.sql` | 共用同一 schema | + +两进程不要同时写同一库抢同一 `queued`（尚无领取互斥）。切换时停掉一侧再起另一侧。 + +## 运行 + +```bash +# 需要 Go 1.21+、MySQL（库 eightbf_analysis） +export EBA_DB_USER=eba +export EBA_DB_PASSWORD=eba_local_dev +# 或直接： +# export EBA_DB_DSN='eba:eba_local_dev@tcp(127.0.0.1:3306)/eightbf_analysis?parseTime=true&loc=Asia%2FShanghai&charset=utf8mb4' + +export EBA_DF_BASE_URL=http://127.0.0.1:8001 +export ANALYSIS_MODE=live # demo 则假进度 +# export ANALYSIS_AUTH_ENABLED=true + +cd /project/eightbf-analysis-go +go run ./cmd/server +# 默认 :18097 +``` + +构建： + +```bash +go build -o bin/eightbf-analysis-go ./cmd/server +./bin/eightbf-analysis-go +``` + +## 接口（与 Java 一致） + +- `GET /api/v1/analysis/health` +- `POST /api/v1/analysis/jobs` +- `GET /api/v1/analysis/jobs/{id}` +- `GET /api/v1/analysis/jobs?taskId=&latest=` +- `POST /api/v1/analysis/jobs/{id}/cancel|resume|plan|inject` +- `GET /api/v1/analysis/jobs/{id}/result` + +身份头：`X-Account-Id`、`X-Analysis-Role`。 + +## 目录 + +```text +cmd/server 入口 +internal/api HTTP +internal/service 编排 / 鉴权投影 / DemoWorker +internal/worker LiveWorker / LiveChainRunner / 提示词 +internal/deerflow 登录、SSE、模型校验 +internal/repo MySQL +internal/domain 作业实体与状态常量 +internal/config 环境变量 +docs/ 设计要求（自 Java 同步）+ env.sample +``` + +## 日志 + +遵循设计要求 §11：业务日志带 `jobId=`；禁止打印 token/密码/完整 dialogues。 +当前写 stdout，正式环境用 systemd/Docker 收集或挂滚动文件。 diff --git a/cmd/server/main.go b/cmd/server/main.go new file mode 100644 index 0000000..c06af05 --- /dev/null +++ b/cmd/server/main.go @@ -0,0 +1,78 @@ +package main + +import ( + "database/sql" + "log" + "net/http" + "os" + "strconv" + "time" + + _ "github.com/go-sql-driver/mysql" + + "git.gozeros.com/wsk/eightbf-analysis-go/internal/api" + "git.gozeros.com/wsk/eightbf-analysis-go/internal/config" + "git.gozeros.com/wsk/eightbf-analysis-go/internal/deerflow" + "git.gozeros.com/wsk/eightbf-analysis-go/internal/repo" + "git.gozeros.com/wsk/eightbf-analysis-go/internal/service" + "git.gozeros.com/wsk/eightbf-analysis-go/internal/worker" +) + +func main() { + cfg := config.Load() + log.Printf("eightbf-analysis-go starting port=%d mode=%s auth=%v demo=%v", + cfg.Port, cfg.AnalysisMode, cfg.AuthEnabled, cfg.DemoEnabled) + + db, err := sql.Open("mysql", cfg.DBDSN) + if err != nil { + log.Fatalf("open db: %v", err) + } + db.SetMaxOpenConns(20) + db.SetMaxIdleConns(5) + db.SetConnMaxLifetime(time.Hour) + if err := db.Ping(); err != nil { + log.Fatalf("ping db: %v（请检查 EBA_DB_DSN / EBA_DB_*）", err) + } + + jobRepo := repo.NewJobRepo(db) + if schema, err := os.ReadFile("schema.sql"); err != nil { + log.Printf("未找到 schema.sql，跳过自动建表: %v", err) + } else { + jobRepo.ApplySchema(string(schema)) + } + + access := &service.AccessService{AuthEnabled: cfg.AuthEnabled} + jobs := &service.JobService{Repo: jobRepo, Access: access} + h := &api.Handler{Jobs: jobs} + + mux := http.NewServeMux() + h.Routes(mux) + + if cfg.AnalysisMode == "live" && !cfg.DemoEnabled { + auth := deerflow.NewAuth(cfg) + df := deerflow.NewClient(cfg, auth) + runner := &worker.LiveChainRunner{DF: df, Repo: jobRepo, Coord: cfg.DeerFlowCoordinator} + lw := &worker.LiveWorker{ + Repo: jobRepo, + Runner: runner, + PollEvery: time.Duration(cfg.LivePollMS) * time.Millisecond, + } + lw.Start() + log.Printf("LiveWorker 已启动 poll=%dms", cfg.LivePollMS) + } else if cfg.DemoEnabled || cfg.AnalysisMode == "demo" { + dw := &service.DemoWorker{ + Repo: jobRepo, + Every: time.Duration(cfg.DemoTickMS) * time.Millisecond, + } + dw.Start() + log.Printf("DemoWorker 已启动 tick=%dms", cfg.DemoTickMS) + } else { + log.Printf("警告：未启动任何 Worker（mode=%s demo=%v）", cfg.AnalysisMode, cfg.DemoEnabled) + } + + addr := ":" + strconv.Itoa(cfg.Port) + log.Printf("监听 %s （健康检查 GET /api/v1/analysis/health）", addr) + if err := http.ListenAndServe(addr, api.WithCORS(mux)); err != nil { + log.Fatal(err) + } +} diff --git a/docs/env.sample b/docs/env.sample new file mode 100644 index 0000000..7b9de08 --- /dev/null +++ b/docs/env.sample @@ -0,0 +1,24 @@ +# eightbf-analysis / eightbf-analysis-go 内网环境变量样例 +# 勿把真实密码提交进 Git。 + +# MySQL（正式必填） +# Java 可用 JDBC；Go 优先 EBA_DB_DSN，也能解析 EBA_DB_URL(jdbc) +EBA_DB_URL=jdbc:mysql://:3306/eightbf_analysis?useUnicode=true&characterEncoding=utf8&useSSL=false&serverTimezone=Asia/Shanghai&allowPublicKeyRetrieval=true +EBA_DB_USER=eba +EBA_DB_PASSWORD= +# Go 推荐： +# EBA_DB_DSN=eba:xxx@tcp(mysql-host:3306)/eightbf_analysis?parseTime=true&loc=Asia%2FShanghai&charset=utf8mb4&multiStatements=true + +# DeerFlow Gateway（ANALYSIS_MODE=live 时必填） +EBA_DF_BASE_URL=http://:8001 +EBA_DF_USER=admin +EBA_DF_PASSWORD= + +# 可选：改端口 +# SERVER_PORT=18097 + +# Go 运行模式 +# ANALYSIS_MODE=live +# ANALYSIS_AUTH_ENABLED=false +# EBA_WORKER_DEMO_ENABLED=false +# 假进度：ANALYSIS_MODE=demo diff --git a/docs/对齐说明.md b/docs/对齐说明.md new file mode 100644 index 0000000..5f83acb --- /dev/null +++ b/docs/对齐说明.md @@ -0,0 +1,28 @@ +# Go 版与 Java 版对齐说明 + +设计权威：[服务端开发设计要求.md](./服务端开发设计要求.md)（与 Java 仓同步）。 + +## 已对齐 + +- API 路径、状态码、`ApiResponse`、身份头 +- 作业状态机、一开一场、cancel_flag、plan 检查点 +- 人工干预 / resume / inject / patch plan（关干预自动放行） +- 岗位投影（position_owner 裁气泡） +- Live：模型校验 → init → 总控派活 → 席位 → 可选卡点 → 收口；失败不当成功 +- Demo 假进度 Worker +- 环境变量 `EBA_*`；Go 额外 `EBA_DB_DSN`、`ANALYSIS_MODE`、`ANALYSIS_AUTH_ENABLED` + +## 已知差异 / 注意 + +1. **进程互斥**：勿与 Java 实例同时抢同一 MySQL 的 `queued`（尚未做多实例领取锁）。 +2. **schema 启动**：Go 启动读工作目录 `schema.sql`；部署时保证 cwd 或改绝对路径。 +3. **SSE 细节**：DeerFlow 帧格式若有变，优先对照 Java `DeerFlowClient` 再改本仓 `internal/deerflow`。 +4. **日志**：先 stdout；滚动文件按设计要求 §11 后续加 lumberjack 即可。 + +## 切换建议 + +1. 停 Java 进程 +2. 确认同一 `EBA_DB_*` / DeerFlow +3. 起 Go：`go run ./cmd/server` 或 systemd +4. Nginx `/analysis-api` 仍指 `:18097` +5. 健康检查 + 开一场研讨验收 diff --git a/docs/服务端开发设计要求.md b/docs/服务端开发设计要求.md new file mode 100644 index 0000000..c5012ff --- /dev/null +++ b/docs/服务端开发设计要求.md @@ -0,0 +1,379 @@ +# eightbf-analysis 服务端开发设计要求 + +> 适用范围：本仓库（8BF「多岗位会商」后台作业服务，Spring Boot 2.0.9 + JDK8）。 +> 目标：关浏览器后会商仍能跑完；状态以服务端为准；产品前端只经 `/analysis-api` 访问本服务。 +> 配套：根 [README.md](../README.md)、[内网迁移.md](./内网迁移.md)、源码包注释（`package-info` / 各层类注释）。 + +本文是**开发约束与设计验收标准**，不是空泛愿景。改功能前先对照；评审以本文为准。 + +--- + +## 1. 定位与边界 + +### 1.1 本服务负责 + +| 职责 | 说明 | +|------|------| +| 会商作业生命周期 | 创建 / 查询 / 取消 / 续跑 / 热更新 / 注入 / 结果 | +| 服务端执行 | Demo 假进度或 Live 真模型链条；关页不断 | +| 状态与流式内容落库 | `status`、`plan`、`dialogues`、`result` 等以 MySQL 为准 | +| 对接 DeerFlow | 登录、建线程、消费 SSE；失败策略由本服务决定 | +| 岗位可见性投影 | 按账号裁剪对话后再返回前端 | + +### 1.2 本服务不负责 + +- 产品登录 / SSO / 账号体系本体（只认请求头里的账号与角色） +- 岗位会商「DeerFlow joinStream」那条链路（岗位侧保持原样，不经本服务挂起） +- 浏览器直连 DeerFlow `roundtable-jobs` 跑 8BF（**禁止**作为正式路径） +- Fusion 其它 API（`/zl_consumer`、`/deerflow-api` 等） + +### 1.3 硬原则（不可破） + +1. **状态权威在服务端**：前端轮询展示；刷新必须带草稿里的 `jobId`，禁止用「task 下最新一场」冒充重连。 +2. **一开一场**：每次「开始研讨」新建作业；同一 `taskId` 可并存多场；不自动取消旧场。 +3. **浏览器可走，作业不能丢**：取消只靠 `cancel_flag` + Worker 协作停；不要假设 SSE 连着用户标签页。 +4. **失败不当成功**：模型不可用、席位空输出、流无终帧、自动换模 → 整场 `error`；禁止刷完链条再标 `done` 解锁前端下一阶段。 +5. **热更新以库为准**：人工干预 / 模型 / inject 写在库里；Runner 每席与 flush 前从库合并，禁止内存态盖掉用户 PATCH。 + +--- + +## 2. 总体架构 + +```text +浏览器 → Nginx /analysis-api → eightbf-analysis (:18097) + ├─ web（HTTP + DTO） + ├─ service（编排 / 鉴权投影 / 落库 / DemoWorker） + ├─ worker（LiveWorker 调度 + LiveChainRunner 执行） + ├─ deerflow（Auth / Client / SSE） + └─ MySQL eba_analysis_job + ↑ + DeerFlow Gateway（仅服务端访问） +``` + +### 2.1 分层要求 + +| 包 | 允许 | 禁止 | +|----|------|------| +| `web` | 参数校验、HTTP 状态码、组 Caller、调 service | 直接 JDBC；直接调 DeerFlow；写业务状态机细节 | +| `service` | 作业编排、权限、仓储、Demo 推进 | 拼 DeerFlow SSE；复制一套状态机与 Runner 打架 | +| `worker` | 真模型调度与链条；读写作业检查点 | 暴露新 HTTP；绕过 Repository 改库 | +| `deerflow` | HTTP/SSE、token、模型校验、流式折算 | 知道「作业业务规则」以外的前端约定；改作业 status | +| `domain` | 实体、状态常量 | Spring 注解、IO | + +依赖方向只能向内：`web → service → domain`；`worker → service/domain + deerflow`。禁止 `deerflow` 依赖 `web`。 + +### 2.2 运行模式 + +| 配置 | 行为 | +|------|------| +| `analysis.mode=live` + `eba.worker.demo-enabled=false` | **正式/联调默认**：LiveWorker 领 `queued`，LiveChainRunner 调 DeerFlow | +| `analysis.mode=demo` + `eba.worker.demo-enabled=true` | 仅演示假发言；不得当内网验收真模型 | + +两模式互斥清晰：禁止「一边假进度一边又去打 DeerFlow」。 + +--- + +## 3. 领域与状态机 + +### 3.1 作业（AnalysisJob） + +一场作业 = 服务端上的一场研讨录像带。关键袋子： + +| 字段 | 用途 | 开发约束 | +|------|------|----------| +| `id` | 主键；草稿必须存 | 新建生成；禁止按 task 覆盖 | +| `status` / `phase` | 生命周期 / 细阶段 | 状态字符串只用 `JobStatus` 常量 | +| `plan` | 检查点与开关 | 至少：`seatIndex`、`manualIntervention`、`pendingInject`、`awaitReason`、`seedMessage` | +| `dialogues` | 气泡流式展示 | 增量追加；终帧勿无脑整段覆盖丢掉 `` | +| `threadIds` | DeerFlow 线程 | 有则续跑跳过 init | +| `access` | 岗位账号限制 | 结构与前端约定一致 | +| `progress` / `result` | 进度文案 / 终态汇总 | `error` 时清掉误导性 summary / 共识 | + +### 3.2 状态流转 + +```text +queued → running → awaiting_input ⇄ queued（resume） + → done + → error + → cancelled +``` + +| 状态 | 含义 | 允许的写操作 | +|------|------|----------------| +| `queued` | 等 Worker 领取 | cancel；inject；patch plan | +| `running` | 执行中 | cancel；inject；patch plan | +| `awaiting_input` | 澄清或人工卡点 | resume；cancel；inject；patch plan（关干预可自动放行） | +| `done` / `error` / `cancelled` | 终态 | **禁止** resume / inject / 改计划 | + +实现要求： + +- 取消：先 `cancel_flag=1`，Runner/Demo 每拍检查；再落 `cancelled`。 +- 续跑：仅 `awaiting_input → queued`；检查点用 `plan.seatIndex`，勿重新从 0 开场除非产品明确要求。 +- 人工干预默认安全：`manualIntervention` 缺省视为 **true**（宁可多卡一次）。 + +### 3.3 真模型链条语义 + +顺序固定： + +1. 校验席位 ≥ 2、模型在 DeerFlow 列表中（未知模型 **拒绝**，禁止静默换模）。 +2. 无 `threadIds` 则 init；有则续跑。 +3. 对每个席位：总控派活 → 该席输出 →（若开人工干预且非末席）进入 `awaiting_input`。 +4. 全席成功后总控收口；交付不全或对话已有 error 气泡 → 不得 `done`。 + +一席失败立刻 `error` 停场，禁止「继续跑完再汇总错误」。 + +--- + +## 4. API 设计要求 + +### 4.1 约定 + +- 前缀：`/api/v1/analysis` +- 包体：`ApiResponse` → `{ code, message, data }`；成功 `code=0` +- HTTP：`201` 创建；`404` 找不到；`403` 无权限；`400` 状态/参数不允许 +- 身份头：`X-Account-Id`、`X-Analysis-Role`（`admin` / `task_owner` / `position_owner` / `viewer`） +- 探活：`GET /health` **不鉴权** + +### 4.2 接口清单（保持稳定） + +| 方法 | 路径 | 说明 | +|------|------|------| +| POST | `/jobs` | 新建；返回投影后快照 | +| GET | `/jobs/{id}` | 轮询 / 重连主路径 | +| GET | `/jobs?taskId=&latest=` | 兼容查询；**正式重连勿依赖此接口替代 jobId** | +| POST | `/jobs/{id}/cancel` | 仅责任人/创建人/管理员 | +| POST | `/jobs/{id}/resume` | 澄清回答或人工「继续」 | +| POST | `/jobs/{id}/plan` | 热更新人工干预 / 模型 | +| POST | `/jobs/{id}/inject` | 运行中补充；空文本 400 | +| GET | `/jobs/{id}/result` | 裁剪字段的结果视图 | +| GET | `/health` | 探活 | + +### 4.3 兼容与演进 + +- 字段名尽量与前端 `mapJob` 一致，少做二次映射。 +- 新增字段：只增不改语义；废弃字段先双写/文档标明，再删。 +- 破坏性变更（状态字符串、必填头、错误码语义）必须同步改 code-zk 客户端与本文，并在 PR 说明迁移步骤。 + +--- + +## 5. 鉴权与投影 + +配置：`analysis.auth.enabled`（本机可 false；内网建议 true）。 + +| 角色 | 读 | resume/inject/改 plan | cancel | +|------|----|------------------------|--------| +| admin / task_owner / 创建人 / access 任务责任人 | 全量 | 是 | 是 | +| position_owner（绑定了席位） | 本席 + system/user/leader | 是 | **否**（除非兼责任人） | +| 无账号且强制鉴权 | 否 | 否 | 否 | + +投影规则（即使本机关强制鉴权，声明了 `position_owner` 也要裁剪，便于验收）： + +- 岗位账号：只见 `system` / `user` / `leader` + 自己 `agentId` 的 special 气泡。 +- `result.seatDeliveries` 同步裁剪。 + +新接口若返回作业快照，必须走 `JobAccessService.projectForCaller`，禁止裸库对象出站。 + +--- + +## 6. 持久化要求 + +- 表：`eba_analysis_job`（见 `schema.sql`）；大字段 JSON 文本列。 +- 正式库：**MySQL utf8mb4**；H2 仅应急联调。 +- 写库路径统一经 `JobRepository`；流式 flush 可节流（约 ≥150ms），但取消旗、状态跃迁必须及时。 +- 索引：`task_id` / `status` / `owner_account_id` 保持；新增查询条件先补索引再上线。 +- 启动执行 `schema.sql` 允许 `continue-on-error`（重复索引）；**禁止**靠静默失败掩盖真实 DDL 错误而不验收。 + +--- + +## 7. DeerFlow 集成要求 + +1. **仅服务端**持有 Gateway 账号；口令走环境变量，不写死进正式配置提交物。 +2. Token 复用并提前刷新；并发登录要有同步，避免打爆登录接口。 +3. `assertModelAvailable`：指定模型不在列表 → 开跑前失败；禁止依赖 Gateway「自动换模」。 +4. SSE：识别 `event:error`、`model_switch`、无终帧；一律映射为 `invalidDelivery` / 业务失败。 +5. 流式展示：`reasoning_content` 与正文一并折成 `…` + 正文，保证前端两边都能流。 +6. 超时：`stream-idle-timeout-ms` 可配；取消检查嵌在读循环里。 +7. 本服务不得把 DeerFlow 内部错误原文无差别甩给用户；要带「当前模型」提示，便于换模重开。 + +--- + +## 8. Worker 与并发 + +- LiveWorker：FIFO 领 `queued`；先占坑改 `running` 再异步执行；默认并发上限 **2**（可配置化时须有文档）。 +- 单机进程内线程池即可；多实例部署前必须先设计「领取互斥」（乐观锁 / `FOR UPDATE` / 外部队列），否则禁止双活抢同一作业。 +- DemoWorker 与 LiveWorker 开关清晰；生产关闭 demo。 + +--- + +## 9. 配置与环境 + +| 项 | 要求 | +|----|------| +| 端口 | 默认 `18097`；经 Nginx `/analysis-api` 反代 | +| 数据源 | `EBA_DB_*` 覆盖；正式禁用默认弱口令 | +| DeerFlow | `EBA_DF_*` 覆盖 | +| 鉴权 | 内网打开 `analysis.auth.enabled=true` | +| 密钥 | 不进 Git；参考 `docs/env.sample` | + +变更配置项必须更新：`application.yml` 注释、`env.sample`、内网迁移文档。 + +--- + +## 10. 编码与注释规范 + +- 语言：JDK **8**；禁止使用仅高版本语法（除非整体升级并改构建）。 +- 注释：新代码须**详细中文注释**（为什么 / 约定 / 边界），禁止只复述变量名；有用旧注释不得顺手删。 +- 异常：业务用明确文案；`SecurityException` → 403；`IllegalStateException` / `IllegalArgumentException` → 400。 +- 日志：见 **§11 日志与收集**；关键作业必须带 `jobId=`；禁止打 token、密码、完整用户隐私对话到 INFO（必要时 DEBUG 且脱敏）。 +- 测试：至少覆盖状态不允许的操作、权限拒绝、模型校验失败、取消中断；无单测的改动须在 PR 写清手工验收步骤。 + +--- + +## 11. 日志与收集 + +本节是运维可落地的约束：先保证「能查、能关联、不泄密」，再谈集中平台。 + +### 11.1 目标 + +| 目标 | 说明 | +|------|------| +| 排障 | 任意一场会商能用 `jobId`（必要时再加 `requestId` / `taskId`）串起创建→领取→DeerFlow→卡点→结束 | +| 审计 | 谁 cancel / resume / inject / 改 plan 可追溯（带账号头，不带口令） | +| 采集 | 日志形态稳定，便于本机落盘或内网 Filebeat/Vector/rsyslog 采集 | +| 安全 | 密钥与大段会商正文默认不上日志；完整内容以库表为准 | + +### 11.2 输出与落盘（应用侧） + +1. **默认打 stdout/stderr**（systemd / Docker / 进程托管都能直接收）。 +2. 正式环境建议同时挂 **滚动文件**（便于采集 agent 扫目录），例如： + - 路径：`/var/log/eightbf-analysis/app.log`（可用环境变量覆盖） + - 滚动：按天或按大小（如 100MB × 14 份）；禁止无限单文件撑爆磁盘 +3. 使用标准 Logback（Spring Boot 默认即可）；自定义 `logback-spring.xml` 时须文档化路径与滚动策略。 +4. 日志级别： + - 生产默认：`com.eightbf.analysis=INFO`，框架 `WARN` + - 排障临时：单 logger 升 `DEBUG`，用完改回；禁止长期全局 DEBUG + +### 11.3 格式（方便采集与检索） + +每条业务日志尽量同构，文本键值即可（先不强制 JSON，内网解析成本更低）： + +```text +2026-11-02 23:40:01.123 INFO [...] c.e.a.worker.LiveWorker - 开始真模型会商 jobId=job_xxx requestId=req_yyy +``` + +**推荐固定键（有则必带）：** + +| 键 | 何时 | +|----|------| +| `jobId=` | 一切与作业相关的日志 | +| `requestId=` | 创建、与前端对账 | +| `taskId=` | 仅当本次操作明确带了 task（勿刷屏） | +| `accountId=` | cancel / resume / inject / patchPlan（来自请求头） | +| `phase=` / `status=` | 状态跃迁、暂停等待 | +| `model=` | 模型校验失败、开跑、换模拒绝 | + +禁止：无键的散文导致无法 grep；同一事件重复刷屏（流式 delta **不要**每字打日志）。 + +### 11.4 必打事件（最低集合） + +| 级别 | 事件 | +|------|------| +| INFO | 作业创建；LiveWorker 领取并开始；进入 `awaiting_input`（澄清/人工）；resume 重新入队；会商 `done`；`cancelled` | +| INFO | DeerFlow 登录成功（只打过期秒数，**不打 token**）；init 完成（席位数） | +| WARN | Worker/Runner 未捕获异常；DeerFlow HTTP≥400；模型不可用；流无终帧 / `model_switch` 拒绝；取消后仍异常退出 | +| ERROR | 进程级故障（数据源不可用、反复登录失败等） | + +HTTP 访问明细优先靠 **Nginx access log**（`/analysis-api`）；应用内不必再镜像每一条轮询 GET（会淹没真正故障）。 + +### 11.5 脱敏与禁止项 + +**禁止写入日志：** + +- DeerFlow / DB 密码、Bearer token、Cookie +- 完整 `dialogues` / 模型长回复 / inject 全文（库里已有；日志最多截断摘要 ≤200 字且仅 WARN/ERROR） +- 身份证等敏感 PII（本服务一般无；若 payload 带入勿原样打印） + +**允许：** 错误摘要、模型名、agentId、席位序号、HTTP 状态码、耗时毫秒。 + +### 11.6 采集与留存（运维侧） + +按内网条件选一种，**应用不绑定某一家商业 APM**： + +```text +eightbf-analysis + → stdout 和/或 /var/log/eightbf-analysis/*.log + → Filebeat / Vector / rsyslog + → 内网 Loki / ELK / 现有日志平台 +Nginx /analysis-api access/error + → 同上（务必带上游 upstream 耗时，便于区分网关与本服务） +``` + +要求： + +1. 采集配置与本服务**同主机交付说明**写进内网迁移或运维手册（路径、用户、权限）。 +2. 应用日志与 Nginx 日志**时间对齐**（统一时区 `Asia/Shanghai`）。 +3. 留存：建议热数据 ≥14 天，冷存按等保/单位规定；磁盘告警纳入主机巡检。 +4. 检索演练：上线后用一场真实 `jobId` 能在平台搜到「创建→开始→结束/失败」完整链，算采集验收通过。 + +### 11.7 与业务库的分工 + +| 数据 | 放哪 | +|------|------| +| 会商正文、气泡、检查点 | **MySQL**（权威） | +| 何时开始/停/失败、对接 DeerFlow 是否成功 | **日志**（排障） | +| 进度百分比给用户看 | `progress` 字段，不是日志 | + +不要把日志当第二份 dialogues 备份。 + +### 11.8 可观测（日志之外，可排期） + +最低仍须保留：`GET /api/v1/analysis/health`。 + +建议后续：作业耗时、DeerFlow 错误码计数、queued 堆积告警（指标/告警通道另定，不替代日志）。 + +内网功能验收仍见 [内网迁移.md](./内网迁移.md)。 + +--- + + +## 12. 变更流程（本仓） + +1. **先方案**：非琐碎改动先说清影响面（状态机 / API / 前端约定 / DeerFlow），用户批准后再写代码。 +2. **范围克制**：只改任务所需；不顺手重构无关层。 +3. **联调**：本服务 + code-zk `/analysis-api`；`sim:build` 后走一遍开研讨 / 干预 / 刷新。 +4. **提交**：仅用户要求时 commit/push；勿夹带密钥与 `target/`、`data/`。 +5. **文档同步**：行为变了必须改本文相关条款 + README/内网迁移（若涉及部署）。 + +--- + +## 13. 验收检查表（开发自测） + +- [ ] 新建两场同 `taskId`，互不覆盖；刷新只跟各自 `jobId` +- [ ] live 模式关浏览器后作业仍推进或停在明确卡点 +- [ ] 取消后 SSE/推进停止，状态 `cancelled` +- [ ] 人工干预开：每席（末席前）进入 `awaiting_input`；关且在卡点时自动续跑 +- [ ] inject 后下一席派活带上补充；气泡可见用户原文 +- [ ] 未知模型开跑失败，不进入 done +- [ ] `position_owner` 看不到其它席 special 气泡；不能 cancel +- [ ] 失败场无「可解锁下一阶段」的假 summary / 100% 共识 +- [ ] 一场作业可用 `jobId` 在日志中串起创建→开始→结束/失败；无 token/密码泄露 +- [ ] 正式环境日志有滚动策略；采集路径写进运维说明并能检索 +- [ ] 注释与本文冲突时：以代码行为为准并回修文档 + +--- + +## 14. 反例（明确禁止） + +- 浏览器直连 DeerFlow 跑 8BF 正式会商 +- 用 `GET ?taskId=` 代替草稿 `jobId` 做刷新重连主路径 +- 信任 DeerFlow 自动换模并当成功 +- flush 时用旧内存 `manualIntervention`/`model` 盖掉库里用户热改 +- 终态仍允许 inject/resume 而不报错 +- 多实例无互斥双活抢作业 +- 把演示 DemoWorker 开进生产当真模型 +- 把 token / 密码 / 完整 dialogues 打进 INFO；流式 delta 逐字打日志 +- 只写控制台、无滚动、无采集说明就当「生产可运维」 + +--- + +*文档版本与仓库代码同步维护；重大行为变更请改本节并在提交说明中引用。* diff --git a/go.mod b/go.mod new file mode 100644 index 0000000..c6af3d9 --- /dev/null +++ b/go.mod @@ -0,0 +1,7 @@ +module git.gozeros.com/wsk/eightbf-analysis-go + +go 1.26.5 + +require github.com/go-sql-driver/mysql v1.8.1 + +require filippo.io/edwards25519 v1.1.0 // indirect diff --git a/go.sum b/go.sum new file mode 100644 index 0000000..19dbcec --- /dev/null +++ b/go.sum @@ -0,0 +1,4 @@ +filippo.io/edwards25519 v1.1.0 h1:FNf4tywRC1HmFuKW5xopWpigGjJKiJSV0Cqo0cJWDaA= +filippo.io/edwards25519 v1.1.0/go.mod h1:BxyFTGdWcka3PhytdK4V28tE5sGfRvvvRV7EaN4VDT4= +github.com/go-sql-driver/mysql v1.8.1 h1:LedoTUt/eveggdHS9qUFC1EFSa8bU2+1pZjSRpvNJ1Y= +github.com/go-sql-driver/mysql v1.8.1/go.mod h1:wEBSXgmK//2ZFJyE+qWnIsVGmvmEKlqwuVSjsCm7DZg= diff --git a/internal/api/handler.go b/internal/api/handler.go new file mode 100644 index 0000000..2f12ae3 --- /dev/null +++ b/internal/api/handler.go @@ -0,0 +1,189 @@ +package api + +import ( + "encoding/json" + "errors" + "net/http" + "strings" + + "git.gozeros.com/wsk/eightbf-analysis-go/internal/service" +) + +// Handler HTTP 入口，路径与 Java AnalysisJobController 对齐。 +type Handler struct { + Jobs *service.JobService +} + +func (h *Handler) Routes(mux *http.ServeMux) { + mux.HandleFunc("GET /api/v1/analysis/health", h.health) + mux.HandleFunc("POST /api/v1/analysis/jobs", h.create) + mux.HandleFunc("GET /api/v1/analysis/jobs/{id}", h.get) + mux.HandleFunc("GET /api/v1/analysis/jobs", h.getByTask) + mux.HandleFunc("POST /api/v1/analysis/jobs/{id}/cancel", h.cancel) + mux.HandleFunc("POST /api/v1/analysis/jobs/{id}/resume", h.resume) + mux.HandleFunc("POST /api/v1/analysis/jobs/{id}/plan", h.patchPlan) + mux.HandleFunc("POST /api/v1/analysis/jobs/{id}/inject", h.inject) + mux.HandleFunc("GET /api/v1/analysis/jobs/{id}/result", h.result) +} + +func (h *Handler) caller(r *http.Request) service.Caller { + return service.NewCaller(r.Header.Get("X-Account-Id"), r.Header.Get("X-Analysis-Role")) +} + +func (h *Handler) health(w http.ResponseWriter, _ *http.Request) { + writeJSON(w, http.StatusOK, map[string]interface{}{ + "ok": true, + "service": "eightbf-analysis-go", + }) +} + +func (h *Handler) create(w http.ResponseWriter, r *http.Request) { + var body service.CreateJobRequest + _ = json.NewDecoder(r.Body).Decode(&body) + job, err := h.Jobs.Create(&body, h.caller(r)) + if err != nil { + h.mapErr(w, err) + return + } + Created(w, job) +} + +func (h *Handler) get(w http.ResponseWriter, r *http.Request) { + id := r.PathValue("id") + job, err := h.Jobs.Get(id, h.caller(r)) + if err != nil { + h.mapErr(w, err) + return + } + if job == nil { + Fail(w, http.StatusNotFound, 404, "job not found") + return + } + OK(w, job) +} + +func (h *Handler) getByTask(w http.ResponseWriter, r *http.Request) { + taskID := strings.TrimSpace(r.URL.Query().Get("taskId")) + if taskID == "" { + Fail(w, http.StatusBadRequest, 400, "taskId required") + return + } + latest := r.URL.Query().Get("latest") == "true" + job, err := h.Jobs.GetByTaskID(taskID, latest, h.caller(r)) + if err != nil { + h.mapErr(w, err) + return + } + if job == nil { + Fail(w, http.StatusNotFound, 404, "job not found for taskId") + return + } + OK(w, job) +} + +func (h *Handler) cancel(w http.ResponseWriter, r *http.Request) { + job, err := h.Jobs.Cancel(r.PathValue("id"), h.caller(r)) + if err != nil { + h.mapErr(w, err) + return + } + if job == nil { + Fail(w, http.StatusNotFound, 404, "job not found") + return + } + OK(w, job) +} + +func (h *Handler) resume(w http.ResponseWriter, r *http.Request) { + var body service.ResumeJobRequest + _ = json.NewDecoder(r.Body).Decode(&body) + job, err := h.Jobs.Resume(r.PathValue("id"), &body, h.caller(r)) + if err != nil { + h.mapErr(w, err) + return + } + if job == nil { + Fail(w, http.StatusNotFound, 404, "job not found") + return + } + OK(w, job) +} + +func (h *Handler) patchPlan(w http.ResponseWriter, r *http.Request) { + var body service.PatchJobPlanRequest + _ = json.NewDecoder(r.Body).Decode(&body) + job, err := h.Jobs.PatchPlan(r.PathValue("id"), &body, h.caller(r)) + if err != nil { + h.mapErr(w, err) + return + } + if job == nil { + Fail(w, http.StatusNotFound, 404, "job not found") + return + } + OK(w, job) +} + +func (h *Handler) inject(w http.ResponseWriter, r *http.Request) { + var body service.InjectJobRequest + _ = json.NewDecoder(r.Body).Decode(&body) + job, err := h.Jobs.Inject(r.PathValue("id"), &body, h.caller(r)) + if err != nil { + h.mapErr(w, err) + return + } + if job == nil { + Fail(w, http.StatusNotFound, 404, "job not found") + return + } + OK(w, job) +} + +func (h *Handler) result(w http.ResponseWriter, r *http.Request) { + job, err := h.Jobs.Get(r.PathValue("id"), h.caller(r)) + if err != nil { + h.mapErr(w, err) + return + } + if job == nil { + Fail(w, http.StatusNotFound, 404, "job not found") + return + } + OK(w, map[string]interface{}{ + "id": job.ID, + "status": job.Status, + "result": job.Result, + "error": job.Error, + "dialogues": job.Dialogues, + }) +} + +func (h *Handler) mapErr(w http.ResponseWriter, err error) { + var be *service.BizError + if errors.As(err, &be) { + switch { + case errors.Is(be.Kind, service.ErrForbidden): + Fail(w, http.StatusForbidden, 403, be.Msg) + case errors.Is(be.Kind, service.ErrBadState), errors.Is(be.Kind, service.ErrBadArg): + Fail(w, http.StatusBadRequest, 400, be.Msg) + default: + Fail(w, http.StatusInternalServerError, 500, be.Msg) + } + return + } + Fail(w, http.StatusInternalServerError, 500, err.Error()) +} + +// WithCORS 本机联调放开跨域（经反代同源时无影响）。 +func WithCORS(next http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Access-Control-Allow-Origin", "*") + w.Header().Set("Access-Control-Allow-Methods", "GET, POST, PUT, DELETE, OPTIONS") + w.Header().Set("Access-Control-Allow-Headers", "*") + if r.Method == http.MethodOptions { + w.WriteHeader(http.StatusNoContent) + return + } + next.ServeHTTP(w, r) + }) +} diff --git a/internal/api/response.go b/internal/api/response.go new file mode 100644 index 0000000..ae81152 --- /dev/null +++ b/internal/api/response.go @@ -0,0 +1,31 @@ +package api + +import ( + "encoding/json" + "net/http" +) + +// ApiResponse 与 Java 一致：{ code, message, data }；成功 code=0。 +type ApiResponse struct { + Code int `json:"code"` + Message string `json:"message"` + Data interface{} `json:"data"` +} + +func writeJSON(w http.ResponseWriter, status int, body interface{}) { + w.Header().Set("Content-Type", "application/json; charset=utf-8") + w.WriteHeader(status) + _ = json.NewEncoder(w).Encode(body) +} + +func OK(w http.ResponseWriter, data interface{}) { + writeJSON(w, http.StatusOK, ApiResponse{Code: 0, Message: "ok", Data: data}) +} + +func Created(w http.ResponseWriter, data interface{}) { + writeJSON(w, http.StatusCreated, ApiResponse{Code: 0, Message: "ok", Data: data}) +} + +func Fail(w http.ResponseWriter, httpStatus, code int, msg string) { + writeJSON(w, httpStatus, ApiResponse{Code: code, Message: msg, Data: nil}) +} diff --git a/internal/config/config.go b/internal/config/config.go new file mode 100644 index 0000000..ccbbd4a --- /dev/null +++ b/internal/config/config.go @@ -0,0 +1,144 @@ +package config + +import ( + "os" + "strconv" + "strings" +) + +// Config 运行配置（环境变量优先，与 Java EBA_* / analysis.* 对齐）。 +type Config struct { + Port int + + // MySQL：优先 EBA_DB_DSN；否则用 HOST/PORT/NAME/USER/PASSWORD 拼，或解析 JDBC URL。 + DBDSN string + + AnalysisMode string // live | demo + AuthEnabled bool + DemoEnabled bool + LivePollMS int + DemoTickMS int + + DeerFlowBaseURL string + DeerFlowUser string + DeerFlowPassword string + DeerFlowCoordinator string + DeerFlowStreamIdleTimeout int // 毫秒 +} + +// Load 从环境变量加载；缺省贴近本机联调。 +func Load() Config { + c := Config{ + Port: envInt("SERVER_PORT", 18097), + AnalysisMode: env("ANALYSIS_MODE", "live"), + AuthEnabled: envBool("ANALYSIS_AUTH_ENABLED", false), + DemoEnabled: envBool("EBA_WORKER_DEMO_ENABLED", false), + LivePollMS: envInt("EBA_WORKER_LIVE_POLL_MS", 1000), + DemoTickMS: envInt("EBA_WORKER_TICK_MS", 400), + DeerFlowBaseURL: strings.TrimRight(env("EBA_DF_BASE_URL", "http://127.0.0.1:8001"), "/"), + DeerFlowUser: env("EBA_DF_USER", "admin"), + DeerFlowPassword: env("EBA_DF_PASSWORD", "123ewq"), + DeerFlowCoordinator: env("EBA_DF_COORDINATOR", "roundtable-coordinator"), + DeerFlowStreamIdleTimeout: envInt("EBA_DF_STREAM_IDLE_TIMEOUT_MS", 180000), + } + c.DBDSN = resolveDSN() + // live 默认关掉 demo；显式 ANALYSIS_MODE=demo 时打开假进度 + if c.AnalysisMode == "demo" { + c.DemoEnabled = true + } + if c.AnalysisMode == "live" && !envExplicit("EBA_WORKER_DEMO_ENABLED") { + c.DemoEnabled = false + } + return c +} + +func resolveDSN() string { + if d := strings.TrimSpace(os.Getenv("EBA_DB_DSN")); d != "" { + return ensureMultiStatements(d) + } + user := env("EBA_DB_USER", "eba") + pass := env("EBA_DB_PASSWORD", "eba_local_dev") + jdbc := strings.TrimSpace(os.Getenv("EBA_DB_URL")) + if jdbc != "" { + if dsn, ok := jdbcToDSN(jdbc, user, pass); ok { + return dsn + } + } + host := env("EBA_DB_HOST", "127.0.0.1") + port := env("EBA_DB_PORT", "3306") + name := env("EBA_DB_NAME", "eightbf_analysis") + return user + ":" + pass + "@tcp(" + host + ":" + port + ")/" + name + + "?parseTime=true&loc=Asia%2FShanghai&charset=utf8mb4&multiStatements=true" +} + +// jdbcToDSN 把 Java 的 jdbc:mysql://host:port/db?... 转成 Go driver DSN。 +func jdbcToDSN(jdbc, user, pass string) (string, bool) { + s := strings.TrimPrefix(jdbc, "jdbc:mysql://") + if s == jdbc { + return "", false + } + hostPath := s + query := "" + if i := strings.Index(s, "?"); i >= 0 { + hostPath = s[:i] + query = s[i+1:] + } + host := hostPath + db := "eightbf_analysis" + if i := strings.Index(hostPath, "/"); i >= 0 { + host = hostPath[:i] + db = hostPath[i+1:] + } + params := "parseTime=true&loc=Asia%2FShanghai&charset=utf8mb4&multiStatements=true" + _ = query // Java JDBC 专用参数不照搬 + return user + ":" + pass + "@tcp(" + host + ")/" + db + "?" + params, true +} + +func ensureMultiStatements(dsn string) string { + if strings.Contains(dsn, "multiStatements=") { + return dsn + } + if strings.Contains(dsn, "?") { + return dsn + "&multiStatements=true" + } + return dsn + "?multiStatements=true" +} + +func env(k, def string) string { + if v := strings.TrimSpace(os.Getenv(k)); v != "" { + return v + } + return def +} + +func envExplicit(k string) bool { + _, ok := os.LookupEnv(k) + return ok +} + +func envBool(k string, def bool) bool { + v := strings.TrimSpace(os.Getenv(k)) + if v == "" { + return def + } + switch strings.ToLower(v) { + case "1", "true", "yes", "on": + return true + case "0", "false", "no", "off": + return false + default: + return def + } +} + +func envInt(k string, def int) int { + v := strings.TrimSpace(os.Getenv(k)) + if v == "" { + return def + } + n, err := strconv.Atoi(v) + if err != nil { + return def + } + return n +} diff --git a/internal/deerflow/auth.go b/internal/deerflow/auth.go new file mode 100644 index 0000000..efafef0 --- /dev/null +++ b/internal/deerflow/auth.go @@ -0,0 +1,72 @@ +package deerflow + +import ( + "bytes" + "encoding/json" + "fmt" + "io" + "net/http" + "sync" + "time" + + "git.gozeros.com/wsk/eightbf-analysis-go/internal/config" +) + +// Auth 向 DeerFlow 换 Bearer，过期前复用。 +type Auth struct { + cfg config.Config + mu sync.Mutex + token string + expiresAt time.Time + httpClient *http.Client +} + +func NewAuth(cfg config.Config) *Auth { + return &Auth{ + cfg: cfg, + httpClient: &http.Client{Timeout: 30 * time.Second}, + } +} + +func (a *Auth) BearerToken() (string, error) { + a.mu.Lock() + defer a.mu.Unlock() + now := time.Now() + if a.token != "" && a.expiresAt.After(now.Add(2*time.Minute)) { + return a.token, nil + } + body, _ := json.Marshal(map[string]string{ + "username": a.cfg.DeerFlowUser, + "password": a.cfg.DeerFlowPassword, + }) + req, err := http.NewRequest(http.MethodPost, a.cfg.DeerFlowBaseURL+"/api/v1/auth/login/username", bytes.NewReader(body)) + if err != nil { + return "", err + } + req.Header.Set("Content-Type", "application/json") + resp, err := a.httpClient.Do(req) + if err != nil { + return "", err + } + defer resp.Body.Close() + raw, _ := io.ReadAll(resp.Body) + if resp.StatusCode >= 400 { + return "", fmt.Errorf("DeerFlow 登录失败 HTTP %d: %s", resp.StatusCode, string(raw)) + } + var parsed struct { + AccessToken string `json:"access_token"` + ExpiresIn int `json:"expires_in"` + } + if err := json.Unmarshal(raw, &parsed); err != nil { + return "", err + } + if parsed.AccessToken == "" { + return "", fmt.Errorf("DeerFlow 登录响应无 access_token") + } + if parsed.ExpiresIn <= 0 { + parsed.ExpiresIn = 3600 + } + a.token = parsed.AccessToken + a.expiresAt = now.Add(time.Duration(parsed.ExpiresIn) * time.Second) + return a.token, nil +} diff --git a/internal/deerflow/client.go b/internal/deerflow/client.go new file mode 100644 index 0000000..c0e4331 --- /dev/null +++ b/internal/deerflow/client.go @@ -0,0 +1,503 @@ +package deerflow + +import ( + "bufio" + "bytes" + "encoding/json" + "fmt" + "io" + "log" + "net/http" + "strings" + "time" + + "git.gozeros.com/wsk/eightbf-analysis-go/internal/config" +) + +// Client 调 DeerFlow：模型校验 / init / run/stream。 +type Client struct { + cfg config.Config + auth *Auth +} + +func NewClient(cfg config.Config, auth *Auth) *Client { + return &Client{cfg: cfg, auth: auth} +} + +// AssertModelAvailable 未知模型拒绝开跑（禁止依赖网关静默换模）。 +func (c *Client) AssertModelAvailable(model string) error { + want := strings.TrimSpace(model) + if want == "" || strings.EqualFold(want, "demo") { + return nil + } + token, err := c.auth.BearerToken() + if err != nil { + return err + } + req, err := http.NewRequest(http.MethodGet, c.cfg.DeerFlowBaseURL+"/api/models", nil) + if err != nil { + return err + } + req.Header.Set("Accept", "application/json") + req.Header.Set("Authorization", "Bearer "+token) + client := &http.Client{Timeout: 30 * time.Second} + resp, err := client.Do(req) + if err != nil { + return err + } + defer resp.Body.Close() + raw, _ := io.ReadAll(resp.Body) + if resp.StatusCode >= 400 { + return fmt.Errorf("无法校验模型列表 HTTP %d: %s", resp.StatusCode, string(raw)) + } + var root map[string]interface{} + _ = json.Unmarshal(raw, &root) + var arr []interface{} + if m, ok := root["models"].([]interface{}); ok { + arr = m + } else { + var list []interface{} + if err := json.Unmarshal(raw, &list); err == nil { + arr = list + } + } + if arr == nil { + return fmt.Errorf("DeerFlow /api/models 响应无 models 列表，拒绝使用未知模型「%s」", want) + } + for _, item := range arr { + m, ok := item.(map[string]interface{}) + if !ok { + continue + } + name := fmt.Sprint(m["name"]) + id := fmt.Sprint(m["model"]) + display := fmt.Sprint(m["display_name"]) + if want == name || want == id || want == display { + return nil + } + } + return fmt.Errorf("所选模型「%s」不在可用列表中（或不可用）。请换可用模型后重新开始会商，不会自动改用其它模型", want) +} + +// InitMultiAgent 建多席线程，返回 agentName → threadId。 +func (c *Client) InitMultiAgent(agentNames []string, mainAgent, model string) (map[string]string, error) { + body := map[string]interface{}{ + "agent_names": agentNames, + "main_agent_name": mainAgent, + } + if m := strings.TrimSpace(model); m != "" && !strings.EqualFold(m, "demo") { + body["model"] = m + } + threadIDs := map[string]string{} + var errMsg string + err := c.readSSE("/api/multi-agent/init", body, nil, func(event string, parsed map[string]interface{}) (bool, error) { + ev := event + if ev == "" { + ev = fmt.Sprint(parsed["event"]) + } + switch ev { + case "seat_ready", "coordinator_ready": + agent := strField(parsed, "agent_name", "main_agent_name") + tid := strField(parsed, "thread_id") + if agent != "" && tid != "" { + threadIDs[agent] = tid + } + case "init_done": + if tids, ok := parsed["thread_ids"].(map[string]interface{}); ok { + for k, v := range tids { + threadIDs[k] = fmt.Sprint(v) + } + } + return true, nil + case "error": + errMsg = strField(parsed, "detail") + if errMsg == "" { + errMsg = "init error" + } + return true, nil + } + return false, nil + }) + if err != nil { + return nil, err + } + if errMsg != "" { + return nil, fmt.Errorf("多智能体初始化失败：%s", errMsg) + } + if len(threadIDs) == 0 { + return nil, fmt.Errorf("多智能体初始化未返回 thread_ids") + } + log.Printf("DeerFlow init 完成 seats=%d", len(threadIDs)) + return threadIDs, nil +} + +// RunStream 跑一轮流式会商。 +func (c *Client) RunStream( + agentType string, + threadIDs map[string]string, + agentName, displayName, newMessage, model string, + synthesisMode bool, + onText TextDeltaHandler, + cancel CancelCheck, +) (*StreamResult, error) { + tidNode := map[string]string{} + for k, v := range threadIDs { + tidNode[k] = v + } + body := map[string]interface{}{ + "agent_type": agentType, + "d_agent_thread_id": tidNode, + "agent_name": agentName, + "new_message": newMessage, + "thinking_enabled": true, + "reasoning_effort": "medium", + "subagent_enabled": false, + "seat_skill_directive": false, + "synthesis_mode": synthesisMode, + "suppress_pre_broadcast": true, + "parallel_no_file": false, + "peer_deliveries": []interface{}{}, + } + if displayName != "" { + body["display_name"] = displayName + } + if m := strings.TrimSpace(model); m != "" && !strings.EqualFold(m, "demo") { + body["model"] = m + } + + result := &StreamResult{} + var contentAcc strings.Builder + lastEvent := "" + terminal := false + display := &streamDisplayAssembler{} + + err := c.readSSE("/api/multi-agent/run/stream", body, cancel, func(event string, parsed map[string]interface{}) (bool, error) { + if event != "" { + lastEvent = event + } + if cancel != nil && cancel() { + result.Aborted = true + return true, nil + } + if lastEvent == "error" { + result.InvalidDelivery = true + detail := strField(parsed, "detail", "message", "content") + if detail != "" { + result.Content = detail + } else if contentAcc.Len() > 0 { + result.Content = contentAcc.String() + } else { + result.Content = "模型运行报错" + } + terminal = true + return true, nil + } + if status, _ := parsed["status"].(string); status == "model_switch" { + result.InvalidDelivery = true + failed := strField(parsed, "failed_model") + next := strField(parsed, "next_model") + reason := strField(parsed, "reason_label", "reason") + if reason == "" { + reason = "调用失败" + } + msg := "所选模型" + if failed != "" { + msg += "「" + failed + "」" + } + msg += "异常（" + reason + "）" + if next != "" { + msg += "，已拒绝自动切换到「" + next + "」" + } + msg += "。请换可用模型后重新开始会商。" + result.Content = msg + terminal = true + return true, nil + } + if isFinalStatusFrame(parsed) { + if closeThink := display.closeThinkIfOpen(); closeThink != "" { + contentAcc.WriteString(closeThink) + if onText != nil { + _ = onText(closeThink) + } + } + if c, ok := parsed["content"].(string); ok { + result.Content = c + } + if n, ok := parsed["agent_name"].(string); ok { + result.AgentName = n + } + switch st := parsed["status"].(type) { + case string: + if st == "clarification" { + result.Clarification = true + result.ClarificationDetail = strField(parsed, "content") + } else if st == "invalid_delivery" || st == "error" { + result.InvalidDelivery = true + } + case []interface{}: + for _, row := range st { + arr, ok := row.([]interface{}) + if !ok || len(arr) < 2 { + continue + } + result.Dispatched = append(result.Dispatched, [2]string{fmt.Sprint(arr[0]), fmt.Sprint(arr[1])}) + } + } + if result.Content == "" { + result.Content = contentAcc.String() + } + terminal = true + return true, nil + } + if lastEvent == "messages" || lastEvent == "" { + delta := display.push(parsed) + if delta != "" { + contentAcc.WriteString(delta) + if onText != nil { + if err := onText(delta); err != nil { + return true, err + } + } + } + } + return false, nil + }) + if err != nil { + return result, err + } + if result.Content == "" { + result.Content = contentAcc.String() + } + if !result.Aborted && !terminal && !result.Clarification { + result.InvalidDelivery = true + if strings.TrimSpace(result.Content) == "" { + result.Content = "流式输出异常结束（无终态帧）" + } + } + return result, nil +} + +func (c *Client) readSSE( + path string, + body map[string]interface{}, + cancel CancelCheck, + onData func(event string, parsed map[string]interface{}) (stop bool, err error), +) error { + token, err := c.auth.BearerToken() + if err != nil { + return err + } + raw, _ := json.Marshal(body) + req, err := http.NewRequest(http.MethodPost, c.cfg.DeerFlowBaseURL+path, bytes.NewReader(raw)) + if err != nil { + return err + } + req.Header.Set("Content-Type", "application/json") + req.Header.Set("Accept", "text/event-stream") + req.Header.Set("Authorization", "Bearer "+token) + client := &http.Client{Timeout: time.Duration(c.cfg.DeerFlowStreamIdleTimeout) * time.Millisecond} + resp, err := client.Do(req) + if err != nil { + return err + } + defer resp.Body.Close() + if resp.StatusCode >= 400 { + b, _ := io.ReadAll(resp.Body) + return fmt.Errorf("DeerFlow %s HTTP %d: %s", path, resp.StatusCode, string(b)) + } + scanner := bufio.NewScanner(resp.Body) + // 单行可能很长 + buf := make([]byte, 0, 64*1024) + scanner.Buffer(buf, 4*1024*1024) + curEvent := "" + for scanner.Scan() { + if cancel != nil && cancel() { + return nil + } + line := scanner.Text() + if strings.HasPrefix(line, "event:") { + curEvent = strings.TrimSpace(strings.TrimPrefix(line, "event:")) + continue + } + if !strings.HasPrefix(line, "data:") { + continue + } + payload := strings.TrimSpace(strings.TrimPrefix(line, "data:")) + if payload == "" { + continue + } + var parsed interface{} + if err := json.Unmarshal([]byte(payload), &parsed); err != nil { + continue + } + switch t := parsed.(type) { + case map[string]interface{}: + stop, err := onData(curEvent, t) + if err != nil || stop { + return err + } + case []interface{}: + // messages 帧常为数组，包一层给 assembler + stop, err := onData(curEvent, map[string]interface{}{"_arr": t}) + if err != nil || stop { + return err + } + // 同时把数组直接交给 push 路径：用特殊 key 不够；改为在 onData 外再处理 + _ = t + } + } + return scanner.Err() +} + +func isFinalStatusFrame(v map[string]interface{}) bool { + if v == nil { + return false + } + if _, ok := v["status"]; !ok { + return false + } + if _, ok := v["content"]; !ok { + return false + } + switch st := v["status"].(type) { + case string: + return true + case []interface{}: + for _, item := range st { + arr, ok := item.([]interface{}) + if !ok || len(arr) < 2 { + return false + } + } + return true + default: + return false + } +} + +func strField(m map[string]interface{}, keys ...string) string { + for _, k := range keys { + if v, ok := m[k]; ok && v != nil { + s := strings.TrimSpace(fmt.Sprint(v)) + if s != "" && s != "" { + return s + } + } + } + return "" +} + +// —— 流式展示折算（reasoning → ）—— + +type streamDisplayAssembler struct { + thinkOpen bool + reasoningSeen string +} + +func (d *streamDisplayAssembler) push(parsed map[string]interface{}) string { + var arr []interface{} + if a, ok := parsed["_arr"].([]interface{}); ok { + arr = a + } else { + // 兼容：有些路径直接把数组当根；此处若不是 AI chunk 则空 + return "" + } + if len(arr) == 0 { + return "" + } + obj, ok := arr[0].(map[string]interface{}) + if !ok { + return "" + } + typ := strings.ToLower(fmt.Sprint(obj["type"])) + if typ != "ai" && typ != "aimessagechunk" && typ != "aimessage" { + return "" + } + var reasoning, answer strings.Builder + switch c := obj["content"].(type) { + case string: + answer.WriteString(c) + case []interface{}: + for _, item := range c { + if s, ok := item.(string); ok { + answer.WriteString(s) + continue + } + m, ok := item.(map[string]interface{}) + if !ok { + continue + } + bt := strings.ToLower(fmt.Sprint(m["type"])) + val := firstText(m, "thinking", "text", "content") + if val == "" { + continue + } + if strings.Contains(bt, "think") || strings.Contains(bt, "reason") { + reasoning.WriteString(val) + } else { + answer.WriteString(val) + } + } + } + if fromKw := extractReasoning(obj); fromKw != "" { + if strings.HasPrefix(fromKw, d.reasoningSeen) { + more := fromKw[len(d.reasoningSeen):] + if more != "" { + reasoning.WriteString(more) + } + d.reasoningSeen = fromKw + } else if !strings.Contains(d.reasoningSeen, fromKw) { + reasoning.WriteString(fromKw) + d.reasoningSeen += fromKw + } + } + var out strings.Builder + if reasoning.Len() > 0 { + if !d.thinkOpen { + out.WriteString("") + d.thinkOpen = true + } + out.WriteString(reasoning.String()) + } + if answer.Len() > 0 { + if d.thinkOpen { + out.WriteString("") + d.thinkOpen = false + } + out.WriteString(answer.String()) + } + return out.String() +} + +func (d *streamDisplayAssembler) closeThinkIfOpen() string { + if !d.thinkOpen { + return "" + } + d.thinkOpen = false + return "" +} + +func extractReasoning(obj map[string]interface{}) string { + for _, k := range []string{"reasoning_content", "reasoning"} { + if s, ok := obj[k].(string); ok && s != "" { + return s + } + } + if kw, ok := obj["additional_kwargs"].(map[string]interface{}); ok { + for _, k := range []string{"reasoning_content", "reasoning"} { + if s, ok := kw[k].(string); ok && s != "" { + return s + } + } + } + return "" +} + +func firstText(m map[string]interface{}, keys ...string) string { + for _, k := range keys { + if s, ok := m[k].(string); ok && s != "" { + return s + } + } + return "" +} diff --git a/internal/deerflow/types.go b/internal/deerflow/types.go new file mode 100644 index 0000000..62aaac5 --- /dev/null +++ b/internal/deerflow/types.go @@ -0,0 +1,18 @@ +package deerflow + +// StreamResult 一次 run/stream 汇总（与 Java StreamResult 对齐）。 +type StreamResult struct { + Content string + AgentName string + Dispatched [][2]string // [agentId, task] + Clarification bool + ClarificationDetail string + InvalidDelivery bool + Aborted bool +} + +// TextDeltaHandler 流式正文增量（可含 ）。 +type TextDeltaHandler func(delta string) error + +// CancelCheck 读 SSE 前询问是否已取消。 +type CancelCheck func() bool diff --git a/internal/domain/job.go b/internal/domain/job.go new file mode 100644 index 0000000..ff89f5e --- /dev/null +++ b/internal/domain/job.go @@ -0,0 +1,56 @@ +package domain + +import "time" + +// AnalysisJob 一场会商作业（服务端「研讨录像带」）。 +// JSON 字段名与前端 mapJob / Java 实体对齐，少做二次映射。 +type AnalysisJob struct { + ID string `json:"id"` + RequestID string `json:"requestId,omitempty"` + TaskID string `json:"taskId,omitempty"` + DraftID string `json:"draftId,omitempty"` + DraftKey string `json:"draftKey,omitempty"` + Status string `json:"status"` + Phase string `json:"phase,omitempty"` + Cycle int `json:"cycle"` + ActiveAgentID string `json:"activeAgentId,omitempty"` + ConsensusPercentage int `json:"consensusPercentage"` + PendingClarification string `json:"pendingClarification,omitempty"` + OrchestrationMode string `json:"orchestrationMode,omitempty"` + Model string `json:"model,omitempty"` + Error string `json:"error,omitempty"` + OwnerAccountID string `json:"ownerAccountId,omitempty"` + Agents []map[string]interface{} `json:"agents"` + Dialogues []map[string]interface{} `json:"dialogues"` + ThreadIDs map[string]interface{} `json:"threadIds,omitempty"` + Plan map[string]interface{} `json:"plan"` + Access map[string]interface{} `json:"access"` + Progress map[string]interface{} `json:"progress"` + Result map[string]interface{} `json:"result,omitempty"` + CreatedAt *time.Time `json:"createdAt,omitempty"` + UpdatedAt *time.Time `json:"updatedAt,omitempty"` +} + +// EnsurePlan plan 为空时补可写 Map。 +func (j *AnalysisJob) EnsurePlan() map[string]interface{} { + if j.Plan == nil { + j.Plan = map[string]interface{}{} + } + return j.Plan +} + +// EnsureProgress progress 为空时补可写 Map。 +func (j *AnalysisJob) EnsureProgress() map[string]interface{} { + if j.Progress == nil { + j.Progress = map[string]interface{}{} + } + return j.Progress +} + +// EnsureDialogues dialogues 为空时补切片。 +func (j *AnalysisJob) EnsureDialogues() []map[string]interface{} { + if j.Dialogues == nil { + j.Dialogues = []map[string]interface{}{} + } + return j.Dialogues +} diff --git a/internal/domain/status.go b/internal/domain/status.go new file mode 100644 index 0000000..1389188 --- /dev/null +++ b/internal/domain/status.go @@ -0,0 +1,21 @@ +package domain + +// 会商作业状态（与 Java JobStatus / 前端约定一致，勿自创别名）。 +const ( + StatusQueued = "queued" + StatusRunning = "running" + StatusAwaitingInput = "awaiting_input" + StatusDone = "done" + StatusError = "error" + StatusCancelled = "cancelled" +) + +// IsActive 是否仍可能被 worker 推进（含等人中）。 +func IsActive(status string) bool { + return status == StatusQueued || status == StatusRunning || status == StatusAwaitingInput +} + +// IsTerminal 是否已结束，不应再 resume / inject / 改计划。 +func IsTerminal(status string) bool { + return status == StatusDone || status == StatusError || status == StatusCancelled +} diff --git a/internal/repo/job_repo.go b/internal/repo/job_repo.go new file mode 100644 index 0000000..d0d4d8d --- /dev/null +++ b/internal/repo/job_repo.go @@ -0,0 +1,298 @@ +package repo + +import ( + "database/sql" + "encoding/json" + "time" + + "git.gozeros.com/wsk/eightbf-analysis-go/internal/domain" +) + +// JobRepo 会商作业表读写（eba_analysis_job）。大字段整段 JSON 文本存。 +type JobRepo struct { + DB *sql.DB +} + +func NewJobRepo(db *sql.DB) *JobRepo { + return &JobRepo{DB: db} +} + +// EnsureSchema 执行建表脚本（幂等；重复索引错误由调用方忽略亦可）。 +func (r *JobRepo) EnsureSchema(schemaSQL string) error { + _, err := r.DB.Exec(schemaSQL) + return err +} + +func (r *JobRepo) Insert(job *domain.AnalysisJob) error { + now := time.Now() + job.CreatedAt = &now + job.UpdatedAt = &now + _, err := r.DB.Exec(`INSERT INTO eba_analysis_job( + id,request_id,task_id,draft_id,draft_key,status,phase,cycle,active_agent_id, + consensus_percentage,pending_clarification,orchestration_mode,model,error_message, + owner_account_id,agents_json,chain_json,plan_json,access_json,thread_ids_json, + progress_json,result_json,cancel_flag,created_at,updated_at + ) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,0,?,?)`, + job.ID, nullStr(job.RequestID), nullStr(job.TaskID), nullStr(job.DraftID), nullStr(job.DraftKey), + job.Status, nullStr(job.Phase), job.Cycle, nullStr(job.ActiveAgentID), + job.ConsensusPercentage, nullStr(job.PendingClarification), nullStr(job.OrchestrationMode), + nullStr(job.Model), nullStr(job.Error), nullStr(job.OwnerAccountID), + mustJSON(job.Agents), mustJSON(job.Dialogues), mustJSON(job.Plan), mustJSON(job.Access), + mustJSON(job.ThreadIDs), mustJSON(job.Progress), mustJSON(job.Result), + now, now, + ) + return err +} + +// UpdateProgress 全量刷新进度相关列；cancelled 时同步 cancel_flag=1。 +func (r *JobRepo) UpdateProgress(job *domain.AnalysisJob) error { + now := time.Now() + job.UpdatedAt = &now + cancel := 0 + if job.Status == domain.StatusCancelled { + cancel = 1 + } + _, err := r.DB.Exec(`UPDATE eba_analysis_job SET status=?, phase=?, cycle=?, active_agent_id=?, + consensus_percentage=?, pending_clarification=?, model=?, error_message=?, owner_account_id=?, + agents_json=?, chain_json=?, plan_json=?, access_json=?, thread_ids_json=?, + progress_json=?, result_json=?, cancel_flag=?, updated_at=? WHERE id=?`, + job.Status, nullStr(job.Phase), job.Cycle, nullStr(job.ActiveAgentID), + job.ConsensusPercentage, nullStr(job.PendingClarification), nullStr(job.Model), + nullStr(job.Error), nullStr(job.OwnerAccountID), + mustJSON(job.Agents), mustJSON(job.Dialogues), mustJSON(job.Plan), mustJSON(job.Access), + mustJSON(job.ThreadIDs), mustJSON(job.Progress), mustJSON(job.Result), + cancel, now, job.ID, + ) + return err +} + +func (r *JobRepo) FindByID(id string) (*domain.AnalysisJob, error) { + rows, err := r.DB.Query(`SELECT * FROM eba_analysis_job WHERE id=?`, id) + if err != nil { + return nil, err + } + defer rows.Close() + if !rows.Next() { + return nil, nil + } + return scanJob(rows) +} + +func (r *JobRepo) FindActiveByTaskID(taskID string) (*domain.AnalysisJob, error) { + if taskID == "" { + return nil, nil + } + rows, err := r.DB.Query(`SELECT * FROM eba_analysis_job WHERE task_id=? AND status IN (?,?,?) + ORDER BY created_at DESC LIMIT 1`, + taskID, domain.StatusQueued, domain.StatusRunning, domain.StatusAwaitingInput) + if err != nil { + return nil, err + } + defer rows.Close() + if !rows.Next() { + return nil, nil + } + return scanJob(rows) +} + +func (r *JobRepo) FindLatestByTaskID(taskID string) (*domain.AnalysisJob, error) { + if taskID == "" { + return nil, nil + } + rows, err := r.DB.Query(`SELECT * FROM eba_analysis_job WHERE task_id=? ORDER BY created_at DESC LIMIT 1`, taskID) + if err != nil { + return nil, err + } + defer rows.Close() + if !rows.Next() { + return nil, nil + } + return scanJob(rows) +} + +func (r *JobRepo) FindActiveJobs() ([]*domain.AnalysisJob, error) { + rows, err := r.DB.Query(`SELECT * FROM eba_analysis_job WHERE status IN (?,?,?) ORDER BY created_at ASC`, + domain.StatusQueued, domain.StatusRunning, domain.StatusAwaitingInput) + if err != nil { + return nil, err + } + defer rows.Close() + return scanJobs(rows) +} + +func (r *JobRepo) FindQueuedJobs() ([]*domain.AnalysisJob, error) { + rows, err := r.DB.Query(`SELECT * FROM eba_analysis_job WHERE status=? ORDER BY created_at ASC`, domain.StatusQueued) + if err != nil { + return nil, err + } + defer rows.Close() + return scanJobs(rows) +} + +func (r *JobRepo) MarkCancel(id string) (bool, error) { + res, err := r.DB.Exec(`UPDATE eba_analysis_job SET cancel_flag=1, updated_at=? WHERE id=? AND status IN (?,?,?)`, + time.Now(), id, domain.StatusQueued, domain.StatusRunning, domain.StatusAwaitingInput) + if err != nil { + return false, err + } + n, _ := res.RowsAffected() + return n > 0, nil +} + +func (r *JobRepo) IsCancelFlag(id string) (bool, error) { + var v sql.NullInt64 + err := r.DB.QueryRow(`SELECT cancel_flag FROM eba_analysis_job WHERE id=?`, id).Scan(&v) + if err == sql.ErrNoRows { + return false, nil + } + if err != nil { + return false, err + } + return v.Valid && v.Int64 == 1, nil +} + +func scanJobs(rows *sql.Rows) ([]*domain.AnalysisJob, error) { + var out []*domain.AnalysisJob + for rows.Next() { + j, err := scanJob(rows) + if err != nil { + return nil, err + } + out = append(out, j) + } + return out, rows.Err() +} + +func scanJob(rows *sql.Rows) (*domain.AnalysisJob, error) { + cols, err := rows.Columns() + if err != nil { + return nil, err + } + vals := make([]interface{}, len(cols)) + ptrs := make([]interface{}, len(cols)) + for i := range vals { + ptrs[i] = &vals[i] + } + if err := rows.Scan(ptrs...); err != nil { + return nil, err + } + m := map[string]interface{}{} + for i, c := range cols { + m[c] = vals[i] + } + j := &domain.AnalysisJob{} + j.ID = asString(m["id"]) + j.RequestID = asString(m["request_id"]) + j.TaskID = asString(m["task_id"]) + j.DraftID = asString(m["draft_id"]) + j.DraftKey = asString(m["draft_key"]) + j.Status = asString(m["status"]) + j.Phase = asString(m["phase"]) + j.Cycle = asInt(m["cycle"]) + j.ActiveAgentID = asString(m["active_agent_id"]) + j.ConsensusPercentage = asInt(m["consensus_percentage"]) + j.PendingClarification = asString(m["pending_clarification"]) + j.OrchestrationMode = asString(m["orchestration_mode"]) + j.Model = asString(m["model"]) + j.Error = asString(m["error_message"]) + j.OwnerAccountID = asString(m["owner_account_id"]) + j.Agents = readList(asString(m["agents_json"])) + j.Dialogues = readList(asString(m["chain_json"])) + j.Plan = readMap(asString(m["plan_json"])) + j.Access = readMap(asString(m["access_json"])) + j.ThreadIDs = readMap(asString(m["thread_ids_json"])) + j.Progress = readMap(asString(m["progress_json"])) + j.Result = readMap(asString(m["result_json"])) + if t := asTime(m["created_at"]); t != nil { + j.CreatedAt = t + } + if t := asTime(m["updated_at"]); t != nil { + j.UpdatedAt = t + } + return j, nil +} + +func mustJSON(v interface{}) interface{} { + if v == nil { + return nil + } + b, err := json.Marshal(v) + if err != nil { + return nil + } + return string(b) +} + +func readMap(jsonStr string) map[string]interface{} { + if jsonStr == "" { + return nil + } + var m map[string]interface{} + if err := json.Unmarshal([]byte(jsonStr), &m); err != nil { + return nil + } + return m +} + +func readList(jsonStr string) []map[string]interface{} { + if jsonStr == "" { + return []map[string]interface{}{} + } + var list []map[string]interface{} + if err := json.Unmarshal([]byte(jsonStr), &list); err != nil { + return []map[string]interface{}{} + } + return list +} + +func nullStr(s string) interface{} { + if s == "" { + return nil + } + return s +} + +func asString(v interface{}) string { + if v == nil { + return "" + } + switch t := v.(type) { + case string: + return t + case []byte: + return string(t) + default: + return "" + } +} + +func asInt(v interface{}) int { + if v == nil { + return 0 + } + switch t := v.(type) { + case int64: + return int(t) + case int32: + return int(t) + case int: + return t + case []byte: + var n int + _ = json.Unmarshal(t, &n) + return n + default: + return 0 + } +} + +func asTime(v interface{}) *time.Time { + if v == nil { + return nil + } + if t, ok := v.(time.Time); ok { + tt := t + return &tt + } + return nil +} diff --git a/internal/repo/schema.go b/internal/repo/schema.go new file mode 100644 index 0000000..03a714c --- /dev/null +++ b/internal/repo/schema.go @@ -0,0 +1,47 @@ +package repo + +import ( + "log" + "strings" +) + +// ApplySchema 按语句拆开执行；重复索引等可忽略错误不阻断启动。 +func (r *JobRepo) ApplySchema(schemaSQL string) { + for _, stmt := range splitSQL(schemaSQL) { + if _, err := r.DB.Exec(stmt); err != nil { + msg := err.Error() + // 1061 Duplicate key name / 已存在表 + if strings.Contains(msg, "1061") || strings.Contains(msg, "1050") || + strings.Contains(msg, "already exists") { + continue + } + log.Printf("schema 语句失败（已跳过）: %v | sql=%.80s…", err, stmt) + } + } +} + +func splitSQL(s string) []string { + var out []string + var b strings.Builder + lines := strings.Split(s, "\n") + for _, line := range lines { + trim := strings.TrimSpace(line) + if trim == "" || strings.HasPrefix(trim, "--") { + continue + } + b.WriteString(line) + b.WriteByte('\n') + if strings.HasSuffix(trim, ";") { + stmt := strings.TrimSpace(b.String()) + stmt = strings.TrimSuffix(stmt, ";") + if stmt != "" { + out = append(out, stmt) + } + b.Reset() + } + } + if rest := strings.TrimSpace(b.String()); rest != "" { + out = append(out, rest) + } + return out +} diff --git a/internal/service/access.go b/internal/service/access.go new file mode 100644 index 0000000..b632556 --- /dev/null +++ b/internal/service/access.go @@ -0,0 +1,217 @@ +package service + +import ( + "fmt" + "strings" + + "git.gozeros.com/wsk/eightbf-analysis-go/internal/domain" +) + +const ( + RoleAdmin = "admin" + RoleTaskOwner = "task_owner" + RolePositionOwner = "position_owner" + RoleViewer = "viewer" +) + +// Caller 当前请求调用方（来自 X-Account-Id / X-Analysis-Role）。 +type Caller struct { + AccountID string + Role string +} + +func NewCaller(accountID, role string) Caller { + r := strings.TrimSpace(role) + if r == "" { + r = RoleViewer + } + return Caller{AccountID: strings.TrimSpace(accountID), Role: r} +} + +// AccessService 岗位账号限制与对话投影。 +type AccessService struct { + AuthEnabled bool +} + +func (a *AccessService) CanMutate(job *domain.AnalysisJob, c Caller) bool { + if !a.AuthEnabled { + return true + } + if c.AccountID == "" { + return false + } + if c.Role == RoleAdmin || c.Role == RoleTaskOwner { + return true + } + if c.AccountID == strings.TrimSpace(job.OwnerAccountID) { + return true + } + if a.isListedTaskOwner(job, c.AccountID) { + return true + } + return c.Role == RolePositionOwner && len(a.boundSeats(job, c.AccountID)) > 0 +} + +func (a *AccessService) CanCancel(job *domain.AnalysisJob, c Caller) bool { + if c.Role == RolePositionOwner { + if c.AccountID == strings.TrimSpace(job.OwnerAccountID) || a.isListedTaskOwner(job, c.AccountID) { + return true + } + return false + } + if !a.AuthEnabled { + return true + } + if c.AccountID == "" { + return false + } + if c.Role == RoleAdmin || c.Role == RoleTaskOwner { + return true + } + if c.AccountID == strings.TrimSpace(job.OwnerAccountID) { + return true + } + return a.isListedTaskOwner(job, c.AccountID) +} + +func (a *AccessService) CanRead(job *domain.AnalysisJob, c Caller) bool { + if !a.AuthEnabled { + return true + } + if c.AccountID == "" { + return false + } + if a.CanMutate(job, c) || a.CanCancel(job, c) { + return true + } + return len(a.boundSeats(job, c.AccountID)) > 0 +} + +// ProjectForCaller 按角色裁剪对话后返回（责任人全量；岗位只见本席+总控/系统）。 +func (a *AccessService) ProjectForCaller(job *domain.AnalysisJob, c Caller) *domain.AnalysisJob { + if job == nil { + return nil + } + if c.AccountID == "" { + return job + } + if c.Role == RoleAdmin || c.Role == RoleTaskOwner || + c.AccountID == strings.TrimSpace(job.OwnerAccountID) || + a.isListedTaskOwner(job, c.AccountID) { + return job + } + if c.Role != RolePositionOwner && !a.AuthEnabled { + return job + } + seats := a.boundSeats(job, c.AccountID) + copyJob := *job + if len(seats) == 0 { + var filtered []map[string]interface{} + for _, d := range job.Dialogues { + role := asStr(d["role"]) + if role == "system" || role == "user" { + filtered = append(filtered, d) + } + } + copyJob.Dialogues = filtered + if job.Result != nil { + r := cloneMap(job.Result) + delete(r, "seatDeliveries") + copyJob.Result = r + } + return ©Job + } + var filtered []map[string]interface{} + for _, d := range job.Dialogues { + role := asStr(d["role"]) + agentID := asStr(d["agentId"]) + if role == "system" || role == "user" || role == "leader" { + filtered = append(filtered, d) + } else if seats[agentID] { + filtered = append(filtered, d) + } + } + copyJob.Dialogues = filtered + if job.Result != nil { + r := cloneMap(job.Result) + if raw, ok := r["seatDeliveries"].([]interface{}); ok { + var kept []interface{} + for _, o := range raw { + m, ok := o.(map[string]interface{}) + if !ok { + continue + } + if seats[asStr(m["agentId"])] { + kept = append(kept, m) + } + } + r["seatDeliveries"] = kept + } + copyJob.Result = r + } + return ©Job +} + +func (a *AccessService) boundSeats(job *domain.AnalysisJob, accountID string) map[string]bool { + out := map[string]bool{} + if job == nil || accountID == "" || job.Access == nil { + return out + } + raw, ok := job.Access["seatAccounts"] + if !ok { + return out + } + m, ok := raw.(map[string]interface{}) + if !ok { + return out + } + for k, v := range m { + if accountInList(v, accountID) { + out[k] = true + } + } + return out +} + +func (a *AccessService) isListedTaskOwner(job *domain.AnalysisJob, accountID string) bool { + if job == nil || job.Access == nil { + return false + } + return accountInList(job.Access["taskOwnerAccountIds"], accountID) || + accountInList(job.Access["adminAccountIds"], accountID) +} + +func accountInList(raw interface{}, accountID string) bool { + if raw == nil || accountID == "" { + return false + } + switch t := raw.(type) { + case []interface{}: + for _, o := range t { + if strings.TrimSpace(asStr(o)) == accountID { + return true + } + } + return false + default: + return strings.TrimSpace(asStr(raw)) == accountID + } +} + +func cloneMap(m map[string]interface{}) map[string]interface{} { + out := make(map[string]interface{}, len(m)) + for k, v := range m { + out[k] = v + } + return out +} + +func asStr(v interface{}) string { + if v == nil { + return "" + } + if s, ok := v.(string); ok { + return s + } + return strings.TrimSpace(fmt.Sprint(v)) +} diff --git a/internal/service/access_test.go b/internal/service/access_test.go new file mode 100644 index 0000000..bc26caa --- /dev/null +++ b/internal/service/access_test.go @@ -0,0 +1,50 @@ +package service + +import ( + "testing" + + "git.gozeros.com/wsk/eightbf-analysis-go/internal/domain" +) + +func TestPositionOwnerProjection(t *testing.T) { + a := &AccessService{AuthEnabled: false} + job := &domain.AnalysisJob{ + OwnerAccountID: "owner1", + Access: map[string]interface{}{ + "seatAccounts": map[string]interface{}{ + "seat-a": []interface{}{"acc-a"}, + }, + }, + Dialogues: []map[string]interface{}{ + {"role": "leader", "agentId": "coord", "content": "派活"}, + {"role": "special", "agentId": "seat-a", "content": "本席"}, + {"role": "special", "agentId": "seat-b", "content": "它席"}, + {"role": "user", "content": "我说"}, + }, + } + out := a.ProjectForCaller(job, NewCaller("acc-a", RolePositionOwner)) + if len(out.Dialogues) != 3 { + t.Fatalf("期望 3 条（leader+本席+user），得到 %d", len(out.Dialogues)) + } + for _, d := range out.Dialogues { + if asStr(d["agentId"]) == "seat-b" { + t.Fatal("岗位责任人不应看到它席气泡") + } + } +} + +func TestPositionOwnerCannotCancel(t *testing.T) { + a := &AccessService{AuthEnabled: false} + job := &domain.AnalysisJob{ + OwnerAccountID: "owner1", + Access: map[string]interface{}{ + "seatAccounts": map[string]interface{}{"seat-a": []interface{}{"acc-a"}}, + }, + } + if a.CanCancel(job, NewCaller("acc-a", RolePositionOwner)) { + t.Fatal("岗位责任人默认不能 cancel") + } + if !a.CanMutate(job, NewCaller("acc-a", RolePositionOwner)) { + t.Fatal("岗位责任人应可 resume/inject") + } +} diff --git a/internal/service/demo_worker.go b/internal/service/demo_worker.go new file mode 100644 index 0000000..ac0f2a3 --- /dev/null +++ b/internal/service/demo_worker.go @@ -0,0 +1,277 @@ +package service + +import ( + "log" + "strconv" + "time" + + "git.gozeros.com/wsk/eightbf-analysis-go/internal/domain" + "git.gozeros.com/wsk/eightbf-analysis-go/internal/repo" +) + +// DemoWorker 假进度推进（仅 ANALYSIS_MODE=demo）。 +type DemoWorker struct { + Repo *repo.JobRepo + Every time.Duration +} + +var demoLines = []string{ + "【演示】我先说观点：当前应先收窄目标，再谈方案细节。", + "【演示】节奏上建议短周期验证，避免一次铺太开。", + "【演示】依赖和数据口径要尽早对齐，否则后面容易返工。", + "【演示】成本先压到能证明价值的最小投入。", + "【演示】主要风险是范围蔓延和决策偏慢，需要明确拍板人。", + "【演示】我同意形成一版可执行共识，可以进入总结。", +} + +const charsPerTick = 8 + +func (w *DemoWorker) Start() { + if w.Every <= 0 { + w.Every = 400 * time.Millisecond + } + go func() { + t := time.NewTicker(w.Every) + defer t.Stop() + for range t.C { + list, err := w.Repo.FindActiveJobs() + if err != nil { + continue + } + for _, job := range list { + if err := w.advance(job); err != nil { + log.Printf("demo worker advance failed jobId=%s: %v", job.ID, err) + job.Status = domain.StatusError + job.Phase = "error" + job.Error = err.Error() + _ = w.Repo.UpdateProgress(job) + } + } + } + }() +} + +func (w *DemoWorker) advance(job *domain.AnalysisJob) error { + if ok, _ := w.Repo.IsCancelFlag(job.ID); ok { + finishCurrentBubble(job, "cancelled") + job.Status = domain.StatusCancelled + job.Phase = "cancelled" + job.Error = "cancelled" + p := job.EnsureProgress() + p["message"] = "cancelled" + return w.Repo.UpdateProgress(job) + } + if job.Status == domain.StatusAwaitingInput { + return nil + } + agents := job.Agents + if agents == nil { + agents = []map[string]interface{}{} + } + agentCount := len(agents) + if agentCount < 1 { + agentCount = 1 + } + dialogues := job.EnsureDialogues() + progress := job.EnsureProgress() + + if job.Status == domain.StatusQueued { + job.Status = domain.StatusRunning + job.Phase = "analyzing" + job.Cycle = 1 + progress["percent"] = 5 + progress["message"] = "席位开始发言" + progress["lineIndex"] = 0 + progress["charIndex"] = 0 + applyAgentStatus(agents, 0, "speaking") + job.ActiveAgentID = agentIDAt(agents, 0) + job.Agents = agents + openDemoBubble(job, dialogues, agents, 0, "席位输出中…") + // openDemoBubble 已写回 job.Dialogues；切勿再用旧 dialogues 切片盖掉新气泡 + return w.Repo.UpdateProgress(job) + } + + lineIndex := toIntVal(progress["lineIndex"], 0) + charIndex := toIntVal(progress["charIndex"], 0) + + if lineIndex < len(demoLines) { + speakerIdx := lineIndex % agentCount + wantAgent := agentIDAt(agents, speakerIdx) + if job.ActiveAgentID != wantAgent || (charIndex == 0 && !hasStreaming(dialogues)) { + if charIndex == 0 { + finishCurrentBubble(job, "") + for i := range agents { + st := "listening" + if i == speakerIdx { + st = "speaking" + } + applyAgentStatus(agents, i, st) + } + job.ActiveAgentID = wantAgent + openDemoBubble(job, dialogues, agents, speakerIdx, "席位输出中…") + dialogues = job.Dialogues + } + } + line := demoLines[lineIndex] + end := charIndex + charsPerTick + if end > len([]rune(line)) { + end = len([]rune(line)) + } + rs := []rune(line) + if end > charIndex { + appendStreaming(dialogues, string(rs[charIndex:end])) + charIndex = end + } + if charIndex >= len(rs) { + appendStreaming(dialogues, "\n\n") + finishCurrentBubble(job, "") + lineIndex++ + charIndex = 0 + } + totalChars := 0 + for _, l := range demoLines { + totalChars += len([]rune(l)) + } + doneChars := 0 + for i := 0; i < lineIndex && i < len(demoLines); i++ { + doneChars += len([]rune(demoLines[i])) + } + doneChars += charIndex + percent := 5 + (doneChars*85)/max(totalChars, 1) + if percent > 90 { + percent = 90 + } + progress["lineIndex"] = lineIndex + progress["charIndex"] = charIndex + progress["percent"] = percent + progress["message"] = "发言中 " + itoa(percent) + "%" + job.Cycle = 1 + lineIndex/max(agentCount, 1) + job.ConsensusPercentage = min(95, percent) + job.Agents = agents + job.Dialogues = dialogues + return w.Repo.UpdateProgress(job) + } + + finishCurrentBubble(job, "") + summary := "【演示总结】以上均为占位发言，用于验证「关浏览器后任务仍继续 + 页面流式展示」。" + + "接入真实模型后，这里会换成总控根据各席发言生成的真正总结。" + + "（job=" + job.ID + "）" + dialogues = append(dialogues, map[string]interface{}{ + "id": "dlg_" + job.ID + "_leader", "role": "leader", "sender": "总控协调", + "agentId": "coordinator", "content": summary, "status": "done", "phaseLabel": "全案盖印", + }) + for i := range agents { + applyAgentStatus(agents, i, "done") + } + job.ActiveAgentID = "" + job.Agents = agents + job.Status = domain.StatusDone + job.Phase = "done" + job.ConsensusPercentage = 100 + progress["percent"] = 100 + progress["message"] = "completed" + job.Dialogues = dialogues + job.Result = map[string]interface{}{ + "summary": summary, + "consensusPercentage": 100, + "recommendations": []string{"确认产品边界与里程碑", "补充关键风险与资源假设", "安排下一轮执行评审"}, + } + return w.Repo.UpdateProgress(job) +} + +func openDemoBubble(job *domain.AnalysisJob, dialogues, agents []map[string]interface{}, idx int, phase string) { + bubble := map[string]interface{}{ + "id": "dlg_" + job.ID + "_" + itoa(len(dialogues)+1), "role": "special", + "sender": agentNameAt(agents, idx), "agentId": agentIDAt(agents, idx), + "content": "", "status": "streaming", "phaseLabel": phase, + } + job.Dialogues = append(dialogues, bubble) +} + +func finishCurrentBubble(job *domain.AnalysisJob, override string) { + for i := len(job.Dialogues) - 1; i >= 0; i-- { + b := job.Dialogues[i] + if asStr(b["status"]) == "streaming" { + st := "done" + if override != "" { + st = override + } + b["status"] = st + b["phaseLabel"] = "" + return + } + } +} + +func appendStreaming(dialogues []map[string]interface{}, chunk string) { + for i := len(dialogues) - 1; i >= 0; i-- { + b := dialogues[i] + if asStr(b["status"]) == "streaming" { + b["content"] = asStr(b["content"]) + chunk + return + } + } +} + +func hasStreaming(dialogues []map[string]interface{}) bool { + for i := len(dialogues) - 1; i >= 0; i-- { + if asStr(dialogues[i]["status"]) == "streaming" { + return true + } + } + return false +} + +func agentIDAt(agents []map[string]interface{}, idx int) string { + if len(agents) == 0 { + return "" + } + return asStr(agents[idx%len(agents)]["id"]) +} + +func agentNameAt(agents []map[string]interface{}, idx int) string { + if len(agents) == 0 { + return "席位" + } + a := agents[idx%len(agents)] + if n := asStr(a["name"]); n != "" { + return n + } + return asStr(a["id"]) +} + +func applyAgentStatus(agents []map[string]interface{}, idx int, status string) { + if idx < 0 || idx >= len(agents) { + return + } + agents[idx]["status"] = status +} + +func toIntVal(v interface{}, def int) int { + switch t := v.(type) { + case int: + return t + case int64: + return int(t) + case float64: + return int(t) + default: + return def + } +} + +func itoa(n int) string { return strconv.Itoa(n) } + +func max(a, b int) int { + if a > b { + return a + } + return b +} + +func min(a, b int) int { + if a < b { + return a + } + return b +} diff --git a/internal/service/errors.go b/internal/service/errors.go new file mode 100644 index 0000000..1a7f55b --- /dev/null +++ b/internal/service/errors.go @@ -0,0 +1,23 @@ +package service + +import "errors" + +// 业务错误：HTTP 层按类型映射 403/400。 +var ( + ErrForbidden = errors.New("forbidden") + ErrBadState = errors.New("bad state") + ErrBadArg = errors.New("bad argument") +) + +// BizError 带给人看的文案。 +type BizError struct { + Kind error + Msg string +} + +func (e *BizError) Error() string { return e.Msg } +func (e *BizError) Unwrap() error { return e.Kind } + +func Forbidden(msg string) error { return &BizError{Kind: ErrForbidden, Msg: msg} } +func BadState(msg string) error { return &BizError{Kind: ErrBadState, Msg: msg} } +func BadArg(msg string) error { return &BizError{Kind: ErrBadArg, Msg: msg} } diff --git a/internal/service/job_service.go b/internal/service/job_service.go new file mode 100644 index 0000000..bec430a --- /dev/null +++ b/internal/service/job_service.go @@ -0,0 +1,352 @@ +package service + +import ( + "crypto/rand" + "encoding/hex" + "strconv" + "strings" + + "git.gozeros.com/wsk/eightbf-analysis-go/internal/domain" + "git.gozeros.com/wsk/eightbf-analysis-go/internal/repo" +) + +// CreateJobRequest 创建会商作业入参。 +type CreateJobRequest struct { + RequestID string `json:"requestId"` + TaskID string `json:"taskId"` + DraftID string `json:"draftId"` + DraftKey string `json:"draftKey"` + Model string `json:"model"` + OrchestrationMode string `json:"orchestrationMode"` + Title string `json:"title"` + Description string `json:"description"` + OwnerAccountID string `json:"ownerAccountId"` + ManualIntervention *bool `json:"manualIntervention"` + Agents []map[string]interface{} `json:"agents"` + Payload map[string]interface{} `json:"payload"` + Access map[string]interface{} `json:"access"` +} + +type ResumeJobRequest struct { + Answer string `json:"answer"` + Reason string `json:"reason"` +} + +type PatchJobPlanRequest struct { + ManualIntervention *bool `json:"manualIntervention"` + Model *string `json:"model"` +} + +type InjectJobRequest struct { + Text string `json:"text"` +} + +// JobService 会商作业编排（不直接调大模型）。 +type JobService struct { + Repo *repo.JobRepo + Access *AccessService +} + +func (s *JobService) Create(req *CreateJobRequest, caller Caller) (*domain.AnalysisJob, error) { + if req == nil { + req = &CreateJobRequest{} + } + manual := true + if req.ManualIntervention != nil { + manual = *req.ManualIntervention + } + owner := blankToNull(req.OwnerAccountID) + if owner == "" && caller.AccountID != "" { + owner = caller.AccountID + } + mode := blankToNull(req.OrchestrationMode) + if mode == "" { + mode = "roundtable" + } + job := &domain.AnalysisJob{ + ID: "job_" + randHex(8), + RequestID: firstNonEmpty(blankToNull(req.RequestID), "req_"+randHex(6)), + TaskID: blankToNull(req.TaskID), + DraftID: blankToNull(req.DraftID), + DraftKey: blankToNull(req.DraftKey), + Status: domain.StatusQueued, + Phase: "queued", + OrchestrationMode: mode, + Model: strings.TrimSpace(req.Model), + OwnerAccountID: owner, + Agents: normalizeAgents(req.Agents), + Dialogues: []map[string]interface{}{}, + Access: orEmptyMap(req.Access), + Plan: map[string]interface{}{ + "seatIndex": 0, + "manualIntervention": manual, + "seedMessage": nil, + "pendingInject": nil, + "awaitReason": nil, + }, + Progress: map[string]interface{}{ + "percent": 0, + "message": "queued", + "title": req.Title, + "description": req.Description, + }, + } + if req.Payload != nil { + job.Progress["payload"] = req.Payload + } + if err := s.Repo.Insert(job); err != nil { + return nil, err + } + return s.Access.ProjectForCaller(job, caller), nil +} + +func (s *JobService) Get(id string, caller Caller) (*domain.AnalysisJob, error) { + job, err := s.Repo.FindByID(id) + if err != nil || job == nil { + return job, err + } + if !s.Access.CanRead(job, caller) { + return nil, Forbidden("无权查看该会商作业") + } + return s.Access.ProjectForCaller(job, caller), nil +} + +func (s *JobService) GetByTaskID(taskID string, latestOnly bool, caller Caller) (*domain.AnalysisJob, error) { + var job *domain.AnalysisJob + var err error + if latestOnly { + job, err = s.Repo.FindLatestByTaskID(taskID) + } else { + job, err = s.Repo.FindActiveByTaskID(taskID) + if err == nil && job == nil { + job, err = s.Repo.FindLatestByTaskID(taskID) + } + } + if err != nil || job == nil { + return job, err + } + if !s.Access.CanRead(job, caller) { + return nil, Forbidden("无权查看该会商作业") + } + return s.Access.ProjectForCaller(job, caller), nil +} + +func (s *JobService) Cancel(id string, caller Caller) (*domain.AnalysisJob, error) { + job, err := s.Repo.FindByID(id) + if err != nil || job == nil { + return job, err + } + if !s.Access.CanCancel(job, caller) { + return nil, Forbidden("无权取消该会商作业（仅任务责任人/创建人/管理员）") + } + if domain.IsTerminal(job.Status) { + return s.Access.ProjectForCaller(job, caller), nil + } + _, _ = s.Repo.MarkCancel(id) + job.Status = domain.StatusCancelled + job.Phase = "cancelled" + job.Error = "cancelled by client" + p := job.EnsureProgress() + p["message"] = "cancelled" + if err := s.Repo.UpdateProgress(job); err != nil { + return nil, err + } + return s.Access.ProjectForCaller(job, caller), nil +} + +func (s *JobService) Resume(id string, body *ResumeJobRequest, caller Caller) (*domain.AnalysisJob, error) { + job, err := s.Repo.FindByID(id) + if err != nil || job == nil { + return job, err + } + if !s.Access.CanMutate(job, caller) { + return nil, Forbidden("无权续跑该会商作业") + } + if job.Status != domain.StatusAwaitingInput { + return nil, BadState("当前状态不是等待输入，无法 resume（status=" + job.Status + ")") + } + answer := "" + if body != nil { + answer = strings.TrimSpace(body.Answer) + } + plan := job.EnsurePlan() + if answer != "" { + appendUserDialogue(job, answer) + prev := asStr(plan["pendingInject"]) + merged := answer + if strings.TrimSpace(prev) != "" { + merged = prev + "\n\n" + answer + } + plan["pendingInject"] = "用户对澄清的回复：\n" + merged + } + plan["awaitReason"] = nil + job.PendingClarification = "" + job.Status = domain.StatusQueued + job.Phase = "resume_queued" + p := job.EnsureProgress() + p["message"] = "已收到补充，准备继续" + if err := s.Repo.UpdateProgress(job); err != nil { + return nil, err + } + fresh, err := s.Repo.FindByID(id) + if err != nil { + return nil, err + } + return s.Access.ProjectForCaller(fresh, caller), nil +} + +func (s *JobService) PatchPlan(id string, body *PatchJobPlanRequest, caller Caller) (*domain.AnalysisJob, error) { + job, err := s.Repo.FindByID(id) + if err != nil || job == nil { + return job, err + } + if !s.Access.CanMutate(job, caller) { + return nil, Forbidden("无权修改该会商作业设置") + } + if domain.IsTerminal(job.Status) { + return nil, BadState("作业已结束，无法修改设置") + } + if body == nil { + return s.Access.ProjectForCaller(job, caller), nil + } + touched := false + plan := job.EnsurePlan() + progress := job.EnsureProgress() + if body.Model != nil { + model := strings.TrimSpace(*body.Model) + job.Model = model + if model == "" { + progress["message"] = "已清除模型指定（用默认）" + } else { + progress["message"] = "已切换模型为 " + model + } + touched = true + } + if body.ManualIntervention != nil { + manual := *body.ManualIntervention + plan["manualIntervention"] = manual + if manual { + progress["message"] = "已开启人工干预（下一席结束后将等待确认）" + } else { + progress["message"] = "已关闭人工干预（将自动连跑）" + } + touched = true + atGate := job.Status == domain.StatusAwaitingInput && asStr(plan["awaitReason"]) == "manual_continue" + if !manual && atGate { + plan["awaitReason"] = nil + job.PendingClarification = "" + job.Status = domain.StatusQueued + job.Phase = "resume_queued" + progress["message"] = "已关闭人工干预，自动继续下一席" + } + } + if !touched { + return s.Access.ProjectForCaller(job, caller), nil + } + if err := s.Repo.UpdateProgress(job); err != nil { + return nil, err + } + fresh, err := s.Repo.FindByID(id) + if err != nil { + return nil, err + } + return s.Access.ProjectForCaller(fresh, caller), nil +} + +func (s *JobService) Inject(id string, body *InjectJobRequest, caller Caller) (*domain.AnalysisJob, error) { + job, err := s.Repo.FindByID(id) + if err != nil || job == nil { + return job, err + } + if !s.Access.CanMutate(job, caller) { + return nil, Forbidden("无权向该会商注入消息") + } + text := "" + if body != nil { + text = strings.TrimSpace(body.Text) + } + if text == "" { + return nil, BadArg("注入内容不能为空") + } + if domain.IsTerminal(job.Status) { + return nil, BadState("作业已结束，无法注入") + } + appendUserDialogue(job, text) + plan := job.EnsurePlan() + prev := asStr(plan["pendingInject"]) + merged := text + if strings.TrimSpace(prev) != "" { + merged = prev + "\n\n" + text + } + plan["pendingInject"] = "用户向会商插入的补充说明：\n" + merged + p := job.EnsureProgress() + p["message"] = "已收到用户补充，将在下一轮派活时带入" + if err := s.Repo.UpdateProgress(job); err != nil { + return nil, err + } + fresh, err := s.Repo.FindByID(id) + if err != nil { + return nil, err + } + return s.Access.ProjectForCaller(fresh, caller), nil +} + +func appendUserDialogue(job *domain.AnalysisJob, text string) { + list := job.EnsureDialogues() + bubble := map[string]interface{}{ + "id": "dlg_" + job.ID + "_user_" + strconv.Itoa(len(list)+1), + "role": "user", + "sender": "我", + "content": text, + "status": "done", + "phaseLabel": "", + } + job.Dialogues = append(list, bubble) +} + +func normalizeAgents(agents []map[string]interface{}) []map[string]interface{} { + if len(agents) > 0 { + out := make([]map[string]interface{}, 0, len(agents)) + for _, a := range agents { + copy := cloneMap(a) + if _, ok := copy["id"]; !ok { + if aid, ok := copy["agentId"]; ok { + copy["id"] = aid + } + } + if _, ok := copy["status"]; !ok { + copy["status"] = "idle" + } + out = append(out, copy) + } + return out + } + return []map[string]interface{}{ + {"id": "local-flow-stub-intel", "name": "联调桩-情报采集", "status": "idle"}, + {"id": "local-flow-stub-plan", "name": "联调桩-筹划规划", "status": "idle"}, + } +} + +func blankToNull(s string) string { + return strings.TrimSpace(s) +} + +func firstNonEmpty(a, b string) string { + if a != "" { + return a + } + return b +} + +func orEmptyMap(m map[string]interface{}) map[string]interface{} { + if m == nil { + return map[string]interface{}{} + } + return m +} + +func randHex(nBytes int) string { + b := make([]byte, nBytes) + _, _ = rand.Read(b) + return hex.EncodeToString(b) +} diff --git a/internal/worker/live_runner.go b/internal/worker/live_runner.go new file mode 100644 index 0000000..43c6ef8 --- /dev/null +++ b/internal/worker/live_runner.go @@ -0,0 +1,623 @@ +package worker + +import ( + "fmt" + "log" + "strings" + "sync" + "time" + "unicode/utf8" + + "git.gozeros.com/wsk/eightbf-analysis-go/internal/deerflow" + "git.gozeros.com/wsk/eightbf-analysis-go/internal/domain" + "git.gozeros.com/wsk/eightbf-analysis-go/internal/repo" +) + +// LiveChainRunner 真模型按链条顺序执行（被 LiveWorker 异步调用）。 +type LiveChainRunner struct { + DF *deerflow.Client + Repo *repo.JobRepo + Coord string + + mu sync.Mutex + lastFlush time.Time +} + +func (r *LiveChainRunner) Run(job *domain.AnalysisJob) { + jobID := job.ID + defer func() { + if rec := recover(); rec != nil { + log.Printf("真模型会商 panic jobId=%s: %v", jobID, rec) + r.fail(job, fmt.Sprint(rec)) + } + }() + + agents := job.Agents + if len(agents) < 2 { + r.fail(job, "有效席位不足 2 个，无法启动真模型会商") + return + } + var agentNames []string + for _, a := range agents { + id := asStr(a["id"]) + if id != "" { + agentNames = append(agentNames, id) + } + } + if len(agentNames) < 2 { + r.fail(job, "席位 agentId 无效（需要 DeerFlow 的 hyphen-case id）") + return + } + + plan := ensurePlan(job) + r.pullModelFromDB(job) + if err := r.DF.AssertModelAvailable(job.Model); err != nil { + r.fail(job, err.Error()) + return + } + job.Status = domain.StatusRunning + job.Phase = "running" + r.flush(job) + + if r.cancelled(jobID) { + r.markCancelled(job) + return + } + + threadIDs := loadThreadIDs(job) + if len(threadIDs) == 0 { + putProgress(job, 2, "正在初始化 DeerFlow 席位线程…") + r.flush(job) + r.pullModelFromDB(job) + tids, err := r.DF.InitMultiAgent(agentNames, r.Coord, job.Model) + if err != nil { + r.fail(job, err.Error()) + return + } + threadIDs = tids + store := map[string]interface{}{} + for k, v := range tids { + store[k] = v + } + job.ThreadIDs = store + putProgress(job, 8, "席位线程已就绪，开始链条研讨") + r.flush(job) + } else { + pct := toInt(job.Progress["percent"], 8) + if pct < 8 { + pct = 8 + } + putProgress(job, pct, "从检查点继续会商") + r.flush(job) + } + + seed := asStr(plan["seedMessage"]) + if seed == "" { + seed = BuildSeed(job.Progress) + plan["seedMessage"] = seed + } + if inj := consumePendingInject(plan); inj != "" { + seed = seed + "\n\n" + inj + plan["seedMessage"] = seed + } + + total := len(agentNames) + start := toInt(plan["seatIndex"], 0) + if start < 0 { + start = 0 + } + seatDeliveries := loadDeliveries(job) + lastLeader := "" + + for i := start; i < total; i++ { + if r.cancelled(jobID) { + r.markCancelled(job) + return + } + r.pullManualFromDB(job, plan) + r.pullInjectFromDB(job, plan) + r.pullModelFromDB(job) + model := job.Model + if mid := consumePendingInject(plan); mid != "" { + seed = seed + "\n\n" + mid + plan["seedMessage"] = seed + } + plan["seatIndex"] = i + job.Plan = plan + + seat := agents[i] + seatID := asStr(seat["id"]) + seatName := firstNonEmpty(asStr(seat["name"]), seatID) + job.ActiveAgentID = r.Coord + job.Cycle = i + 1 + applyAgentSpeaking(agents, seatID, false) + job.Agents = agents + + leaderBubbleID := openBubble(job, "leader", "总控协调", r.Coord, "链条派活中") + r.flush(job) + leaderRes, err := r.DF.RunStream("leader", threadIDs, r.Coord, "总控协调", + BuildLeaderTask(seed, i, total, seatID, seatName, i == 0), + model, false, + r.deltaHandler(job, leaderBubbleID), + func() bool { return r.cancelled(jobID) }, + ) + if err != nil { + finishBubble(job, leaderBubbleID, err.Error(), "error") + r.fail(job, buildModelHint("总控派活失败", model, err.Error())) + return + } + if r.cancelled(jobID) || leaderRes.Aborted { + finishBubble(job, leaderBubbleID, leaderRes.Content, "stopped") + r.markCancelled(job) + return + } + if leaderRes.Clarification { + finishBubble(job, leaderBubbleID, leaderRes.Content, "clarification") + detail := leaderRes.ClarificationDetail + if detail == "" { + detail = leaderRes.Content + } + r.pauseForInput(job, plan, i, "clarification", detail) + return + } + leaderBad := leaderRes.InvalidDelivery || strings.TrimSpace(leaderRes.Content) == "" || looksLikeModelFailure(leaderRes.Content) + if leaderBad { + finishBubble(job, leaderBubbleID, leaderRes.Content, "error") + r.fail(job, buildModelHint("总控派活失败", model, leaderRes.Content)) + return + } + finishBubble(job, leaderBubbleID, leaderRes.Content, "done") + lastLeader = leaderRes.Content + r.flush(job) + + task := PickChainTask(leaderRes.Dispatched, leaderRes.Content, seatID, seatName) + job.ActiveAgentID = seatID + applyAgentSpeaking(agents, seatID, true) + job.Agents = agents + seatBubbleID := openBubble(job, "special", seatName, seatID, "席位输出中…") + r.flush(job) + seatRes, err := r.DF.RunStream("special", threadIDs, seatID, seatName, task, model, false, + r.deltaHandler(job, seatBubbleID), + func() bool { return r.cancelled(jobID) }, + ) + if err != nil { + finishBubble(job, seatBubbleID, err.Error(), "error") + r.fail(job, buildModelHint("席位「"+seatName+"」输出失败", model, err.Error())) + return + } + if r.cancelled(jobID) || seatRes.Aborted { + finishBubble(job, seatBubbleID, seatRes.Content, "stopped") + r.markCancelled(job) + return + } + bad := seatRes.InvalidDelivery || strings.TrimSpace(seatRes.Content) == "" || looksLikeModelFailure(seatRes.Content) + status := "done" + if bad { + status = "error" + } + finishBubble(job, seatBubbleID, seatRes.Content, status) + if bad { + r.fail(job, buildModelHint("席位「"+seatName+"」输出失败", model, seatRes.Content)) + return + } + seatDeliveries = append(seatDeliveries, map[string]interface{}{ + "sender": seatName, + "agentId": seatID, + "content": seatRes.Content, + }) + saveDeliveries(job, seatDeliveries) + pct := (i + 1) * 85 / total + if pct > 90 { + pct = 90 + } + job.ConsensusPercentage = pct + putProgress(job, pct, fmt.Sprintf("已完成 %d/%d 席", i+1, total)) + plan["seatIndex"] = i + 1 + job.Plan = plan + r.flush(job) + + r.pullManualFromDB(job, plan) + if readManual(plan) && i+1 < total { + r.pauseForInput(job, plan, i+1, "manual_continue", + fmt.Sprintf("人工干预：第 %d/%d 席已完成，请确认后继续下一席。", i+1, total)) + return + } + } + + if r.cancelled(jobID) { + r.markCancelled(job) + return + } + job.ActiveAgentID = r.Coord + r.pullModelFromDB(job) + closeModel := job.Model + closeID := openBubble(job, "leader", "总控协调", r.Coord, "全案盖印") + r.flush(job) + closeRes, err := r.DF.RunStream("leader", threadIDs, r.Coord, "总控协调", Closeout, closeModel, true, + r.deltaHandler(job, closeID), + func() bool { return r.cancelled(jobID) }, + ) + if err != nil { + finishBubble(job, closeID, err.Error(), "error") + r.fail(job, buildModelHint("全案收口失败", closeModel, err.Error())) + return + } + if r.cancelled(jobID) || closeRes.Aborted { + finishBubble(job, closeID, closeRes.Content, "stopped") + r.markCancelled(job) + return + } + if closeRes.InvalidDelivery || looksLikeModelFailure(closeRes.Content) || + (strings.TrimSpace(closeRes.Content) == "" && strings.TrimSpace(lastLeader) == "") { + finishBubble(job, closeID, closeRes.Content, "error") + r.fail(job, buildModelHint("全案收口失败", closeModel, closeRes.Content)) + return + } + if len(seatDeliveries) < total { + finishBubble(job, closeID, closeRes.Content, "error") + r.fail(job, fmt.Sprintf("会商未完整交付（成功 %d/%d 席），请检查模型后重新开始。", len(seatDeliveries), total)) + return + } + if dialoguesHaveError(job) { + finishBubble(job, closeID, closeRes.Content, "error") + r.fail(job, "会商过程中已有失败发言，不能进入收口/下一阶段。请换可用模型后重新开始。") + return + } + summary := closeRes.Content + if strings.TrimSpace(summary) == "" { + summary = lastLeader + } + finishBubble(job, closeID, summary, "done") + for _, a := range agents { + a["status"] = "done" + } + job.Agents = agents + job.ActiveAgentID = "" + job.Status = domain.StatusDone + job.Phase = "done" + job.ConsensusPercentage = 100 + putProgress(job, 100, "completed") + job.Result = map[string]interface{}{ + "summary": summary, + "consensusPercentage": 100, + "seatDeliveries": seatDeliveries, + } + plan["awaitReason"] = nil + job.Plan = plan + r.flush(job) + log.Printf("真模型会商完成 jobId=%s", jobID) +} + +func (r *LiveChainRunner) pauseForInput(job *domain.AnalysisJob, plan map[string]interface{}, next int, reason, detail string) { + plan["seatIndex"] = next + plan["awaitReason"] = reason + job.Plan = plan + job.Status = domain.StatusAwaitingInput + job.Phase = reason + job.PendingClarification = detail + msg := "等待用户澄清" + if reason == "manual_continue" { + msg = "等待人工确认继续" + } + putProgress(job, job.ConsensusPercentage, msg) + r.flush(job) + log.Printf("作业暂停等待输入 jobId=%s reason=%s", job.ID, reason) +} + +func (r *LiveChainRunner) deltaHandler(job *domain.AnalysisJob, bubbleID string) deerflow.TextDeltaHandler { + return func(delta string) error { + appendBubble(job, bubbleID, delta) + r.throttleFlush(job) + return nil + } +} + +func (r *LiveChainRunner) throttleFlush(job *domain.AnalysisJob) { + r.mu.Lock() + defer r.mu.Unlock() + if time.Since(r.lastFlush) >= 150*time.Millisecond { + r.flushLocked(job) + } +} + +func (r *LiveChainRunner) flush(job *domain.AnalysisJob) { + r.mu.Lock() + defer r.mu.Unlock() + r.flushLocked(job) +} + +func (r *LiveChainRunner) flushLocked(job *domain.AnalysisJob) { + plan := ensurePlan(job) + r.pullManualFromDB(job, plan) + r.pullModelFromDB(job) + injectConsumed := plan["_injectConsumed"] == true + if !injectConsumed { + if strings.TrimSpace(asStr(plan["pendingInject"])) == "" { + r.pullInjectFromDB(job, plan) + } + } + delete(plan, "_injectConsumed") + job.Plan = plan + _ = r.Repo.UpdateProgress(job) + r.lastFlush = time.Now() +} + +func (r *LiveChainRunner) cancelled(jobID string) bool { + ok, err := r.Repo.IsCancelFlag(jobID) + return err == nil && ok +} + +func (r *LiveChainRunner) markCancelled(job *domain.AnalysisJob) { + job.Status = domain.StatusCancelled + job.Phase = "cancelled" + job.Error = "cancelled" + putProgress(job, job.ConsensusPercentage, "cancelled") + r.flush(job) +} + +func (r *LiveChainRunner) fail(job *domain.AnalysisJob, message string) { + job.Status = domain.StatusError + job.Phase = "error" + job.Error = message + job.ConsensusPercentage = 0 + if job.Result != nil { + delete(job.Result, "summary") + job.Result["consensusPercentage"] = 0 + job.Result["failed"] = true + } + putProgress(job, 0, "error") + r.flush(job) +} + +func (r *LiveChainRunner) pullModelFromDB(job *domain.AnalysisJob) { + fresh, err := r.Repo.FindByID(job.ID) + if err != nil || fresh == nil { + return + } + job.Model = fresh.Model +} + +func (r *LiveChainRunner) pullManualFromDB(job *domain.AnalysisJob, plan map[string]interface{}) { + fresh, err := r.Repo.FindByID(job.ID) + if err != nil || fresh == nil || fresh.Plan == nil { + return + } + if v, ok := fresh.Plan["manualIntervention"]; ok { + plan["manualIntervention"] = v + } +} + +func (r *LiveChainRunner) pullInjectFromDB(job *domain.AnalysisJob, plan map[string]interface{}) { + fresh, err := r.Repo.FindByID(job.ID) + if err != nil || fresh == nil || fresh.Plan == nil { + return + } + dbText := strings.TrimSpace(asStr(fresh.Plan["pendingInject"])) + if dbText == "" { + return + } + local := strings.TrimSpace(asStr(plan["pendingInject"])) + if local == "" { + plan["pendingInject"] = dbText + } else if !strings.Contains(local, dbText) { + plan["pendingInject"] = local + "\n\n" + dbText + } +} + +func consumePendingInject(plan map[string]interface{}) string { + s := strings.TrimSpace(asStr(plan["pendingInject"])) + plan["pendingInject"] = nil + plan["_injectConsumed"] = true + return s +} + +func ensurePlan(job *domain.AnalysisJob) map[string]interface{} { + p := job.EnsurePlan() + if _, ok := p["seatIndex"]; !ok { + p["seatIndex"] = 0 + } + if _, ok := p["manualIntervention"]; !ok { + p["manualIntervention"] = true + } + return p +} + +func putProgress(job *domain.AnalysisJob, percent int, message string) { + p := job.EnsureProgress() + p["percent"] = percent + p["message"] = message +} + +func loadThreadIDs(job *domain.AnalysisJob) map[string]string { + out := map[string]string{} + for k, v := range job.ThreadIDs { + if v != nil { + out[k] = fmt.Sprint(v) + } + } + return out +} + +func loadDeliveries(job *domain.AnalysisJob) []map[string]interface{} { + if job.Result == nil { + return nil + } + raw, ok := job.Result["seatDeliveries"] + if !ok { + return nil + } + switch t := raw.(type) { + case []map[string]interface{}: + return t + case []interface{}: + var out []map[string]interface{} + for _, o := range t { + if m, ok := o.(map[string]interface{}); ok { + out = append(out, m) + } + } + return out + default: + return nil + } +} + +func saveDeliveries(job *domain.AnalysisJob, list []map[string]interface{}) { + if job.Result == nil { + job.Result = map[string]interface{}{} + } + job.Result["seatDeliveries"] = list +} + +func readManual(plan map[string]interface{}) bool { + if plan == nil { + return true + } + v, ok := plan["manualIntervention"] + if !ok || v == nil { + return true + } + switch t := v.(type) { + case bool: + return t + default: + return !strings.EqualFold(strings.TrimSpace(fmt.Sprint(t)), "false") + } +} + +func openBubble(job *domain.AnalysisJob, role, sender, agentID, phase string) string { + list := job.EnsureDialogues() + id := fmt.Sprintf("dlg_%s_%d", job.ID, len(list)+1) + job.Dialogues = append(list, map[string]interface{}{ + "id": id, "role": role, "sender": sender, "agentId": agentID, + "content": "", "status": "streaming", "phaseLabel": phase, + }) + return id +} + +func appendBubble(job *domain.AnalysisJob, bubbleID, delta string) { + if delta == "" { + return + } + for _, b := range job.Dialogues { + if asStr(b["id"]) == bubbleID { + b["content"] = asStr(b["content"]) + delta + return + } + } +} + +func finishBubble(job *domain.AnalysisJob, bubbleID, finalContent, status string) { + for _, b := range job.Dialogues { + if asStr(b["id"]) != bubbleID { + continue + } + if strings.TrimSpace(finalContent) != "" { + prev := asStr(b["content"]) + b["content"] = mergeKeepThinking(prev, finalContent) + } + if status == "" { + status = "done" + } + b["status"] = status + b["phaseLabel"] = "" + return + } +} + +func mergeKeepThinking(streamed, finalContent string) string { + if strings.Contains(finalContent, "") || strings.Contains(finalContent, "") { + return finalContent + } + open := strings.Index(streamed, "") + if open < 0 { + return finalContent + } + close := strings.LastIndex(streamed, "") + if close >= open { + return streamed[:close+len("")] + "\n" + finalContent + } + return streamed + "\n" + finalContent +} + +func applyAgentSpeaking(agents []map[string]interface{}, speakingID string, seatTurn bool) { + for _, a := range agents { + id := asStr(a["id"]) + if seatTurn && id == speakingID { + a["status"] = "speaking" + } else { + a["status"] = "listening" + } + } +} + +func dialoguesHaveError(job *domain.AnalysisJob) bool { + for _, b := range job.Dialogues { + if strings.EqualFold(asStr(b["status"]), "error") { + return true + } + } + return false +} + +func looksLikeModelFailure(content string) bool { + t := strings.TrimSpace(content) + if t == "" { + return false + } + lower := strings.ToLower(t) + needles := []string{"model_not_found", "invalid model", "does not exist", "no such model", + "litellm", "aperror", "apierror", "authenticationerror", "rate_limit", "context_length"} + for _, n := range needles { + if strings.Contains(lower, n) { + return true + } + } + cn := []string{"大模型本次调用未能成功完成", "请稍后再试或更换模型", "大模型因鉴权失败", + "大模型服务暂时繁忙", "大模型响应暂时中断", "大模型当前响应异常", "已被熔断保护", + "已拒绝自动切换", "模型不存在", "无可用模型", "模型不可用", "鉴权失败"} + for _, n := range cn { + if strings.Contains(t, n) { + return true + } + } + if utf8.RuneCountInString(t) < 400 && len(t) >= 5 && t[0] == '[' && t[4] == ']' { + return true + } + return false +} + +func buildModelHint(head, model, detail string) string { + sb := head + if strings.TrimSpace(model) != "" { + sb += "（当前模型：" + strings.TrimSpace(model) + "）" + } + sb += "。请换可用模型后重新开始会商。" + d := strings.TrimSpace(detail) + if d != "" { + if utf8.RuneCountInString(d) > 180 { + rs := []rune(d) + d = string(rs[:180]) + "…" + } + sb += " 详情：" + d + } + return sb +} + +func toInt(v interface{}, def int) int { + switch t := v.(type) { + case int: + return t + case int64: + return int(t) + case float64: + return int(t) + case string: + var n int + _, err := fmt.Sscanf(t, "%d", &n) + if err == nil { + return n + } + } + return def +} diff --git a/internal/worker/live_worker.go b/internal/worker/live_worker.go new file mode 100644 index 0000000..53cff27 --- /dev/null +++ b/internal/worker/live_worker.go @@ -0,0 +1,64 @@ +package worker + +import ( + "log" + "sync/atomic" + "time" + + "git.gozeros.com/wsk/eightbf-analysis-go/internal/domain" + "git.gozeros.com/wsk/eightbf-analysis-go/internal/repo" +) + +// LiveWorker 定时扫 queued，最多并发 2 场交给 LiveChainRunner。 +type LiveWorker struct { + Repo *repo.JobRepo + Runner *LiveChainRunner + PollEvery time.Duration + inflight int32 +} + +func (w *LiveWorker) Start() { + if w.PollEvery <= 0 { + w.PollEvery = time.Second + } + go func() { + t := time.NewTicker(w.PollEvery) + defer t.Stop() + for range t.C { + w.poll() + } + }() +} + +func (w *LiveWorker) poll() { + if atomic.LoadInt32(&w.inflight) >= 2 { + return + } + queued, err := w.Repo.FindQueuedJobs() + if err != nil { + log.Printf("LiveWorker 扫队列失败: %v", err) + return + } + for _, job := range queued { + if atomic.LoadInt32(&w.inflight) >= 2 { + break + } + job.Status = domain.StatusRunning + job.Phase = "claimed" + if err := w.Repo.UpdateProgress(job); err != nil { + log.Printf("LiveWorker 占坑失败 jobId=%s: %v", job.ID, err) + continue + } + atomic.AddInt32(&w.inflight, 1) + jobID := job.ID + go func() { + defer atomic.AddInt32(&w.inflight, -1) + fresh, err := w.Repo.FindByID(jobID) + if err != nil || fresh == nil { + return + } + log.Printf("开始真模型会商 jobId=%s", jobID) + w.Runner.Run(fresh) + }() + } +} diff --git a/internal/worker/prompts.go b/internal/worker/prompts.go new file mode 100644 index 0000000..cc001a8 --- /dev/null +++ b/internal/worker/prompts.go @@ -0,0 +1,119 @@ +package worker + +import ( + "fmt" + "strings" +) + +// 链条提示词（与 Java ChainPrompts / 前端历史行为对齐）。 + +const Lang = "【语言要求】你的全部可见输出（派活说明、任务描述、分析、结论）必须使用简体中文；" + + "禁止用英文句子作答。agent_name / 技能名等标识符可保持原文，正文一律中文。" + +const Closeout = "各链条席位已按顺序完成发言,请你综合所有席位的交付,输出最终方案与结论。不要再派活、不要再向任何席位分配任务。\n" + Lang + +func BuildSeed(progress map[string]interface{}) string { + if progress == nil { + return "请基于用户任务意图，调度子智能体完成研讨。" + } + var sb strings.Builder + title := asStr(progress["title"]) + desc := asStr(progress["description"]) + if payload, ok := progress["payload"].(map[string]interface{}); ok { + if intent, ok := payload["intent"].(map[string]interface{}); ok { + obj := firstNonEmpty(asStr(intent["objective"]), asStr(intent["summary"]), title) + if obj != "" { + sb.WriteString(obj) + } + if list, ok := intent["constraints"].([]interface{}); ok && len(list) > 0 { + sb.WriteString("\n\n约束条件:\n") + for _, c := range list { + sb.WriteString("- ") + sb.WriteString(fmt.Sprint(c)) + sb.WriteByte('\n') + } + } + if list, ok := intent["assumptions"].([]interface{}); ok && len(list) > 0 { + sb.WriteString("\n关键假设:\n") + for _, a := range list { + sb.WriteString("- ") + sb.WriteString(fmt.Sprint(a)) + sb.WriteByte('\n') + } + } + } + } + if sb.Len() == 0 { + sb.WriteString(title) + if desc != "" { + if sb.Len() > 0 { + sb.WriteByte('\n') + } + sb.WriteString(desc) + } + } + if sb.Len() == 0 { + return "请基于用户此前澄清的任务意图，调度子智能体完成研讨。" + } + sb.WriteString("\n\n用户已经过任务理解智能体澄清,以上信息可以直接作为研讨依据。请基于上述目标与约束,在圆桌内调度合适的子智能体完成研讨,不要再反问用户。") + return sb.String() +} + +func BuildLeaderTask(seed string, position, total int, seatAgentID, seatName string, isFirst bool) string { + ordinal := fmt.Sprintf("%d/%d", position+1, total) + if isFirst { + return "这是一次【按业务链条顺序】进行的多智能体圆桌研讨。任务背景如下：\n" + + seed + "\n\n" + + "本次研讨严格按预设链条顺序进行，共 " + fmt.Sprint(total) + " 个席位，现在是第 " + ordinal + " 个席位。\n" + + "请你作为总控，通过 agent_orchestration 技能**仅**向智能体「" + seatName + + "」（agent_name: " + seatAgentID + "）派活，结合任务背景为它分配聚焦其职责的本轮任务。\n" + + "不要派给其他任何智能体，也不要自己直接作答或给出最终结论。\n" + + "派活说明与交给该席位的任务正文必须是简体中文（可写成：本轮派活给「" + + seatName + "」……），不要输出英文派活句。\n" + Lang + } + return "链条研讨继续，现在轮到第 " + ordinal + " 个席位。\n" + + "请你作为总控，结合此前各席位已完成的交付，通过 agent_orchestration 技能**仅**向智能体「" + + seatName + "」（agent_name: " + seatAgentID + + "）派活，为它分配承接前序结论、聚焦其职责的本轮任务。\n" + + "不要派给其他任何智能体，也不要自己直接作答或给出最终结论。\n" + + "派活说明与任务正文必须是简体中文。\n" + Lang +} + +func PickChainTask(dispatched [][2]string, leaderContent, seatAgentID, seatName string) string { + wanted := strings.ToLower(seatAgentID) + for _, row := range dispatched { + if strings.ToLower(row[0]) == wanted && strings.TrimSpace(row[1]) != "" { + return withLang(row[1]) + } + } + for _, row := range dispatched { + if strings.TrimSpace(row[1]) != "" { + return withLang(row[1]) + } + } + if strings.TrimSpace(leaderContent) != "" { + return withLang("请你作为「" + seatName + "」，基于以下总控说明与此前各席位的交付，从你的专业职责出发给出你的分析与交付：\n" + strings.TrimSpace(leaderContent)) + } + return withLang("请你作为「" + seatName + "」，基于任务背景与此前各席位的交付，从你的专业职责出发给出你的分析与交付。") +} + +func withLang(body string) string { return Lang + "\n" + body } + +func asStr(v interface{}) string { + if v == nil { + return "" + } + if s, ok := v.(string); ok { + return strings.TrimSpace(s) + } + return strings.TrimSpace(fmt.Sprint(v)) +} + +func firstNonEmpty(parts ...string) string { + for _, p := range parts { + if strings.TrimSpace(p) != "" { + return strings.TrimSpace(p) + } + } + return "" +} diff --git a/internal/worker/prompts_test.go b/internal/worker/prompts_test.go new file mode 100644 index 0000000..de9f60e --- /dev/null +++ b/internal/worker/prompts_test.go @@ -0,0 +1,25 @@ +package worker + +import ( + "strings" + "testing" +) + +func TestBuildSeedAndPickTask(t *testing.T) { + seed := BuildSeed(map[string]interface{}{ + "title": "测试任务", + "payload": map[string]interface{}{ + "intent": map[string]interface{}{ + "objective": "完成联调", + "constraints": []interface{}{"必须中文"}, + }, + }, + }) + if !strings.Contains(seed, "完成联调") { + t.Fatalf("seed 异常: %s", seed) + } + task := PickChainTask([][2]string{{"agent-x", "请分析风险"}}, "总控全文", "agent-x", "风险岗") + if !strings.Contains(task, "请分析风险") || !strings.Contains(task, "简体中文") { + t.Fatalf("pick 任务异常: %s", task) + } +} diff --git a/schema.sql b/schema.sql new file mode 100644 index 0000000..b60be06 --- /dev/null +++ b/schema.sql @@ -0,0 +1,35 @@ +-- 会商作业主表：一条记录 = 一次后台会商 +-- 本机 / 正式默认走 MySQL（utf8mb4）；H2 profile 仍可跑同一份（LONGTEXT≈CLOB） +CREATE TABLE IF NOT EXISTS eba_analysis_job ( + id VARCHAR(64) PRIMARY KEY, + request_id VARCHAR(64), + task_id VARCHAR(128), + draft_id VARCHAR(64), + draft_key VARCHAR(255), + status VARCHAR(32) NOT NULL, + phase VARCHAR(32), + cycle INT DEFAULT 0, + active_agent_id VARCHAR(128), + consensus_percentage INT DEFAULT 0, + pending_clarification LONGTEXT, + orchestration_mode VARCHAR(32), + model VARCHAR(128), + error_message LONGTEXT, + -- 创建人账号；岗位限制与检查点见 access_json / plan_json + owner_account_id VARCHAR(128), + agents_json LONGTEXT, + chain_json LONGTEXT, -- dialogues 气泡列表 + plan_json LONGTEXT, -- 检查点：seatIndex、manualIntervention、pendingInject 等 + access_json LONGTEXT, -- 权限：taskOwnerAccountIds、seatAccounts、adminAccountIds + thread_ids_json LONGTEXT, + progress_json LONGTEXT, + result_json LONGTEXT, + cancel_flag TINYINT DEFAULT 0, + created_at TIMESTAMP NULL DEFAULT CURRENT_TIMESTAMP, + updated_at TIMESTAMP NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP +); + +-- MySQL 无「IF NOT EXISTS」索引时，重复执行靠 continue-on-error 忽略 +CREATE INDEX idx_eba_job_task ON eba_analysis_job(task_id); +CREATE INDEX idx_eba_job_status ON eba_analysis_job(status); +CREATE INDEX idx_eba_job_owner ON eba_analysis_job(owner_account_id); diff --git a/scripts/smoke-demo.sh b/scripts/smoke-demo.sh new file mode 100755 index 0000000..29456a3 --- /dev/null +++ b/scripts/smoke-demo.sh @@ -0,0 +1,21 @@ +#!/usr/bin/env bash +# 对本机已启动的 Go 服务做 demo 冒烟（默认 :18097）。 +# 注意：勿与 Java 实例同时抢同一库的 queued。 +set -euo pipefail +BASE="${BASE:-http://127.0.0.1:18097/api/v1/analysis}" +echo "== health ==" +curl -sS "$BASE/health" | tee /tmp/eba-smoke-health.json +echo +CREATE=$(curl -sS -X POST "$BASE/jobs" -H 'Content-Type: application/json' \ + -d '{"taskId":"smoke-'"$(date +%s)"'","title":"smoke","manualIntervention":true,"agents":[{"id":"a1","name":"甲"},{"id":"a2","name":"乙"}]}') +JOB=$(echo "$CREATE" | python3 -c 'import sys,json;print(json.load(sys.stdin)["data"]["id"])') +echo "job=$JOB" +for i in $(seq 1 40); do + J=$(curl -sS "$BASE/jobs/$JOB") + ST=$(echo "$J" | python3 -c 'import sys,json;print(json.load(sys.stdin)["data"]["status"])') + PCT=$(echo "$J" | python3 -c 'import sys,json;print(json.load(sys.stdin)["data"].get("progress",{}).get("percent",0))') + echo "tick=$i status=$ST percent=$PCT" + [ "$ST" = "done" ] && break + sleep 0.5 +done +echo OK