Cyrene_v2/internal/services/gateway/voice.go

package gateway

import (
	"context"
	"encoding/base64"
	"encoding/json"
	"fmt"
	"log/slog"
	"strings"

	"git.yeij.top/AskaEth/Cyrene_v2/internal/logger"
	"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/loop"
	"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model"
	"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/multimodal/voice"
	voicetts "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/voice/tts"
)

// VoiceService 为 voice_input WS 路由运行语音转文本与
// 文本转语音。*voice.Chain（降级 provider 链）与 *voice.Service
// （后台工作池）都满足它。它容忍 nil：未注入服务时
// voice 路由返回明确的错误。
type VoiceService interface {
	// STT 把音频转写为文本。
	STT(ctx context.Context, audio voice.Audio) (string, error)
	// TTS 从文本合成音频。每请求的情绪/语气指令可以
	// 经 ctx（voice.WithTTSInstruction）串下来，以在单次调用中覆盖
	// 已配置的 VOICE_TTS_INSTRUCTION。
	TTS(ctx context.Context, text string) (voice.Audio, error)
	// CloneVoice 从公开参考音频 URL 创建自定义音色（声音复刻
	// create_voice）并返回其 voice_id。
	CloneVoice(ctx context.Context, referenceURL, prefix string) (string, error)
}

// voiceProviderHeader 是二进制 /api/v1/audio/speech 响应上携带 provider
// 档位的响应头，因为那里无法添加 JSON 信号字段。
const voiceProviderHeader = "X-Voice-Provider"

// voiceProviderReporter 是 VoiceService 的可选扩展：一个能报告
// 结果是哪个 provider 档位（cloud / local / edge-tts / espeak / silent）
// 产出的服务。*voice.Chain 与 *voice.Service 实现了它；未实现的
// 自定义 VoiceService 仍被支持，回退到朴素的
// STT/TTS 方法且不报告档位。
type voiceProviderReporter interface {
	STTWithProvider(ctx context.Context, audio voice.Audio) (string, string, error)
	TTSWithProvider(ctx context.Context, text string) (voice.Audio, string, error)
}

// sttWithProvider 运行 STT 并返回转写文本加上服务它的
// provider 档位（注入的服务无法报告时为 ""）。
func sttWithProvider(ctx context.Context, v VoiceService, audio voice.Audio) (string, string, error) {
	if r, ok := v.(voiceProviderReporter); ok {
		return r.STTWithProvider(ctx, audio)
	}
	text, err := v.STT(ctx, audio)
	return text, "", err
}

// ttsWithProvider 运行 TTS 并返回音频加上服务它的
// provider 档位。
func ttsWithProvider(ctx context.Context, v VoiceService, text string) (voice.Audio, string, error) {
	if r, ok := v.(voiceProviderReporter); ok {
		return r.TTSWithProvider(ctx, text)
	}
	audio, err := v.TTS(ctx, text)
	return audio, "", err
}

// providerSignal 为语音结果返回 "provider" 降级信号：结果为
// 空时返回实际 provider 档位，档位未知时返回 "silent"，
// 而结果携带真实文本/音频时返回 ""（省略该字段）。它
// 只标记「没有真实语音/音频」；它绝不是业务错误。
func providerSignal(rung string, empty bool) string {
	if !empty {
		return ""
	}
	if rung == "" {
		return "silent"
	}
	return rung
}

// voiceInputRequest 是 voice_input 帧解码后的载荷。Audio 是
// base64（或 data: URI）音频块；Format 是其编码（"wav"/"mp3"/"ogg"）；
// Text 存在时覆盖 STT（已自行转写的客户端可跳过
// 识别）；SessionID 默认 "web"；DeviceID 是客户端产出的
// 端标识（design/16 §4.5），串进持久化的语音回合。
type voiceInputRequest struct {
	Audio     string `json:"audio"`
	Format    string `json:"format"`
	Text      string `json:"text"`
	SessionID string `json:"sessionId"`
	DeviceID  string `json:"deviceId,omitempty"`
	// ClientMsgID 是客户端自带的幂等键（协同 G4），与文本 chat 帧**同形**：
	// 同一个非空 clientMsgId 只承认第一条，重复到达不再落库、不再起回合
	// （回一条 chat_duplicate）。它可选，省略时行为与加入该字段之前逐字相同。
	ClientMsgID string `json:"clientMsgId,omitempty"`
}

