From b9a1e48a587e22cabaa2719d593954fa7623ed9d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Igor=20=C5=A0=C4=87eki=C4=87?= Date: Sat, 29 Aug 2026 20:46:02 +0200 Subject: [PATCH] feat(sessions): bound history transport for harness reads --- .../web/src/lib/session-ingest-client.test.ts | 187 +++++++++++++++++- apps/web/src/lib/session-ingest-client.ts | 30 +++ .../routers/cli-sessions-v2-router.test.ts | 151 +++++++++++++- .../web/src/routers/cli-sessions-v2-router.ts | 24 ++- 4 files changed, 381 insertions(+), 11 deletions(-) diff --git a/apps/web/src/lib/session-ingest-client.test.ts b/apps/web/src/lib/session-ingest-client.test.ts index b58da47d45..b9c971af80 100644 --- a/apps/web/src/lib/session-ingest-client.test.ts +++ b/apps/web/src/lib/session-ingest-client.test.ts @@ -779,6 +779,191 @@ describe('fetchSessionMessagesPage', () => { ], }; + describe('bounded transport', () => { + const options = { limit: 50, bounded: true }; + const headers = { 'content-type': 'application/json' }; + const page = { + kiloSessionId: validSessionId, + history: { messages: [storedMessage], nextCursor: 'opaque-cursor', omittedItemCount: 0 }, + }; + const largePage = { + ...page, + history: { + ...page.history, + messages: [ + { + ...storedMessage, + parts: [ + ...storedMessage.parts, + { + id: 'prt_file_01', + sessionID: validSessionId, + messageID: 'msg_user_01', + type: 'file', + mime: 'text/plain', + url: '界'.repeat(350_000), + }, + ], + }, + ], + }, + }; + const largeBody = JSON.stringify({ success: true, ...largePage }); + + beforeEach(() => mockCaptureException.mockReset()); + afterEach(() => { + jest.restoreAllMocks(); + jest.useRealTimers(); + }); + + it('accepts exactly 1 MiB and preserves the page, cursor, and authenticated request', async () => { + const body = JSON.stringify({ success: true, ...page }); + mockFetch.mockResolvedValue( + new Response(body + ' '.repeat(1_048_576 - Buffer.byteLength(body)), { headers }) + ); + await expect( + fetchSessionMessagesPage(validSessionId, 'user_123', { + ...options, + before: 'opaque-cursor', + }) + ).resolves.toEqual(page); + expect(mockFetch).toHaveBeenCalledWith( + `https://ingest.test.example.com/api/session/${validSessionId}/messages?limit=50&before=opaque-cursor`, + expect.objectContaining({ + headers: { Authorization: 'Bearer mock-jwt-token', Accept: 'application/json' }, + }) + ); + }); + + describe.each([200, 404, 503])('HTTP %s', status => { + it.each(['declared', 'streamed', 'understated'])( + 'rejects %s overflow before decoding and cancels the stream', + async length => { + let cancelled = false; + const bytes = new TextEncoder().encode( + length === 'declared' ? JSON.stringify({ success: true, ...page }) : largeBody + ); + expect(largeBody.length).toBeLessThan(1_048_576); + const stream = new ReadableStream({ + start(controller) { + controller.enqueue(bytes.subarray(0, 600_000)); + controller.enqueue(bytes.subarray(600_000)); + if (length === 'declared') controller.close(); + }, + cancel() { + cancelled = true; + }, + }); + mockFetch.mockResolvedValue( + new Response(stream, { + status, + headers: { + ...headers, + ...(length === 'streamed' + ? {} + : { 'content-length': length === 'declared' ? '1048577' : '10' }), + }, + }) + ); + const decode = jest.spyOn(TextDecoder.prototype, 'decode'); + await expect( + fetchSessionMessagesPage(validSessionId, 'user_123', options) + ).rejects.toThrow('size limit'); + expect(cancelled).toBe(true); + expect(decode).not.toHaveBeenCalled(); + expect(mockCaptureException).not.toHaveBeenCalled(); + } + ); + }); + + it.each([undefined, false])('keeps oversized legacy pages when bounded=%s', async bounded => { + mockFetch.mockResolvedValue(new Response(largeBody)); + await expect( + fetchSessionMessagesPage(validSessionId, 'user_123', { limit: 50, bounded }) + ).resolves.toEqual(largePage); + }); + + it.each([ + null, + { messages: [], nextCursor: null, omittedItemCount: 0 }, + { kind: 'retryable_failure', phase: 'page_parts' }, + { kind: 'too_large', maximumBytes: 8_388_608, phase: 'message_scan' }, + { kind: 'invalid_data' }, + ])('preserves the worker outcome %j', async history => { + mockFetch.mockResolvedValue( + Response.json({ success: true, kiloSessionId: validSessionId, history }) + ); + await expect(fetchSessionMessagesPage(validSessionId, 'user_123', options)).resolves.toEqual({ + kiloSessionId: validSessionId, + history, + }); + }); + + it('returns null for a bounded 404 without exposing its body', async () => { + mockFetch.mockResolvedValue(new Response('private response', { status: 404 })); + await expect( + fetchSessionMessagesPage(validSessionId, 'user_123', options) + ).resolves.toBeNull(); + expect(mockCaptureException).not.toHaveBeenCalled(); + }); + + it.each([ + { + status: 503, + body: 'private response', + message: 'Session ingest messages page failed: 503', + }, + { + status: 200, + body: 'private response', + message: 'Session ingest messages page returned an unexpected response', + }, + { + status: 200, + body: JSON.stringify({ success: true, ...page, history: { kind: 'private response' } }), + message: 'Session ingest messages page returned an unexpected response', + }, + ])('sanitizes HTTP $status errors and malformed output', async ({ status, body, message }) => { + mockFetch.mockResolvedValue( + new Response(body, { status, statusText: 'private response', headers }) + ); + await expect( + fetchSessionMessagesPage(validSessionId, 'user_123', options) + ).rejects.toMatchObject({ message }); + expect(mockCaptureException).not.toHaveBeenCalled(); + }); + + it.each(['headers', 'success body', 'error body'])( + 'ends stalled %s at 30 seconds', + async phase => { + jest.useFakeTimers(); + let cancelled = false; + let signal: AbortSignal | undefined; + mockFetch.mockImplementation((_url, init: RequestInit) => { + signal = init.signal ?? undefined; + return phase === 'headers' + ? new Promise(() => {}) + : Promise.resolve( + new Response( + new ReadableStream({ + cancel() { + cancelled = true; + }, + }), + { status: phase === 'error body' ? 503 : 200, headers } + ) + ); + }); + const result = fetchSessionMessagesPage(validSessionId, 'user_123', options); + const rejection = expect(result).rejects.toThrow('timed out'); + await jest.advanceTimersByTimeAsync(30_000); + await rejection; + expect(signal?.aborted).toBe(true); + expect(cancelled).toBe(phase !== 'headers'); + } + ); + }); + it('returns the bounded page and the opaque next cursor', async () => { mockFetch.mockResolvedValue({ ok: true, @@ -894,7 +1079,7 @@ describe('fetchSessionMessagesPage', () => { await expect( fetchSessionMessagesPage(validSessionId, 'user_123', { limit: 50 }) - ).rejects.toThrow(/Session ingest messages page failed/); + ).rejects.toThrow('Session ingest messages page failed: 500 Internal Server Error - boom'); expect(mockCaptureException).toHaveBeenCalledWith( expect.any(Error), expect.objectContaining({ diff --git a/apps/web/src/lib/session-ingest-client.ts b/apps/web/src/lib/session-ingest-client.ts index b63f3c2e52..7823e824aa 100644 --- a/apps/web/src/lib/session-ingest-client.ts +++ b/apps/web/src/lib/session-ingest-client.ts @@ -4,6 +4,10 @@ import { captureException } from '@sentry/nextjs'; import { z } from 'zod'; import { INTERNAL_API_SECRET, SESSION_INGEST_WORKER_URL } from '@/lib/config.server'; import { generateInternalServiceToken } from '@/lib/tokens'; +import { + boundRepositoryResponse, + withRepositoryReadDeadline, +} from '@/lib/integrations/core/repository-read-limits'; import type { User } from '@kilocode/db/schema'; import { kiloSdkMessageHistorySchema, @@ -128,6 +132,8 @@ export type SessionMessagesPageOptions = { limit?: number; /** Opaque cursor returned by a previous page; requires a positive limit. */ before?: string; + /** Opt into the shared 1 MiB/30-second transport limits. */ + bounded?: boolean; }; export type SessionMessagesPageResult = { @@ -168,6 +174,30 @@ export async function fetchSessionMessagesPage( }`; const token = generateInternalServiceToken(userId); + // Omitted bounds preserve the deployed transport until legacy callers retire. + if (options.bounded) { + return withRepositoryReadDeadline(options, async signal => { + const response = await boundRepositoryResponse( + await fetch(url, { + headers: { Authorization: `Bearer ${token}`, Accept: 'application/json' }, + signal, + }), + signal + ); + if (response.status === 404) return null; + if (!response.ok) { + throw new Error(`Session ingest messages page failed: ${response.status}`); + } + const parsed = SessionMessagesPageResponseSchema.safeParse( + await response.json().catch(() => undefined) + ); + if (!parsed.success) { + throw new Error('Session ingest messages page returned an unexpected response'); + } + return { kiloSessionId: parsed.data.kiloSessionId, history: parsed.data.history }; + }); + } + const response = await fetch(url, { headers: { Authorization: `Bearer ${token}` }, }); diff --git a/apps/web/src/routers/cli-sessions-v2-router.test.ts b/apps/web/src/routers/cli-sessions-v2-router.test.ts index d999dfb4c5..1fb26ecc2b 100644 --- a/apps/web/src/routers/cli-sessions-v2-router.test.ts +++ b/apps/web/src/routers/cli-sessions-v2-router.test.ts @@ -14,7 +14,13 @@ import { eq, and, inArray } from 'drizzle-orm'; import type { User, Organization } from '@kilocode/db/schema'; import * as githubAdapter from '@/lib/integrations/platforms/github/adapter'; import { TRPCError } from '@trpc/server'; -import { parseGitHubOwnerRepo, parseGitHubPrUrl } from '@/routers/cli-sessions-v2-router'; +import { + cliSessionsV2Router, + parseGitHubOwnerRepo, + parseGitHubPrUrl, +} from '@/routers/cli-sessions-v2-router'; +import { createCallerFactory } from '@/lib/trpc/init'; +import { PgDialect } from 'drizzle-orm/pg-core'; import type { fetchSessionMessagesPage as FetchSessionMessagesPageType } from '@/lib/session-ingest-client'; import { notifyCliSessionRenamed } from '@/lib/cloud-agent/session-events'; import { captureException } from '@sentry/nextjs'; @@ -120,6 +126,149 @@ let otherUser: User; let adminUser: User; let testOrganization: Organization; +describe('bounded session history router', () => { + const user = { id: 'oauth/history-owner', is_admin: false } as User; + const session = { + session_id: 'ses_messages_page_test_1234', + kilo_user_id: user.id, + organization_id: null as string | null, + cloud_agent_session_id: 'agent_history', + }; + const page = { kiloSessionId: session.session_id, history: null }; + const fetchPage = jest.mocked( + jest.requireMock<{ fetchSessionMessagesPage: typeof FetchSessionMessagesPageType }>( + '@/lib/session-ingest-client' + ).fetchSessionMessagesPage + ); + const caller = createCallerFactory(cliSessionsV2Router)({ user }); + const limit = jest.fn(); + const where = jest.fn((_condition: ReturnType) => ({ limit })); + const query = { from: () => ({ where }) } as ReturnType; + + beforeEach(() => { + session.organization_id = null; + limit.mockReset().mockResolvedValue([session]); + where.mockClear(); + jest.spyOn(db, 'select').mockReturnValue(query); + fetchPage.mockReset().mockResolvedValue(page); + mockGetSession.mockReset().mockResolvedValue({ latestEventId: 42 }); + }); + afterEach(() => jest.restoreAllMocks()); + + it.each([undefined, false, true])( + 'forwards bounded=%s and preserves legacy defaults and watermarks', + async bounded => { + await expect( + caller.getSessionMessagesPage({ session_id: session.session_id, bounded }) + ).resolves.toEqual({ + ...page, + watermarkEventId: bounded ? null : 42, + }); + expect(fetchPage).toHaveBeenCalledWith(session.session_id, user.id, { + limit: 50, + ...(bounded !== undefined ? { bounded } : {}), + }); + expect(new PgDialect().sqlToQuery(where.mock.calls[0][0]).params).toEqual([ + session.session_id, + user.id, + ]); + expect(mockGetSession).toHaveBeenCalledTimes(bounded ? 0 : 1); + } + ); + + it('forwards a validated cursor without reading the watermark', async () => { + const cursor = btoa(JSON.stringify({ id: 'msg_user_01', time: 1761000000100 })).replace( + /=+$/, + '' + ); + await expect( + caller.getSessionMessagesPage({ + session_id: session.session_id, + bounded: true, + limit: 25, + cursor, + }) + ).resolves.toEqual({ ...page, watermarkEventId: null }); + expect(fetchPage).toHaveBeenCalledWith(session.session_id, user.id, { + bounded: true, + limit: 25, + before: cursor, + }); + expect(mockGetSession).not.toHaveBeenCalled(); + }); + + it.each([{ limit: 0 }, { limit: 101 }, { limit: 1.5 }, { cursor: 'bad' }])( + 'retains input validation for %j', + async input => { + await expect( + caller.getSessionMessagesPage({ session_id: session.session_id, bounded: true, ...input }) + ).rejects.toMatchObject({ code: 'BAD_REQUEST' }); + expect(fetchPage).not.toHaveBeenCalled(); + } + ); + + it.each([undefined, true])( + 'denies missing or foreign sessions before transport when bounded=%s', + async bounded => { + limit.mockResolvedValue([]); + await expect( + caller.getSessionMessagesPage({ session_id: session.session_id, bounded }) + ).rejects.toMatchObject({ code: 'NOT_FOUND', message: 'Session not found' }); + expect(fetchPage).not.toHaveBeenCalled(); + expect(mockGetSession).not.toHaveBeenCalled(); + } + ); + + it.each(['member', 'removed'])( + 'retains organization authorization for %s membership', + async membership => { + session.organization_id = '11111111-1111-4111-8111-111111111111'; + const membershipQuery = { + from: jest.fn().mockReturnThis(), + innerJoin: jest.fn().mockReturnThis(), + where: jest.fn().mockResolvedValue(membership === 'member' ? [{ role: 'member' }] : []), + } as ReturnType; + jest.spyOn(db, 'select').mockReturnValueOnce(query).mockReturnValue(membershipQuery); + const result = caller.getSessionMessagesPage({ + session_id: session.session_id, + bounded: true, + }); + if (membership === 'member') { + await expect(result).resolves.toEqual({ ...page, watermarkEventId: null }); + } else { + await expect(result).rejects.toMatchObject({ code: 'UNAUTHORIZED' }); + expect(fetchPage).not.toHaveBeenCalled(); + } + } + ); + + it.each([undefined, true])( + 'preserves the public error without bounded raw logging when bounded=%s', + async bounded => { + const upstream = new Error('private response Bearer private-credential'); + const log = jest.spyOn(console, 'error').mockImplementation(() => {}); + fetchPage.mockRejectedValue(upstream); + const error = await caller + .getSessionMessagesPage({ session_id: session.session_id, bounded }) + .catch(error => error); + expect(error).toMatchObject({ + code: 'INTERNAL_SERVER_ERROR', + message: 'Failed to fetch session messages page', + }); + expect(error.cause).toBe(bounded ? undefined : upstream); + if (bounded) expect(log).not.toHaveBeenCalled(); + else expect(log).toHaveBeenCalledWith(expect.any(String), upstream.message); + } + ); + + it('keeps the worker not-found error in bounded mode', async () => { + fetchPage.mockResolvedValue(null); + await expect( + caller.getSessionMessagesPage({ session_id: session.session_id, bounded: true }) + ).rejects.toMatchObject({ code: 'NOT_FOUND', message: 'Session not found' }); + }); +}); + describe('cli-sessions-v2-router', () => { beforeEach(() => { afterCallbacks.length = 0; diff --git a/apps/web/src/routers/cli-sessions-v2-router.ts b/apps/web/src/routers/cli-sessions-v2-router.ts index 680a08e7a2..19f56cc655 100644 --- a/apps/web/src/routers/cli-sessions-v2-router.ts +++ b/apps/web/src/routers/cli-sessions-v2-router.ts @@ -475,6 +475,7 @@ const GetSessionMessagesPageInputSchema = z .max(MAX_KILO_SDK_MESSAGE_HISTORY_PAGE_SIZE) .default(DEFAULT_KILO_SDK_MESSAGE_PAGE_SIZE), cursor: z.string().min(1).optional(), + bounded: z.boolean().optional(), }) .superRefine((params, ctx) => { if (params.cursor === undefined) return; @@ -1041,9 +1042,10 @@ export const cliSessionsV2Router = createTRPCRouter({ // on its first WebSocket connect instead of `replay=false`. Cursor // pages skip the Cloud Agent read — the watermark is only seeded once. // Failures are swallowed and return null so the page endpoint is - // never blocked on an optional watermark read. + // never blocked on an optional watermark read. Bounded history reads do + // not seed an event stream; keep the old path until legacy callers retire. let watermarkEventId: number | null = null; - if (!input.cursor && session.cloud_agent_session_id) { + if (!input.bounded && !input.cursor && session.cloud_agent_session_id) { try { const authToken = generateApiToken(ctx.user); const client = createCloudAgentNextClient(authToken); @@ -1062,21 +1064,25 @@ export const cliSessionsV2Router = createTRPCRouter({ result = await fetchSessionMessagesPage(input.session_id, ctx.user.id, { limit: input.limit, ...(input.cursor !== undefined ? { before: input.cursor } : {}), + ...(input.bounded !== undefined ? { bounded: input.bounded } : {}), }); } catch (error) { // Match the existing `getSessionMessages` error contract: surface a // stable INTERNAL_SERVER_ERROR so the mobile client can map the // outcome without inferring retry semantics from the worker's - // text. The client already calls `captureException`; we do not - // double-capture here. - console.error( - `Failed to fetch session messages page for session ${input.session_id}:`, - error instanceof Error ? error.message : error - ); + // text. The legacy client already calls `captureException`. + // Bounded reads must not send upstream bodies or credentials to logs, + // including the Sentry middleware's error cause. + if (!input.bounded) { + console.error( + `Failed to fetch session messages page for session ${input.session_id}:`, + error instanceof Error ? error.message : error + ); + } throw new TRPCError({ code: 'INTERNAL_SERVER_ERROR', message: 'Failed to fetch session messages page', - cause: error, + cause: input.bounded ? undefined : error, }); }