diff --git a/Docs/api-client.md b/Docs/api-client.md index cd030d2..6c9eed9 100644 --- a/Docs/api-client.md +++ b/Docs/api-client.md @@ -23,7 +23,7 @@ |---|------|------|------|:---:|------| | 1 | `/api/v1/auth/login` | POST | 否 | ✅ | `{username,password}` → `{data:{token,expiresAt,role}}`(24h,role=owner) | | 2 | `/api/v1/health` | GET | 否 | ✅ | 健康检查 `{data:{status:"ok"}}` | -| 3 | `/api/v1/chat` | POST | JWT | ✅ | `{text,image?,deviceId?,stream?}` → `{data:{reply}}`;`stream:true` 时经 WS 触发流式。`image` 为 base64 data: URI 或 http(s) URL,入视觉路由(design/06)——**真视觉需配 `LLM_VISION_MODEL`/凭据,无凭据时降级为主模型直接看图或视觉·OCR 工具文本描述**;`deviceId` 为**客户端生成的具体端标识**(design/16 §4.5,如 `web:chrome:a1b2c3`,可选),落库到消息、便于主人知道从哪个端发来。**会话编排(design/15)**:非流式且该会话「正忙」时**不阻塞**,返回 `202` `{data:{queued:true,queueDepth}}`(稍后按上下文续接);待处理队列满时返回 `409` `{error:{code:"chat_busy"}}` | +| 3 | `/api/v1/chat` | POST | JWT | ✅ | `{text,image?,deviceId?,mode?,stream?}` → `{data:{reply}}`;`stream:true` 时经 WS 触发流式。`image` 为 base64 data: URI 或 http(s) URL,入视觉路由(design/06)——**真视觉需配 `LLM_VISION_MODEL`/凭据,无凭据时降级为主模型直接看图或视觉·OCR 工具文本描述**;`deviceId` 为**客户端生成的具体端标识**(design/16 §4.5,如 `web:chrome:a1b2c3`,可选),落库到消息、便于主人知道从哪个端发来;`mode` 为**行为模式**(design/01 §3 / design/16 §5.1,如 `chat`/`code`/`home`/`docs`/…,可选),随该轮次落库——IM 恒 `chat`、缺省回落 `chat`,供前端按模式渲染气泡(仅 `chat` 才解析 `

` 拆气泡)。**会话编排(design/15)**:非流式且该会话「正忙」时**不阻塞**,返回 `202` `{data:{queued:true,queueDepth}}`(稍后按上下文续接);待处理队列满时返回 `409` `{error:{code:"chat_busy"}}` | | 4 | `/api/v1/sessions` | GET | JWT | ✅ | `{data:[session...]}`。**会话持久化(design/16)**:注入 HistoryStore 后返回持久化会话元数据 `{data:[{id,type("main"|"sub"),title?,createdAt}]}`(主会话 `id="web"` 恒在首位,次级 `web:`);未注入时退化回渠道会话列表 | | 5 | `/api/v1/sessions` | POST | JWT(owner/admin) | ✅ | **新建次级会话** → `201 {data:{sessionId}}`(`sessionId="web:"`);未注入 HistoryStore 回 `501 not_implemented` | | 6 | `/api/v1/sessions/{id}/messages` | GET | JWT(owner/admin) | ✅ | **热分页拉历史**:`?limit=&before=&platform=` 默认 `limit=50`(上限 200)、`before`=上一页最旧 `createdAt`(毫秒),按时间倒序 `{data:[{id,sessionId,role,content,mode?,mood?,image?,thinking?,platform?,deviceId?,senderName?,createdAt}]}`;`platform` 为**可选**来源平台过滤(design/16 §4.4)——Web 端不传(=全量共享主会话),IM 适配器传**本平台**(如 `telegram`)只拉本渠道消息;`deviceId` 为**客户端具体端标识**(design/16 §4.5,如 `web:chrome:a1b2c3` / `weixin-ilink:conf0:wx_uid`);`senderName` 为**第三方发送者昵称**(design/16 §4.6,群聊/陌生人私聊消息带,主人私聊为空,`omitempty`);非法 `limit`/`before` 回 `400`;未注入 HistoryStore 回 `501` | @@ -204,6 +204,7 @@ - 语音:上行 `voice_input`(base64 音频或已转写 text),下行 `voice_output`(合成音频 base64 + text)。 ## 5. 变更记录 +- `v0.20`(**A3 消息 mode 落库**):`POST /api/v1/chat` 与 `ws.chat` 上行新增**可选 `mode`**(行为模式,design/01 §3 / design/16 §5.1)。后端把每条持久化消息的 `Mode` **真实写入**(IM 路径恒 `chat`;Web 路径经 ctx 透传本轮 `mode`,缺省回落 `chat`;恢复/续接路径沿用发起轮次的 `mode`),消息 DTO、`message_persisted`、`channel_persisted` 的 `mode` 不再恒空,供前端按模式渲染气泡(**仅 `chat` 解析 `