// SetVoice 注入 voice_input WS 路由所用的语音 STT/TTS 服务。
// 它是网关暴露的注入点，使组合根能接线 *voice.Chain 或
// *voice.Service 而无需触碰 Deps。nil 值会禁用该
// 路由（它以明确错误应答）。
func (s *Server) SetVoice(v VoiceService) {
	s.voice = v
	s.hub.voice = v
}

// handleVoiceInput 路由一个 voice_input 帧。未注入语音
// 服务时它以明确错误应答（绝不崩溃）；否则它异步跑这个语音
// 回合（STT → 编排器 → 结算时的 TTS），使读泵绝不被
// provider 或模型延迟阻塞。
func (c *wsConn) handleVoiceInput(in wsInbound) {
	if c.h.voice == nil {
		c.enqueueJSON(wsEvent{Type: "error", Data: map[string]any{"message": "voice unavailable"}})
		return
	}
	var req voiceInputRequest
	_ = json.Unmarshal(in.Data, &req)
	if strings.TrimSpace(req.Audio) == "" && strings.TrimSpace(req.Text) == "" {
		c.enqueueJSON(wsEvent{Type: "error", Data: map[string]any{"message": "voice audio or text required"}})
		return
	}
	go c.h.runVoiceTurn(c, req)
}

// voiceLogSource 是语音路径的日志来源标签，使这条路径上的 WARN 能被追溯到
// 具体子功能（AGENTS §10 的「日志必须标来源」）。
const voiceLogSource = "gateway/voice"

// runVoiceTurn 驱动一个语音回合。转写（STT）仍在这里完成 —— 识别是语音入口
// 自己的事；而**回合**从此刻起与文本回合走同一条路：进编排器 → 落库 → 流式 →
// 跨端同步（协同 C1）。此前它直调 runner.Run，于是语音问过的事不进历史、不跨端、
// 也不参与 busy/pending。
//
// 音频回复不在本函数等待：回合是异步完成的（可能排队、可能被中断），因此 TTS 与
// 定向回帧由收尾回调（voiceReplySink）在回合**结算**时完成 —— 那是编排器唯一
// 知道这个回合产出了什么的地方（与 S3.5 回写文档结论同一收尾点）。
func (h *hub) runVoiceTurn(c *wsConn, req voiceInputRequest) {
	ctx := context.Background()
	// 把客户端的 deviceID 与可选幂等键串下来，使持久化的语音回合被发送端标注
	//（design/16 §4.5）并让重复到达的 clientMsgId 被判重（协同 G4）—— 与
	// ws_hub_turn.go 的文本路径同一机制。
	ctx = withClientMsgHint(withDeviceID(ctx, req.DeviceID), clientMsgHint{ClientMsgID: req.ClientMsgID})
	if req.SessionID == "" {
		req.SessionID = "web"
	}
	transcript := strings.TrimSpace(req.Text)
	if transcript == "" {
		if strings.TrimSpace(req.Audio) == "" {
			h.voiceError(c, "voice audio or text required")
			return
		}
		audio, err := decodeVoiceAudio(req.Audio, req.Format)
		if err != nil {
			h.voiceLog().Warn("voice audio decode", "err", err, "session", req.SessionID)
			h.voiceError(c, "voice audio could not be decoded")
			return
		}
		stt, err := h.voice.STT(ctx, audio)
		if err != nil {
			h.voiceLog().Warn("voice stt", "err", err, "session", req.SessionID)
			h.voiceError(c, "voice transcription failed")
			return
		}
		transcript = strings.TrimSpace(stt)
	}
	if transcript == "" {
		h.voiceError(c, "no speech recognized")
		return
	}
	if h.runner == nil {
		h.voiceError(c, "chat service is not configured")
		return
	}
	// Web 主会话是主人的私密对话，因此它绑定全访问 env：记忆读+写，行为不变。
	// 角色来自该会话的绑定（design/22 §4.5）。
	// 语音收尾要用**生效人格**选音色（会话绑定 → 全局默认人格），而 env 只带绑定
	// （loop 自己回退）。这里**复用同一次会话读取**，不为选音色多查一次库。
	env, turnPersona := h.envResolver.ownerWithPersona(ctx, req.SessionID)
	msgs := []model.Message{model.TextMessage(model.RoleUser, transcript)}
	// 两条「怎么回到这条连接」的通路都由本连接给出：收尾回调（音频回复，C1）与
	// 定向拒绝回调（队列满，G5）。编排器只把它们当成本回合的传输线索。
	ctx = withVoiceReplySink(ctx, h.voiceReplySink(c, req.SessionID, req.DeviceID, turnPersona.Effective, transcript))
	ctx = withPendingFullNotifier(ctx, c.notifyPendingFull)
	if h.orchestrator != nil {
		h.orchestrator.Submit(ctx, req.SessionID, msgs, env)
		return
	}
	h.runVoiceTurnDirect(ctx, c, req.SessionID, req.DeviceID, turnPersona.Effective, transcript, env)
}

