diff --git a/.gitignore b/.gitignore index 1999b4ac7..bbb1e2038 100644 --- a/.gitignore +++ b/.gitignore @@ -134,3 +134,5 @@ apps/stage-tamagotchi/electron.vite.config.*.mjs # Tools - Obsidian .obsidian/ + +docs/ai/context/verifications/ diff --git a/apps/server/src/app.ts b/apps/server/src/app.ts index c3caa5ade..72a91526b 100644 --- a/apps/server/src/app.ts +++ b/apps/server/src/app.ts @@ -63,6 +63,7 @@ import { createRequestLogService } from './services/request-log' import { createStripeService } from './services/stripe' import { createUserDeletionService } from './services/user-deletion' import { ApiError, createInternalError, createUnauthorizedError } from './utils/error' +import { nanoid } from './utils/id' import { getTrustedOrigin } from './utils/origin' interface AppDeps { @@ -113,7 +114,12 @@ export async function buildApp(deps: AppDeps) { // WebSocket setup — must be registered BEFORE bodyLimit middleware const { injectWebSocket, upgradeWebSocket } = createNodeWebSocket({ app }) - const chatWsSetup = createChatWsHandlers(deps.chatService, deps.redis, deps.otel?.engagement ?? null) + // Per-process stable id used by the chat-ws sub callback to skip echoes of + // its own publishes. Falls back to a random nanoid when ops do not provide + // SERVER_INSTANCE_ID, which is fine because we only need uniqueness across + // simultaneously-running api instances, not across restarts. + const instanceId = process.env.SERVER_INSTANCE_ID || nanoid() + const chatWsSetup = createChatWsHandlers(deps.chatService, deps.redis, instanceId, deps.otel?.engagement ?? null) app.get('/ws/chat', upgradeWebSocket(async (c) => { const token = c.req.query('token') diff --git a/apps/server/src/routes/chat-ws/index.ts b/apps/server/src/routes/chat-ws/index.ts index ad115babe..737800a64 100644 --- a/apps/server/src/routes/chat-ws/index.ts +++ b/apps/server/src/routes/chat-ws/index.ts @@ -52,6 +52,7 @@ function broadcastToLocalDevices(userId: string, excludeCtx: HonoWsInvocableEven export function createChatWsHandlers( chatService: ChatService, redis: Redis, + instanceId: string, metrics?: EngagementMetrics | null, ) { // TODO: Separate connection lifecycle, cross-instance broadcast, and RPC orchestration into smaller modules. @@ -62,8 +63,16 @@ export function createChatWsHandlers( sub.on('message', (_channel: string, message: string) => { try { const data = parseChatBroadcastMessage(message) - // Deliver to all local connections of this user (no excludeCtx since the - // sender is on a different instance) + // Skip messages we ourselves published. ioredis pub/sub delivers to + // every subscriber, including the publishing connection — without + // this filter the publisher's local peers would receive each message + // twice (once via in-process broadcastToLocalDevices, once via the + // sub callback) and the sender's own ctx would receive an unwanted + // echo. + if (data.originInstanceId === instanceId) + return + // Cross-instance delivery: hand off to local peers of this user. + // No excludeCtx because the sender lives on a different instance. broadcastToLocalDevices(data.userId, null, newMessages, data.payload) } catch (err) { @@ -92,7 +101,7 @@ export function createChatWsHandlers( /** Publish a broadcast message so other instances can deliver it. */ function publishBroadcast(userId: string, payload: Parameters[1]) { const channel = userChatBroadcastRedisKey(userId) - const message = createChatBroadcastMessage(userId, payload) + const message = createChatBroadcastMessage(userId, payload, instanceId) redis.publish(channel, JSON.stringify(message)).catch((err) => { log.withError(err).error('Failed to publish broadcast message') }) diff --git a/apps/server/src/utils/chat-broadcast.ts b/apps/server/src/utils/chat-broadcast.ts index fbb4d91fe..98d851a37 100644 --- a/apps/server/src/utils/chat-broadcast.ts +++ b/apps/server/src/utils/chat-broadcast.ts @@ -8,6 +8,16 @@ export interface ChatBroadcastPayload { export interface ChatBroadcastMessage { userId: string payload: ChatBroadcastPayload + /** + * Stable identifier of the api instance that published this broadcast. + * + * The subscribing instance compares it with its own instance id and skips + * delivery to local peers when they match — the publisher already + * delivered locally via `broadcastToLocalDevices` and re-delivering would + * echo every message twice on the originating instance (and once extra on + * the sender's own ctx). + */ + originInstanceId: string } function assertNonEmptyString(value: unknown, fieldName: string): string { @@ -24,9 +34,26 @@ function assertFiniteNumber(value: unknown, fieldName: string): number { return value } +/** + * Build a normalized chat broadcast message ready for `redis.publish`. + * + * Use when: + * - The chat-ws route has just persisted new messages and needs to fan them + * out to other api instances over the user's pub/sub channel. + * + * Expects: + * - `originInstanceId` is the publisher's stable instance id (env or nanoid + * fallback). It must be non-empty so the echo-skip filter on the + * subscriber side is reliable. + * + * Returns: + * - The validated message object; callers `JSON.stringify` it before + * `redis.publish`. + */ export function createChatBroadcastMessage( userId: string, payload: ChatBroadcastPayload, + originInstanceId: string, ): ChatBroadcastMessage { return { userId: assertNonEmptyString(userId, 'chat broadcast userId'), @@ -36,9 +63,24 @@ export function createChatBroadcastMessage( fromSeq: assertFiniteNumber(payload.fromSeq, 'chat broadcast payload.fromSeq'), toSeq: assertFiniteNumber(payload.toSeq, 'chat broadcast payload.toSeq'), }, + originInstanceId: assertNonEmptyString(originInstanceId, 'chat broadcast originInstanceId'), } } +/** + * Parse a raw redis pub/sub message back into a validated broadcast message. + * + * Use when: + * - A subscribing api instance received a message and needs to decide + * whether to deliver it to local peers. + * + * Expects: + * - The raw message is JSON produced by `createChatBroadcastMessage`. + * + * Returns: + * - A fully validated `ChatBroadcastMessage`. Throws on schema violations so + * bad messages do not silently corrupt the local registry. + */ export function parseChatBroadcastMessage(raw: string): ChatBroadcastMessage { let parsed: unknown try { @@ -67,6 +109,7 @@ export function parseChatBroadcastMessage(raw: string): ChatBroadcastMessage { fromSeq: assertFiniteNumber(payloadRecord.fromSeq, 'chat broadcast payload.fromSeq'), toSeq: assertFiniteNumber(payloadRecord.toSeq, 'chat broadcast payload.toSeq'), }, + assertNonEmptyString(message.originInstanceId, 'chat broadcast originInstanceId'), ) } diff --git a/apps/server/src/utils/tests/chat-broadcast.test.ts b/apps/server/src/utils/tests/chat-broadcast.test.ts index 56867113d..ee7d1ae9f 100644 --- a/apps/server/src/utils/tests/chat-broadcast.test.ts +++ b/apps/server/src/utils/tests/chat-broadcast.test.ts @@ -9,7 +9,7 @@ describe('chat broadcast utils', () => { messages: [{ id: 'msg-1' }], fromSeq: 3, toSeq: 4, - })).toEqual({ + }, 'instance-A')).toEqual({ userId: 'user-1', payload: { chatId: 'chat-1', @@ -17,6 +17,7 @@ describe('chat broadcast utils', () => { fromSeq: 3, toSeq: 4, }, + originInstanceId: 'instance-A', }) }) @@ -26,7 +27,14 @@ describe('chat broadcast utils', () => { messages: [], fromSeq: 1, toSeq: 1, - })).toThrow('chat broadcast userId must be a non-empty string') + }, 'instance-A')).toThrow('chat broadcast userId must be a non-empty string') + + expect(() => createChatBroadcastMessage('user-1', { + chatId: 'chat-1', + messages: [], + fromSeq: 1, + toSeq: 1, + }, '')).toThrow('chat broadcast originInstanceId must be a non-empty string') }) it('parses a valid broadcast message payload', () => { @@ -38,6 +46,7 @@ describe('chat broadcast utils', () => { fromSeq: 9, toSeq: 12, }, + originInstanceId: 'instance-B', }))).toEqual({ userId: 'user-2', payload: { @@ -46,6 +55,7 @@ describe('chat broadcast utils', () => { fromSeq: 9, toSeq: 12, }, + originInstanceId: 'instance-B', }) }) @@ -59,6 +69,7 @@ describe('chat broadcast utils', () => { fromSeq: 1, toSeq: 1, }, + originInstanceId: 'instance-A', }))).toThrow('chat broadcast userId must be a non-empty string') expect(() => parseChatBroadcastMessage(JSON.stringify({ userId: 'user-1', @@ -68,6 +79,31 @@ describe('chat broadcast utils', () => { fromSeq: 1, toSeq: 1, }, + originInstanceId: 'instance-A', }))).toThrow('chat broadcast payload.messages must be an array') + + // commit 88744602f — chat broadcast loopback echoes + // + // ROOT CAUSE: + // + // Earlier broadcast messages did not carry originInstanceId, so the + // sub callback could not distinguish "this came from another instance" + // from "this came from us via redis loopback". Without it the sender's + // own peers received every message twice (once from in-process fanout, + // once from the sub callback re-delivering the publish). + // + // We fixed this by requiring originInstanceId on every wire message and + // having the sub callback compare against its own instanceId before + // delivering. The parse step rejects messages missing it so a stale + // publisher cannot bypass the dedup. + expect(() => parseChatBroadcastMessage(JSON.stringify({ + userId: 'user-1', + payload: { + chatId: 'chat-1', + messages: [], + fromSeq: 1, + toSeq: 1, + }, + }))).toThrow('chat broadcast originInstanceId must be a non-empty string') }) }) diff --git a/packages/i18n/src/locales/en/stage.yaml b/packages/i18n/src/locales/en/stage.yaml index ad01945c0..5af4fee49 100644 --- a/packages/i18n/src/locales/en/stage.yaml +++ b/packages/i18n/src/locales/en/stage.yaml @@ -7,6 +7,13 @@ chat: core-system: Core System you: You reasoning: Reasoning + sessions: + title: Conversations + new: + New + empty: No conversations yet + cloud-badge: Synced to cloud + new-chat-fallback: New chat + delete: Delete conversation message: Say something... send-mode: title: Send key diff --git a/packages/i18n/src/locales/zh-Hans/stage.yaml b/packages/i18n/src/locales/zh-Hans/stage.yaml index 5c36bd90c..c8ab21b30 100644 --- a/packages/i18n/src/locales/zh-Hans/stage.yaml +++ b/packages/i18n/src/locales/zh-Hans/stage.yaml @@ -7,6 +7,13 @@ chat: core-system: 核心系统 you: 你 reasoning: 思考 + sessions: + title: 会话 + new: + 新建 + empty: 还没有会话 + cloud-badge: 已同步到云端 + new-chat-fallback: 新会话 + delete: 删除会话 message: 说点什么... send-mode: title: 发送键值 diff --git a/packages/stage-layouts/src/components/Layouts/MobileInteractiveArea.vue b/packages/stage-layouts/src/components/Layouts/MobileInteractiveArea.vue index d52f28848..35040119c 100644 --- a/packages/stage-layouts/src/components/Layouts/MobileInteractiveArea.vue +++ b/packages/stage-layouts/src/components/Layouts/MobileInteractiveArea.vue @@ -4,6 +4,7 @@ import type { ChatProvider } from '@xsai-ext/providers/utils' import { useThreeViewControl } from '@proj-airi/stage-ui-three' import { ChatHistory, HearingConfigDialog } from '@proj-airi/stage-ui/components' +import { ChatSessionsDrawer } from '@proj-airi/stage-ui/components/scenarios/chat' import { useAudioAnalyzer } from '@proj-airi/stage-ui/composables' import { useAudioContext } from '@proj-airi/stage-ui/stores/audio' import { useChatOrchestratorStore } from '@proj-airi/stage-ui/stores/chat' @@ -45,6 +46,7 @@ function handleDeleteMessage(index: number) { const messageInput = ref('') const isComposing = ref(false) const backgroundDialogOpen = ref(false) +const sessionsDrawerOpen = ref(false) const screenSafeArea = useScreenSafeArea() const providersStore = useProvidersStore() @@ -173,6 +175,16 @@ onMounted(() => {
+ + {
+ + + + +