From a6acd606e2698db13b4932af12bb76c7100eb495 Mon Sep 17 00:00:00 2001 From: ding113 Date: Sun, 2 Aug 2026 01:32:21 +0800 Subject: [PATCH 1/3] fix: align prefix session monitoring and usage logs --- src/actions/active-sessions.ts | 15 ++- .../_components/usage-logs-table.test.tsx | 22 +++++ .../logs/_components/usage-logs-table.tsx | 21 +++- .../virtualized-logs-table.test.tsx | 12 +++ .../_components/virtualized-logs-table.tsx | 21 +++- src/repository/_shared/usage-log-filters.ts | 5 +- src/repository/activity-stream.ts | 24 +++-- src/repository/usage-logs.ts | 44 ++++++++- .../active-sessions-monitoring.test.ts | 99 +++++++++++++++++++ .../repository/activity-stream-replay.test.ts | 17 +++- .../usage-logs-sessionid-filter.test.ts | 26 +++++ .../usage-logs-sessionid-suggestions.test.ts | 21 ++++ 12 files changed, 306 insertions(+), 21 deletions(-) create mode 100644 tests/unit/actions/active-sessions-monitoring.test.ts diff --git a/src/actions/active-sessions.ts b/src/actions/active-sessions.ts index a88e25056..25188acc6 100644 --- a/src/actions/active-sessions.ts +++ b/src/actions/active-sessions.ts @@ -425,8 +425,13 @@ export async function getAllSessions( if (cached) { logger.debug("[SessionCache] All sessions cache hit"); + const { SessionTracker } = await import("@/lib/session-tracker"); + // 过滤：管理员可查看所有，普通用户只能查看自己的 const filteredCached = isAdmin ? cached : cached.filter((s) => s.userId === currentUserId); + const concurrentCounts = await SessionTracker.getObservedConcurrentCountBatch( + filteredCached.map((s) => s.sessionId) + ); // 分离活跃和非活跃（5 分钟内有请求为活跃） const now = Date.now(); @@ -460,9 +465,10 @@ export async function getAllSessions( s.totalCacheCreationTokens + s.totalCacheReadTokens, costUsd: s.totalCostUsd, - status: "completed", + status: (concurrentCounts.get(s.sessionId) ?? 0) > 0 ? "in_progress" : "completed", durationMs: s.totalDurationMs, requestCount: s.requestCount, + concurrentCount: concurrentCounts.get(s.sessionId) ?? 0, }; if (lastRequestTime >= fiveMinutesAgo) { @@ -520,6 +526,10 @@ export async function getAllSessions( const { aggregateMultipleSessionStats } = await import("@/repository/message"); const sessionsData = await aggregateMultipleSessionStats(allSessionIds); + const concurrentCounts = await SessionTracker.getObservedConcurrentCountBatch( + sessionsData.map((s) => s.sessionId) + ); + // 4. 写入缓存 setActiveSessionsCache(sessionsData, cacheKey); @@ -560,9 +570,10 @@ export async function getAllSessions( s.totalCacheCreationTokens + s.totalCacheReadTokens, costUsd: s.totalCostUsd, - status: "completed", + status: (concurrentCounts.get(s.sessionId) ?? 0) > 0 ? "in_progress" : "completed", durationMs: s.totalDurationMs, requestCount: s.requestCount, + concurrentCount: concurrentCounts.get(s.sessionId) ?? 0, }; if (lastRequestTime >= fiveMinutesAgo) { diff --git a/src/app/[locale]/dashboard/logs/_components/usage-logs-table.test.tsx b/src/app/[locale]/dashboard/logs/_components/usage-logs-table.test.tsx index dfa68d367..0d758d475 100644 --- a/src/app/[locale]/dashboard/logs/_components/usage-logs-table.test.tsx +++ b/src/app/[locale]/dashboard/logs/_components/usage-logs-table.test.tsx @@ -606,6 +606,28 @@ describe("usage-logs-table multiplier badge", () => { }); container.remove(); }); + + test("shows the client session ID and keeps the canonical prefix identity in the tooltip", () => { + const html = renderToStaticMarkup( + {}} + isPending={false} + /> + ); + + expect(html).toContain('data-session-id="client-session-id"'); + expect(html).toContain("client-session-id"); + expect(html).toContain("pfx:scope:fingerprint"); + }); }); describe("usage-logs-table pricing resolution", () => { diff --git a/src/app/[locale]/dashboard/logs/_components/usage-logs-table.tsx b/src/app/[locale]/dashboard/logs/_components/usage-logs-table.tsx index 072cfa481..95588cc7c 100644 --- a/src/app/[locale]/dashboard/logs/_components/usage-logs-table.tsx +++ b/src/app/[locale]/dashboard/logs/_components/usage-logs-table.tsx @@ -186,15 +186,30 @@ export function UsageLogsTable({

- {log.sessionId} + {(log.sourceSessionIds?.length + ? log.sourceSessionIds + : [log.sourceSessionId] + ) + .filter((id): id is string => Boolean(id)) + .map((id) => ( + + {id} + + ))} + {log.sessionId && + !log.sourceSessionIds?.includes(log.sessionId) && ( + + {log.sessionId} + + )}

diff --git a/src/app/[locale]/dashboard/logs/_components/virtualized-logs-table.test.tsx b/src/app/[locale]/dashboard/logs/_components/virtualized-logs-table.test.tsx index b7a6bd093..88e15b216 100644 --- a/src/app/[locale]/dashboard/logs/_components/virtualized-logs-table.test.tsx +++ b/src/app/[locale]/dashboard/logs/_components/virtualized-logs-table.test.tsx @@ -196,6 +196,18 @@ function renderTableWithLog(overrides: Partial) { return renderToStaticMarkup(); } +test("shows the client session ID and canonical prefix identity in the virtualized tooltip", () => { + const html = renderTableWithLog({ + sessionId: "pfx:scope:fingerprint", + sourceSessionId: "client-session-id", + sourceSessionIds: ["client-session-id", "client-session-id-2"], + }); + + expect(html).toContain("client-session-id"); + expect(html).toContain("client-session-id-2"); + expect(html).toContain("pfx:scope:fingerprint"); +}); + function renderCostTooltipWithLog(overrides: Partial) { const html = renderTableWithLog(overrides); const container = document.createElement("div"); diff --git a/src/app/[locale]/dashboard/logs/_components/virtualized-logs-table.tsx b/src/app/[locale]/dashboard/logs/_components/virtualized-logs-table.tsx index 2502e7b77..b4ab45411 100644 --- a/src/app/[locale]/dashboard/logs/_components/virtualized-logs-table.tsx +++ b/src/app/[locale]/dashboard/logs/_components/virtualized-logs-table.tsx @@ -838,15 +838,30 @@ export function VirtualizedLogsTable({

- {log.sessionId} + {(log.sourceSessionIds?.length + ? log.sourceSessionIds + : [log.sourceSessionId] + ) + .filter((id): id is string => Boolean(id)) + .map((id) => ( + + {id} + + ))} + {log.sessionId && + !log.sourceSessionIds?.includes(log.sessionId) && ( + + {log.sessionId} + + )}

diff --git a/src/repository/_shared/usage-log-filters.ts b/src/repository/_shared/usage-log-filters.ts index 274de058b..6edc5a8f0 100644 --- a/src/repository/_shared/usage-log-filters.ts +++ b/src/repository/_shared/usage-log-filters.ts @@ -142,7 +142,10 @@ export function buildUsageLogConditions(filters: UsageLogFilterParams): SQL[] { const trimmedSessionId = filters.sessionId?.trim(); if (trimmedSessionId) { conditions.push( - sql`COALESCE(${messageRequest.sessionIdentity}, ${messageRequest.sessionId}) = ${trimmedSessionId}` + sql`( + COALESCE(${messageRequest.sessionIdentity}, ${messageRequest.sessionId}) = ${trimmedSessionId} + OR ${messageRequest.sessionId} = ${trimmedSessionId} + )` ); } diff --git a/src/repository/activity-stream.ts b/src/repository/activity-stream.ts index 9043d8629..a31cde7b4 100644 --- a/src/repository/activity-stream.ts +++ b/src/repository/activity-stream.ts @@ -1,10 +1,14 @@ "use server"; -import { and, desc, eq, inArray, isNull, notInArray, sql } from "drizzle-orm"; +import { and, desc, eq, inArray, isNull, notInArray, or, sql } from "drizzle-orm"; import { db } from "@/drizzle/db"; import { keys as keysTable, messageRequest, providers, users } from "@/drizzle/schema"; import { logger } from "@/lib/logger"; +const messageSessionIdentity = sql< + string | null +>`COALESCE(${messageRequest.sessionIdentity}, ${messageRequest.sessionId})`; + /** * 活动流条目（单个请求记录） */ @@ -63,7 +67,7 @@ export async function findRecentActivityStream(limit = 20): Promise`ROW_NUMBER() OVER (PARTITION BY ${messageRequest.sessionId} ORDER BY ${messageRequest.createdAt} DESC)`, + rowNum: sql`ROW_NUMBER() OVER (PARTITION BY ${messageSessionIdentity} ORDER BY ${messageRequest.createdAt} DESC)`, }) .from(messageRequest) .leftJoin(users, eq(messageRequest.userId, users.id)) @@ -100,7 +104,10 @@ export async function findRecentActivityStream(limit = 20): Promise 0) { - conditions.push(notInArray(messageRequest.sessionId, excludedSessionIds)); + conditions.push( + notInArray(messageSessionIdentity, excludedSessionIds), + notInArray(messageRequest.sessionId, excludedSessionIds) + ); } const recentRequests = await db .select({ id: messageRequest.id, - sessionId: messageRequest.sessionId, + sessionId: messageSessionIdentity, userName: users.name, userId: messageRequest.userId, keyId: keysTable.id, diff --git a/src/repository/usage-logs.ts b/src/repository/usage-logs.ts index 14acf9e44..752ac65f0 100644 --- a/src/repository/usage-logs.ts +++ b/src/repository/usage-logs.ts @@ -65,12 +65,21 @@ const messageSessionIdentity = sql< const ledgerSessionIdentity = sql< string | null >`COALESCE(${usageLedger.sessionIdentity}, ${usageLedger.sessionId})`; +const messageSourceSessionIds = sql` + ARRAY_AGG(${messageRequest.sessionId}) FILTER (WHERE ${messageRequest.sessionId} IS NOT NULL) + OVER (PARTITION BY ${messageSessionIdentity}) +`; +const ledgerSourceSessionIds = sql` + ARRAY_AGG(${usageLedger.sessionId}) FILTER (WHERE ${usageLedger.sessionId} IS NOT NULL) + OVER (PARTITION BY ${ledgerSessionIdentity}) +`; export interface UsageLogRow { id: number; createdAt: Date | null; sessionId: string | null; // Public Session identity sourceSessionId: string | null; // Physical Session source for request-scoped readback + sourceSessionIds?: string[]; // All physical client session IDs grouped under the public identity requestSequence: number | null; // Request Sequence（Session 内请求序号） userName: string; keyName: string; @@ -218,6 +227,9 @@ export async function findUsageLogsBatch( createdAtRaw: sql`to_char(${messageRequest.createdAt} AT TIME ZONE 'UTC', 'YYYY-MM-DD"T"HH24:MI:SS.US"Z"')`, sessionId: messageSessionIdentity, sourceSessionId: messageRequest.sessionId, + sourceSessionIds: sql< + string[] + >`ARRAY_AGG(${messageRequest.sessionId}) FILTER (WHERE ${messageRequest.sessionId} IS NOT NULL) OVER (PARTITION BY ${messageSessionIdentity})`, requestSequence: messageRequest.requestSequence, userName: users.name, keyName: keysTable.name, @@ -295,6 +307,7 @@ export async function findUsageLogsBatch( return { ...row, + sourceSessionIds: row.sourceSessionIds ? [...new Set(row.sourceSessionIds)] : undefined, requestSequence: row.requestSequence ?? null, totalTokens: totalRowTokens, cacheCreation5mInputTokens: row.cacheCreation5mInputTokens, @@ -340,7 +353,12 @@ export async function findUsageLogsBatch( const trimmedSessionId = filters.sessionId?.trim(); if (trimmedSessionId) { - ledgerConditions.push(eq(ledgerSessionIdentity, trimmedSessionId)); + ledgerConditions.push( + sql`( + ${ledgerSessionIdentity} = ${trimmedSessionId} + OR ${usageLedger.sessionId} = ${trimmedSessionId} + )` + ); } if (filters.startTime !== undefined) { @@ -404,6 +422,7 @@ export async function findUsageLogsBatch( createdAtRaw: sql`to_char(${usageLedger.createdAt} AT TIME ZONE 'UTC', 'YYYY-MM-DD"T"HH24:MI:SS.US"Z"')`, sessionId: ledgerSessionIdentity, sourceSessionId: usageLedger.sessionId, + sourceSessionIds: ledgerSourceSessionIds, userId: usageLedger.userId, userName: users.name, key: usageLedger.key, @@ -462,6 +481,7 @@ export async function findUsageLogsBatch( createdAt: row.createdAt, sessionId: row.sessionId, sourceSessionId: row.sourceSessionId, + sourceSessionIds: row.sourceSessionIds ? [...new Set(row.sourceSessionIds)] : undefined, requestSequence: null, userName: row.userName ?? `User #${row.userId}`, keyName: row.keyName ?? row.key, @@ -1023,6 +1043,7 @@ function mapUsageLogRowFromMessageResult(row: { createdAt: Date | null; sessionId: string | null; sourceSessionId: string | null; + sourceSessionIds?: string[]; requestSequence: number | null; userName: string; keyName: string; @@ -1079,6 +1100,7 @@ function mapUsageLogRowFromMessageResult(row: { return { ...row, + sourceSessionIds: row.sourceSessionIds ? [...new Set(row.sourceSessionIds)] : undefined, requestSequence: row.requestSequence ?? null, totalTokens: totalRowTokens, costUsd: row.costUsd?.toString() ?? null, @@ -1098,6 +1120,7 @@ function mapUsageLogRowFromLedgerResult(row: { createdAt: Date | null; sessionId: string | null; sourceSessionId: string | null; + sourceSessionIds?: string[]; userId: number; userName: string | null; key: string; @@ -1138,6 +1161,7 @@ function mapUsageLogRowFromLedgerResult(row: { createdAt: row.createdAt, sessionId: row.sessionId, sourceSessionId: row.sourceSessionId, + sourceSessionIds: row.sourceSessionIds ? [...new Set(row.sourceSessionIds)] : undefined, requestSequence: null, userName: row.userName ?? `User #${row.userId}`, keyName: row.keyName ?? row.key, @@ -1199,6 +1223,7 @@ export async function findReadonlyUsageLogsBatchForKey( createdAtRaw: sql`to_char(${messageRequest.createdAt} AT TIME ZONE 'UTC', 'YYYY-MM-DD"T"HH24:MI:SS.US"Z"')`, sessionId: messageSessionIdentity, sourceSessionId: messageRequest.sessionId, + sourceSessionIds: messageSourceSessionIds, requestSequence: messageRequest.requestSequence, userName: users.name, keyName: keysTable.name, @@ -1252,6 +1277,7 @@ export async function findReadonlyUsageLogsBatchForKey( createdAtRaw: sql`to_char(${usageLedger.createdAt} AT TIME ZONE 'UTC', 'YYYY-MM-DD"T"HH24:MI:SS.US"Z"')`, sessionId: ledgerSessionIdentity, sourceSessionId: usageLedger.sessionId, + sourceSessionIds: ledgerSourceSessionIds, userId: usageLedger.userId, userName: users.name, key: usageLedger.key, @@ -1459,6 +1485,7 @@ export async function findUsageLogsWithDetails(filters: UsageLogFilters): Promis createdAt: messageRequest.createdAt, sessionId: messageSessionIdentity, // Public Session identity sourceSessionId: messageRequest.sessionId, // Physical Session source + sourceSessionIds: messageSourceSessionIds, requestSequence: messageRequest.requestSequence, // Request Sequence userName: users.name, keyName: keysTable.name, @@ -1646,7 +1673,10 @@ export async function findUsageLogSessionIdSuggestions( EXCLUDE_WARMUP_CONDITION, sql`${messageSessionIdentity} IS NOT NULL`, sql`length(${messageSessionIdentity}) > 0`, - sql`${messageSessionIdentity} LIKE ${pattern} ESCAPE '\\'`, + sql`( + ${messageSessionIdentity} LIKE ${pattern} ESCAPE '\\' + OR ${messageRequest.sessionId} LIKE ${pattern} ESCAPE '\\' + )`, ]; if (userId !== undefined) { @@ -1664,6 +1694,7 @@ export async function findUsageLogSessionIdSuggestions( const baseQuery = db .select({ sessionId: messageSessionIdentity, + sourceSessionId: messageRequest.sessionId, firstSeen: sql`min(${messageRequest.createdAt})`, }) .from(messageRequest); @@ -1675,11 +1706,16 @@ export async function findUsageLogSessionIdSuggestions( const results = await query .where(and(...conditions)) - .groupBy(messageSessionIdentity) + .groupBy(messageSessionIdentity, messageRequest.sessionId) .orderBy(desc(sql`min(${messageRequest.createdAt})`)) .limit(limit); - return results.map((r) => r.sessionId).filter((id): id is string => Boolean(id)); + const suggestions = new Set(); + for (const row of results) { + if (row.sessionId) suggestions.add(row.sessionId); + if (row.sourceSessionId) suggestions.add(row.sourceSessionId); + } + return [...suggestions].slice(0, limit); } /** diff --git a/tests/unit/actions/active-sessions-monitoring.test.ts b/tests/unit/actions/active-sessions-monitoring.test.ts new file mode 100644 index 000000000..7a2bde9a6 --- /dev/null +++ b/tests/unit/actions/active-sessions-monitoring.test.ts @@ -0,0 +1,99 @@ +import { beforeEach, describe, expect, test, vi } from "vitest"; + +const getSessionMock = vi.fn(); +const getActiveSessionsCacheMock = vi.fn(); +const setActiveSessionsCacheMock = vi.fn(); +const getObservedActiveSessionsMock = vi.fn(); +const getObservedConcurrentCountBatchMock = vi.fn(); +const getAllSessionIdsMock = vi.fn(); +const aggregateMultipleSessionStatsMock = vi.fn(); + +vi.mock("@/lib/auth", () => ({ getSession: getSessionMock })); +vi.mock("@/lib/cache/session-cache", () => ({ + getActiveSessionsCache: getActiveSessionsCacheMock, + setActiveSessionsCache: setActiveSessionsCacheMock, + getSessionDetailsCache: vi.fn(() => null), + setSessionDetailsCache: vi.fn(), +})); +vi.mock("@/lib/logger", () => ({ + logger: { + trace: vi.fn(), + debug: vi.fn(), + info: vi.fn(), + warn: vi.fn(), + error: vi.fn(), + }, +})); +vi.mock("@/lib/session-manager", () => ({ + SessionManager: { + getAllSessionIds: getAllSessionIdsMock, + }, +})); +vi.mock("@/lib/session-tracker", () => ({ + SessionTracker: { + getObservedActiveSessions: getObservedActiveSessionsMock, + getObservedConcurrentCountBatch: getObservedConcurrentCountBatchMock, + }, +})); +vi.mock("@/repository/message", () => ({ + aggregateMultipleSessionStats: aggregateMultipleSessionStatsMock, +})); + +const SESSION_STATS = { + sessionId: "pfx:scope:fingerprint", + sessionIdentityKind: "prefix_affinity" as const, + sessionFingerprint: "fingerprint", + requestCount: 2, + totalCostUsd: "0.01", + totalInputTokens: 10, + totalOutputTokens: 5, + totalCacheCreationTokens: 0, + totalCacheReadTokens: 0, + totalDurationMs: 100, + firstRequestAt: new Date("2026-08-02T00:00:00.000Z"), + lastRequestAt: new Date("2026-08-02T00:01:00.000Z"), + providers: [], + models: ["model"], + userName: "user", + userId: 1, + keyName: "key", + keyId: 1, + userAgent: "client", + apiType: "chat", + cacheTtlApplied: null, +}; + +describe("getAllSessions monitoring status", () => { + beforeEach(() => { + vi.clearAllMocks(); + getSessionMock.mockResolvedValue({ user: { id: 1, role: "admin" } }); + getActiveSessionsCacheMock.mockReturnValue(null); + getObservedActiveSessionsMock.mockResolvedValue([SESSION_STATS.sessionId]); + getAllSessionIdsMock.mockResolvedValue([]); + aggregateMultipleSessionStatsMock.mockResolvedValue([SESSION_STATS]); + getObservedConcurrentCountBatchMock.mockResolvedValue(new Map([[SESSION_STATS.sessionId, 1]])); + }); + + test("marks an observed prefix session as in progress", async () => { + const { getAllSessions } = await import("@/actions/active-sessions"); + + const result = await getAllSessions(1, 1, 20); + + expect(result).toEqual( + expect.objectContaining({ + ok: true, + data: expect.objectContaining({ + active: [ + expect.objectContaining({ + sessionId: SESSION_STATS.sessionId, + status: "in_progress", + concurrentCount: 1, + }), + ], + totalActive: 1, + }), + }) + ); + expect(getObservedConcurrentCountBatchMock).toHaveBeenCalledWith([SESSION_STATS.sessionId]); + }); +}); diff --git a/tests/unit/repository/activity-stream-replay.test.ts b/tests/unit/repository/activity-stream-replay.test.ts index 6b88f0de4..636964729 100644 --- a/tests/unit/repository/activity-stream-replay.test.ts +++ b/tests/unit/repository/activity-stream-replay.test.ts @@ -5,7 +5,7 @@ const activeSessionIdsMock = vi.fn<() => Promise>(); vi.mock("@/lib/session-tracker", () => ({ SessionTracker: { - getActiveSessions: activeSessionIdsMock, + getObservedActiveSessions: activeSessionIdsMock, }, })); @@ -81,4 +81,19 @@ describe("activity stream Replay exclusion", () => { expectReplayExcluded(boundary.whereConditions[0]); }); + + it("reads canonical prefix identities from the observed session tracker", async () => { + activeSessionIdsMock.mockResolvedValueOnce(["pfx:scope:fingerprint"]); + const boundary = installDbBoundary([ + { ...REQUEST_ROW, sessionId: "physical-session", rowNum: 1 }, + ]); + const { findRecentActivityStream } = await import("@/repository/activity-stream"); + + await findRecentActivityStream(1); + + const condition = new PgDialect().sqlToQuery(boundary.whereConditions[0] as never); + expect(condition.sql).toContain("session_identity"); + expect(condition.sql).toContain("session_id"); + expect(activeSessionIdsMock).toHaveBeenCalledOnce(); + }); }); diff --git a/tests/unit/repository/usage-logs-sessionid-filter.test.ts b/tests/unit/repository/usage-logs-sessionid-filter.test.ts index 59b87856a..9300c75eb 100644 --- a/tests/unit/repository/usage-logs-sessionid-filter.test.ts +++ b/tests/unit/repository/usage-logs-sessionid-filter.test.ts @@ -115,6 +115,31 @@ describe("Usage logs sessionId filter", () => { expect(ledgerWhereSql).not.toContain(" abc "); }); + test("findUsageLogsBatch: sessionId should match canonical and physical identities", async () => { + vi.resetModules(); + + const whereArgs: unknown[] = []; + const selectMock = vi.fn(() => createThenableQuery([], whereArgs)); + + vi.doMock("@/drizzle/db", () => ({ + db: { + select: selectMock, + execute: vi.fn(async () => ({ count: 0 })), + }, + })); + vi.doMock("@/lib/ledger-fallback", () => ({ + isLedgerOnlyMode: vi.fn(async () => false), + })); + + const { findUsageLogsBatch } = await import("@/repository/usage-logs"); + await findUsageLogsBatch({ sessionId: "client-session" }); + + const primaryWhereSql = sqlToString(whereArgs[0]).toLowerCase(); + expect(primaryWhereSql).toContain("session_identity"); + expect(primaryWhereSql).toContain("session_id"); + expect(primaryWhereSql).toContain(" or "); + }); + test("findUsageLogsBatch: hasMore 为 true 时缺失 createdAtRaw 应直接报错，避免静默截断", async () => { vi.resetModules(); @@ -125,6 +150,7 @@ describe("Usage logs sessionId filter", () => { createdAt: new Date("2026-03-21T00:00:00Z"), createdAtRaw: null, sessionId: null, + sourceSessionIds: [], requestSequence: null, userName: "u", keyName: "k", diff --git a/tests/unit/repository/usage-logs-sessionid-suggestions.test.ts b/tests/unit/repository/usage-logs-sessionid-suggestions.test.ts index 0114be6cc..1b53d48fd 100644 --- a/tests/unit/repository/usage-logs-sessionid-suggestions.test.ts +++ b/tests/unit/repository/usage-logs-sessionid-suggestions.test.ts @@ -133,6 +133,27 @@ describe("Usage logs sessionId suggestions", () => { expect(limitArgs).toEqual([20]); }); + test("returns both canonical and client session identities", async () => { + vi.resetModules(); + + const selectMock = vi.fn(() => + createThenableQuery([ + { + sessionId: "pfx:scope:fingerprint", + sourceSessionId: "client-session", + firstSeen: new Date("2026-01-01T00:00:00Z"), + }, + ]) + ); + vi.doMock("@/drizzle/db", () => ({ db: { select: selectMock } })); + + const { findUsageLogSessionIdSuggestions } = await import("@/repository/usage-logs"); + await expect(findUsageLogSessionIdSuggestions({ term: "client", limit: 20 })).resolves.toEqual([ + "pfx:scope:fingerprint", + "client-session", + ]); + }); + test("term 含 %/_/\\\\：应按字面量前缀匹配（需转义）", async () => { vi.resetModules(); From cf809808eb2030d581dda5bae9920e01e0875220 Mon Sep 17 00:00:00 2001 From: ding113 Date: Sun, 2 Aug 2026 03:23:26 +0800 Subject: [PATCH 2/3] fix: address prefix session review feedback --- src/actions/active-sessions.ts | 16 +- src/actions/usage-logs.ts | 6 +- .../_components/usage-logs-table.test.tsx | 33 +- .../logs/_components/usage-logs-table.tsx | 20 +- .../virtualized-logs-table.test.tsx | 40 ++- .../_components/virtualized-logs-table.tsx | 34 +- src/lib/api-client/v1/actions/usage-logs.ts | 2 + src/repository/activity-stream.ts | 5 +- src/repository/usage-logs.ts | 308 ++++++++++++++---- .../active-sessions-monitoring.test.ts | 63 +++- .../usage-logs-export-retry-count.test.ts | 12 + .../repository/activity-stream-replay.test.ts | 23 +- .../usage-logs-replay-projection.test.ts | 9 +- .../usage-logs-sessionid-filter.test.ts | 169 ++++++++++ .../usage-logs-sessionid-suggestions.test.ts | 79 ++++- 15 files changed, 693 insertions(+), 126 deletions(-) diff --git a/src/actions/active-sessions.ts b/src/actions/active-sessions.ts index 25188acc6..215d974d9 100644 --- a/src/actions/active-sessions.ts +++ b/src/actions/active-sessions.ts @@ -442,6 +442,7 @@ export async function getAllSessions( for (const s of filteredCached) { const lastRequestTime = s.lastRequestAt ? new Date(s.lastRequestAt).getTime() : 0; + const concurrentCount = concurrentCounts.get(s.sessionId) ?? 0; const sessionInfo: ActiveSessionInfo = { sessionId: s.sessionId, sessionIdentityKind: s.sessionIdentityKind, @@ -465,13 +466,14 @@ export async function getAllSessions( s.totalCacheCreationTokens + s.totalCacheReadTokens, costUsd: s.totalCostUsd, - status: (concurrentCounts.get(s.sessionId) ?? 0) > 0 ? "in_progress" : "completed", + status: concurrentCount > 0 ? "in_progress" : "completed", durationMs: s.totalDurationMs, requestCount: s.requestCount, - concurrentCount: concurrentCounts.get(s.sessionId) ?? 0, + concurrentCount, }; - if (lastRequestTime >= fiveMinutesAgo) { + const isConcurrent = concurrentCount > 0; + if (isConcurrent || lastRequestTime >= fiveMinutesAgo) { active.push(sessionInfo); } else { inactive.push(sessionInfo); @@ -547,6 +549,7 @@ export async function getAllSessions( for (const s of filteredSessions) { const lastRequestTime = s.lastRequestAt ? new Date(s.lastRequestAt).getTime() : 0; + const concurrentCount = concurrentCounts.get(s.sessionId) ?? 0; const sessionInfo: ActiveSessionInfo = { sessionId: s.sessionId, sessionIdentityKind: s.sessionIdentityKind, @@ -570,13 +573,14 @@ export async function getAllSessions( s.totalCacheCreationTokens + s.totalCacheReadTokens, costUsd: s.totalCostUsd, - status: (concurrentCounts.get(s.sessionId) ?? 0) > 0 ? "in_progress" : "completed", + status: concurrentCount > 0 ? "in_progress" : "completed", durationMs: s.totalDurationMs, requestCount: s.requestCount, - concurrentCount: concurrentCounts.get(s.sessionId) ?? 0, + concurrentCount, }; - if (lastRequestTime >= fiveMinutesAgo) { + const isConcurrent = concurrentCount > 0; + if (isConcurrent || lastRequestTime >= fiveMinutesAgo) { active.push(sessionInfo); } else { inactive.push(sessionInfo); diff --git a/src/actions/usage-logs.ts b/src/actions/usage-logs.ts index 026412b97..d51594c26 100644 --- a/src/actions/usage-logs.ts +++ b/src/actions/usage-logs.ts @@ -157,7 +157,10 @@ async function buildUsageLogsExport( progress: Pick ) => Promise | void ): Promise { - const initialResult = await findUsageLogsWithDetails({ ...filters, page: 1, pageSize: 1 }); + const initialResult = await findUsageLogsWithDetails( + { ...filters, page: 1, pageSize: 1 }, + { includeSourceSessionIds: false } + ); let estimatedTotalRows = initialResult.total; if (estimatedTotalRows === 0) { @@ -178,6 +181,7 @@ async function buildUsageLogsExport( ...filters, cursor, limit: USAGE_LOGS_EXPORT_BATCH_SIZE, + includeSourceSessionIds: false, }); if (batch.logs.length > 0) { diff --git a/src/app/[locale]/dashboard/logs/_components/usage-logs-table.test.tsx b/src/app/[locale]/dashboard/logs/_components/usage-logs-table.test.tsx index 0d758d475..fab07462a 100644 --- a/src/app/[locale]/dashboard/logs/_components/usage-logs-table.test.tsx +++ b/src/app/[locale]/dashboard/logs/_components/usage-logs-table.test.tsx @@ -24,7 +24,9 @@ vi.mock("@/components/ui/tooltip", () => ({ TooltipProvider: ({ children }: { children?: ReactNode }) =>
{children}
, Tooltip: ({ children }: { children?: ReactNode }) =>
{children}
, TooltipTrigger: ({ children }: { children?: ReactNode }) =>
{children}
, - TooltipContent: ({ children }: { children?: ReactNode }) =>
{children}
, + TooltipContent: ({ children }: { children?: ReactNode }) => ( +
{children}
+ ), })); vi.mock("@/components/ui/relative-time", () => ({ @@ -628,6 +630,35 @@ describe("usage-logs-table multiplier badge", () => { expect(html).toContain("client-session-id"); expect(html).toContain("pfx:scope:fingerprint"); }); + + test("does not duplicate the canonical identity when it is the source fallback", () => { + const html = renderToStaticMarkup( + {}} + isPending={false} + /> + ); + const container = document.createElement("div"); + container.innerHTML = html; + const tooltip = [...container.querySelectorAll('[data-slot="tooltip-content"]')].find((node) => + node.textContent?.includes("same-session-id") + ); + + expect( + [...(tooltip?.querySelectorAll("span") ?? [])].filter( + (node) => node.textContent === "same-session-id" + ) + ).toHaveLength(1); + }); }); describe("usage-logs-table pricing resolution", () => { diff --git a/src/app/[locale]/dashboard/logs/_components/usage-logs-table.tsx b/src/app/[locale]/dashboard/logs/_components/usage-logs-table.tsx index 95588cc7c..f95911d54 100644 --- a/src/app/[locale]/dashboard/logs/_components/usage-logs-table.tsx +++ b/src/app/[locale]/dashboard/logs/_components/usage-logs-table.tsx @@ -161,6 +161,9 @@ export function UsageLogsTable({ : Number(actualCostMultiplier); const hasCostBadge = multiplier != null && Number.isFinite(multiplier) && multiplier !== 1; + const displayedSourceSessionIds = ( + log.sourceSessionIds?.length ? log.sourceSessionIds : [log.sourceSessionId] + ).filter((id): id is string => Boolean(id)); return (

- {(log.sourceSessionIds?.length - ? log.sourceSessionIds - : [log.sourceSessionId] - ) - .filter((id): id is string => Boolean(id)) - .map((id) => ( - - {id} - - ))} + {displayedSourceSessionIds.map((id) => ( + + {id} + + ))} {log.sessionId && - !log.sourceSessionIds?.includes(log.sessionId) && ( + !displayedSourceSessionIds.includes(log.sessionId) && ( {log.sessionId} diff --git a/src/app/[locale]/dashboard/logs/_components/virtualized-logs-table.test.tsx b/src/app/[locale]/dashboard/logs/_components/virtualized-logs-table.test.tsx index 88e15b216..3d523ed83 100644 --- a/src/app/[locale]/dashboard/logs/_components/virtualized-logs-table.test.tsx +++ b/src/app/[locale]/dashboard/logs/_components/virtualized-logs-table.test.tsx @@ -8,6 +8,7 @@ import type { UsageLogRow } from "@/repository/usage-logs"; import type { RoutingTraceV1 } from "@/types/routing-trace"; let mockLogs: UsageLogRow[] = []; +let mockSourceSessionIdsByIdentity: Record = {}; let mockIsLoading = false; let mockIsError = false; let mockError: unknown = null; @@ -24,7 +25,16 @@ vi.mock("@tanstack/react-query", () => ({ useInfiniteQuery: (options: unknown) => { useInfiniteQuerySpy(options); return { - data: { pages: [{ logs: mockLogs, nextCursor: null, hasMore: false }] }, + data: { + pages: [ + { + logs: mockLogs, + sourceSessionIdsByIdentity: mockSourceSessionIdsByIdentity, + nextCursor: null, + hasMore: false, + }, + ], + }, fetchNextPage: vi.fn(), hasNextPage: mockHasNextPage, isFetchingNextPage: mockIsFetchingNextPage, @@ -192,22 +202,46 @@ function renderTableWithLog(overrides: Partial) { mockHasNextPage = false; mockIsFetchingNextPage = false; mockLogs = [makeLog({ id: 1, ...overrides })]; + mockSourceSessionIdsByIdentity = {}; return renderToStaticMarkup(); } test("shows the client session ID and canonical prefix identity in the virtualized tooltip", () => { - const html = renderTableWithLog({ + renderTableWithLog({ sessionId: "pfx:scope:fingerprint", sourceSessionId: "client-session-id", - sourceSessionIds: ["client-session-id", "client-session-id-2"], }); + mockSourceSessionIdsByIdentity = { + "pfx:scope:fingerprint": ["client-session-id", "client-session-id-2"], + }; + const html = renderToStaticMarkup( + + ); expect(html).toContain("client-session-id"); expect(html).toContain("client-session-id-2"); expect(html).toContain("pfx:scope:fingerprint"); }); +test("does not duplicate the canonical identity when it is the source fallback", () => { + const html = renderTableWithLog({ + sessionId: "same-session-id", + sourceSessionId: "same-session-id", + }); + const container = document.createElement("div"); + container.innerHTML = html; + const tooltip = [...container.querySelectorAll('[data-slot="tooltip-content"]')].find((node) => + node.textContent?.includes("same-session-id") + ); + + expect( + [...(tooltip?.querySelectorAll("span") ?? [])].filter( + (node) => node.textContent === "same-session-id" + ) + ).toHaveLength(1); +}); + function renderCostTooltipWithLog(overrides: Partial) { const html = renderTableWithLog(overrides); const container = document.createElement("div"); diff --git a/src/app/[locale]/dashboard/logs/_components/virtualized-logs-table.tsx b/src/app/[locale]/dashboard/logs/_components/virtualized-logs-table.tsx index b4ab45411..cc2dd282c 100644 --- a/src/app/[locale]/dashboard/logs/_components/virtualized-logs-table.tsx +++ b/src/app/[locale]/dashboard/logs/_components/virtualized-logs-table.tsx @@ -203,6 +203,13 @@ export function VirtualizedLogsTable({ // Flatten all pages into a single array const pages = data?.pages; const allLogs = useMemo(() => pages?.flatMap((page) => page.logs) ?? [], [pages]); + const sourceSessionIdsByIdentity = useMemo>( + () => + Object.fromEntries( + pages?.flatMap((page) => Object.entries(page.sourceSessionIdsByIdentity ?? {})) ?? [] + ), + [pages] + ); const filtersResetKey = useMemo(() => JSON.stringify(filters), [filters]); const previousFiltersResetKeyRef = useRef(filtersResetKey); @@ -781,6 +788,16 @@ export function VirtualizedLogsTable({ } const isNonBilling = isNonBillingEndpoint(log.endpoint); + const mappedSourceSessionIds = log.sessionId + ? sourceSessionIdsByIdentity[log.sessionId] + : undefined; + const displayedSourceSessionIds = ( + mappedSourceSessionIds?.length + ? mappedSourceSessionIds + : log.sourceSessionIds?.length + ? log.sourceSessionIds + : [log.sourceSessionId] + ).filter((id): id is string => Boolean(id)); const _isWarmupSkipped = log.blockedBy === "warmup"; return (

- {(log.sourceSessionIds?.length - ? log.sourceSessionIds - : [log.sourceSessionId] - ) - .filter((id): id is string => Boolean(id)) - .map((id) => ( - - {id} - - ))} + {displayedSourceSessionIds.map((id) => ( + + {id} + + ))} {log.sessionId && - !log.sourceSessionIds?.includes(log.sessionId) && ( + !displayedSourceSessionIds.includes(log.sessionId) && ( {log.sessionId} diff --git a/src/lib/api-client/v1/actions/usage-logs.ts b/src/lib/api-client/v1/actions/usage-logs.ts index 483d5a02e..42a9626c2 100644 --- a/src/lib/api-client/v1/actions/usage-logs.ts +++ b/src/lib/api-client/v1/actions/usage-logs.ts @@ -112,6 +112,7 @@ function toLegacyUsageLogsPage(body: unknown): UsageLogsBatchResult { const page = body as { logs?: UsageLogsBatchResult["logs"]; items?: UsageLogsBatchResult["logs"]; + sourceSessionIdsByIdentity?: UsageLogsBatchResult["sourceSessionIdsByIdentity"]; pageInfo?: { nextCursor?: UsageLogsBatchResult["nextCursor"] | string; hasMore?: boolean; @@ -129,6 +130,7 @@ function toLegacyUsageLogsPage(body: unknown): UsageLogsBatchResult { }; return { logs: page.logs ?? page.items ?? [], + sourceSessionIdsByIdentity: page.sourceSessionIdsByIdentity, nextCursor: normalizeLegacyCursor(page.nextCursor ?? page.pageInfo?.nextCursor), hasMore: page.hasMore ?? page.pageInfo?.hasMore ?? false, ...((page.total ?? page.pageInfo?.total !== undefined) diff --git a/src/repository/activity-stream.ts b/src/repository/activity-stream.ts index a31cde7b4..add860348 100644 --- a/src/repository/activity-stream.ts +++ b/src/repository/activity-stream.ts @@ -154,10 +154,7 @@ export async function findRecentActivityStream(limit = 20): Promise 0) { - conditions.push( - notInArray(messageSessionIdentity, excludedSessionIds), - notInArray(messageRequest.sessionId, excludedSessionIds) - ); + conditions.push(notInArray(messageSessionIdentity, excludedSessionIds)); } const recentRequests = await db diff --git a/src/repository/usage-logs.ts b/src/repository/usage-logs.ts index 752ac65f0..3199b3f8a 100644 --- a/src/repository/usage-logs.ts +++ b/src/repository/usage-logs.ts @@ -1,6 +1,7 @@ import "server-only"; -import { and, desc, eq, gte, isNull, lt, sql } from "drizzle-orm"; +import type { SQL } from "drizzle-orm"; +import { and, desc, eq, gte, inArray, isNull, lt, sql } from "drizzle-orm"; import { db } from "@/drizzle/db"; import { keys as keysTable, messageRequest, providers, usageLedger, users } from "@/drizzle/schema"; import { TTLMap } from "@/lib/cache/ttl-map"; @@ -59,20 +60,153 @@ function buildLedgerUsageLogConditions(replayFilter: UsageLogReplayFilter | unde return conditions; } +function buildLedgerSessionIdCondition(sessionId: string) { + return sql`( + ${ledgerSessionIdentity} = ${sessionId} + OR ${usageLedger.sessionId} = ${sessionId} + )`; +} + const messageSessionIdentity = sql< string | null >`COALESCE(${messageRequest.sessionIdentity}, ${messageRequest.sessionId})`; const ledgerSessionIdentity = sql< string | null >`COALESCE(${usageLedger.sessionIdentity}, ${usageLedger.sessionId})`; -const messageSourceSessionIds = sql` - ARRAY_AGG(${messageRequest.sessionId}) FILTER (WHERE ${messageRequest.sessionId} IS NOT NULL) - OVER (PARTITION BY ${messageSessionIdentity}) -`; -const ledgerSourceSessionIds = sql` - ARRAY_AGG(${usageLedger.sessionId}) FILTER (WHERE ${usageLedger.sessionId} IS NOT NULL) - OVER (PARTITION BY ${ledgerSessionIdentity}) -`; + +interface UsageLogSourceSessionScope { + userId?: number; + keyId?: number; + keyString?: string; +} + +async function loadMessageSourceSessionIds( + sessionIds: string[], + scope: UsageLogSourceSessionScope +): Promise> { + if (sessionIds.length === 0) return new Map(); + + const conditions = [ + isNull(messageRequest.deletedAt), + inArray(messageSessionIdentity, sessionIds), + ]; + if (scope.userId !== undefined) conditions.push(eq(messageRequest.userId, scope.userId)); + if (scope.keyId !== undefined) conditions.push(eq(keysTable.id, scope.keyId)); + if (scope.keyString !== undefined) conditions.push(eq(messageRequest.key, scope.keyString)); + + const baseQuery = db + .select({ + sessionId: messageSessionIdentity, + sourceSessionIds: sql` + ARRAY_AGG(DISTINCT ${messageRequest.sessionId}) + FILTER (WHERE ${messageRequest.sessionId} IS NOT NULL) + `, + }) + .from(messageRequest); + const query = + scope.keyId !== undefined + ? baseQuery.innerJoin(keysTable, eq(messageRequest.key, keysTable.key)) + : baseQuery; + const rows = await query.where(and(...conditions)).groupBy(messageSessionIdentity); + + return new Map( + rows + .filter((row): row is { sessionId: string; sourceSessionIds: string[] } => + Boolean(row.sessionId) + ) + .map((row) => [row.sessionId, row.sourceSessionIds ?? []]) + ); +} + +async function loadLedgerSourceSessionIds( + sessionIds: string[], + scope: UsageLogSourceSessionScope +): Promise> { + if (sessionIds.length === 0) return new Map(); + + const conditions = [inArray(ledgerSessionIdentity, sessionIds)]; + if (scope.userId !== undefined) conditions.push(eq(usageLedger.userId, scope.userId)); + if (scope.keyId !== undefined) conditions.push(eq(keysTable.id, scope.keyId)); + if (scope.keyString !== undefined) conditions.push(eq(usageLedger.key, scope.keyString)); + + const baseQuery = db + .select({ + sessionId: ledgerSessionIdentity, + sourceSessionIds: sql` + ARRAY_AGG(DISTINCT ${usageLedger.sessionId}) + FILTER (WHERE ${usageLedger.sessionId} IS NOT NULL) + `, + }) + .from(usageLedger); + const query = + scope.keyId !== undefined + ? baseQuery.innerJoin(keysTable, eq(usageLedger.key, keysTable.key)) + : baseQuery; + const rows = await query.where(and(...conditions)).groupBy(ledgerSessionIdentity); + + return new Map( + rows + .filter((row): row is { sessionId: string; sourceSessionIds: string[] } => + Boolean(row.sessionId) + ) + .map((row) => [row.sessionId, row.sourceSessionIds ?? []]) + ); +} + +async function hydrateUsageLogSourceSessionIds( + logs: UsageLogRow[], + scope: UsageLogSourceSessionScope, + sources: { message: boolean; ledger: boolean } +): Promise { + const sessionIds = [ + ...new Set(logs.map((log) => log.sessionId).filter((id): id is string => Boolean(id))), + ]; + if (sessionIds.length === 0) return logs; + + const [messageSources, ledgerSources] = await Promise.all([ + sources.message ? loadMessageSourceSessionIds(sessionIds, scope) : Promise.resolve(new Map()), + sources.ledger ? loadLedgerSourceSessionIds(sessionIds, scope) : Promise.resolve(new Map()), + ]); + + return logs.map((log) => { + if (!log.sessionId) return log; + const sourceSessionIds = [ + ...new Set([ + ...(messageSources.get(log.sessionId) ?? []), + ...(ledgerSources.get(log.sessionId) ?? []), + ]), + ]; + return sourceSessionIds.length > 0 ? { ...log, sourceSessionIds } : log; + }); +} + +async function loadUsageLogSourceSessionIdsByIdentity( + logs: UsageLogRow[], + scope: UsageLogSourceSessionScope, + sources: { message: boolean; ledger: boolean } +): Promise> { + const sessionIds = [ + ...new Set(logs.map((log) => log.sessionId).filter((id): id is string => Boolean(id))), + ]; + if (sessionIds.length === 0) return {}; + + const [messageSources, ledgerSources] = await Promise.all([ + sources.message ? loadMessageSourceSessionIds(sessionIds, scope) : Promise.resolve(new Map()), + sources.ledger ? loadLedgerSourceSessionIds(sessionIds, scope) : Promise.resolve(new Map()), + ]); + + const sourceIdsByIdentity: Array<[string, string[]]> = []; + for (const sessionId of sessionIds) { + const sourceSessionIds = [ + ...new Set([ + ...(messageSources.get(sessionId) ?? []), + ...(ledgerSources.get(sessionId) ?? []), + ]), + ]; + if (sourceSessionIds.length > 0) sourceIdsByIdentity.push([sessionId, sourceSessionIds]); + } + return Object.fromEntries(sourceIdsByIdentity); +} export interface UsageLogRow { id: number; @@ -170,6 +304,7 @@ export interface UsageLogsPaginatedResult { */ export interface UsageLogsBatchResult { logs: UsageLogRow[]; + sourceSessionIdsByIdentity?: Record; nextCursor: { createdAt: string; id: number } | null; hasMore: boolean; } @@ -180,6 +315,8 @@ export interface UsageLogsBatchResult { export interface UsageLogBatchFilters extends Omit { cursor?: { createdAt: string; id: number }; limit?: number; + /** Export callers can skip the UI-only source identity hydration query. */ + includeSourceSessionIds?: boolean; } /** @@ -189,7 +326,7 @@ export interface UsageLogBatchFilters extends Omit { - const { userId, keyId, providerId, cursor, limit = 50 } = filters; + const { userId, keyId, providerId, cursor, limit = 50, includeSourceSessionIds = true } = filters; const safeLimit = Math.min(100, Math.max(1, limit)); // Build query conditions @@ -227,9 +364,6 @@ export async function findUsageLogsBatch( createdAtRaw: sql`to_char(${messageRequest.createdAt} AT TIME ZONE 'UTC', 'YYYY-MM-DD"T"HH24:MI:SS.US"Z"')`, sessionId: messageSessionIdentity, sourceSessionId: messageRequest.sessionId, - sourceSessionIds: sql< - string[] - >`ARRAY_AGG(${messageRequest.sessionId}) FILTER (WHERE ${messageRequest.sessionId} IS NOT NULL) OVER (PARTITION BY ${messageSessionIdentity})`, requestSequence: messageRequest.requestSequence, userName: users.name, keyName: keysTable.name, @@ -307,7 +441,6 @@ export async function findUsageLogsBatch( return { ...row, - sourceSessionIds: row.sourceSessionIds ? [...new Set(row.sourceSessionIds)] : undefined, requestSequence: row.requestSequence ?? null, totalTokens: totalRowTokens, cacheCreation5mInputTokens: row.cacheCreation5mInputTokens, @@ -326,7 +459,19 @@ export async function findUsageLogsBatch( }); if (logs.length > 0) { - return { logs, nextCursor, hasMore }; + const sourceSessionIdsByIdentity = includeSourceSessionIds + ? await loadUsageLogSourceSessionIdsByIdentity( + logs, + { userId: filters.userId, keyId: filters.keyId }, + { message: true, ledger: false } + ) + : undefined; + return { + logs, + sourceSessionIdsByIdentity, + nextCursor, + hasMore, + }; } if (!(await isLedgerOnlyMode())) { @@ -353,12 +498,7 @@ export async function findUsageLogsBatch( const trimmedSessionId = filters.sessionId?.trim(); if (trimmedSessionId) { - ledgerConditions.push( - sql`( - ${ledgerSessionIdentity} = ${trimmedSessionId} - OR ${usageLedger.sessionId} = ${trimmedSessionId} - )` - ); + ledgerConditions.push(buildLedgerSessionIdCondition(trimmedSessionId)); } if (filters.startTime !== undefined) { @@ -422,7 +562,6 @@ export async function findUsageLogsBatch( createdAtRaw: sql`to_char(${usageLedger.createdAt} AT TIME ZONE 'UTC', 'YYYY-MM-DD"T"HH24:MI:SS.US"Z"')`, sessionId: ledgerSessionIdentity, sourceSessionId: usageLedger.sessionId, - sourceSessionIds: ledgerSourceSessionIds, userId: usageLedger.userId, userName: users.name, key: usageLedger.key, @@ -481,7 +620,6 @@ export async function findUsageLogsBatch( createdAt: row.createdAt, sessionId: row.sessionId, sourceSessionId: row.sourceSessionId, - sourceSessionIds: row.sourceSessionIds ? [...new Set(row.sourceSessionIds)] : undefined, requestSequence: null, userName: row.userName ?? `User #${row.userId}`, keyName: row.keyName ?? row.key, @@ -523,7 +661,18 @@ export async function findUsageLogsBatch( }; }); - return { logs: fallbackLogs, nextCursor: ledgerNextCursor, hasMore: ledgerHasMore }; + return { + logs: fallbackLogs, + sourceSessionIdsByIdentity: includeSourceSessionIds + ? await loadUsageLogSourceSessionIdsByIdentity( + fallbackLogs, + { userId: filters.userId, keyId: filters.keyId }, + { message: false, ledger: true } + ) + : undefined, + nextCursor: ledgerNextCursor, + hasMore: ledgerHasMore, + }; } interface UsageLogSlimFilters { @@ -796,7 +945,7 @@ function buildKeyLedgerConditions( const trimmedSessionId = filters.sessionId?.trim(); if (trimmedSessionId) { - conditions.push(eq(ledgerSessionIdentity, trimmedSessionId)); + conditions.push(buildLedgerSessionIdCondition(trimmedSessionId)); } if (filters.startTime) { @@ -1208,7 +1357,7 @@ function mapUsageLogRowFromLedgerResult(row: { export async function findReadonlyUsageLogsBatchForKey( filters: Omit & { keyString: string } ): Promise { - const { keyString, cursor, limit = 50 } = filters; + const { keyString, cursor, limit = 50, includeSourceSessionIds = true } = filters; const safeLimit = Math.min(100, Math.max(1, limit)); const fetchLimit = safeLimit + 1; @@ -1223,7 +1372,6 @@ export async function findReadonlyUsageLogsBatchForKey( createdAtRaw: sql`to_char(${messageRequest.createdAt} AT TIME ZONE 'UTC', 'YYYY-MM-DD"T"HH24:MI:SS.US"Z"')`, sessionId: messageSessionIdentity, sourceSessionId: messageRequest.sessionId, - sourceSessionIds: messageSourceSessionIds, requestSequence: messageRequest.requestSequence, userName: users.name, keyName: keysTable.name, @@ -1277,7 +1425,6 @@ export async function findReadonlyUsageLogsBatchForKey( createdAtRaw: sql`to_char(${usageLedger.createdAt} AT TIME ZONE 'UTC', 'YYYY-MM-DD"T"HH24:MI:SS.US"Z"')`, sessionId: ledgerSessionIdentity, sourceSessionId: usageLedger.sessionId, - sourceSessionIds: ledgerSourceSessionIds, userId: usageLedger.userId, userName: users.name, key: usageLedger.key, @@ -1336,8 +1483,16 @@ export async function findReadonlyUsageLogsBatchForKey( "findReadonlyUsageLogsBatchForKey" ); + const logs = pageRows.map(({ createdAtRaw: _createdAtRaw, ...log }) => log); return { - logs: pageRows.map(({ createdAtRaw: _createdAtRaw, ...log }) => log), + logs, + sourceSessionIdsByIdentity: includeSourceSessionIds + ? await loadUsageLogSourceSessionIdsByIdentity( + logs, + { keyString }, + { message: messageRows.length > 0, ledger: ledgerRows.length > 0 } + ) + : undefined, nextCursor, hasMore, }; @@ -1417,7 +1572,10 @@ export async function getDistinctEndpointsForKey(keyString: string): Promise { +export async function findUsageLogsWithDetails( + filters: UsageLogFilters, + options: { includeSourceSessionIds?: boolean } = {} +): Promise { const { userId, keyId, providerId, page = 1, pageSize = 50 } = filters; const safePage = page > 0 ? page : 1; @@ -1485,7 +1643,6 @@ export async function findUsageLogsWithDetails(filters: UsageLogFilters): Promis createdAt: messageRequest.createdAt, sessionId: messageSessionIdentity, // Public Session identity sourceSessionId: messageRequest.sessionId, // Physical Session source - sourceSessionIds: messageSourceSessionIds, requestSequence: messageRequest.requestSequence, // Request Sequence userName: users.name, keyName: keysTable.name, @@ -1586,7 +1743,14 @@ export async function findUsageLogsWithDetails(filters: UsageLogFilters): Promis }); return { - logs, + logs: + options.includeSourceSessionIds === false + ? logs + : await hydrateUsageLogSourceSessionIds( + logs, + { userId: filters.userId, keyId: filters.keyId }, + { message: true, ledger: false } + ), total, summary: { totalRequests, @@ -1668,54 +1832,62 @@ export async function findUsageLogSessionIdSuggestions( if (!trimmedTerm) return []; const pattern = `${escapeLike(trimmedTerm)}%`; - const conditions = [ - isNull(messageRequest.deletedAt), - EXCLUDE_WARMUP_CONDITION, - sql`${messageSessionIdentity} IS NOT NULL`, - sql`length(${messageSessionIdentity}) > 0`, - sql`( - ${messageSessionIdentity} LIKE ${pattern} ESCAPE '\\' - OR ${messageRequest.sessionId} LIKE ${pattern} ESCAPE '\\' - )`, - ]; + const sharedConditions = [isNull(messageRequest.deletedAt), EXCLUDE_WARMUP_CONDITION]; if (userId !== undefined) { - conditions.push(eq(messageRequest.userId, userId)); + sharedConditions.push(eq(messageRequest.userId, userId)); } if (keyId !== undefined) { - conditions.push(eq(keysTable.id, keyId)); + sharedConditions.push(eq(keysTable.id, keyId)); } if (providerId !== undefined) { - conditions.push(eq(messageRequest.providerId, providerId)); + sharedConditions.push(eq(messageRequest.providerId, providerId)); } - const baseQuery = db - .select({ - sessionId: messageSessionIdentity, - sourceSessionId: messageRequest.sessionId, - firstSeen: sql`min(${messageRequest.createdAt})`, - }) - .from(messageRequest); + const queryCandidates = async ( + candidate: SQL | typeof messageRequest.sessionId + ) => { + const baseQuery = db + .select({ + sessionId: candidate, + firstSeen: sql`max(${messageRequest.createdAt})`, + }) + .from(messageRequest); + const query = + keyId !== undefined + ? baseQuery.innerJoin(keysTable, eq(messageRequest.key, keysTable.key)) + : baseQuery; - const query = - keyId !== undefined - ? baseQuery.innerJoin(keysTable, eq(messageRequest.key, keysTable.key)) - : baseQuery; + return query + .where( + and( + ...sharedConditions, + sql`${candidate} IS NOT NULL`, + sql`length(${candidate}) > 0`, + sql`${candidate} LIKE ${pattern} ESCAPE '\\'` + ) + ) + .groupBy(candidate) + .orderBy(desc(sql`max(${messageRequest.createdAt})`)) + .limit(limit); + }; - const results = await query - .where(and(...conditions)) - .groupBy(messageSessionIdentity, messageRequest.sessionId) - .orderBy(desc(sql`min(${messageRequest.createdAt})`)) - .limit(limit); - - const suggestions = new Set(); - for (const row of results) { - if (row.sessionId) suggestions.add(row.sessionId); - if (row.sourceSessionId) suggestions.add(row.sourceSessionId); + const [canonicalResults, physicalResults] = await Promise.all([ + queryCandidates(messageSessionIdentity), + queryCandidates(messageRequest.sessionId), + ]); + const bySessionId = new Map(); + for (const row of [...canonicalResults, ...physicalResults]) { + if (!row.sessionId || !row.firstSeen) continue; + const current = bySessionId.get(row.sessionId); + if (!current || row.firstSeen > current) bySessionId.set(row.sessionId, row.firstSeen); } - return [...suggestions].slice(0, limit); + return [...bySessionId.entries()] + .sort((a, b) => b[1].getTime() - a[1].getTime()) + .slice(0, limit) + .map(([sessionId]) => sessionId); } /** @@ -1815,7 +1987,7 @@ export async function findUsageLogsStats( const trimmedSessionId = filters.sessionId?.trim(); if (trimmedSessionId) { - conditions.push(eq(ledgerSessionIdentity, trimmedSessionId)); + conditions.push(buildLedgerSessionIdCondition(trimmedSessionId)); } if (filters.startTime !== undefined) { diff --git a/tests/unit/actions/active-sessions-monitoring.test.ts b/tests/unit/actions/active-sessions-monitoring.test.ts index 7a2bde9a6..8ea721715 100644 --- a/tests/unit/actions/active-sessions-monitoring.test.ts +++ b/tests/unit/actions/active-sessions-monitoring.test.ts @@ -1,4 +1,4 @@ -import { beforeEach, describe, expect, test, vi } from "vitest"; +import { afterEach, beforeEach, describe, expect, test, vi } from "vitest"; const getSessionMock = vi.fn(); const getActiveSessionsCacheMock = vi.fn(); @@ -63,8 +63,12 @@ const SESSION_STATS = { cacheTtlApplied: null, }; +const NOW = new Date("2026-08-02T00:02:00.000Z"); + describe("getAllSessions monitoring status", () => { beforeEach(() => { + vi.useFakeTimers(); + vi.setSystemTime(NOW); vi.clearAllMocks(); getSessionMock.mockResolvedValue({ user: { id: 1, role: "admin" } }); getActiveSessionsCacheMock.mockReturnValue(null); @@ -74,6 +78,10 @@ describe("getAllSessions monitoring status", () => { getObservedConcurrentCountBatchMock.mockResolvedValue(new Map([[SESSION_STATS.sessionId, 1]])); }); + afterEach(() => { + vi.useRealTimers(); + }); + test("marks an observed prefix session as in progress", async () => { const { getAllSessions } = await import("@/actions/active-sessions"); @@ -96,4 +104,57 @@ describe("getAllSessions monitoring status", () => { ); expect(getObservedConcurrentCountBatchMock).toHaveBeenCalledWith([SESSION_STATS.sessionId]); }); + + test("keeps an in-progress session active when its last request is older than five minutes", async () => { + aggregateMultipleSessionStatsMock.mockResolvedValue([ + { + ...SESSION_STATS, + lastRequestAt: new Date("2026-08-01T23:00:00.000Z"), + }, + ]); + + const { getAllSessions } = await import("@/actions/active-sessions"); + const result = await getAllSessions(1, 1, 20); + + expect(result).toEqual( + expect.objectContaining({ + ok: true, + data: expect.objectContaining({ + active: [ + expect.objectContaining({ + sessionId: SESSION_STATS.sessionId, + status: "in_progress", + concurrentCount: 1, + }), + ], + inactive: [], + totalActive: 1, + totalInactive: 0, + }), + }) + ); + }); + + test("keeps a cached in-progress session in the active page", async () => { + getActiveSessionsCacheMock.mockReturnValue([ + { + ...SESSION_STATS, + lastRequestAt: new Date("2026-08-01T23:00:00.000Z"), + }, + ]); + + const { getAllSessions } = await import("@/actions/active-sessions"); + const result = await getAllSessions(1, 1, 20); + + expect(result).toEqual( + expect.objectContaining({ + ok: true, + data: expect.objectContaining({ + active: [expect.objectContaining({ status: "in_progress", concurrentCount: 1 })], + inactive: [], + }), + }) + ); + expect(aggregateMultipleSessionStatsMock).not.toHaveBeenCalled(); + }); }); diff --git a/tests/unit/actions/usage-logs-export-retry-count.test.ts b/tests/unit/actions/usage-logs-export-retry-count.test.ts index 0c97c2587..5aa479632 100644 --- a/tests/unit/actions/usage-logs-export-retry-count.test.ts +++ b/tests/unit/actions/usage-logs-export-retry-count.test.ts @@ -265,6 +265,10 @@ describe("Usage logs CSV export retryCount", () => { const result = await exportUsageLogs({}); expect(result.ok).toBe(true); + expect(findUsageLogsWithDetailsMock).toHaveBeenCalledWith( + expect.objectContaining({ page: 1, pageSize: 1 }), + { includeSourceSessionIds: false } + ); const csv = result.data; const csvNoBom = csv.replace(/^\uFEFF/, ""); const lines = csvNoBom @@ -311,6 +315,14 @@ describe("Usage logs CSV export retryCount", () => { expect(result.ok).toBe(true); expect(findUsageLogsBatchMock).toHaveBeenCalledTimes(2); + expect(findUsageLogsBatchMock).toHaveBeenNthCalledWith( + 1, + expect.objectContaining({ includeSourceSessionIds: false }) + ); + expect(findUsageLogsBatchMock).toHaveBeenNthCalledWith( + 2, + expect.objectContaining({ includeSourceSessionIds: false }) + ); const csvNoBom = result.data.replace(/^\uFEFF/, ""); const lines = csvNoBom diff --git a/tests/unit/repository/activity-stream-replay.test.ts b/tests/unit/repository/activity-stream-replay.test.ts index 636964729..fc76d06f0 100644 --- a/tests/unit/repository/activity-stream-replay.test.ts +++ b/tests/unit/repository/activity-stream-replay.test.ts @@ -9,9 +9,13 @@ vi.mock("@/lib/session-tracker", () => ({ }, })); -function installDbBoundary(rows: readonly unknown[]) { +function installDbBoundary(rows: readonly unknown[] | readonly (readonly unknown[])[]) { const whereConditions: unknown[] = []; + let selectIndex = 0; const select = vi.fn(() => { + const selectedRows = Array.isArray(rows[0]) + ? ((rows as readonly (readonly unknown[])[])[selectIndex++] ?? []) + : (rows as readonly unknown[]); const query = { from: vi.fn(() => query), leftJoin: vi.fn(() => query), @@ -20,7 +24,7 @@ function installDbBoundary(rows: readonly unknown[]) { return query; }), orderBy: vi.fn(() => query), - limit: vi.fn(async () => rows), + limit: vi.fn(async () => selectedRows), }; return query; }); @@ -96,4 +100,19 @@ describe("activity stream Replay exclusion", () => { expect(condition.sql).toContain("session_id"); expect(activeSessionIdsMock).toHaveBeenCalledOnce(); }); + + it("does not exclude fallback rows through the physical session ID", async () => { + activeSessionIdsMock.mockResolvedValueOnce(["pfx:scope:fingerprint"]); + const boundary = installDbBoundary([ + [{ ...REQUEST_ROW, sessionId: "pfx:scope:fingerprint", rowNum: 1 }], + [], + ]); + const { findRecentActivityStream } = await import("@/repository/activity-stream"); + + await findRecentActivityStream(2); + + const condition = new PgDialect().sqlToQuery(boundary.whereConditions[1] as never); + expect(condition.sql.toLowerCase()).toContain("coalesce"); + expect(condition.sql).not.toContain('and "message_request"."session_id" not in'); + }); }); diff --git a/tests/unit/repository/usage-logs-replay-projection.test.ts b/tests/unit/repository/usage-logs-replay-projection.test.ts index 5e2093649..86878c079 100644 --- a/tests/unit/repository/usage-logs-replay-projection.test.ts +++ b/tests/unit/repository/usage-logs-replay-projection.test.ts @@ -8,6 +8,7 @@ function createThenableQuery(result: T, whereArgs?: unknown[]) { query.innerJoin = vi.fn(() => query); query.leftJoin = vi.fn(() => query); query.orderBy = vi.fn(() => query); + query.groupBy = vi.fn(() => query); query.limit = vi.fn(() => query); query.offset = vi.fn(() => query); query.where = vi.fn((condition: unknown) => { @@ -77,7 +78,10 @@ function makeReplayRow(overrides: Record = {}) { describe("findUsageLogsBatch Replay projection", () => { test("projects Replay provenance from message_request", async () => { vi.resetModules(); - const selectMock = vi.fn(() => createThenableQuery([makeReplayRow()])); + const selectMock = vi + .fn() + .mockImplementationOnce(() => createThenableQuery([makeReplayRow()])) + .mockImplementationOnce(() => createThenableQuery([])); vi.doMock("@/drizzle/db", () => ({ db: { select: selectMock } })); vi.doMock("@/lib/ledger-fallback", () => ({ @@ -104,7 +108,8 @@ describe("findUsageLogsBatch Replay projection", () => { const selectMock = vi .fn() .mockImplementationOnce(() => createThenableQuery([])) - .mockImplementationOnce(() => createThenableQuery([ledgerRow])); + .mockImplementationOnce(() => createThenableQuery([ledgerRow])) + .mockImplementationOnce(() => createThenableQuery([])); vi.doMock("@/drizzle/db", () => ({ db: { select: selectMock } })); vi.doMock("@/lib/ledger-fallback", () => ({ diff --git a/tests/unit/repository/usage-logs-sessionid-filter.test.ts b/tests/unit/repository/usage-logs-sessionid-filter.test.ts index 9300c75eb..e3dc1b9e3 100644 --- a/tests/unit/repository/usage-logs-sessionid-filter.test.ts +++ b/tests/unit/repository/usage-logs-sessionid-filter.test.ts @@ -258,6 +258,148 @@ describe("Usage logs sessionId filter", () => { expect(query?.limit).toHaveBeenCalledWith(101); }); + test("findUsageLogsBatch: returns distinct source IDs once per page identity", async () => { + vi.resetModules(); + + const projections: unknown[] = []; + const selectMock = vi.fn((projection: unknown) => { + projections.push(projection); + const call = projections.length; + if (call === 1) { + return createThenableQuery([ + { + id: 101, + createdAt: new Date("2026-03-21T00:00:00Z"), + createdAtRaw: "2026-03-21T00:00:00.000000Z", + sessionId: "pfx:scope:fingerprint", + sourceSessionId: "client-old", + requestSequence: 1, + userName: "u", + keyName: "k", + providerName: "p", + model: "m", + originalModel: "m", + actualResponseModel: null, + endpoint: "/v1/messages", + statusCode: 200, + inputTokens: 1, + outputTokens: 1, + cacheCreationInputTokens: 0, + cacheReadInputTokens: 0, + cacheCreation5mInputTokens: 0, + cacheCreation1hInputTokens: 0, + cacheTtlApplied: null, + costUsd: "0.01", + costMultiplier: null, + groupCostMultiplier: null, + costBreakdown: null, + hedgeLosers: null, + durationMs: 10, + tfftMs: 5, + firstByteMs: 5, + errorMessage: null, + providerChain: null, + routingTrace: null, + blockedBy: null, + blockedReason: null, + isReplay: false, + replaySourceRequestId: null, + userAgent: null, + clientIp: null, + messagesCount: null, + context1mApplied: null, + swapCacheTtlApplied: null, + specialSettings: null, + }, + ]); + } + if (call === 2) { + return createThenableQuery([ + { + sessionId: "pfx:scope:fingerprint", + sourceSessionIds: ["client-new", "client-old"], + }, + ]); + } + return createThenableQuery([]); + }); + + vi.doMock("@/drizzle/db", () => ({ db: { select: selectMock } })); + vi.doMock("@/lib/ledger-fallback", () => ({ + isLedgerOnlyMode: vi.fn(async () => false), + })); + + const { findUsageLogsBatch } = await import("@/repository/usage-logs"); + const result = await findUsageLogsBatch({ cursor: { createdAt: "2026-03-22", id: 102 } }); + + expect(result.logs[0]?.sourceSessionIds).toBeUndefined(); + expect(result.sourceSessionIdsByIdentity).toEqual({ + "pfx:scope:fingerprint": ["client-new", "client-old"], + }); + expect(sqlToString(projections[0]).toLowerCase()).not.toContain("array_agg"); + expect(selectMock).toHaveBeenCalledTimes(2); + }); + + test("findUsageLogsBatch: skips source ID hydration for export callers", async () => { + vi.resetModules(); + + const selectMock = vi.fn(() => + createThenableQuery([ + { + id: 101, + createdAt: new Date("2026-03-21T00:00:00Z"), + createdAtRaw: "2026-03-21T00:00:00.000000Z", + sessionId: "pfx:scope:fingerprint", + sourceSessionId: "client-old", + requestSequence: 1, + userName: "u", + keyName: "k", + providerName: "p", + model: "m", + originalModel: "m", + actualResponseModel: null, + endpoint: "/v1/messages", + statusCode: 200, + inputTokens: 1, + outputTokens: 1, + cacheCreationInputTokens: 0, + cacheReadInputTokens: 0, + cacheCreation5mInputTokens: 0, + cacheCreation1hInputTokens: 0, + cacheTtlApplied: null, + costUsd: "0.01", + costMultiplier: null, + groupCostMultiplier: null, + costBreakdown: null, + hedgeLosers: null, + durationMs: 10, + tfftMs: 5, + firstByteMs: 5, + errorMessage: null, + providerChain: null, + routingTrace: null, + blockedBy: null, + blockedReason: null, + isReplay: false, + replaySourceRequestId: null, + userAgent: null, + clientIp: null, + messagesCount: null, + context1mApplied: null, + swapCacheTtlApplied: null, + specialSettings: null, + }, + ]) + ); + vi.doMock("@/drizzle/db", () => ({ db: { select: selectMock } })); + + const { findUsageLogsBatch } = await import("@/repository/usage-logs"); + const result = await findUsageLogsBatch({ includeSourceSessionIds: false }); + + expect(result.logs[0]?.sourceSessionIds).toBeUndefined(); + expect(selectMock).toHaveBeenCalledTimes(1); + }); + test("findUsageLogsForKeyBatch: hasMore 为 true 时缺失 createdAtRaw 应直接报错，避免静默截断", async () => { vi.resetModules(); @@ -529,4 +671,31 @@ describe("Usage logs sessionId filter", () => { expect(whereSql).toContain("abc"); expect(whereSql).not.toContain(" abc "); }); + + test("ledger filters match canonical and physical session identities", async () => { + vi.resetModules(); + + const whereArgs: unknown[] = []; + const selectMock = vi.fn(() => createThenableQuery([], whereArgs)); + + vi.doMock("@/drizzle/db", () => ({ + db: { + select: selectMock, + execute: vi.fn(async () => ({ count: 0 })), + }, + })); + vi.doMock("@/lib/ledger-fallback", () => ({ + isLedgerOnlyMode: vi.fn(async () => true), + })); + + const { findUsageLogsForKeyBatch, findUsageLogsStats } = await import( + "@/repository/usage-logs" + ); + await findUsageLogsForKeyBatch({ keyString: "key", sessionId: "client-session" }); + await findUsageLogsStats({ sessionId: "client-session" }); + + const ledgerSql = whereArgs.map((arg) => sqlToString(arg).toLowerCase()).join("\n"); + expect(ledgerSql).toContain("session_identity"); + expect(ledgerSql.match(/ or /g)?.length ?? 0).toBeGreaterThanOrEqual(2); + }); }); diff --git a/tests/unit/repository/usage-logs-sessionid-suggestions.test.ts b/tests/unit/repository/usage-logs-sessionid-suggestions.test.ts index 1b53d48fd..995e7432a 100644 --- a/tests/unit/repository/usage-logs-sessionid-suggestions.test.ts +++ b/tests/unit/repository/usage-logs-sessionid-suggestions.test.ts @@ -83,7 +83,7 @@ describe("Usage logs sessionId suggestions", () => { expect(selectMock).not.toHaveBeenCalled(); }); - test("term 应 trim 并按 MIN(created_at) 倒序，limit 生效", async () => { + test("term 应 trim 并按最近 created_at 倒序，limit 生效", async () => { vi.resetModules(); const whereArgs: unknown[] = []; @@ -128,32 +128,78 @@ describe("Usage logs sessionId suggestions", () => { expect(orderByArgs.length).toBeGreaterThan(0); const orderSql = sqlToString(orderByArgs[0]).toLowerCase(); - expect(orderSql).toContain("min"); + expect(orderSql).toContain("max"); - expect(limitArgs).toEqual([20]); + expect(limitArgs).toEqual([20, 20]); }); - test("returns both canonical and client session identities", async () => { + test("returns only candidate identities that match the searched prefix", async () => { vi.resetModules(); - const selectMock = vi.fn(() => - createThenableQuery([ - { - sessionId: "pfx:scope:fingerprint", - sourceSessionId: "client-session", - firstSeen: new Date("2026-01-01T00:00:00Z"), - }, - ]) - ); + const selectMock = vi + .fn() + .mockImplementationOnce(() => createThenableQuery([])) + .mockImplementationOnce(() => + createThenableQuery([ + { + sessionId: "client-session", + firstSeen: new Date("2026-01-01T00:00:00Z"), + }, + ]) + ); vi.doMock("@/drizzle/db", () => ({ db: { select: selectMock } })); const { findUsageLogSessionIdSuggestions } = await import("@/repository/usage-logs"); await expect(findUsageLogSessionIdSuggestions({ term: "client", limit: 20 })).resolves.toEqual([ - "pfx:scope:fingerprint", "client-session", ]); }); + test("deduplicates canonical and physical candidates before applying the final limit", async () => { + vi.resetModules(); + + const selectMock = vi + .fn() + .mockImplementationOnce(() => + createThenableQuery([ + { sessionId: "session-shared", firstSeen: new Date("2026-01-03T00:00:00Z") }, + { sessionId: "session-canonical", firstSeen: new Date("2026-01-01T00:00:00Z") }, + ]) + ) + .mockImplementationOnce(() => + createThenableQuery([ + { sessionId: "session-shared", firstSeen: new Date("2026-01-02T00:00:00Z") }, + { sessionId: "session-physical", firstSeen: new Date("2026-01-02T12:00:00Z") }, + ]) + ); + vi.doMock("@/drizzle/db", () => ({ db: { select: selectMock } })); + + const { findUsageLogSessionIdSuggestions } = await import("@/repository/usage-logs"); + await expect(findUsageLogSessionIdSuggestions({ term: "session-", limit: 2 })).resolves.toEqual( + ["session-shared", "session-physical"] + ); + }); + + test("ignores candidates whose latest createdAt is NULL", async () => { + vi.resetModules(); + + const selectMock = vi + .fn() + .mockImplementationOnce(() => + createThenableQuery([ + { sessionId: "session-null", firstSeen: null }, + { sessionId: "session-valid", firstSeen: new Date("2026-01-02T00:00:00Z") }, + ]) + ) + .mockImplementationOnce(() => createThenableQuery([])); + vi.doMock("@/drizzle/db", () => ({ db: { select: selectMock } })); + + const { findUsageLogSessionIdSuggestions } = await import("@/repository/usage-logs"); + await expect( + findUsageLogSessionIdSuggestions({ term: "session-", limit: 20 }) + ).resolves.toEqual(["session-valid"]); + }); + test("term 含 %/_/\\\\：应按字面量前缀匹配（需转义）", async () => { vi.resetModules(); @@ -190,7 +236,7 @@ describe("Usage logs sessionId suggestions", () => { const { findUsageLogSessionIdSuggestions } = await import("@/repository/usage-logs"); await findUsageLogSessionIdSuggestions({ term: "abc", limit: 500 }); - expect(limitArgs).toEqual([50]); + expect(limitArgs).toEqual([50, 50]); }); test("keyId 未提供时不应 innerJoin(keysTable)", async () => { @@ -205,6 +251,7 @@ describe("Usage logs sessionId suggestions", () => { const { findUsageLogSessionIdSuggestions } = await import("@/repository/usage-logs"); await findUsageLogSessionIdSuggestions({ term: "abc", limit: 20 }); + expect(selectMock).toHaveBeenCalledTimes(2); expect(query.innerJoin).not.toHaveBeenCalled(); }); @@ -220,6 +267,6 @@ describe("Usage logs sessionId suggestions", () => { const { findUsageLogSessionIdSuggestions } = await import("@/repository/usage-logs"); await findUsageLogSessionIdSuggestions({ term: "abc", keyId: 2, limit: 20 }); - expect(query.innerJoin).toHaveBeenCalledTimes(1); + expect(query.innerJoin).toHaveBeenCalledTimes(2); }); }); From de49db97dcf97d030fdd210c1b9c3127ee6966b8 Mon Sep 17 00:00:00 2001 From: "claude[bot]" <41898282+claude[bot]@users.noreply.github.com> Date: Sat, 1 Aug 2026 19:39:58 +0000 Subject: [PATCH 3/3] fix(build): polyfill markAsUncloneable for Bun Docker build undici >= 8 destructures markAsUncloneable from node:worker_threads without a fallback. Bun (the Docker build stage runtime) does not implement this Node.js 23+ API, so webidl.util.markAsUncloneable is undefined and next build crashes during page-data collection for the /v1 and /v1beta proxy routes. Add a side-effect polyfill that defines markAsUncloneable as a no-op when it is missing, restoring the safe fallback that undici 7.x shipped. The guard ensures the real Node.js implementation is used unchanged in production. Verified: bun run build passes with undici 8.9.0 installed (the failing scenario). typecheck and lint clean. Co-Authored-By: Claude Opus 4.6 --- src/app/v1/[...route]/route.ts | 1 + src/app/v1beta/[...route]/route.ts | 1 + src/lib/polyfills/worker-threads.ts | 13 +++++++++++++ 3 files changed, 15 insertions(+) create mode 100644 src/lib/polyfills/worker-threads.ts diff --git a/src/app/v1/[...route]/route.ts b/src/app/v1/[...route]/route.ts index 3a3c82922..88651c055 100644 --- a/src/app/v1/[...route]/route.ts +++ b/src/app/v1/[...route]/route.ts @@ -1,4 +1,5 @@ import "@/lib/polyfills/file"; +import "@/lib/polyfills/worker-threads"; import { Hono } from "hono"; import { handle } from "hono/vercel"; import { registerCors } from "@/app/v1/_lib/cors"; diff --git a/src/app/v1beta/[...route]/route.ts b/src/app/v1beta/[...route]/route.ts index 2c70283fe..b1f2f16a1 100644 --- a/src/app/v1beta/[...route]/route.ts +++ b/src/app/v1beta/[...route]/route.ts @@ -1,4 +1,5 @@ import "@/lib/polyfills/file"; +import "@/lib/polyfills/worker-threads"; import { Hono } from "hono"; import { handle } from "hono/vercel"; import { registerCors } from "@/app/v1/_lib/cors"; diff --git a/src/lib/polyfills/worker-threads.ts b/src/lib/polyfills/worker-threads.ts new file mode 100644 index 000000000..e1162726e --- /dev/null +++ b/src/lib/polyfills/worker-threads.ts @@ -0,0 +1,13 @@ +import { createRequire } from "node:module"; + +const nodeRequire = createRequire(import.meta.url); +const workerThreads = nodeRequire("node:worker_threads") as { + markAsUncloneable?: (...args: unknown[]) => void; +}; + +// undici >= 8 destructures markAsUncloneable from node:worker_threads without +// a fallback. Bun (Docker build stage) does not implement this Node.js 23+ API, +// so next build crashes during page data collection. +if (typeof workerThreads.markAsUncloneable !== "function") { + workerThreads.markAsUncloneable = function markAsUncloneable() {}; +}