diff --git a/.DS_Store b/.DS_Store new file mode 100644 index 0000000..5008ddf Binary files /dev/null and b/.DS_Store differ diff --git a/sundynix-gateway/internal/handler/task_handler.go b/sundynix-gateway/internal/handler/task_handler.go index a1b431c..64b8653 100644 --- a/sundynix-gateway/internal/handler/task_handler.go +++ b/sundynix-gateway/internal/handler/task_handler.go @@ -36,45 +36,64 @@ func New(db *store.Postgres, cache *store.Redis, bus *nats.Bus, blob *blob.Store return &Handler{db: db, cache: cache, bus: bus, blob: blob, pay: payment.NewManager()} } -// preflight 是「会烧钱的执行」提交前的统一关卡:当日 token 预算 → 计费租户 → 积分硬拦截。 -// 返回计费租户;ok=false 表示已写过响应,调用方直接 return。 -// -// 抽出来是因为这套关卡曾经只长在 SubmitTask 上,报告生成(GenerateReport)是另一条路径、 -// 一直停在最初的「发个 NATS」——于是报告绕过了预算、不记计费租户、余额为 0 也照生成。 -// 两条路径共用同一个函数,才不会再各长各的。 -func (h *Handler) preflight(c *gin.Context) (string, bool) { +// preflightBlock 是关卡未通过时的拒绝信息(HTTP 状态 + 响应体)。core 返回它、由具体调用方 +// 决定怎么把它变成回应(HTTP 写 JSON / 语音会话取 error 文案播报)。 +type preflightBlock struct { + Status int + Body gin.H +} + +func (b *preflightBlock) message() string { + if b == nil { + return "" + } + if m, ok := b.Body["error"].(string); ok { + return m + } + return "提交被拦截" +} + +// preflightCore 是「会烧钱的执行」提交前统一关卡的**无 gin 内核**:当日 token 预算 → 暂停管控 +// → 计费租户解析 → 积分硬拦截。返回计费租户;block 非空表示被拦(HTTP 与语音两条入口共用它, +// 谁也别再各长各的关卡——见记忆 execution-single-entry)。 +func (h *Handler) preflightCore(ctx context.Context, uid, tid string) (billingTenant string, block *preflightBlock) { // 成本护栏:单用户当日 token 日预算门控(USER_DAILY_TOKEN_BUDGET,0=不限)。 if budget := userDailyTokenBudget(); budget > 0 { - uid := userID(c) - used := h.cache.GetUsage(c.Request.Context(), uid, time.Now().Format("20060102")) + used := h.cache.GetUsage(ctx, uid, time.Now().Format("20060102")) if used >= int64(budget) { - c.JSON(http.StatusPaymentRequired, gin.H{ + return "", &preflightBlock{http.StatusPaymentRequired, gin.H{ "error": "已达当日 token 预算上限", "used": used, "budget": budget, - }) - return "", false + }} } } // 暂停管控:活跃租户(工作区)被暂停 → 拒绝提交。否则「暂停」只是个装了没接线的开关。 - if tid := tenantID(c); h.db.TenantSuspended(c.Request.Context(), tid) { - c.JSON(http.StatusForbidden, gin.H{"error": "租户已被暂停,暂无法提交任务"}) - return "", false + if h.db.TenantSuspended(ctx, tid) { + return "", &preflightBlock{http.StatusForbidden, gin.H{"error": "租户已被暂停,暂无法提交任务"}} } // 计费目标:数据落在活跃租户(工作区),但消耗记到"计费租户"——owner/共享计费→活跃租户, // 否则→本人个人租户(各付各的)。硬拦截与用量都按计费租户走。 - billingTenant := h.db.ResolveBillingTenantID(c.Request.Context(), userID(c), tenantID(c)) + billingTenant = h.db.ResolveBillingTenantID(ctx, uid, tid) // 计费租户与活跃租户不同(共享计费分叉)时,计费租户被暂停也拦——别让暂停的组织被人借道烧积分。 - if billingTenant != "" && billingTenant != tenantID(c) && h.db.TenantSuspended(c.Request.Context(), billingTenant) { - c.JSON(http.StatusForbidden, gin.H{"error": "计费租户已被暂停,暂无法提交任务"}) - return "", false + if billingTenant != "" && billingTenant != tid && h.db.TenantSuspended(ctx, billingTenant) { + return "", &preflightBlock{http.StatusForbidden, gin.H{"error": "计费租户已被暂停,暂无法提交任务"}} } // 积分硬拦截(默认关;开关 credit_enforce):计费租户积分余额 ≤0 则拒绝,提示充值。 - if billingTenant != "" && h.db.CreditEnforceEnabled(c.Request.Context()) { - if h.db.TenantBalance(c.Request.Context(), billingTenant) <= 0 { - c.JSON(http.StatusPaymentRequired, gin.H{"error": "租户积分余额不足,请充值后再试", "balance_micro": 0}) - return "", false + if billingTenant != "" && h.db.CreditEnforceEnabled(ctx) { + if h.db.TenantBalance(ctx, billingTenant) <= 0 { + return "", &preflightBlock{http.StatusPaymentRequired, gin.H{"error": "租户积分余额不足,请充值后再试", "balance_micro": 0}} } } - return billingTenant, true + return billingTenant, nil +} + +// preflight 是 preflightCore 的 gin 薄封装:ok=false 表示已写过响应,调用方直接 return。 +func (h *Handler) preflight(c *gin.Context) (string, bool) { + bt, block := h.preflightCore(c.Request.Context(), userID(c), tenantID(c)) + if block != nil { + c.JSON(block.Status, block.Body) + return "", false + } + return bt, true } // launch 把一次执行真正发出去,并接上「执行」该有的全套基建: @@ -82,16 +101,23 @@ func (h *Handler) preflight(c *gin.Context) (string, bool) { // 报告生成此前只 PublishTask,这两样都没有,所以报告既进不了运行历史, // 切个页面回来也彻底找不回——它明明在后端好好地跑完了。 func (h *Handler) launch(c *gin.Context, task *contract.Task) error { + return h.launchCore(c.Request.Context(), userID(c), task) +} + +// launchCore 是 launch 的**无 gin 内核**:落库 + Publish + 起 token/轨迹录像。 +// 语音会话(无 gin.Context)也走它,与 HTTP 提交共用同一条发射流程。 +// 注意:ctx 只用于落库与 Publish(同步、瞬时完成),录像器自持后台 ctx,不受此 ctx 生命周期影响。 +func (h *Handler) launchCore(ctx context.Context, uid string, task *contract.Task) error { // 持久化任务提交。DB 降级(nil)时 SaveTask 返 nil 静默跳过(开发态本就无库,不阻断); // 但 DB 活着却写失败 → 真故障,绝不能吞:一旦 PublishTask 发出去,任务就在后端跑了, // 却不进运行历史、复盘不了、报告类的会彻底"丢"(用户切页面回来找不回)。 // 宁可这里失败上浮 5xx 让用户重试,也不发一个"看不见的执行"。落库在 Publish 之前, // 失败时还没发布,中止是干净的。 - if err := h.db.SaveTask(c.Request.Context(), userID(c), task.ID, string(task.Graph)); err != nil { + if err := h.db.SaveTask(ctx, uid, task.ID, string(task.Graph)); err != nil { log.Printf("[gateway] save task %s failed: %v", task.ID, err) return fmt.Errorf("任务落库失败,请重试: %w", err) } - if err := h.bus.PublishTask(c.Request.Context(), task); err != nil { + if err := h.bus.PublishTask(ctx, task); err != nil { return err } // 从提交即开始把 token 流 + 执行轨迹录进 Redis Stream(订阅早于 dispatcher 产出)→ diff --git a/sundynix-gateway/internal/handler/voice.go b/sundynix-gateway/internal/handler/voice.go index 86267a3..08691bd 100644 --- a/sundynix-gateway/internal/handler/voice.go +++ b/sundynix-gateway/internal/handler/voice.go @@ -5,6 +5,7 @@ import ( "encoding/json" "log" "net/http" + "strings" "sync" "time" @@ -47,21 +48,32 @@ func (h *Handler) VoiceStream(c *gin.Context) { } defer conn.Close() - sess := &voiceSession{conn: conn, uid: uid, cfg: cfg} + // 租户/会话在升级时(还握着 gin.Context)一并抓取,供 WS 读循环里提交任务复用共用关卡。 + sess := &voiceSession{ + conn: conn, uid: uid, cfg: cfg, h: h, + tenantID: tenantID(c), sessionID: sessionID(c), + } sess.send(voice.ServerMsg{Type: voice.ServerReady}) sess.run() sess.stopASR() // 连接结束,收掉在跑的识别会话 } // voiceSession 是一次语音会话的外壳:持 WS 连接,跑协议循环。 -// 上行 = 音频→ASR→转写(→下一步 SubmitTask);下行(token流→攒句→TTS→音频)将在 TTS 步接上。 +// 上行 = 音频→ASR→转写→提交任务;下行(token流→攒句→TTS→音频)将在 TTS 步接上。 type voiceSession struct { - conn *websocket.Conn - uid string - cfg voice.Config + conn *websocket.Conn + h *Handler // 复用 preflightCore/launchCore 提交任务 + uid string + tenantID string // 升级时抓取(读循环里无 gin.Context) + sessionID string + cfg voice.Config + writeMu sync.Mutex // gorilla WS 不允许并发写:读循环与 ASR 结果 goroutine 都会 send,须串行化 asr *voice.ASRSession asrCancel context.CancelFunc + + pendingGraph string // 客户端 start 时带的画布编排图(语音触发既有编排),空则按转写现组 + lastFinal string // 最近一次已提交的最终转写,去重连发的重复 final } // send 下发一条控制/事件消息(文本帧,JSON)。并发安全。 @@ -125,13 +137,14 @@ func (s *voiceSession) onControl(m voice.ClientMsg) (done bool) { case voice.ClientBye: return true case voice.ClientStart: + s.pendingGraph = m.Graph // 客户端画布图(可空):本轮若有转写则语音触发它跑 + s.lastFinal = "" s.stopASR() s.startASR() // 新一轮:重开识别 case voice.ClientEnd: if s.asr != nil { - _ = s.asr.Finish() // 告知火山本轮说完,等最终转写(结果流里带 Final=true) + _ = s.asr.Finish() // 告知火山本轮说完,等最终转写(结果流里带 Final=true→提交任务) } - // TODO(上行接线):拿到 Final 转写 → 组 DSL/用 m.Graph → SubmitTask → 回推 task。 case voice.ClientBargeIn: // 打断:停当前 TTS 播放(TTS 接入后处理)。 } @@ -156,10 +169,33 @@ func (s *voiceSession) startASR() { return // 识别流结束/出错 } s.send(voice.ServerMsg{Type: voice.ServerTranscript, Text: r.Text, Final: r.Final}) + if r.Final { + s.onFinalTranscript(r.Text) // 最终转写 → 提交任务 + } else { + s.lastFinal = "" // 新的部分结果=新一轮开口,放行下一次 final 提交 + } } }() } +// onFinalTranscript 拿到一段最终转写就提交一次任务。去重连发的重复 final(同一句 SAUC 可能回多条)。 +// 只在结果 goroutine 里调,lastFinal 无需加锁。 +func (s *voiceSession) onFinalTranscript(text string) { + txt := strings.TrimSpace(text) + if txt == "" || txt == s.lastFinal { + return + } + s.lastFinal = txt + taskID, err := s.submitVoiceTask(txt, s.pendingGraph) + if err != nil { + s.send(voice.ServerMsg{Type: voice.ServerError, Msg: "任务提交失败:" + err.Error()}) + return + } + s.pendingGraph = "" // 画布图一次性消费,避免后续转写重复触发同图 + s.send(voice.ServerMsg{Type: voice.ServerTask, TaskID: taskID}) + // 下行(订阅 token 流→攒句→TTS→音频帧)在 TTS 步接上;此处先只回 task_id 供客户端切运行视图。 +} + // stopASR 收掉当前识别会话(幂等)。 func (s *voiceSession) stopASR() { if s.asr != nil { diff --git a/sundynix-gateway/internal/handler/voice_task.go b/sundynix-gateway/internal/handler/voice_task.go new file mode 100644 index 0000000..6828ffb --- /dev/null +++ b/sundynix-gateway/internal/handler/voice_task.go @@ -0,0 +1,71 @@ +package handler + +import ( + "context" + "encoding/json" + "fmt" + "strings" + + "github.com/sundynix/sundynix-gateway/internal/dsl" + "github.com/sundynix/sundynix-shared/contract" +) + +// 语音上行接线:最终转写 → 组 DSL → 复用 preflightCore/launchCore 关卡 → 提交任务 → 回 task_id。 +// 语音只是"嘴替键盘",一行编排/工具/计费逻辑都不新造:走的正是 HTTP SubmitTask 那条关卡 +// (见记忆 execution-single-entry「提交必走 preflight()+launch() 共用关卡」)。 + +// voiceAgentSystem 是语音默认单 agent 的系统提示词:口语化、简洁、适合朗读(下行要过 TTS)。 +const voiceAgentSystem = "你是用户的语音助手 JARVIS。用简洁、口语化、适合朗读的中文回答," + + "避免冗长的列表和代码块;必要时可调用工具获取信息后再作答。" + +// buildVoiceGraph 把一句转写组成最简可执行图:input(转写) → agent(JARVIS)。 +// 与前端画布 exportDsl 同构(kind=input/agent、config.text/system),dispatcher 直接吃。 +func buildVoiceGraph(query string) json.RawMessage { + g := map[string]any{ + "version": "voice-1", + "nodes": []map[string]any{ + {"id": "voice_in", "kind": "input", "config": map[string]any{"text": query}}, + {"id": "voice_agent", "kind": "agent", "config": map[string]any{"system": voiceAgentSystem}}, + }, + "edges": []map[string]any{ + {"source": "voice_in", "target": "voice_agent"}, + }, + } + b, _ := json.Marshal(g) + return b +} + +// submitVoiceTask 提交一次语音任务。graphOverride 非空时用客户端画布图(语音触发既有编排), +// 否则用转写现组的单 agent 图。返回 task_id。 +func (s *voiceSession) submitVoiceTask(transcript, graphOverride string) (string, error) { + transcript = strings.TrimSpace(transcript) + if transcript == "" && graphOverride == "" { + return "", fmt.Errorf("空转写") + } + ctx := context.Background() // WS 会话长生命周期,不绑单条请求 ctx + + var raw json.RawMessage + if strings.TrimSpace(graphOverride) != "" { + raw = json.RawMessage(graphOverride) // 语音触发画布上的既有编排图 + } else { + raw = buildVoiceGraph(transcript) + } + task, err := dsl.ParseAndAssemble(raw) + if err != nil { + return "", err + } + + // 共用关卡:预算 / 暂停 / 计费租户 / 积分硬拦截。被拦时把文案上抛(供语音播报/回传)。 + billingTenant, block := s.h.preflightCore(ctx, s.uid, s.tenantID) + if block != nil { + return "", fmt.Errorf("%s", block.message()) + } + task.Meta[contract.MetaUserID] = s.uid + task.Meta[contract.MetaTenantID] = billingTenant + task.Meta[contract.MetaSessionID] = s.sessionID + + if err := s.h.launchCore(ctx, s.uid, task); err != nil { + return "", err + } + return task.ID, nil +} diff --git a/sundynix-gateway/internal/handler/voice_task_test.go b/sundynix-gateway/internal/handler/voice_task_test.go new file mode 100644 index 0000000..c873472 --- /dev/null +++ b/sundynix-gateway/internal/handler/voice_task_test.go @@ -0,0 +1,72 @@ +package handler + +import ( + "encoding/json" + "strings" + "testing" + + "github.com/sundynix/sundynix-gateway/internal/dsl" +) + +// buildVoiceGraph 的产物必须是 dsl.ParseAndAssemble 能吃下的合法图,且带上转写文本。 +func TestBuildVoiceGraph_Valid(t *testing.T) { + const q = "帮我查一下明天上海的天气" + raw := buildVoiceGraph(q) + + // 1) 能通过 DSL 解析与拓扑校验(与 HTTP SubmitTask 同一条解析)。 + task, err := dsl.ParseAndAssemble(raw) + if err != nil { + t.Fatalf("语音图未通过 DSL 校验: %v", err) + } + if task.ID == "" { + t.Fatal("task.ID 为空") + } + + // 2) 图里带着转写文本(input 节点)与 JARVIS 系统提示(agent 节点)。 + var g struct { + Nodes []struct { + ID string `json:"id"` + Kind string `json:"kind"` + Config map[string]any `json:"config"` + } `json:"nodes"` + Edges []struct { + Source, Target string + } `json:"edges"` + } + if err := json.Unmarshal(raw, &g); err != nil { + t.Fatalf("反解语音图失败: %v", err) + } + if len(g.Nodes) != 2 || len(g.Edges) != 1 { + t.Fatalf("期望 2 节点 1 边,得 %d 节点 %d 边", len(g.Nodes), len(g.Edges)) + } + var gotInput, gotAgent bool + for _, n := range g.Nodes { + switch n.Kind { + case "input": + gotInput = true + if text, _ := n.Config["text"].(string); text != q { + t.Errorf("input.text=%q,期望 %q", text, q) + } + case "agent": + gotAgent = true + if sys, _ := n.Config["system"].(string); !strings.Contains(sys, "JARVIS") { + t.Errorf("agent.system 未含 JARVIS 提示: %q", sys) + } + } + } + if !gotInput || !gotAgent { + t.Fatalf("缺 input(%v)/agent(%v) 节点", gotInput, gotAgent) + } + // 边必须连 input→agent(否则 compose 编译后 agent 收不到输入)。 + if g.Edges[0].Source != "voice_in" || g.Edges[0].Target != "voice_agent" { + t.Errorf("边应为 voice_in→voice_agent,得 %s→%s", g.Edges[0].Source, g.Edges[0].Target) + } +} + +// 空转写不该组图触发(提交侧兜底:submitVoiceTask 空转写返错)——这里只校验组图函数对空串仍产出结构。 +func TestBuildVoiceGraph_EmptyStillStructured(t *testing.T) { + raw := buildVoiceGraph("") + if _, err := dsl.ParseAndAssemble(raw); err != nil { + t.Fatalf("空转写图仍应结构合法: %v", err) + } +}