fix(web): keep member filters out of proximity queries

Preserve time bounds for group proximity analysis without pre-filtering the message stream to a single sender.

apps/web-wasm/src/components/session/SessionInsights.test.ts

@@ -0,0 +1,10 @@
+import assert from 'node:assert/strict'
+import { readFileSync } from 'node:fs'
+import test from 'node:test'
+
+const source = readFileSync(new URL('./SessionInsights.vue', import.meta.url), 'utf8')
+
+test('keeps member filter invalidation scoped to the active insight subtree', () => {
+  assert.equal(source.includes('abortAnalyticsRequests'), false)
+  assert.equal(source.match(/:key="selectedMemberId \?\? 'all'"/g)?.length, 3)
+})

apps/web-wasm/src/components/session/SessionInsights.vue

@@ -74,12 +74,14 @@ const viewTimeFilter = computed(() => ({ ...props.timeFilter, memberId: selected
         />
         <TypeAnalysisView
           v-else-if="activeSubTab === 'type-analysis'"
+          :key="selectedMemberId ?? 'all'"
           :session-id="props.sessionId"
           :session-name="props.sessionName"
           :time-filter="viewTimeFilter"
         />
         <TimeAnalysisView
           v-else-if="activeSubTab === 'time-analysis'"
+          :key="selectedMemberId ?? 'all'"
           :session-id="props.sessionId"
           :session-name="props.sessionName"
           :time-filter="viewTimeFilter"
@@ -100,6 +102,7 @@ const viewTimeFilter = computed(() => ({ ...props.timeFilter, memberId: selected
         />
         <GroupRelationshipInsights
           v-else-if="activeSubTab === 'group-relationships'"
+          :key="selectedMemberId ?? 'all'"
           :session-id="props.sessionId"
           :time-filter="viewTimeFilter"
         />

apps/web-wasm/src/components/session/insights/GroupRelationshipInsights.vue

@@ -6,6 +6,7 @@ import ClusterView from '@/components/analysis/relationships/ClusterView.vue'
 import InteractionView from '@/components/analysis/relationships/InteractionView.vue'
 import MentionRankingView from '@/components/analysis/relationships/MentionRankingView.vue'
 import type { TimeFilter } from '@openchatlab/shared-types'
+import { getProximityTimeFilter } from './group-relationship-filter'
 
 const props = defineProps<{
   sessionId: string
@@ -14,6 +15,7 @@ const props = defineProps<{
 
 const { t } = useI18n()
 const activeSubTab = ref('mention-graph')
+const proximityTimeFilter = computed(() => getProximityTimeFilter(props.timeFilter))
 const subTabs = computed(() => [
   {
     id: 'mention-graph',
@@ -52,7 +54,7 @@ const subTabs = computed(() => [
         <ClusterView
           v-else-if="activeSubTab === 'proximity'"
           :session-id="props.sessionId"
-          :time-filter="props.timeFilter"
+          :time-filter="proximityTimeFilter"
         />
       </Transition>
     </div>

apps/web-wasm/src/components/session/insights/group-relationship-filter.test.ts

@@ -0,0 +1,17 @@
+import assert from 'node:assert/strict'
+import { describe, it } from 'node:test'
+import { getProximityTimeFilter } from './group-relationship-filter'
+
+describe('getProximityTimeFilter', () => {
+  it('keeps time bounds but removes the member filter', () => {
+    assert.deepEqual(getProximityTimeFilter({ startTs: 10, endTs: 20, memberId: 3 }), {
+      startTs: 10,
+      endTs: 20,
+    })
+  })
+
+  it('returns no filter when only a member is selected', () => {
+    assert.equal(getProximityTimeFilter({ memberId: 3 }), undefined)
+    assert.equal(getProximityTimeFilter(), undefined)
+  })
+})

apps/web-wasm/src/components/session/insights/group-relationship-filter.ts

@@ -0,0 +1,11 @@
+import type { TimeFilter } from '@openchatlab/shared-types'
+
+export function getProximityTimeFilter(filter?: TimeFilter): TimeFilter | undefined {
+  if (!filter) return undefined
+
+  const timeFilter: TimeFilter = {}
+  if (filter.startTs !== undefined) timeFilter.startTs = filter.startTs
+  if (filter.endTs !== undefined) timeFilter.endTs = filter.endTs
+
+  return timeFilter.startTs === undefined && timeFilter.endTs === undefined ? undefined : timeFilter
+}

src/services/data/browser.test.ts

@@ -6,8 +6,22 @@ import type {
   WebRuntimeTaskResult,
   WebRuntimeTaskType,
 } from '@openchatlab/web-runtime'
+import { abortAnalyticsRequests } from '../utils/http'
 import { createBrowserDataAdapter } from './browser'
 
+interface Deferred<T> {
+  promise: Promise<T>
+  resolve: (value: T) => void
+}
+
+function createDeferred<T>(): Deferred<T> {
+  let resolve!: (value: T) => void
+  const promise = new Promise<T>((resolvePromise) => {
+    resolve = resolvePromise
+  })
+  return { promise, resolve }
+}
+
 describe('BrowserDataAdapter', () => {
   it('maps catalog sessions and forwards session mutations through RPC', async () => {
     const requests: Array<{ type: WebRuntimeTaskType; payload: unknown }> = []
@@ -232,6 +246,48 @@ describe('BrowserDataAdapter', () => {
     )
   })
 
+  it('ignores stale Worker analysis results after the analytics epoch changes', async () => {
+    const loads = [createDeferred<unknown>(), createDeferred<unknown>()]
+    const requestOptions: Array<RpcRequestOptions | undefined> = []
+    let loadIndex = 0
+    const rpc = {
+      request<T extends WebRuntimeTaskType>(
+        type: T,
+        _payload: WebRuntimeTaskPayload<T>,
+        options?: RpcRequestOptions
+      ): Promise<WebRuntimeTaskResult<T>> {
+        assert.equal(type, 'analysis.hourly')
+        requestOptions.push(options)
+        return loads[loadIndex++]!.promise as Promise<WebRuntimeTaskResult<T>>
+      },
+      dispose: () => undefined,
+    }
+    const adapter = createBrowserDataAdapter(rpc)
+
+    let staleSettled = false
+    void adapter.getHourlyActivity('session-one', { startTs: 1 }).then(
+      () => {
+        staleSettled = true
+      },
+      () => {
+        staleSettled = true
+      }
+    )
+
+    abortAnalyticsRequests()
+    const latest = adapter.getHourlyActivity('session-one', { startTs: 2 })
+
+    assert.equal(requestOptions[0]?.signal?.aborted, true)
+    assert.equal(requestOptions[1]?.signal?.aborted, false)
+
+    loads[0]!.resolve([{ hour: 1, messageCount: 1 }])
+    await new Promise<void>((resolve) => setImmediate(resolve))
+    assert.equal(staleSettled, false)
+
+    loads[1]!.resolve([{ hour: 2, messageCount: 2 }])
+    assert.deepEqual(await latest, [{ hour: 2, messageCount: 2 }])
+  })
+
   it('rejects unsupported DataAdapter capabilities instead of returning fake data', async () => {
     const rpc = {
       request: () => Promise.reject(new Error('not used')),

src/services/data/browser.ts

@@ -1,4 +1,10 @@
-import { sessionDatabaseFilename, type BrowserSessionCatalogItem } from '@openchatlab/web-runtime'
+import {
+  sessionDatabaseFilename,
+  type BrowserSessionCatalogItem,
+  type WebRuntimeTaskPayload,
+  type WebRuntimeTaskResult,
+  type WebRuntimeTaskType,
+} from '@openchatlab/web-runtime'
 import type { AnalysisSession, MessageType } from '@/types/base'
 import type {
   ClusterGraphData,
@@ -16,6 +22,7 @@ import type { LanguagePreferenceResult } from '@/types/quotes/languagePreference
 import type { MemberMonthlyTrend, WordFrequencyParams, WordFrequencyResult } from '@openchatlab/core'
 import type { TimeFilter } from '@openchatlab/shared-types'
 import type { BrowserRuntimeRpcPort } from '../browser-runtime/types'
+import { withAnalyticsRequestEpoch } from '../utils/http'
 import type { DataAdapter } from './types'
 
 type BrowserSessionDataAdapter = Pick<
@@ -50,6 +57,13 @@ type BrowserSessionDataAdapter = Pick<
 export class BrowserDataAdapter implements BrowserSessionDataAdapter {
   constructor(private readonly rpc: BrowserRuntimeRpcPort) {}
 
+  private requestAnalysis<T extends WebRuntimeTaskType>(
+    type: T,
+    payload: WebRuntimeTaskPayload<T>
+  ): Promise<WebRuntimeTaskResult<T>> {
+    return withAnalyticsRequestEpoch((signal) => this.rpc.request(type, payload, { signal }))
+  }
+
   async getSessions(): Promise<AnalysisSession[]> {
     return (await this.rpc.request('session.list', undefined)).map(mapSession)
   }
@@ -68,15 +82,15 @@ export class BrowserDataAdapter implements BrowserSessionDataAdapter {
   }
 
   getHourlyActivity(sessionId: string, filter?: TimeFilter): Promise<HourlyActivity[]> {
-    return this.rpc.request('analysis.hourly', { sessionId, filter })
+    return this.requestAnalysis('analysis.hourly', { sessionId, filter })
   }
 
   getDailyActivity(sessionId: string, filter?: TimeFilter): Promise<DailyActivity[]> {
-    return this.rpc.request('analysis.daily', { sessionId, filter })
+    return this.requestAnalysis('analysis.daily', { sessionId, filter })
   }
 
   getWeekdayActivity(sessionId: string, filter?: TimeFilter): Promise<WeekdayActivity[]> {
-    return this.rpc.request('analysis.weekday', { sessionId, filter })
+    return this.requestAnalysis('analysis.weekday', { sessionId, filter })
   }
 
   getTimeRange(sessionId: string): Promise<{ start: number; end: number } | null> {
@@ -88,45 +102,45 @@ export class BrowserDataAdapter implements BrowserSessionDataAdapter {
   }
 
   getMemberActivity(sessionId: string, filter?: TimeFilter): Promise<MemberActivity[]> {
-    return this.rpc.request('analysis.members', { sessionId, filter })
+    return this.requestAnalysis('analysis.members', { sessionId, filter })
   }
 
   async getMessageTypeDistribution(
     sessionId: string,
     filter?: TimeFilter
   ): Promise<Array<{ type: MessageType; count: number }>> {
-    return (await this.rpc.request('analysis.messageTypes', { sessionId, filter })).map((item) => ({
+    return (await this.requestAnalysis('analysis.messageTypes', { sessionId, filter })).map((item) => ({
       type: item.type as MessageType,
       count: item.count,
     }))
   }
 
   getMessageLengthDistribution(sessionId: string, filter?: TimeFilter) {
-    return this.rpc.request('analysis.messageLengths', { sessionId, filter })
+    return this.requestAnalysis('analysis.messageLengths', { sessionId, filter })
   }
 
   getTextStats(sessionId: string, filter?: TimeFilter) {
-    return this.rpc.request('analysis.textStats', { sessionId, filter })
+    return this.requestAnalysis('analysis.textStats', { sessionId, filter })
   }
 
   getLongMessageCount(sessionId: string, filter?: TimeFilter, minLength?: number) {
-    return this.rpc.request('analysis.longMessages', { sessionId, filter, minLength })
+    return this.requestAnalysis('analysis.longMessages', { sessionId, filter, minLength })
   }
 
   getTextLengthPercentiles(sessionId: string, filter?: TimeFilter) {
-    return this.rpc.request('analysis.textPercentiles', { sessionId, filter })
+    return this.requestAnalysis('analysis.textPercentiles', { sessionId, filter })
   }
 
   getMonthlyActivity(sessionId: string, filter?: TimeFilter): Promise<MonthlyActivity[]> {
-    return this.rpc.request('analysis.monthly', { sessionId, filter })
+    return this.requestAnalysis('analysis.monthly', { sessionId, filter })
   }
 
   getYearlyActivity(sessionId: string, filter?: TimeFilter): Promise<Array<{ year: number; messageCount: number }>> {
-    return this.rpc.request('analysis.yearly', { sessionId, filter })
+    return this.requestAnalysis('analysis.yearly', { sessionId, filter })
   }
 
   getMemberMonthlyTrend(sessionId: string, filter?: TimeFilter): Promise<MemberMonthlyTrend[]> {
-    return this.rpc.request('analysis.memberMonthlyTrend', { sessionId, filter })
+    return this.requestAnalysis('analysis.memberMonthlyTrend', { sessionId, filter })
   }
 
   getMembers(sessionId: string): Promise<MemberWithStats[]> {
@@ -134,15 +148,15 @@ export class BrowserDataAdapter implements BrowserSessionDataAdapter {
   }
 
   getMentionAnalysis(sessionId: string, filter?: TimeFilter): Promise<MentionAnalysis> {
-    return this.rpc.request('analysis.mentions', { sessionId, filter })
+    return this.requestAnalysis('analysis.mentions', { sessionId, filter })
   }
 
   getMentionGraph(sessionId: string, filter?: TimeFilter) {
-    return this.rpc.request('analysis.mentionGraph', { sessionId, filter })
+    return this.requestAnalysis('analysis.mentionGraph', { sessionId, filter })
   }
 
   getClusterGraph(sessionId: string, filter?: TimeFilter, options?: ClusterGraphOptions): Promise<ClusterGraphData> {
-    return this.rpc.request('analysis.clusterGraph', { sessionId, filter, options })
+    return this.requestAnalysis('analysis.clusterGraph', { sessionId, filter, options })
   }
 
   getRelationshipStats(
@@ -150,7 +164,7 @@ export class BrowserDataAdapter implements BrowserSessionDataAdapter {
     filter?: TimeFilter,
     options?: { perseveranceThreshold?: number }
   ): Promise<RelationshipStats> {
-    return this.rpc.request('analysis.relationship', { sessionId, filter, options })
+    return this.requestAnalysis('analysis.relationship', { sessionId, filter, options })
   }
 
   async getLanguagePreferenceAnalysis(
@@ -158,7 +172,7 @@ export class BrowserDataAdapter implements BrowserSessionDataAdapter {
     locale: string,
     filter?: TimeFilter
   ): Promise<LanguagePreferenceResult> {
-    return (await this.rpc.request('analysis.languagePreference', {
+    return (await this.requestAnalysis('analysis.languagePreference', {
       sessionId,
       locale,
       filter,
@@ -166,7 +180,7 @@ export class BrowserDataAdapter implements BrowserSessionDataAdapter {
   }
 
   getWordFrequency(sessionId: string, params: Omit<WordFrequencyParams, 'sessionId'>): Promise<WordFrequencyResult> {
-    return this.rpc.request('analysis.wordFrequency', { sessionId, params })
+    return this.requestAnalysis('analysis.wordFrequency', { sessionId, params })
   }
 }
 

src/services/utils/http.ts

@@ -120,7 +120,7 @@ export function abortAnalyticsRequests(): void {
 }
 
 function isAbortError(err: unknown): boolean {
-  return err instanceof DOMException && err.name === 'AbortError'
+  return err instanceof Error && err.name === 'AbortError'
 }
 
 // 已作废的请求永不 settle：避免旧数据覆盖新数据，也避免在调用方触发无意义的错误处理与空态闪烁。
@@ -128,20 +128,30 @@ function neverSettle<T>(): Promise<T> {
   return new Promise<T>(() => {})
 }
 
+/**
+ * Bind an analysis request to the current epoch.
+ *
+ * Besides aborting queued work, the identity check also discards a result when
+ * the underlying runtime cannot interrupt a task that has already started.
+ */
+export function withAnalyticsRequestEpoch<T>(request: (signal: AbortSignal) => Promise<T>): Promise<T> {
+  const controller = _analyticsController
+  return request(controller.signal)
+    .then((result) => (controller === _analyticsController ? result : neverSettle<T>()))
+    .catch((err) => {
+      if (controller !== _analyticsController || isAbortError(err)) return neverSettle<T>()
+      throw err
+    })
+}
+
 /** 绑定当前 epoch 的 GET：被 abortAnalyticsRequests() 取消时静默作废。 */
 export function analyticsGet<T>(path: string): Promise<T> {
-  return get<T>(path, _analyticsController.signal).catch((err) => {
-    if (isAbortError(err)) return neverSettle<T>()
-    throw err
-  })
+  return withAnalyticsRequestEpoch((signal) => get<T>(path, signal))
 }
 
 /** 绑定当前 epoch 的 POST：被 abortAnalyticsRequests() 取消时静默作废。 */
 export function analyticsPost<T>(path: string, body?: unknown): Promise<T> {
-  return post<T>(path, body, _analyticsController.signal).catch((err) => {
-    if (isAbortError(err)) return neverSettle<T>()
-    throw err
-  })
+  return withAnalyticsRequestEpoch((signal) => post<T>(path, body, signal))
 }
 
 export async function del<T = boolean>(path: string): Promise<T> {