// runVoiceTurnDirect 是未接编排器时的旧直跑路径（未接线的构建 / 裸 hub 单元
// 测试）：转写之后的回合直接跑 loop，然后立即合成并回音频。它与编排器路径共用
// 同一个帧构造（emitVoiceOutput），因此两条路径的 voice_output 契约一致。
func (h *hub) runVoiceTurnDirect(ctx context.Context, c *wsConn, sessionID, deviceID, personaID, transcript string, env loop.SessionEnv) {
	res, err := h.runner.Run(loop.WithSessionEnv(ctx, env), sessionID, []model.Message{model.TextMessage(model.RoleUser, transcript)})
	if err != nil {
		h.voiceLog().Warn("voice chat run", "err", err, "session", sessionID, "device_id", deviceID)
		h.voiceError(c, "voice chat failed")
		return
	}
	h.synthesizeVoiceReply(ctx, c, sessionID, deviceID, personaID, transcript, voiceReplyText(res))
}

// voiceReplySink 返回本次语音请求的收尾回调：回合结算后取助手正文 → TTS →
// **定向**回发起它的那根连接。它是 C1 的核心接线点 —— 回合异步完成，只有结算处
// 知道助手正文。
//
// personaID 是**本回合**解析出的角色（`env.PersonaID`，会话无绑定时为空）。它随
// 回调一起传下去，使合成按这一回合的角色选发音人 —— 而不是进程启动时定死一个。
// 它不是身份输入（身份仍由 actor 元数据决定），只决定用哪个上游音色。
//
// 连接已断（客户端走了）时记一条带来源的 WARN，绝不静默丢：那正是「静默成功但
// 没声音」那类缺陷的温床（协同 C2 的教训）。
func (h *hub) voiceReplySink(c *wsConn, sessionID, deviceID, personaID, transcript string) VoiceReplyFunc {
	return func(res *loop.Result) {
		if c.closed() {
			h.voiceLog().Warn("voice reply dropped", "reason", "connection closed", "session", sessionID, "device_id", deviceID)
			return
		}
		// 回合的 ctx 已随回合消亡，因此合成用全新的后台 context：音频是「回复已
		// 落库」之后的增强，不该被一次被取消的回合连带取消。
		h.synthesizeVoiceReply(context.Background(), c, sessionID, deviceID, personaID, transcript, voiceReplyText(res))
	}
}