` 拆气泡**,其它模式原样整条)。 - `v0.19`(**百炼声音复刻 create_voice + TTS instruction 情绪驱动,主 TTS 用昔涟复刻音色**):`POST /api/v1/audio/voice-clone`(**新增**,owner-only + 审计,body `{url:"<公网参考音频URL>", prefix?}` → `{data:{voiceId}}`)——走后端 `voice.DashscopeCloner`(百炼 `voice-enrollment` create_voice,`POST {base}/services/audio/tts/customization`,`input{action:create_voice, target_model:VOICE_TTS_MODEL, prefix, url}`,**只收公网 URL 不收 base64**,返回 `output.voice_id`;**URL 不打日志不写审计**)。`POST /api/v1/audio/speech` 新增**可选 `instruction`**(per-request 情绪/语气指令,如 `"语气开朗俏皮,像女朋友撒娇"`),经 ctx 透传给 TTS adapter;**只对复刻/设计音色生效**(`cosyvoice-v3-flash` 复刻音色可任意指令;系统音色 `longxiaochun_v3` 需固定格式,服务端 `canInstruct` 判定后**永不发送**)。config 新增 `VOICE_TTS_INSTRUCTION`(默认 `""`)、`VOICE_CLONE_PREFIX`(默认 `cyrene`,≤10 字母/数字);`voice.Chain`/`voice.Service` 新增 `CloneVoice`,`VoiceCloner` 接口(cloud rung 提供;未配 cloud 回 `voice_unavailable`)。**⚠️ 真建昔涟音色仍待主人提供公网参考音频 URL(当前无)**,且 CosyVoice 声音复刻仅华北2北京地域(确认 key 地域);TTS instruction 用现有复刻音色或先仅单测请求体,避免改系统音色行为。 - `v0.18`(**后端语音接通阿里百炼 STT + TTS 主链路**):`POST /api/v1/audio/transcribe`(**新增**,owner + 审计,body `{audio,format?,deviceId?}` → `{data:{text}}`,走 STT);`POST /api/v1/audio/speech`(**新增**,owner + 审计,body `{text,voice?,format?}` → 音频字节 `audio/`,走 TTS)——**REST 语音源**,原生/移动端经它绕开 WS 4KB 帧限上传/获取音频;WS `voice_input` 上行新增 **`deviceId`**(随语音轮次落库,design/16 §4.5)。后端 `internal/runtime/multimodal/voice` 的 cloud rung 由空 endpoint stub 改为**真实百炼 adapter**:STT `qwen-audio-3.0-asr-flash`(base64 原始字节→文本)、TTS `cosyvoice-v3-flash`(按 `voice_id` 合成,取非流式返回的 24h 音频 URL 下载字节)。**实测**(`DASHSCOPE_API_KEY` 实跑):TTS「你好,我是昔涟」→ 106124 字节 wav;把该 wav 喂给 STT → 文本「你好,我是西联。」(TTS 发音把「昔涟」读得近似「西联」,STT 识别正确);语音链 (cloud→…→silent) 返回真实字节/文本。**已实测修正**:`cosyvoice-v3.5-plus`/`longxiaochun`(v2 音色)走 HTTP `SpeechSynthesizer` 会回 `418`(引擎拒绝),改用 **`cosyvoice-v3-flash` + `longxiaochun_v3`**(v3 音色,各模型音色不可混用)。**声音复刻(响昔涟音色)仍待**:需参考音频(公网 URL)在百炼建自定义音色后把 `voice_id` 填入 `VOICE_TTS_VOICE_ID`(本项目不做 create_voice)。config 新增 `DASHSCOPE_BASE_URL`/`VOICE_STT_\*`/`VOICE_TTS_\*`(`VOICE_TTS_SAMPLE_RATE` 默认 24000)。 - `v0.17`(系统设置安全与存储:吊销全部 API key / 轮换 JWT 密钥 / 备份 / 恢复,design/11 §4 + design/10 §3 模块14):新增 `POST /api/v1/admin/system/{revoke-keys,rotate-jwt,backup,restore}`(均 **owner-only** + 审计)。`revoke-keys` 走 `auth.APIKey.RevokeAll()`(清空 key 表,返回 `{revoked:n}`);`rotate-jwt` 走 `auth.JWTService.RotateSecret`(≥32 位,**新密钥不回显/不落日志**,现有 token 立即失效);`backup` 用 SQLite `VACUUM INTO` 快照到 `/backup-.db`(含已提交 WAL 内容)返回 `{path,size}`;`restore` 先校验路径在 dataDir 内(**`filepath.Rel` 防路径穿越**,逃逸回 `400`)→ **先自动备份当前 DB** → 替换 DB 文件 → 返回 `{restored:true,backup,note:"需重启后生效"}`(运行进程持有 DB 句柄,热替换需重启可见)。gateway 新增 `Deps.DatabasePath`(composition root 传入实际打开的 SQLite 路径)。`编辑环境变量 / 重启服务 / 检查更新` 仍为后置,未实现、不伪造。 diff --git a/internal/services/channelbridge/channelbridge_test.go b/internal/services/channelbridge/channelbridge_test.go index c239c60..c1fd9d3 100644 --- a/internal/services/channelbridge/channelbridge_test.go +++ b/internal/services/channelbridge/channelbridge_test.go @@ -315,14 +315,17 @@ func TestOwnerPrivatePersistsToMainSessionWithPlatform(t *testing.T) { if len(recs) != 2 { t.Fatalf("appends = %d, want 2 (user + assistant)", len(recs)) } - // Both land on the main web session and carry the channel platform + adapter. + // Both land on the main web session and carry the channel platform + adapter, + // and IM messages are always persisted in chat mode (design/16 §5.1). if recs[0].SessionID != history.MainSessionID || recs[0].Role != history.RoleUser || - recs[0].Content != "hello" || recs[0].Platform != "telegram" || recs[0].AdapterID != "telegram" { - t.Fatalf("user record = %+v, want web/telegram/hello/telegram", recs[0]) + recs[0].Content != "hello" || recs[0].Platform != "telegram" || recs[0].AdapterID != "telegram" || + recs[0].Mode != "chat" { + t.Fatalf("user record = %+v, want web/telegram/hello/telegram/chat", recs[0]) } if recs[1].SessionID != history.MainSessionID || recs[1].Role != history.RoleAssistant || - recs[1].Content != "微信回复" || recs[1].Platform != "telegram" || recs[1].AdapterID != "telegram" { - t.Fatalf("assistant record = %+v, want web/telegram/微信回复/telegram", recs[1]) + recs[1].Content != "微信回复" || recs[1].Platform != "telegram" || recs[1].AdapterID != "telegram" || + recs[1].Mode != "chat" { + t.Fatalf("assistant record = %+v, want web/telegram/微信回复/telegram/chat", recs[1]) } // Every persisted message was handed to the broadcast hook. if len(persisted) != 2 { @@ -414,6 +417,10 @@ func TestGroupMessageNotPersistedToMainSession(t *testing.T) { if rec.Platform != "telegram" { t.Fatalf("record[%d] Platform = %q, want telegram", i, rec.Platform) } + // Isolated IM messages are still chat-mode (design/16 §5.1). + if rec.Mode != "chat" { + t.Fatalf("record[%d] Mode = %q, want chat", i, rec.Mode) + } } if chat.sessionID != "sess-g" { t.Fatalf("group loop session id = %q, want sess-g", chat.sessionID) diff --git a/internal/services/channelbridge/persist.go b/internal/services/channelbridge/persist.go index a45e791..907c61d 100644 --- a/internal/services/channelbridge/persist.go +++ b/internal/services/channelbridge/persist.go @@ -14,6 +14,10 @@ import ( // without a refresh. Kept separate from channelbridge.go so the bridge's inbound // pipeline file stays under the module size bound. +// modeChat is the behavior mode written for every IM-persisted message: IM carries +// chat text alone (design/02 §1 单向回流), so no other mode ever applies there. +const modeChat = "chat" + // SetOnPersisted wires (or replaces) the transport hook called after an // owner-private message is persisted into the main web session. The gateway is // built after the bridge in the composition root, so the hook is attached post-New @@ -60,7 +64,7 @@ func (b *Bridge) persistToMain(ctx context.Context, role, content, platform, dev return } full, err := b.history.AppendFull(ctx, history.Message{ - SessionID: history.MainSessionID, Role: role, Content: content, Platform: platform, DeviceID: deviceID, AdapterID: adapterID, + SessionID: history.MainSessionID, Role: role, Content: content, Platform: platform, DeviceID: deviceID, AdapterID: adapterID, Mode: modeChat, }) if err != nil { b.log.Warn("persist main session", "err", err, "role", role, "platform", platform, "device_id", deviceID, "adapter", adapterID) @@ -85,7 +89,7 @@ func (b *Bridge) persistIsolated(ctx context.Context, role, content, deviceID st } full, err := b.history.AppendFull(ctx, history.Message{ SessionID: sess.SessionID, Role: role, Content: content, Platform: platform, - DeviceID: deviceID, SenderName: senderName, AdapterID: adapterID, + DeviceID: deviceID, SenderName: senderName, AdapterID: adapterID, Mode: modeChat, }) if err != nil { b.log.Warn("persist isolated session", "err", err, "role", role, diff --git a/internal/services/gateway/extra_test.go b/internal/services/gateway/extra_test.go index ae0c863..b35f04a 100644 --- a/internal/services/gateway/extra_test.go +++ b/internal/services/gateway/extra_test.go @@ -348,6 +348,33 @@ func TestChatStreamMode(t *testing.T) { } } +func TestChatRequestPersistsMode(t *testing.T) { + t.Parallel() + hist := &fakeHistory{} + _, ts, jwt := buildEnv(t, cfgDefault(), Deps{ + Loop: &stubRunner{res: &loop.Result{Text: "回复", StopReason: "stop"}}, + History: hist, + }) + tok := signAny(t, jwt, "acct", auth.RoleOwner) + // The REST body carries the client's behavior mode (design/16 §5.1); the + // orchestrator persists both the user turn and the assistant reply under it. + resp, body := doReq(t, http.MethodPost, ts.URL+"/api/v1/chat", + map[string]any{"sessionId": "web", "text": "改代码", "mode": "code"}, tok) + if resp.StatusCode != http.StatusOK { + t.Fatalf("status = %d, want 200 (body=%s)", resp.StatusCode, body) + } + recs := hist.records() + if len(recs) != 2 { + t.Fatalf("appends = %d, want 2 (user + assistant)", len(recs)) + } + if recs[0].Role != history.RoleUser || recs[0].Mode != "code" { + t.Fatalf("user record = %+v, want role=user mode=code", recs[0]) + } + if recs[1].Role != history.RoleAssistant || recs[1].Mode != "code" { + t.Fatalf("assistant record = %+v, want role=assistant mode=code", recs[1]) + } +} + func TestChatDefaultsSessionToWeb(t *testing.T) { t.Parallel() cr := &captureRunner{res: &loop.Result{Text: "ok", Steps: 1, StopReason: "stop"}} @@ -882,6 +909,37 @@ func TestHubStreamChat(t *testing.T) { } } +func TestHubStreamChatPersistsMode(t *testing.T) { + t.Parallel() + hist := &fakeHistory{} + hb := newTestHub(&stubRunner{res: &loop.Result{Text: "hi", StopReason: "stop"}}, nil) + hb.orchestrator = NewOrchestrator(OrchestratorOptions{ + Run: func(context.Context, string, []model.Message, loop.SessionEnv) (*loop.Result, error) { + return &loop.Result{Text: "hi", StopReason: "stop"}, nil + }, + History: hist, + }) + // The ws.chat payload carries the client's behavior mode (design/16 §5.1); + // streamChat threads it into the orchestrator, which persists both the user + // turn and the assistant reply under it. + hb.streamChat(context.Background(), map[string]any{"sessionId": "web", "text": "改代码", "mode": "code"}) + + deadline := time.Now().Add(2 * time.Second) + for time.Now().Before(deadline) && len(hist.records()) < 2 { + time.Sleep(time.Millisecond) + } + recs := hist.records() + if len(recs) != 2 { + t.Fatalf("appends = %d, want 2 (user + assistant)", len(recs)) + } + if recs[0].Role != history.RoleUser || recs[0].Mode != "code" { + t.Fatalf("user record = %+v, want role=user mode=code", recs[0]) + } + if recs[1].Role != history.RoleAssistant || recs[1].Mode != "code" { + t.Fatalf("assistant record = %+v, want role=assistant mode=code", recs[1]) + } +} + func TestHubBroadcastStream(t *testing.T) { t.Parallel() hb := newTestHub(nil, nil) diff --git a/internal/services/gateway/rest.go b/internal/services/gateway/rest.go index f3532c5..d0d1aa0 100644 --- a/internal/services/gateway/rest.go +++ b/internal/services/gateway/rest.go @@ -72,11 +72,15 @@ type loginResponse struct { // loop routes it through the vision/OCR tool or the main model (design/06 §0). // DeviceID is the client-produced identifier of the sending browser/device // (design/16 §4.5), e.g. "web:chrome:a1b2c3"; it is optional and persisted as-is. +// Mode is the behavior mode the client sent the turn in (design/01 §3, design/16 +// §5.1); it is optional and persisted per message for bubble rendering, falling +// back to "chat" when omitted. type chatRequest struct { SessionID string `json:"sessionId"` Text string `json:"text"` Image string `json:"image,omitempty"` DeviceID string `json:"deviceId,omitempty"` + Mode string `json:"mode,omitempty"` Stream bool `json:"stream"` } @@ -129,7 +133,7 @@ func (s *Server) handleChat(w http.ResponseWriter, r *http.Request) { return } if req.Stream { - s.publishChatStream(r.Context(), req.SessionID, req.Text, req.Image, req.DeviceID) + s.publishChatStream(r.Context(), req.SessionID, req.Text, req.Image, req.DeviceID, req.Mode) writeJSON(w, http.StatusOK, map[string]any{"stream": true}) return } @@ -144,7 +148,7 @@ func (s *Server) handleChat(w http.ResponseWriter, r *http.Request) { // it binds the full-access env: memory read+write, unchanged behavior. msgs := userChatMessages(req.Text, req.Image) if s.hub.orchestrator != nil { - sr, res := s.hub.orchestrator.SubmitSync(withDeviceID(r.Context(), req.DeviceID), req.SessionID, msgs, loop.OwnerEnv()) + sr, res := s.hub.orchestrator.SubmitSync(withMode(withDeviceID(r.Context(), req.DeviceID), req.Mode), req.SessionID, msgs, loop.OwnerEnv()) if !sr.Accepted { if sr.PendingFull { writeError(w, apperr.Wrap(apperr.ErrConflict, "chat_busy", "会话正忙,待处理消息已满")) diff --git a/internal/services/gateway/session_orchestrator.go b/internal/services/gateway/session_orchestrator.go index 7a73f53..833a27a 100644 --- a/internal/services/gateway/session_orchestrator.go +++ b/internal/services/gateway/session_orchestrator.go @@ -130,11 +130,14 @@ type sessionState struct { // pendingMsg is one queued user turn. It carries its own env so a resume runs // under the same memory-isolation boundary the message arrived with, plus the // client deviceID of the originating turn (design/16 §4.5) so the resumed -// assistant reply is tagged with the same client that sent the message. +// assistant reply is tagged with the same client that sent the message, and the +// turn's behavior mode (design/16 §5.1) so the resumed reply is persisted under +// the same mode as the user message that started it. type pendingMsg struct { messages []model.Message env loop.SessionEnv deviceID string + mode string at time.Time } @@ -219,6 +222,7 @@ func (o *Orchestrator) submitInternal(ctx context.Context, sessionID string, mes o.states[sessionID] = st } deviceID := deviceIDFrom(ctx) + msgMode := modeFrom(ctx) if st.mode != ModeIdle { if len(st.pending) >= o.maxPending { mode := st.mode @@ -226,11 +230,11 @@ func (o *Orchestrator) submitInternal(ctx context.Context, sessionID string, mes o.emitState(sessionID, EventPendingFull, map[string]any{"sessionId": sessionID, "mode": mode, "max": o.maxPending}) return SubmitResult{PendingFull: true}, nil } - st.pending = append(st.pending, pendingMsg{messages: messages, env: env, deviceID: deviceID, at: time.Now()}) + st.pending = append(st.pending, pendingMsg{messages: messages, env: env, deviceID: deviceID, mode: msgMode, at: time.Now()}) depth := len(st.pending) mode := st.mode o.mu.Unlock() - o.persistUserMessages(ctx, sessionID, messages, deviceID) + o.persistUserMessages(ctx, sessionID, messages, deviceID, msgMode) o.emitState(sessionID, EventPendingAck, map[string]any{"sessionId": sessionID, "mode": mode, "queueDepth": depth}) return SubmitResult{Accepted: true, Pending: true, QueueDepth: depth}, nil } @@ -244,32 +248,32 @@ func (o *Orchestrator) submitInternal(ctx context.Context, sessionID string, mes o.mu.Unlock() o.emitState(sessionID, EventSessionBusy, map[string]any{"sessionId": sessionID, "mode": ModeThinking}) - o.persistUserMessages(ctx, sessionID, messages, deviceID) + o.persistUserMessages(ctx, sessionID, messages, deviceID, msgMode) if sync { res, err := o.run(runCtx, sessionID, messages, env) if err != nil { o.log.Error("orchestrator turn", "err", err, "session", sessionID) } - _ = o.completeTurn(sessionID, gen, res, err, deviceID) + _ = o.completeTurn(sessionID, gen, res, err, deviceID, msgMode) return SubmitResult{Accepted: true, Immediate: true, Err: err}, res } - go o.goTurn(sessionID, gen, messages, env, runCtx, cancel, deviceID) + go o.goTurn(sessionID, gen, messages, env, runCtx, cancel, deviceID, msgMode) return SubmitResult{Accepted: true, Immediate: true}, nil } // goTurn runs a single turn on a background goroutine and, when it finishes, // hands the outcome to completeTurn so the orchestrator can resume the queue. // deviceID tags the persistent assistant reply with the client that sent the turn -// (design/16 §4.5). -func (o *Orchestrator) goTurn(sessionID string, gen uint64, messages []model.Message, env loop.SessionEnv, ctx context.Context, cancel context.CancelFunc, deviceID string) { +// (design/16 §4.5) and mode with the turn's behavior mode (design/16 §5.1). +func (o *Orchestrator) goTurn(sessionID string, gen uint64, messages []model.Message, env loop.SessionEnv, ctx context.Context, cancel context.CancelFunc, deviceID, mode string) { defer cancel() res, err := o.run(ctx, sessionID, messages, env) if err != nil && ctx.Err() == nil { o.log.Error("orchestrator turn", "err", err, "session", sessionID) } - _ = o.completeTurn(sessionID, gen, res, err, deviceID) + _ = o.completeTurn(sessionID, gen, res, err, deviceID, mode) } // completeTurn runs after a turn settles (either it finished, errored, or was @@ -278,9 +282,10 @@ func (o *Orchestrator) goTurn(sessionID string, gen uint64, messages []model.Mes // message "remembers" the prior context instead of starting from a blank slate // (design/15 §2.2: incremental, not full replay). A stale generation (an // interrupt or a newer top-level turn superseded this one) is a no-op. -// deviceID is the originating turn's client id, used to tag the persisted reply. -func (o *Orchestrator) completeTurn(sessionID string, gen uint64, res *loop.Result, runErr error, deviceID string) error { - o.persistAssistantReply(sessionID, res, deviceID) +// deviceID is the originating turn's client id and mode its behavior mode, both +// used to tag the persisted reply (design/16 §4.5, §5.1). +func (o *Orchestrator) completeTurn(sessionID string, gen uint64, res *loop.Result, runErr error, deviceID, mode string) error { + o.persistAssistantReply(sessionID, res, deviceID, mode) o.mu.Lock() st := o.states[sessionID] if st == nil || st.gen != gen { @@ -305,7 +310,7 @@ func (o *Orchestrator) completeTurn(sessionID string, gen uint64, res *loop.Resu o.mu.Unlock() o.emitState(sessionID, EventPendingDrain, map[string]any{"sessionId": sessionID, "mode": ModeThinking, "queueDepth": depth}) - go o.goTurn(sessionID, gen, resume, env, runCtx, cancel, next.deviceID) + go o.goTurn(sessionID, gen, resume, env, runCtx, cancel, next.deviceID, next.mode) return nil } diff --git a/internal/services/gateway/session_orchestrator_test.go b/internal/services/gateway/session_orchestrator_test.go index ed6daa8..3e077c3 100644 --- a/internal/services/gateway/session_orchestrator_test.go +++ b/internal/services/gateway/session_orchestrator_test.go @@ -240,6 +240,45 @@ func TestSubmitPersistsUserAndAssistant(t *testing.T) { if recs[1].SessionID != "web" || recs[1].Role != history.RoleAssistant || recs[1].Content == "" { t.Fatalf("assistant record = %+v", recs[1]) } + // No mode on the ctx → both fall back to "chat" (design/16 §5.1). + for i, rec := range recs { + if rec.Mode != "chat" { + t.Fatalf("record[%d] Mode = %q, want chat fallback", i, rec.Mode) + } + } +} + +func TestSubmitPersistsMessageMode(t *testing.T) { + t.Parallel() + run := newGatedRunner(1) + hist := &fakeHistory{} + o := NewOrchestrator(OrchestratorOptions{ + Run: run.Run, + Log: slog.Default(), + History: hist, + }) + run.tickets <- struct{}{} + // The HTTP/WS boundary threads the client's behavior mode through ctx + // (design/01 §3, design/16 §5.1); both the user turn and the assistant reply + // must be persisted under that mode. + ctx := withMode(context.Background(), "code") + sr, res := o.SubmitSync(ctx, "web", []model.Message{userMsg("改代码")}, loop.OwnerEnv()) + if !sr.Accepted || !sr.Immediate { + t.Fatalf("submit = %+v, want Accepted+Immediate", sr) + } + if res == nil { + t.Fatal("sync result is nil") + } + recs := hist.records() + if len(recs) != 2 { + t.Fatalf("appends = %d, want 2 (user + assistant)", len(recs)) + } + if recs[0].Mode != "code" { + t.Fatalf("user Mode = %q, want code", recs[0].Mode) + } + if recs[1].Mode != "code" { + t.Fatalf("assistant Mode = %q, want code", recs[1].Mode) + } } func TestSubmitPersistsDeviceID(t *testing.T) { diff --git a/internal/services/gateway/session_persist.go b/internal/services/gateway/session_persist.go index df5879f..08bcbf0 100644 --- a/internal/services/gateway/session_persist.go +++ b/internal/services/gateway/session_persist.go @@ -13,14 +13,47 @@ import ( // turn settles. Both are silent no-ops when no history store is wired, so the // orchestrator runs unchanged without a store (tests / an unwired gateway). +// modeChat is the behavior mode persisted when a turn carries none (design/01 §3, +// design/16 §5.1). It is also the only mode the IM paths ever write, because IM +// carries chat text alone. +const modeChat = "chat" + +// modeOrDefault returns mode, or modeChat when the caller supplied none, so a +// persisted message never carries an empty behavior mode (design/16 §5.1). +func modeOrDefault(mode string) string { + if mode == "" { + return modeChat + } + return mode +} + +// messageModeContextKey is the context key carrying a turn's behavior mode +// (design/01 §3, design/16 §5.1) from the HTTP/WS boundary into the orchestrator's +// persistence, so each persisted message is tagged with the mode it was sent in. +type messageModeContextKey struct{} + +// withMode returns a context carrying mode so the orchestrator can tag the +// persisted user message and assistant reply. An empty mode is a no-op (the +// persistence layer falls back to modeChat). +func withMode(ctx context.Context, mode string) context.Context { + return context.WithValue(ctx, messageModeContextKey{}, mode) +} + +// modeFrom returns the behavior mode carried by ctx, or "" when none was set. +func modeFrom(ctx context.Context) string { + v, _ := ctx.Value(messageModeContextKey{}).(string) + return v +} + // persistUserMessages writes each RoleUser message of a submitted turn to the // history store so the raw user stream is durable before the turn resolves, then // notifies the transport so every browser of the same account can append it // without a manual refresh (design/16 §3.3, §4.2). deviceID tags the message with -// the sending client (design/16 §4.5). It is a no-op when no history store is -// wired. Rejected turns (empty, pending-full) never reach it, so nothing vacuous -// is persisted. -func (o *Orchestrator) persistUserMessages(ctx context.Context, sessionID string, messages []model.Message, deviceID string) { +// the sending client (design/16 §4.5) and mode is the behavior mode the turn was +// sent in (design/16 §5.1), falling back to modeChat when empty. It is a no-op +// when no history store is wired. Rejected turns (empty, pending-full) never +// reach it, so nothing vacuous is persisted. +func (o *Orchestrator) persistUserMessages(ctx context.Context, sessionID string, messages []model.Message, deviceID, mode string) { if o.history == nil { return } @@ -32,7 +65,7 @@ func (o *Orchestrator) persistUserMessages(ctx context.Context, sessionID string if text == "" && image == "" { continue } - full, err := o.history.AppendFull(ctx, history.Message{SessionID: sessionID, Role: history.RoleUser, Content: text, Image: image, DeviceID: deviceID}) + full, err := o.history.AppendFull(ctx, history.Message{SessionID: sessionID, Role: history.RoleUser, Content: text, Image: image, DeviceID: deviceID, Mode: modeOrDefault(mode)}) if err != nil { o.log.Error("orchestrator persist user", "err", err, "session", sessionID, "device_id", deviceID) continue @@ -43,15 +76,16 @@ func (o *Orchestrator) persistUserMessages(ctx context.Context, sessionID string // persistAssistantReply writes the final assistant text of a settled turn to the // history store, then notifies the transport (design/16 §3.3, §4.2). deviceID tags -// the reply with the client that sent the turn (design/16 §4.5). It is a no-op -// when no history store is wired or the turn produced no text (interrupted/errored -// turns leave res nil). The write runs under a fresh background context so a -// canceled turn context never aborts it. -func (o *Orchestrator) persistAssistantReply(sessionID string, res *loop.Result, deviceID string) { +// the reply with the client that sent the turn (design/16 §4.5) and mode is the +// behavior mode of that turn (design/16 §5.1), falling back to modeChat when +// empty. It is a no-op when no history store is wired or the turn produced no text +// (interrupted/errored turns leave res nil). The write runs under a fresh +// background context so a canceled turn context never aborts it. +func (o *Orchestrator) persistAssistantReply(sessionID string, res *loop.Result, deviceID, mode string) { if o.history == nil || res == nil || res.Text == "" { return } - full, err := o.history.AppendFull(context.Background(), history.Message{SessionID: sessionID, Role: history.RoleAssistant, Content: res.Text, DeviceID: deviceID}) + full, err := o.history.AppendFull(context.Background(), history.Message{SessionID: sessionID, Role: history.RoleAssistant, Content: res.Text, DeviceID: deviceID, Mode: modeOrDefault(mode)}) if err != nil { o.log.Error("orchestrator persist assistant", "err", err, "session", sessionID, "device_id", deviceID) return diff --git a/internal/services/gateway/ws_hub.go b/internal/services/gateway/ws_hub.go index cc3c442..1d23bc3 100644 --- a/internal/services/gateway/ws_hub.go +++ b/internal/services/gateway/ws_hub.go @@ -235,9 +235,10 @@ func (h *hub) streamChat(ctx context.Context, payload any) { // The Web main session is the owner's private conversation (JWT = owner), so // it binds the full-access env: memory read+write, unchanged behavior. msgs := userChatMessages(req.Text, req.Image) - // Thread the client's deviceID into the orchestrator via ctx so both the - // persisted user turn and the assistant reply are tagged (design/16 §4.5). - ctx = withDeviceID(ctx, req.DeviceID) + // Thread the client's deviceID and behavior mode into the orchestrator via ctx + // so both the persisted user turn and the assistant reply are tagged + // (design/16 §4.5, §5.1). + ctx = withMode(withDeviceID(ctx, req.DeviceID), req.Mode) if h.orchestrator != nil { _ = h.orchestrator.Submit(ctx, req.SessionID, msgs, loop.OwnerEnv()) return @@ -312,6 +313,10 @@ type chatStreamRequest struct { // (design/16 §4.5), e.g. "web:chrome:a1b2c3"; it is optional and persisted // alongside the message so the agent knows which end the owner is chatting from. DeviceID string `json:"deviceId,omitempty"` + // Mode is the behavior mode the client sent the turn in (design/01 §3, + // design/16 §5.1); it is optional and persisted per message, falling back to + // "chat" when omitted. + Mode string `json:"mode,omitempty"` } // handleWS upgrades /ws/chat after authenticating the JWT from the query token or @@ -439,13 +444,14 @@ func (c *wsConn) enqueueJSON(ev wsEvent) { } // publishChatStream publishes a ws.chat streaming trigger (REST stream mode). -// deviceID is the sending browser/device id carried on the payload so the ws.chat -// consumer can tag the persisted turn (design/16 §4.5). -func (s *Server) publishChatStream(ctx context.Context, sessionID, text, image, deviceID string) { +// deviceID is the sending browser/device id and mode the behavior mode the client +// sent the turn in, both carried on the payload so the ws.chat consumer can tag +// the persisted turn (design/16 §4.5, §5.1). +func (s *Server) publishChatStream(ctx context.Context, sessionID, text, image, deviceID, mode string) { if s.bus == nil { return } - _ = s.bus.Publish(ctx, wsChatSubject, map[string]any{"sessionId": sessionID, "text": text, "image": image, "deviceId": deviceID}) + _ = s.bus.Publish(ctx, wsChatSubject, map[string]any{"sessionId": sessionID, "text": text, "image": image, "deviceId": deviceID, "mode": mode}) } // wsToken extracts the JWT from the ?token= query or the Authorization header.