diff --git a/Docs/api-client.md b/Docs/api-client.md index 93697ed..2c78bdf 100644 --- a/Docs/api-client.md +++ b/Docs/api-client.md @@ -87,7 +87,7 @@ ## 3. WebSocket `/ws/chat`（实时） - **握手**: `ws://:8080/ws/chat?token=` 或 `Authorization: Bearer`；未认证在握手前回 `401` JSON。 -- **下行（服务端→客户端）**: 经 bus 广播 `{type:"", data:..., turnId?, seq?, sessionId?}`（`turnId`/`seq` 为流式关联字段，见 §3.1；`sessionId` 为**会话持久化（design/16）新增**，能确定会话的事件带上，供客户端按激活会话过滤）——`message.start/delta/end` / `thinking.start/delta/end` / `tool_call.start/delta/end` / `tool_execution.update`（结构化流，design/04 §1.1），以及向后兼容的 `stream_chunk` / `stream_end` / `thinking`(思维链) / `mood` / `reminder` / `device_update` / `tool_used`(工具回显) / `subagent_event` / `graph_update`。**会话编排状态（design/15，均为广播，`data.sessionId` 区分会话）**：`session_busy` / `pending_ack` / `pending_drain` / `session_idle` / `session_interrupted` / `pending_full`（见 §3.2）。语音回复走 `voice_output`（定向回请求连接，非广播）。 +- **下行（服务端→客户端）**: 经 bus 广播 `{type:"", data:..., turnId?, seq?, sessionId?}`（`turnId`/`seq` 为流式关联字段，见 §3.1；`sessionId` 为**会话持久化（design/16）新增**，能确定会话的事件带上，供客户端按激活会话过滤）——`message.start/delta/end` / `thinking.start/delta/end` / `tool_call.start/delta/end` / `tool_execution.update`（结构化流，design/04 §1.1），以及向后兼容的 `stream_chunk` / `stream_end` / `thinking`(思维链) / `mood` / `reminder` / `device_update` / `tool_used`(工具回显) / `subagent_event` / `graph_update` / **`message_persisted`**（会话持久化增量，见 §3.3）。**会话编排状态（design/15，均为广播，`data.sessionId` 区分会话）**：`session_busy` / `pending_ack` / `pending_drain` / `session_idle` / `session_interrupted` / `pending_full`（见 §3.2）。语音回复走 `voice_output`（定向回请求连接，非广播）。 > **`history` 回放事件（design/16，新增）**：WS 握手升级成功后、该连接下一帧即收到 `{type:"history", sessionId:"web", data:{messages:[{id,sessionId,role,content,mode?,mood?,image?,createdAt}]}}`——回放主会话最新 50 条（`thinking` 已剥离，不泄给非 owner；默认主会话，次级会话由客户端按激活会话走 `GET /sessions/{id}/messages` 拉取）。客户端据此在重连/新开连接后补全上下文。 @@ -116,12 +116,18 @@ - **`pending_full`**：单会话 pending 达上限（默认 5）拒绝入队（`{sessionId, mode, max}`）——防无限积压（design/15 §4）。 - **上下文续接**：忙完重开时新回合**带上一轮上下文**（`{mode, queueDepth}` 之外的续接由服务端完成），非清空重来（design/15 §2.2）。 +### 3.3 消息持久化增量事件（`message_persisted`，design/16） +- **触发**：每一条 user / assistant 消息**落库后**由编排层广播（`persistUserMessages` 在提交时写用户消息、`persistAssistantReply` 在回合落定时写助手消息）。 +- **载荷**：`{type:"message_persisted", sessionId, data:{message:{id, sessionId, role, content, mode?, mood?, image?, createdAt}}}`——`role` ∈ `user|assistant`；`id` 为**后端生成的消息 id**（客户端据此去重）；`thinking` 已剥离（design/16 §6）。 +- **客户端处理**：仅当 `data.message.sessionId === activeSessionId`（激活会话）时处理；按 `id` upsert——本端已预建的**同 role 乐观气泡**（本地 `uid` 无后端 `id`）被**归一**（`uid` → 后端 `id`），异端无本地气泡则直接追加，两者最终收敛为**同一条 `id` 相同的消息**，「A 发消息、A 与 B 都无需刷新且各出现一次」。 + ## 4. 客户端接入指引 - **认证**: 先 `POST /api/v1/auth/login` 取 JWT → WS 握手（`?token=`）注入。 - **版本**: 客户端请求头声明 `X-API-Version: 1`。 - **离线**: PWA / 原生端建议缓存 `sessions`/`personas` 列表。 ## 5. 变更记录 +- `v0.11`（跨端实时同步，design/16 §4.2）：WS 下行新增 **`message_persisted`** 事件——每条 user/assistant 消息落库后广播 `{type, sessionId, data:{message:{id,sessionId,role,content,mode?,mood?,image?,createdAt}}}`（`id` 为后端生成，`thinking` 已剥离）。客户端按 `sessionId === activeSessionId` 过滤 + 按 `id` upsert（含本端乐观气泡 `uid`→后端 `id` 归一）去重，实现「A 发消息、A 与 B 都无需刷新且各出现一次」。后端 `history.Store` 新增 `AppendFull` 供编排层取含 `id` 的落库消息。 - `v0.10`（会话消息持久化 — WS 契约，design/16，后端-2）：WS 下行帧新增可选 `sessionId` 字段（能确定会话的事件带上）；新增 **`history` 回放事件**——WS 握手成功即回放主会话最新 50 条（`{type:"history", sessionId:"web", data:{messages:[...]}}`，`thinking` 已剥离）。`stream_end`/`stream_chunk`/会话编排状态帧（`session_busy` 等）在能确定 session 时带 `sessionId`；`message.delta` 等结构化流帧因 `model.StreamEvent` 无 session 上下文不带（即"若知道就带上"边界）。 - `v0.9`（会话消息持久化 + 主/次级会话，design/16，后端-1）：新增 `POST /api/v1/sessions`（新建次级会话）、`GET /api/v1/sessions/{id}/messages?limit=&before=`（游标热分页拉历史，时间倒序）、`DELETE /api/v1/sessions/{id}`（删次级会话及其消息）；`GET /api/v1/sessions` 在注入 `HistoryStore` 后改返回持久化会话元数据数组（主 + 次级，`type` 区分），未注入时保持原渠道会话列表向后兼容。读消息/写会话为 owner/admin（用户侧私有历史，thinking 不达 viewer）；会话元数据列表保持任意认证角色可读。 - `v0.8`（管理面板写操作 CRUD）：补齐管理面板**写端点**（见 §2「管理面板写操作契约」，均 owner-only + 审计）——`POST/PUT /admin/personas`（真实注册 + 持久化 `configs/personas/.yaml`）、`POST /admin/users/apikey` + `DELETE /admin/users/apikey/{id}`（真实签发/撤销）、`POST /admin/iot/{id}/control`（真实设备控制）、`POST /admin/automations` + `/run` + `DELETE`（真实注册/触发/移除规则）、`POST /admin/llm`（临时记录）+ `/admin/llm/test`（真实模型调用）、`POST /admin/channels/{open,close,test}`（真实连接/断开 + 结构性身份识别）、`POST /admin/plugins/uninstall`（真实 registry 移除）、`POST /admin/memories/search`（真实检索）、`GET /admin/sessions/{id}/export`。无存储的写（新增用户 / IoT 场景·规则 / 插件安装）如实返回 `501 not_implemented`。 diff --git a/frontend/web/src/lib/ws.ts b/frontend/web/src/lib/ws.ts index dfbd661..674cefd 100644 --- a/frontend/web/src/lib/ws.ts +++ b/frontend/web/src/lib/ws.ts @@ -1,7 +1,7 @@ import { useAppStore } from "../store/useAppStore"; import { useModeStore } from "../store/useModeStore"; import { getAuthToken } from "./api"; -import { MODES, type AgentInfo, type AwakenState, type FlowGraph, type Mode, type ToolUsed, type WsUpEvent } from "./types"; +import { MODES, type AgentInfo, type AwakenState, type ChatMessage, type FlowGraph, type Mode, type ToolUsed, type WsUpEvent } from "./types"; import { now, uid } from "./util"; /** @@ -156,6 +156,28 @@ function parseGraph(payload: unknown): FlowGraph | null { return { id: typeof r.id === "string" ? r.id : "graph", nodes, edges }; } +/** Build a ChatMessage from a `message_persisted` payload (design/16 §3.3/§4.2): + * the backend broadcast a fully-persisted message carrying the backend-generated + * id. mode is validated against the known modes so a bad value never reaches the + * UI; createdAt falls back to now when the wire omits it. */ +function parsePersistedMessage(v: unknown): ChatMessage | null { + const r = toRecord(v); + if (!r) return null; + const id = typeof r.id === "string" && r.id ? r.id : ""; + const role = r.role; + if (!id || (role !== "user" && role !== "assistant" && role !== "system")) return null; + return { + id, + sessionId: typeof r.sessionId === "string" && r.sessionId ? r.sessionId : undefined, + role, + content: typeof r.content === "string" ? r.content : "", + mode: typeof r.mode === "string" && MODES.includes(r.mode as Mode) ? (r.mode as Mode) : undefined, + mood: typeof r.mood === "string" ? r.mood : undefined, + image: typeof r.image === "string" ? r.image : undefined, + createdAt: typeof r.createdAt === "number" ? r.createdAt : now(), + }; +} + function dispatch(ev: { type: string; data: unknown }): void { const s = useAppStore.getState(); const r = toRecord(ev.data); @@ -191,6 +213,17 @@ function dispatch(ev: { type: string; data: unknown }): void { // do NOT end the stream here (stream_end owns that). break; } + case "message_persisted": { + // design/16 §4.2 — a user or assistant message was persisted and broadcast + // with its backend id. Only surface messages for the active session; the + // store dedups by id and adopts the local optimistic bubble (uid → backend + // id), so both the originating and remote browsers converge on one message. + const msg = parsePersistedMessage(r?.message); + if (msg && msg.sessionId === s.activeSessionId) { + s.appendOrUpdateMessage(msg); + } + break; + } case "thinking.start": { pendingThinking = ""; break; diff --git a/frontend/web/src/store/useAppStore.ts b/frontend/web/src/store/useAppStore.ts index 802deb8..46e4336 100644 --- a/frontend/web/src/store/useAppStore.ts +++ b/frontend/web/src/store/useAppStore.ts @@ -61,7 +61,9 @@ export interface AppStore { setMessages: (list: ChatMessage[]) => void; /** Prepend an older page (chronological) onto the active session's messages. */ prependMessages: (list: ChatMessage[]) => void; - /** Append a new message, or merge by id if it already exists (WS/dedup). */ + /** Append a message, or merge by id if it already exists. On message_persisted + * it also adopts a same-role local optimistic bubble (rewriting its uid to the + * backend id) so a message never renders twice (design/16 §4.3 dedup). */ appendOrUpdateMessage: (msg: ChatMessage) => void; addUserMessage: (text: string, mode?: Mode, image?: string) => void; @@ -135,11 +137,48 @@ export const useAppStore = create()( }), appendOrUpdateMessage: (msg) => set((s) => { + // Exact id match → merge. Keeps local streaming/tools/thinking, takes + // the backend's content/mode/mood as truth. const idx = s.messages.findIndex((m) => m.id === msg.id); - if (idx === -1) return { messages: [...s.messages, msg] }; - const messages = s.messages.slice(); - messages[idx] = { ...messages[idx], ...msg }; - return { messages }; + if (idx !== -1) { + const messages = s.messages.slice(); + messages[idx] = { ...messages[idx], ...msg }; + return { messages }; + } + // No exact id: adopt a local optimistic bubble (no sessionId) of the + // same role, so a message the local browser already rendered + // optimistically is never duplicated when message_persisted arrives + // (design/16 §4.3 dedup by id). Prefer a content match, else the + // nearest by createdAt; scan newest-first. + let best = -1; + let bestDist = Number.MAX_SAFE_INTEGER; + for (let i = s.messages.length - 1; i >= 0; i--) { + const m = s.messages[i]; + if (m.sessionId || m.role !== msg.role || m.id === msg.id) continue; + const dist = m.content === msg.content ? 0 : Math.abs(m.createdAt - msg.createdAt); + if (dist < bestDist) { + bestDist = dist; + best = i; + } + } + if (best !== -1) { + const m = s.messages[best]; + const messages = s.messages.slice(); + messages[best] = { + ...m, + id: msg.id, + sessionId: msg.sessionId ?? m.sessionId, + content: msg.content || m.content, + mode: msg.mode ?? m.mode, + mood: msg.mood ?? m.mood, + image: msg.image ?? m.image, + createdAt: msg.createdAt ?? m.createdAt, + }; + return { messages }; + } + // No local bubble (e.g. a remote browser that never optimistically + // rendered this message): append. + return { messages: [...s.messages, msg] }; }), addUserMessage: (text, mode, image) => diff --git a/internal/runtime/history/store.go b/internal/runtime/history/store.go index 76d80af..14b1477 100644 --- a/internal/runtime/history/store.go +++ b/internal/runtime/history/store.go @@ -87,6 +87,9 @@ type SessionMeta struct { type HistoryStore interface { // Append persists msg, assigning an id and timestamp when the caller omitted them. Append(ctx context.Context, msg Message) error + // AppendFull persists msg like Append but returns the complete stored message + // (with the assigned id/timestamp) so a caller can broadcast it for cross-end sync. + AppendFull(ctx context.Context, msg Message) (Message, error) // List returns up to limit messages for sessionID, newest first. before is a // millisecond timestamp cursor: when nonzero, only messages older than before are // returned (inclusive-free, matching the "before" bound on the frontend). @@ -144,12 +147,28 @@ func NewStore(ctx context.Context, db *sql.DB, log *slog.Logger) (*Store, error) } // Append persists msg, assigning a fresh id and a now-ms CreatedAt when the caller left -// either empty/zero. It returns the error wrapped with the operation name. +// either empty/zero. It returns the error wrapped with the operation name. Callers that +// need the generated id/timestamp back should use AppendFull instead. func (s *Store) Append(ctx context.Context, msg Message) error { + _, err := s.insert(ctx, msg) + return err +} + +// AppendFull persists msg like Append but returns the complete stored message — the id +// and CreatedAt the store assigned when the caller left them empty/zero — so a caller can +// broadcast it for cross-end realtime sync (design/16 §4.2) without a second read. +func (s *Store) AppendFull(ctx context.Context, msg Message) (Message, error) { + return s.insert(ctx, msg) +} + +// insert is the shared Append/AppendFull body. It assigns a fresh id and a now-ms +// CreatedAt when the caller left either empty/zero, then inserts the row and returns the +// resulting message. +func (s *Store) insert(ctx context.Context, msg Message) (Message, error) { if msg.ID == "" { id, err := newID() if err != nil { - return fmt.Errorf("history: append id: %w", err) + return Message{}, fmt.Errorf("history: append id: %w", err) } msg.ID = id } @@ -162,9 +181,9 @@ func (s *Store) Append(ctx context.Context, msg Message) error { msg.ID, msg.SessionID, msg.Role, msg.Content, msg.Mode, msg.Mood, msg.Image, msg.Thinking, msg.CreatedAt) if err != nil { - return fmt.Errorf("history: append %s to %s: %w", msg.Role, msg.SessionID, err) + return Message{}, fmt.Errorf("history: append %s to %s: %w", msg.Role, msg.SessionID, err) } - return nil + return msg, nil } // List returns up to limit messages for sessionID, newest first. When before is nonzero diff --git a/internal/runtime/history/store_test.go b/internal/runtime/history/store_test.go index cb66e30..0046bc7 100644 --- a/internal/runtime/history/store_test.go +++ b/internal/runtime/history/store_test.go @@ -76,6 +76,44 @@ func TestAppendAndListNewestFirst(t *testing.T) { } } +func TestAppendFullReturnsID(t *testing.T) { + t.Parallel() + st := newStore(t) + ctx := context.Background() + + full, err := st.AppendFull(ctx, Message{SessionID: MainSessionID, Role: RoleUser, Content: "hi"}) + if err != nil { + t.Fatalf("append full: %v", err) + } + if full.ID == "" { + t.Fatal("AppendFull did not assign an id") + } + if full.CreatedAt == 0 { + t.Fatal("AppendFull did not assign a created_at") + } + if full.SessionID != MainSessionID || full.Role != RoleUser || full.Content != "hi" { + t.Fatalf("AppendFull returned %+v, want the stored message unchanged plus id/created_at", full) + } + + // The message is actually persisted under the returned id. + msgs, err := st.List(ctx, MainSessionID, 0, 50) + if err != nil { + t.Fatalf("list: %v", err) + } + if len(msgs) != 1 || msgs[0].ID != full.ID { + t.Fatalf("list = %+v, want exactly the persisted message id %q", msgs, full.ID) + } + + // A caller-supplied id/timestamp is preserved, not regenerated. + preset, err := st.AppendFull(ctx, Message{ID: "preset-id", SessionID: MainSessionID, Role: RoleAssistant, Content: "ok", CreatedAt: 42}) + if err != nil { + t.Fatalf("append full preset: %v", err) + } + if preset.ID != "preset-id" || preset.CreatedAt != 42 { + t.Fatalf("AppendFull did not preserve preset id/created_at: %+v", preset) + } +} + func TestListPaginationBefore(t *testing.T) { t.Parallel() st := newStore(t) diff --git a/internal/services/gateway/server.go b/internal/services/gateway/server.go index 34c64af..5a2ba3d 100644 --- a/internal/services/gateway/server.go +++ b/internal/services/gateway/server.go @@ -109,6 +109,10 @@ type MemoryStore interface { // endpoints answer with a clear 501. type HistoryStore interface { Append(ctx context.Context, msg history.Message) error + // AppendFull persists msg like Append but returns the complete stored message + // (with the assigned id/timestamp) so the orchestrator can broadcast it for + // cross-end realtime sync (design/16 §4.2). + AppendFull(ctx context.Context, msg history.Message) (history.Message, error) List(ctx context.Context, sessionID string, before int64, limit int) ([]history.Message, error) Recent(ctx context.Context, sessionID string, n int) ([]history.Message, error) DeleteSession(ctx context.Context, sessionID string) error @@ -455,6 +459,15 @@ func newOrchestrator(s *Server) *Orchestrator { Emit: func(sessionID, typ string, data map[string]any) { s.hub.broadcast(wsEvent{Type: typ, SessionID: sessionID, Data: data}) }, + OnPersisted: func(msg history.Message) { + // Broadcast the complete message so every browser of the same account + // appends it without a manual refresh (design/16 §4.2). The frontend + // dedups by the backend-generated id, and thinking is stripped to stay + // consistent with the WS history replay (design/16 §6). + wire := toHistoryMessage(msg) + wire.Thinking = "" + s.hub.broadcast(wsEvent{Type: "message_persisted", SessionID: msg.SessionID, Data: map[string]any{"message": wire}}) + }, Log: s.log, History: s.history, }) diff --git a/internal/services/gateway/session_orchestrator.go b/internal/services/gateway/session_orchestrator.go index 61204de..e145832 100644 --- a/internal/services/gateway/session_orchestrator.go +++ b/internal/services/gateway/session_orchestrator.go @@ -7,6 +7,7 @@ import ( "sync" "time" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/loop" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model" ) @@ -108,6 +109,11 @@ type OrchestratorOptions struct { // When nil every Append is skipped, so the orchestrator runs without storage // (tests / a gateway built without a history store). History HistoryStore + // OnPersisted, when non-nil, is called with the complete stored message right + // after a user or assistant message is written, so the transport can broadcast + // a `message_persisted` event for cross-end realtime sync (design/16 §4.2). + // Dropped when nil (tests / a gateway built without a transport hook). + OnPersisted func(history.Message) } // sessionState is the per-session message state (design/15 §2). Every mutation @@ -115,7 +121,7 @@ type OrchestratorOptions struct { // lock so a slow model call never blocks Submit. type sessionState struct { mode Mode - gen uint64 // generation: bumped on interrupt and on a fresh top-level turn + gen uint64 // generation: bumped on interrupt and on a fresh top-level turn cancel context.CancelFunc pending []pendingMsg tail []model.Message // bounded conversation tail for context continuation @@ -134,14 +140,15 @@ type pendingMsg struct { // actual run off to a goroutine. It is the single owner of "is this session busy" // for the chat path; the loop/task/subagent engines are reused, not re-created. type Orchestrator struct { - mu sync.Mutex - states map[string]*sessionState - run TurnRunner - emit StateSink - log *slog.Logger - history HistoryStore - maxPending int - maxTail int + mu sync.Mutex + states map[string]*sessionState + run TurnRunner + emit StateSink + log *slog.Logger + history HistoryStore + onPersisted func(history.Message) + maxPending int + maxTail int } // NewOrchestrator builds an orchestrator from opts. Run is required: it is the @@ -161,13 +168,14 @@ func NewOrchestrator(opts OrchestratorOptions) *Orchestrator { opts.Log = slog.New(slog.NewTextHandler(io.Discard, nil)) } return &Orchestrator{ - states: make(map[string]*sessionState), - run: opts.Run, - emit: opts.Emit, - log: opts.Log, - history: opts.History, - maxPending: maxPending, - maxTail: maxTail, + states: make(map[string]*sessionState), + run: opts.Run, + emit: opts.Emit, + log: opts.Log, + history: opts.History, + onPersisted: opts.OnPersisted, + maxPending: maxPending, + maxTail: maxTail, } } @@ -414,6 +422,14 @@ func (o *Orchestrator) emitState(sessionID, typ string, data map[string]any) { } } +// notifyPersisted hands a just-persisted message to the transport hook when one is +// wired, so the gateway can broadcast `message_persisted` for cross-end sync. +func (o *Orchestrator) notifyPersisted(msg history.Message) { + if o.onPersisted != nil { + o.onPersisted(msg) + } +} + // boundedTail returns a fresh copy of the last maxTail messages, so continuation // stays incremental and bounded (design/15 §2.2, §4) and never aliases the input // slice or another state's tail. diff --git a/internal/services/gateway/session_orchestrator_test.go b/internal/services/gateway/session_orchestrator_test.go index 218de0a..e94d74f 100644 --- a/internal/services/gateway/session_orchestrator_test.go +++ b/internal/services/gateway/session_orchestrator_test.go @@ -157,11 +157,22 @@ type fakeHistory struct { appends []history.Message } -func (f *fakeHistory) Append(_ context.Context, msg history.Message) error { +func (f *fakeHistory) Append(ctx context.Context, msg history.Message) error { + _, err := f.AppendFull(ctx, msg) + return err +} + +func (f *fakeHistory) AppendFull(_ context.Context, msg history.Message) (history.Message, error) { f.mu.Lock() + defer f.mu.Unlock() + if msg.ID == "" { + msg.ID = fmt.Sprintf("id-%d", len(f.appends)+1) + } + if msg.CreatedAt == 0 { + msg.CreatedAt = int64(len(f.appends) + 1) + } f.appends = append(f.appends, msg) - f.mu.Unlock() - return nil + return msg, nil } func (f *fakeHistory) List(context.Context, string, int64, int) ([]history.Message, error) { @@ -219,6 +230,44 @@ func TestSubmitPersistsUserAndAssistant(t *testing.T) { } } +func TestSubmitPersistedBroadcast(t *testing.T) { + t.Parallel() + run := newGatedRunner(1) + hist := &fakeHistory{} + var mu sync.Mutex + var persisted []history.Message + o := NewOrchestrator(OrchestratorOptions{ + Run: run.Run, + Log: slog.Default(), + History: hist, + OnPersisted: func(m history.Message) { + mu.Lock() + persisted = append(persisted, m) + mu.Unlock() + }, + }) + run.tickets <- struct{}{} + sr, _ := o.SubmitSync(context.Background(), "web", []model.Message{userMsg("你好")}, loop.OwnerEnv()) + if !sr.Accepted || !sr.Immediate { + t.Fatalf("submit = %+v, want Accepted+Immediate", sr) + } + mu.Lock() + got := append([]history.Message(nil), persisted...) + mu.Unlock() + if len(got) != 2 { + t.Fatalf("persisted = %d, want 2 (user + assistant)", len(got)) + } + if got[0].Role != history.RoleUser || got[0].ID == "" || got[0].SessionID != "web" || got[0].Content != "你好" { + t.Fatalf("user persisted = %+v", got[0]) + } + if got[1].Role != history.RoleAssistant || got[1].ID == "" || got[1].SessionID != "web" || got[1].Content == "" { + t.Fatalf("assistant persisted = %+v", got[1]) + } + if got[0].ID == got[1].ID { + t.Fatalf("user and assistant share id %q", got[0].ID) + } +} + func TestSubmitDoesNotPersistRejectedTurn(t *testing.T) { t.Parallel() hist := &fakeHistory{} diff --git a/internal/services/gateway/session_persist.go b/internal/services/gateway/session_persist.go index d507c04..a451c66 100644 --- a/internal/services/gateway/session_persist.go +++ b/internal/services/gateway/session_persist.go @@ -14,9 +14,11 @@ import ( // orchestrator runs unchanged without a store (tests / an unwired gateway). // persistUserMessages writes each RoleUser message of a submitted turn to the -// history store so the raw user stream is durable before the turn resolves. It is -// a no-op when no history store is wired. Rejected turns (empty, pending-full) -// never reach it, so nothing vacuous is persisted. +// 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). 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) { if o.history == nil { return @@ -29,23 +31,30 @@ func (o *Orchestrator) persistUserMessages(ctx context.Context, sessionID string if text == "" && image == "" { continue } - if err := o.history.Append(ctx, history.Message{SessionID: sessionID, Role: history.RoleUser, Content: text, Image: image}); err != nil { + full, err := o.history.AppendFull(ctx, history.Message{SessionID: sessionID, Role: history.RoleUser, Content: text, Image: image}) + if err != nil { o.log.Error("orchestrator persist user", "err", err, "session", sessionID) + continue } + o.notifyPersisted(full) } } // persistAssistantReply writes the final assistant text of a settled turn to the -// history store. 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. +// history store, then notifies the transport (design/16 §3.3, §4.2). 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) { if o.history == nil || res == nil || res.Text == "" { return } - if err := o.history.Append(context.Background(), history.Message{SessionID: sessionID, Role: history.RoleAssistant, Content: res.Text}); err != nil { + full, err := o.history.AppendFull(context.Background(), history.Message{SessionID: sessionID, Role: history.RoleAssistant, Content: res.Text}) + if err != nil { o.log.Error("orchestrator persist assistant", "err", err, "session", sessionID) + return } + o.notifyPersisted(full) } // messageTextImage returns the concatenated text and the first image URL of a