// synthesizeVoiceReply 把助手正文合成音频并定向回发起这条语音的连接
// （协同 C1/C3）。四条纪律：
//   - TTS 失败或降级为空音频时**照旧回帧**并带既有 provider 降级信号，让客户端
//     能提示「没生成声音」（绝不是只回一条错误帧就完事）；
//   - 没有助手正文（被取消 / 空回复）时不打一次无谓的 TTS，直接回降级帧 ——
//     客户端总能看到这条语音回合的终点，而不是永远等一个不会来的 voice_output；
//   - **发音人按本回合的 personaID 现解析**（`voicetts.WithPersonaID`）：换人格后
//     绝不会继续用老角色的声音（主人 2026-10-26 的硬要求）；
//   - 日志只带会话与设备 id，**绝不记正文/音频**。
func (h *hub) synthesizeVoiceReply(ctx context.Context, c *wsConn, sessionID, deviceID, personaID, transcript, reply string) {
	if strings.TrimSpace(reply) == "" {
		h.voiceLog().Warn("voice reply empty", "reason", "turn produced no text", "session", sessionID, "device_id", deviceID)
		emitVoiceOutput(c, sessionID, deviceID, transcript, "", voice.Audio{}, providerSignal("", true))
		return
	}
	// 把**本回合**的角色交给语音面选发音人；空 personaID = 会话无绑定 ⇒ 语音面
	// 走它的 default 分支（没配 default 就不合成，文本已发出）。
	ctx = voicetts.WithPersonaID(ctx, personaID)
	audi, rung, err := ttsWithProvider(ctx, h.voice, reply)
	if err != nil {
		h.voiceLog().Warn("voice tts", "err", err, "session", sessionID, "device_id", deviceID, "persona", personaID)
		emitVoiceOutput(c, sessionID, deviceID, transcript, reply, voice.Audio{}, providerSignal("", true))
		return
	}
	emitVoiceOutput(c, sessionID, deviceID, transcript, reply, audi, providerSignal(rung, len(audi.Data) == 0))
}

// emitVoiceOutput 定向回一条 voice_output 帧（只有发起这条语音的连接收得到）。
// 载荷在既有四键（audio/format/text/reply）之上**纯增量**地带上 sessionId 与
// deviceId（协同 C3）：多端同时说话时客户端能区分归属。provider 只在没有真实
// 音频时出现（既有降级信号），它不是业务错误。
func emitVoiceOutput(c *wsConn, sessionID, deviceID, transcript, reply string, audi voice.Audio, provider string) {
	data := map[string]any{
		"audio":     base64.StdEncoding.EncodeToString(audi.Data),
		"format":    audi.Format,
		"text":      transcript,
		"reply":     reply,
		"sessionId": sessionID,
	}
	if deviceID != "" {
		data["deviceId"] = deviceID
	}
	if provider != "" {
		data["provider"] = provider
	}
	c.enqueueJSON(wsEvent{Type: wsVoiceOutput, Data: data})
}

// voiceReplyText 返回一个回合的助手正文（没有结果时为空串）。空正文**不拿
// stopReason 顶替**：把 "stop"/"idle" 这类内部状态合成成助手声音，客户端会把它
// 当回复显示；空正文一律走既有的 provider 降级信号，如实说「没生成声音」。
func voiceReplyText(res *loop.Result) string {
	if res == nil {
		return ""
	}
	return res.Text
}

// voiceLog 返回带语音来源标签的 logger（AGENTS §10：日志必须标来源）。
func (h *hub) voiceLog() *slog.Logger {
	return logger.WithSource(h.log, voiceLogSource)
}

// voiceError 为失败的语音回合回复一个带类型的错误帧。
func (h *hub) voiceError(c *wsConn, message string) {
	c.enqueueJSON(wsEvent{Type: "error", Data: map[string]any{"message": message}})
}

// decodeVoiceAudio 把 base64（或 data:）音频载荷转成 voice.Audio。
// 格式在 req 设置时取自 req，否则从 data: URI mime 猜测，
// 再默认到 "wav"。
func decodeVoiceAudio(src, format string) (voice.Audio, error) {
	payload := src
	if strings.HasPrefix(src, "data:") {
		if comma := strings.IndexByte(src, ','); comma >= 0 {
			if format == "" {
				format = formatFromMime(src[:comma])
			}
			payload = src[comma+1:]
		}
	}
	if strings.TrimSpace(payload) == "" {
		return voice.Audio{}, fmt.Errorf("voice: empty audio")
	}
	data, err := base64.StdEncoding.DecodeString(payload)
	if err != nil {
		return voice.Audio{}, fmt.Errorf("voice: decode audio: %w", err)
	}
	if format == "" {
		format = "wav"
	}
	return voice.Audio{Format: format, Data: data}, nil
}

// formatFromMime 从 data: URI 的 mime 段猜测音频格式。
func formatFromMime(meta string) string {
	lower := strings.ToLower(meta)
	for _, f := range []string{"wav", "mp3", "ogg", "webm", "mpeg"} {
		if strings.Contains(lower, f) {
			return f
		}
	}
	return "wav"
}