feat(gateway): 消息mode落库(A3) — 持久化真实写入渲染模式
- POST /api/v1/chat 与 ws.chat 上行新增可选 mode(行为模式 design/01 §3/design/16 §5.1) - Web路径经ctx透传本轮mode到编排器持久化(persistUserMessages/persistAssistantReply),缺省回落chat - pendingMsg记录mode,恢复/续接路径沿用发起轮次mode - IM路径(persistToMain/persistIsolated)恒为chat - 新增/扩充测试:REST/WS/orchestrator/channelbridge 覆盖mode写入 - Docs/api-client.md v0.20 同步mode字段
This commit is contained in:
1 parent
fe5dbcddba
commit
b686d5fec8
9 files changed
+199
-41
No files matched your search
@@ -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)
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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", "会话正忙,待处理消息已满"))
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
|
||||
@@ -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) {
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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.
|
||||
|
||||
Reference in new issue
Block a user