From d309616e0a65fea185d91e6132c69f8410dd3805 Mon Sep 17 00:00:00 2001 From: Tanner Linsley Date: Sun, 4 Oct 2026 22:56:43 -0600 Subject: [PATCH 1/2] perf(chat): shorten admission and model preparation --- docs/tanchat-send-latency.md | 43 +++ harness-tests/core/assistant-mcp-lazy.test.ts | 69 +++++ .../assistant-discovery-runtime.test.ts | 25 +- .../conversation-metadata.test.ts | 96 ++++++ .../conversation-preparation.test.ts | 82 ++--- .../plugin-reference-runtime.test.ts | 2 +- src/chat/server/assistant-instructions.ts | 6 + src/chat/server/assistant-mcp-tools.ts | 3 + src/chat/server/conversation-database.ts | 82 ++--- src/chat/server/conversation-threads.ts | 28 +- src/chat/server/conversation.ts | 293 ++++++++++-------- src/chat/server/memory.ts | 31 +- src/chat/server/run-usage.ts | 81 +++-- src/db/client.ts | 2 +- 14 files changed, 554 insertions(+), 289 deletions(-) create mode 100644 docs/tanchat-send-latency.md create mode 100644 harness-tests/core/assistant-mcp-lazy.test.ts create mode 100644 harness-tests/pending-runtime/conversation-metadata.test.ts diff --git a/docs/tanchat-send-latency.md b/docs/tanchat-send-latency.md new file mode 100644 index 000000000..929be80f5 --- /dev/null +++ b/docs/tanchat-send-latency.md @@ -0,0 +1,43 @@ +# TanChat send latency, October 4, 2026 + +## What changed + +Ordinary messages no longer load the entire MCP connection inventory or skill catalog before calling the model. The model can discover tools and skills when it needs them. Explicit tool selections still load their inventory, and selected skill and plugin versions remain pinned. + +Independent admission checks run together and settle before their database context is released. Bot lifecycle, workspace context, memory preferences, and thread source use fewer database statements. Each authorization snapshot still checks its exact scope against current database state. Post-preparation checks, reset epochs, receipt matching, and runtime tool authorization remain in place. + +Usage receipts, reservations, and counters publish in one statement under the existing transaction and locks. Duplicate sends, concurrent cap enforcement, scheduled usage, funded reservations, and admin policy retain their existing behavior. + +Automatic Kody memory and account guidance remain enabled and overlap with account preferences and thread source reads. Their latency is not represented by the no-credential benchmark below. + +Native PostgreSQL clients allow five connections, matching the existing Workers limit, so independent reads can actually overlap. Copy activation now explicitly waits for its activity publication before checking it. That ordering cannot depend on all queries sharing one socket. + +## Controlled measurements + +The baseline is commit 1bbafeec, the changes merged through PR 1339. Both versions run the actual conversation admission and preparation code against disposable PostgreSQL. The provider is synthetic. Each version runs three messages, the first includes cold client setup. + +| Database connection | Baseline, milliseconds | Candidate, milliseconds | +| ------------------------------------------------ | ---------------------- | ----------------------- | +| Delayed TCP proxy, nominal 80 ms round trip | 5406, 5140, 5148 | 2610, 2210, 2207 | +| Delayed proxy, five connections in both versions | 5358, 4984, 4980 | 2610, 2210, 2207 | +| Same-machine PostgreSQL, no proxy delay | Not measured | 60, 16, 13 | + +Warm preparation in the delayed fixture fell from about 5.14 seconds to 2.21 seconds, about 57 percent. The proxy delays chunks, so protocol startup can incur more than one nominal round trip. The native baseline uses one connection and the candidate uses five. Workers already allowed five. With both versions using five connections, warm preparation fell from 4.98 seconds to 2.21 seconds, about 56 percent. Neither comparison predicts the production improvement. + +These numbers measure the beginning of durable admission to provider dispatch. They exclude HTTP authentication, browser rendering, real model time to first token, and real Kody network reads. They do not establish instant live responses. + +## Remaining production work + +The checked-in production configuration has no Hyperdrive binding. The host already supports HYPERDRIVE.connectionString and otherwise uses DATABASE_URL. A new pool must target this site's database, not the existing unrelated dashboard demo pool. + +The proposed production pool is named tanstack-com, with binding HYPERDRIVE and query caching disabled. Local development must omit that production binding and continue using its optional DATABASE_URL. Creating the pool, binding it, and deploying are separate production changes awaiting authorization. + +Hyperdrive pools connections to reduce TCP, TLS, and authentication setup. It does not remove every SQL round trip or relocate existing Durable Objects. Region placement needs a measurement against the actual database and conversation objects before changing hints. + +Sources: [Hyperdrive connection pooling](https://developers.cloudflare.com/hyperdrive/concepts/how-hyperdrive-works/) and [Durable Object location](https://developers.cloudflare.com/durable-objects/reference/data-location/). + +## Verification + +The runtime suite exercises real PostgreSQL, including access revocation, reset races, retries, copy activation, quota concurrency, funded usage, response preference persistence, and skill discovery through the actual SDK loop. The lazy MCP test also uses the SDK loop with a synthetic transport. + +The local Assistant page rendered from the QA checkout on localhost:3000. No real provider messages were sent during this pass. Full pnpm test passed: all five TypeScript configurations, lint with zero errors and 105 existing warnings, 537 site tests, 2,274 chat tests, and 20 desktop tests. Five optional tests were skipped. The complete PostgreSQL runtime suite passed 268 tests in 35 files. The final metadata scope suite passed 10 tests, including four additional thread-scope rejection cases. diff --git a/harness-tests/core/assistant-mcp-lazy.test.ts b/harness-tests/core/assistant-mcp-lazy.test.ts new file mode 100644 index 000000000..a96e3347c --- /dev/null +++ b/harness-tests/core/assistant-mcp-lazy.test.ts @@ -0,0 +1,69 @@ +import { expect, it, vi } from 'vitest' +import { chat, maxIterations } from '@tanstack/ai' +import { createCloudflareText } from '@tanstack/ai-cloudflare' +import { assistantMcpTools } from '../../src/chat/server/assistant-mcp-tools' +import type { McpConnection } from '../../src/chat/server/mcp' + +it('loads current inventory only when the SDK invokes discovery', async () => { + const connections: McpConnection[] = [] + const loadConnections = vi.fn(async () => { + connections.splice(0, connections.length, { + id: 'current', + label: 'Current service', + url: 'https://example.invalid/mcp', + }) + }) + const tools = assistantMcpTools({ + connections, + loadConnections, + catalog: vi.fn(), + propose: vi.fn(), + read: vi.fn(), + }) + expect(loadConnections).not.toHaveBeenCalled() + let pass = 0 + const fetch = vi.fn(async () => { + const first = pass++ === 0 + const delta = first + ? { + role: 'assistant', + tool_calls: [ + { + index: 0, + id: 'inventory', + type: 'function', + function: { name: 'list_connected_tools', arguments: '{}' }, + }, + ], + } + : { role: 'assistant', content: 'Finished.' } + return new Response( + 'data: ' + + JSON.stringify({ + id: 'synthetic', + choices: [ + { index: 0, delta, finish_reason: first ? 'tool_calls' : 'stop' }, + ], + }) + + '\n\ndata: [DONE]\n\n', + { headers: { 'Content-Type': 'text/event-stream' } }, + ) + }) + const adapter = createCloudflareText('synthetic', { + accountId: 'synthetic', + apiKey: 'synthetic', + fetch, + }) + const results: unknown[] = [] + for await (const chunk of chat({ + adapter, + tools, + messages: [{ role: 'user', content: 'List my connected services.' }], + agentLoopStrategy: maxIterations(2), + })) { + if (chunk.type === 'TOOL_CALL_RESULT') results.push(chunk.content) + } + expect(loadConnections).toHaveBeenCalledOnce() + expect(JSON.stringify(results)).toContain('Current service') + expect(fetch).toHaveBeenCalledTimes(2) +}) diff --git a/harness-tests/pending-runtime/assistant-discovery-runtime.test.ts b/harness-tests/pending-runtime/assistant-discovery-runtime.test.ts index 3bc9c492c..35955c9aa 100644 --- a/harness-tests/pending-runtime/assistant-discovery-runtime.test.ts +++ b/harness-tests/pending-runtime/assistant-discovery-runtime.test.ts @@ -43,7 +43,7 @@ const system = (request: any) => .map((message: any) => message.content) .join('\n') -it('exposes authorized skill metadata on the first request and loads the full skill only after reading', async () => { +it('discovers skill metadata on demand and loads full instructions only after reading', async () => { const h = await conversationHarness() const skillId = crypto.randomUUID() await new Skills(identity).command({ @@ -60,9 +60,11 @@ it('exposes authorized skill metadata on the first request and loads the full sk const requests: any[] = [] const run = vi.fn(async (_model, request) => { requests.push(structuredClone(request)) - return requests.length === 1 - ? reply('read_skill', { skillId, version: 1 }) - : reply() + if (requests.length === 1) + return reply('list_skills', { query: 'test-review' }) + if (requests.length === 2) + return reply('read_skill', { skillId, version: 1 }) + return reply() }) Object.assign(h.env, { AI: { run } }) await h.c.begin({ @@ -72,13 +74,20 @@ it('exposes authorized skill metadata on the first request and loads the full sk }) await h.settle() expect((await h.c.snapshot()).error).toBeUndefined() - expect(requests).toHaveLength(2) - expect(system(requests[0])).toContain(skillId) - expect(system(requests[0])).toContain('Review tests for useful coverage.') + expect(requests).toHaveLength(3) + expect(system(requests[0])).not.toContain(skillId) + expect(names(requests[0])).toContain('list_skills') + expect(JSON.stringify(requests[1].messages)).toContain(skillId) + expect(JSON.stringify(requests[1].messages)).toContain( + 'Review tests for useful coverage.', + ) expect(system(requests[0])).not.toContain( 'INSTRUCTIONS_LOADED_ONLY_AFTER_READ', ) - expect(system(requests[1])).toContain('INSTRUCTIONS_LOADED_ONLY_AFTER_READ') + expect(system(requests[1])).not.toContain( + 'INSTRUCTIONS_LOADED_ONLY_AFTER_READ', + ) + expect(system(requests[2])).toContain('INSTRUCTIONS_LOADED_ONLY_AFTER_READ') expect((await h.c.snapshot()).assistantTask?.loadedSkills).toEqual([ { skillId, version: 1 }, ]) diff --git a/harness-tests/pending-runtime/conversation-metadata.test.ts b/harness-tests/pending-runtime/conversation-metadata.test.ts new file mode 100644 index 000000000..7ef5085d6 --- /dev/null +++ b/harness-tests/pending-runtime/conversation-metadata.test.ts @@ -0,0 +1,96 @@ +import { expect, it } from 'vitest' +import { conversationHarness } from './fixtures/conversation-runtime' +import { conversationThreadContext } from '../../src/chat/server/conversation-threads' +import { Memories } from '../../src/chat/server/memory' +import { + readConversationLifecycle, + readConversationRunContext, +} from '../../src/chat/server/conversation-database' + +it('reads fresh run context and ordered saved actions from a single database snapshot', async () => { + const h = await conversationHarness() + const input = h.input('metadata') + const identity = { workspaceId: 'w', botId: 'b', userId: input.userId } + await h.db`INSERT INTO chat_recipes(id,workspace_id,title,description,code,created_at) VALUES('second','w','Second','','',2),('first','w','First','','',1)` + const result = await readConversationRunContext(identity) + expect(result.bot).toMatchObject({ id: 'b', workspace_id: 'w', name: 'Bot' }) + expect(typeof result.bot.created_at).toBe('number') + expect(result.recipes).toEqual([ + expect.objectContaining({ id: 'first' }), + expect.objectContaining({ id: 'second' }), + ]) + expect((await readConversationRunContext(identity, false)).recipes).toEqual( + [], + ) + await h.db`DELETE FROM chat_memberships WHERE workspace_id='w'` + await expect(readConversationRunContext(identity)).rejects.toThrow( + 'Workspace access is unavailable.', + ) +}) + +it('returns lifecycle nulls and fresh archive state without losing the missing-bot distinction', async () => { + const h = await conversationHarness() + expect(await readConversationLifecycle('missing', 'missing')).toEqual({ + bot: null, + thread: null, + }) + expect(await readConversationLifecycle('b', 'main-conversation')).toEqual({ + bot: { archived_at: null, deleted_at: null }, + thread: null, + }) + await h.db`UPDATE chat_bots SET archived_at=to_timestamp(100) WHERE id='b'` + expect( + await readConversationLifecycle('b', 'main-conversation'), + ).toMatchObject({ bot: { archived_at: 100000 } }) +}) + +it.each(['membership', 'owner', 'workspace', 'deleted'])( + 'does not expose memory preferences after %s access changes', + async (change) => { + const h = await conversationHarness() + const userId = h.input('metadata').userId + const scope = { + workspaceId: 'w', + conversationId: 'main-conversation', + userId, + } + const memories = new Memories(scope) + expect(await memories.preferences()).toEqual({ + enabled: false, + revision: 0, + }) + await h.db`INSERT INTO chat_memory_preferences(conversation_id,enabled,revision,updated_at) VALUES('main-conversation',true,7,1)` + expect(await memories.preferences()).toEqual({ enabled: true, revision: 7 }) + if (change === 'membership') + await h.db`DELETE FROM chat_memberships WHERE workspace_id='w'` + if (change === 'owner') + scope.userId = '00000000-0000-4000-8000-000000000002' + if (change === 'workspace') scope.workspaceId = 'foreign' + if (change === 'deleted') + await h.db`UPDATE chat_bots SET deleted_at=now() WHERE id='b'` + await expect(memories.preferences()).rejects.toThrow( + 'Conversation access is unavailable.', + ) + }, +) + +it.each(['membership', 'owner', 'workspace', 'bot'])( + 'rejects thread context after %s scope changes', + async (change) => { + const h = await conversationHarness() + const identity = { + workspaceId: 'w', + botId: 'b', + conversationId: 'main-conversation', + userId: h.input('metadata').userId, + } + expect(await conversationThreadContext(identity)).toBeUndefined() + if (change === 'membership') + await h.db`DELETE FROM chat_memberships WHERE workspace_id='w'` + if (change === 'owner') + identity.userId = '00000000-0000-4000-8000-000000000002' + if (change === 'workspace') identity.workspaceId = 'foreign' + if (change === 'bot') identity.botId = 'foreign' + await expect(conversationThreadContext(identity)).rejects.toThrow() + }, +) diff --git a/harness-tests/pending-runtime/conversation-preparation.test.ts b/harness-tests/pending-runtime/conversation-preparation.test.ts index 91a408c6f..11e8447aa 100644 --- a/harness-tests/pending-runtime/conversation-preparation.test.ts +++ b/harness-tests/pending-runtime/conversation-preparation.test.ts @@ -7,28 +7,18 @@ import { SkillCatalog } from '../../src/chat/server/skill-catalog' afterEach(() => vi.restoreAllMocks()) -it('reads skill metadata while MCP inventory is pending and waits for both before calling the model', async () => { +it('dispatches a plain response without reading MCP inventory or the skill directory', async () => { const h = await conversationHarness() - let releaseInventory = () => {} - const inventoryGate = new Promise((resolve) => { - releaseInventory = resolve - }) - let releaseDirectory = () => {} - const directoryGate = new Promise((resolve) => { - releaseDirectory = resolve - }) const inventory = vi .spyOn(connections, 'connectedMcpServers') - .mockImplementation(async () => { - await inventoryGate - return [] - }) + .mockRejectedValue( + new Error('Unused inventory must not block a plain response'), + ) const directory = vi .spyOn(SkillCatalog.prototype, 'list') - .mockImplementation(async () => { - await directoryGate - return { items: [] } - }) + .mockRejectedValue( + new Error('Unused directory must not block a plain response'), + ) h.env.AI.run.mockImplementation( async () => new Response( @@ -51,52 +41,18 @@ it('reads skill metadata while MCP inventory is pending and waits for both befor { headers: { 'Content-Type': 'text/event-stream' } }, ), ) - try { - await h.c.begin({ ...h.input('parallel-preparation'), fixture: false }) - await vi.waitFor(() => expect(inventory).toHaveBeenCalledOnce()) - await vi.waitFor(() => expect(directory).toHaveBeenCalledOnce()) - expect(h.env.AI.run).not.toHaveBeenCalled() - releaseInventory() - await Promise.resolve() - expect(h.env.AI.run).not.toHaveBeenCalled() - releaseDirectory() - await h.settle() - expect(h.env.AI.run).toHaveBeenCalledOnce() - } finally { - releaseInventory() - releaseDirectory() - await h.settle() - } -}) - -it('settles the directory read before reporting an inventory failure and never calls the model', async () => { - const h = await conversationHarness() - let releaseDirectory = () => {} - const directoryGate = new Promise((resolve) => { - releaseDirectory = resolve - }) - vi.spyOn(connections, 'connectedMcpServers').mockRejectedValue( - new Error('Synthetic inventory failure'), - ) - const directory = vi - .spyOn(SkillCatalog.prototype, 'list') - .mockImplementation(async () => { - await directoryGate - return { items: [] } - }) - try { - await h.c.begin({ ...h.input('failed-preparation'), fixture: false }) - await vi.waitFor(() => expect(directory).toHaveBeenCalledOnce()) - expect((await h.c.snapshot()).status).toBe('running') - expect(h.env.AI.run).not.toHaveBeenCalled() - releaseDirectory() - await h.settle() - expect((await h.c.snapshot()).error).toBe('Synthetic inventory failure') - expect(h.env.AI.run).not.toHaveBeenCalled() - } finally { - releaseDirectory() - await h.settle() - } + await h.c.begin({ ...h.input('deferred-preparation'), fixture: false }) + await h.settle() + expect(h.env.AI.run).toHaveBeenCalledOnce() + expect(inventory).not.toHaveBeenCalled() + expect(directory).not.toHaveBeenCalled() + expect( + (await h.c.snapshot()).messages.some((message) => + message.parts.some( + (part) => part.type === 'text' && part.content === 'Ready.', + ), + ), + ).toBe(true) }) it('reauthorizes membership inside send admission without requiring a prior identity RPC', async () => { diff --git a/harness-tests/pending-runtime/plugin-reference-runtime.test.ts b/harness-tests/pending-runtime/plugin-reference-runtime.test.ts index c9a4ba174..5401ff274 100644 --- a/harness-tests/pending-runtime/plugin-reference-runtime.test.ts +++ b/harness-tests/pending-runtime/plugin-reference-runtime.test.ts @@ -259,7 +259,7 @@ it('pins a selected old package before dispatch and discovers its exact old skil state.assistantTask?.loadedPlugins, ) expect(h.network).not.toHaveBeenCalled() - expect(discovery).toHaveBeenCalled() + expect(discovery).not.toHaveBeenCalled() discovery.mockRestore() }) diff --git a/src/chat/server/assistant-instructions.ts b/src/chat/server/assistant-instructions.ts index 2c2a1f7e0..b42fc4716 100644 --- a/src/chat/server/assistant-instructions.ts +++ b/src/chat/server/assistant-instructions.ts @@ -299,6 +299,12 @@ export function buildAssistantInstructions( 16000, ), ) + if (!options.skillDirectory && enabled.has('list_skills')) + sections.push({ + id: 'core.skill-discovery', + version: '1', + text: 'Skill metadata loads on demand. When a skill may help the request, search with list_skills, then read its exact id and version with read_skill before following it. An unloaded catalog does not mean no skills exist. Explicitly selected skills and their versions remain part of this request.', + }) if (options.workflowInputs?.length && enabled.has('read_stored_result')) sections.push( dynamicSection( diff --git a/src/chat/server/assistant-mcp-tools.ts b/src/chat/server/assistant-mcp-tools.ts index 6253b9726..ec40a2613 100644 --- a/src/chat/server/assistant-mcp-tools.ts +++ b/src/chat/server/assistant-mcp-tools.ts @@ -70,6 +70,7 @@ export interface AssistantMcpCall { /** A server-neutral model interface. Listing metadata never executes a server tool. */ export function assistantMcpTools(host: { connections: McpConnection[] + loadConnections?(): Promise catalog(connection: McpConnection, refresh: boolean): Promise propose(entry: CatalogEntry, args: Record): Promise read( @@ -95,6 +96,7 @@ export function assistantMcpTools(host: { const identity = host.discoveredEntries?.find((entry) => entry.id === id) if (!identity) throw new Error('List this server first to obtain a current entry ID.') + await host.loadConnections?.() const connection = host.connections.find( (entry) => entry.id === identity.serverId, ) @@ -150,6 +152,7 @@ export function assistantMcpTools(host: { refresh: z.boolean().default(false), }), }).server(async ({ serverId, offset = 0, refresh = false }) => { + await host.loadConnections?.() if (!serverId) return { servers: host.connections diff --git a/src/chat/server/conversation-database.ts b/src/chat/server/conversation-database.ts index 7c66d457e..281ba441e 100644 --- a/src/chat/server/conversation-database.ts +++ b/src/chat/server/conversation-database.ts @@ -11,20 +11,26 @@ export async function readConversationLifecycle( botId: string, conversationId: string, ) { - const [bot] = await db.execute< - { archived_at: number | null; deleted_at: number | null } & Record< - string, - unknown - > - >( - sql`SELECT trunc(extract(epoch FROM archived_at)*1000)::float8 AS archived_at,trunc(extract(epoch FROM deleted_at)*1000)::float8 AS deleted_at FROM chat_bots WHERE id=${botId}`, - ) - const [thread] = await db.execute< - { archived_at: number | null } & Record - >( - sql`SELECT trunc(extract(epoch FROM archived_at)*1000)::float8 AS archived_at FROM chat_conversation_threads WHERE conversation_id=${conversationId}`, - ) - return { bot: bot ?? null, thread: thread ?? null } + const [row] = await db.execute<{ + bot_id: string | null + bot_archived_at: number | null + bot_deleted_at: number | null + thread_id: string | null + thread_archived_at: number | null + }>(sql`SELECT b.id AS bot_id, + trunc(extract(epoch FROM b.archived_at)*1000)::float8 AS bot_archived_at, + trunc(extract(epoch FROM b.deleted_at)*1000)::float8 AS bot_deleted_at, + t.conversation_id AS thread_id, + trunc(extract(epoch FROM t.archived_at)*1000)::float8 AS thread_archived_at + FROM (SELECT 1) anchor + LEFT JOIN chat_bots b ON b.id=${botId} + LEFT JOIN chat_conversation_threads t ON t.conversation_id=${conversationId}`) + return { + bot: row?.bot_id + ? { archived_at: row.bot_archived_at, deleted_at: row.bot_deleted_at } + : null, + thread: row?.thread_id ? { archived_at: row.thread_archived_at } : null, + } } export async function conversationRetryReady(input: { retryId: string @@ -53,30 +59,30 @@ export async function readConversationRunContext( }, loadSavedActions = true, ) { - return db.transaction( - async (tx) => { - const [workspace] = await tx.execute< - { policy: unknown } & Record - >( - sql`SELECT w.policy FROM chat_workspaces w JOIN chat_memberships m ON m.workspace_id=w.id WHERE w.id=${identity.workspaceId} AND m.user_id=${identity.userId}`, - ) - const [bot] = await tx.execute>( - sql`SELECT id,workspace_id,parent_id,name,purpose,trunc(extract(epoch FROM created_at)*1000)::float8 AS created_at FROM chat_bots WHERE id=${identity.botId} AND workspace_id=${identity.workspaceId}`, - ) - if (!workspace || !bot) - throw new Error('Workspace access is unavailable.') - const policy = policySchema.parse(workspace.policy) - const recipes = loadSavedActions - ? Array.from( - await tx.execute>( - sql`SELECT id,workspace_id,title,description,code,created_at::float8 AS created_at FROM chat_recipes WHERE workspace_id=${identity.workspaceId} ORDER BY created_at LIMIT 30`, - ), - ) - : [] - return { bot, policy, recipes, userId: identity.userId } - }, - { isolationLevel: 'repeatable read', accessMode: 'read only' }, - ) + // One statement supplies a coherent snapshot without a transaction handshake. + const [row] = await db.execute<{ + policy: unknown + bot: Bot + recipes: Recipe[] + }>(sql`SELECT w.policy, + jsonb_build_object('id',b.id,'workspace_id',b.workspace_id,'parent_id',b.parent_id, + 'name',b.name,'purpose',b.purpose, + 'created_at',trunc(extract(epoch FROM b.created_at)*1000)::float8) AS bot, + CASE WHEN ${loadSavedActions} THEN COALESCE((SELECT jsonb_agg(r ORDER BY r.created_at) + FROM (SELECT id,workspace_id,title,description,code,created_at::float8 AS created_at + FROM chat_recipes WHERE workspace_id=w.id ORDER BY created_at LIMIT 30) r), + '[]'::jsonb) ELSE '[]'::jsonb END AS recipes + FROM chat_workspaces w + JOIN chat_memberships m ON m.workspace_id=w.id AND m.user_id=${identity.userId} + JOIN chat_bots b ON b.workspace_id=w.id AND b.id=${identity.botId} + WHERE w.id=${identity.workspaceId}`) + if (!row) throw new Error('Workspace access is unavailable.') + return { + bot: row.bot, + policy: policySchema.parse(row.policy), + recipes: row.recipes, + userId: identity.userId, + } } /** Source activateCopyImport publication checks, after the copy and identity have been authorized. */ diff --git a/src/chat/server/conversation-threads.ts b/src/chat/server/conversation-threads.ts index 0d898f913..10e0e0fae 100644 --- a/src/chat/server/conversation-threads.ts +++ b/src/chat/server/conversation-threads.ts @@ -12,7 +12,10 @@ import { type ThreadListItem, type ThreadSource, } from '../core/conversation-threads' -import { resolveConversationIdentity } from '../conversation-identity.server' +import { + resolveConversationIdentity, + ConversationIdentityError, +} from '../conversation-identity.server' import { canonicalCopyJson } from '../core/conversation-copy' import { hash } from './crypto' @@ -136,15 +139,22 @@ const listItem = ( export async function conversationThreadContext( identity: ConversationIdentity, ) { - await resolveConversationIdentity(identity) const [row] = await db.execute<{ - parent_conversation_id: string - source_message_id: string - source_context: string - }>( - sql`SELECT t.parent_conversation_id,t.source_message_id,t.source::text AS source_context ${from} WHERE c.id=${identity.conversationId} AND c.user_id=${identity.userId}::uuid AND b.workspace_id=${identity.workspaceId} AND b.deleted_at IS NULL`, - ) - return row + parent_conversation_id: string | null + source_message_id: string | null + source_context: string | null + }>(sql`SELECT t.parent_conversation_id,t.source_message_id,t.source::text AS source_context + FROM chat_conversations c + JOIN chat_bots b ON b.id=c.bot_id + JOIN chat_memberships m ON m.workspace_id=b.workspace_id AND m.user_id=c.user_id + LEFT JOIN chat_conversation_threads t ON t.conversation_id=c.id + WHERE c.id=${identity.conversationId} AND c.bot_id=${identity.botId} + AND c.user_id=${identity.userId}::uuid AND b.workspace_id=${identity.workspaceId} + AND b.deleted_at IS NULL`) + if (!row) throw new ConversationIdentityError() + return row.parent_conversation_id && + row.source_message_id && + row.source_context ? { parentConversationId: row.parent_conversation_id, sourceMessageId: row.source_message_id, diff --git a/src/chat/server/conversation.ts b/src/chat/server/conversation.ts index 4213d39c7..48aa7e281 100644 --- a/src/chat/server/conversation.ts +++ b/src/chat/server/conversation.ts @@ -23,10 +23,7 @@ import type { CopyFailure } from './conversation-copy-contract' import type { ConversationEnvironment } from './conversation-environment' import type { DurableObjectState } from '@cloudflare/workers-types' import { assistantDeviceTools } from './connected-devices' -import { - assistantToolDiscovery, - readSkillDirectory, -} from './assistant-discovery' +import { assistantToolDiscovery } from './assistant-discovery' import { wakeWorkspaceSync } from './workspace-sync' import { workflowUsage } from '../core/workflow-usage' import type { TaskUsage } from '../core/task-usage' @@ -1926,6 +1923,7 @@ export class Conversation extends DurableObject { } await this.save() const version = this.state.copyReadVersion + await this.activityOutbox.flush() await confirmCopyActivityPublication({ workspaceId: imported.identity.workspaceId, userId: imported.identity.userId, @@ -6638,61 +6636,59 @@ export class Conversation extends DurableObject { ) throw new Error('Write a message or attach a file.') if (this.hasMessage(input.messageId)) return { duplicate: true } - let attachments = await resolveMessageAttachments( - this.env, - { - workspaceId: input.bot.workspace_id, - userId: input.userId, - botId: input.bot.id, - conversationId: this.state.identity?.conversationId, - }, - fileIds, - ) - const resolvedReferences = references.length - ? await resolveMessageReferences( - this.env, - { - workspaceId: input.bot.workspace_id, - userId: input.userId, - botId: input.bot.id, - conversationId: this.state.identity?.conversationId, - }, - references, - { policy: input.policy, fixture: input.fixture }, - ) - : { references: [], attachments: [], skills: [], plugins: [] } - attachments = messageAttachmentsSchema.parse([ - ...new Map( - [...attachments, ...resolvedReferences.attachments].map((file) => [ - file.id, - file, - ]), - ).values(), - ]) - const selected = - !input.systemOne && - !input.proposeToolsOnly && - input.policy.allowChatModels - ? await resolveRunModel(this.env, { - userId: input.userId, - policy: input.policy, - fixture: input.fixture, - selection: input.runModel, - }) - : undefined - input = { - ...input, - runModel: selected?.selection, - memoryRecall: ( - await new Memories({ + const scope = { + workspaceId: input.bot.workspace_id, + userId: input.userId, + botId: input.bot.id, + conversationId: this.state.identity?.conversationId, + } + // Independent reads share the same authorized request scope. Settle all + // work before returning so a failure cannot outlive its database context. + const [filesRead, referencesRead, modelRead, memoryRead] = + await Promise.allSettled([ + resolveMessageAttachments(this.env, scope, fileIds), + references.length + ? resolveMessageReferences(this.env, scope, references, { + policy: input.policy, + fixture: input.fixture, + }) + : { references: [], attachments: [], skills: [], plugins: [] }, + !input.systemOne && + !input.proposeToolsOnly && + input.policy.allowChatModels + ? resolveRunModel(this.env, { + userId: input.userId, + policy: input.policy, + fixture: input.fixture, + selection: input.runModel, + }) + : undefined, + new Memories({ workspaceId: input.bot.workspace_id, userId: input.userId, conversationId: z .string() .min(1) .parse(this.state.identity!.conversationId), - }).preferences() - ).enabled, + }).preferences(), + ]) + if (filesRead.status === 'rejected') throw filesRead.reason + if (referencesRead.status === 'rejected') throw referencesRead.reason + if (modelRead.status === 'rejected') throw modelRead.reason + if (memoryRead.status === 'rejected') throw memoryRead.reason + const resolvedReferences = referencesRead.value + const attachments = messageAttachmentsSchema.parse([ + ...new Map( + [...filesRead.value, ...resolvedReferences.attachments].map( + (file) => [file.id, file], + ), + ).values(), + ]) + const selected = modelRead.value + input = { + ...input, + runModel: selected?.selection, + memoryRecall: memoryRead.value.enabled, } await validateAttachmentRequest( this.env, @@ -9056,13 +9052,67 @@ export class Conversation extends DurableObject { input.text, input.messageId, )) + const identity = this.state.identity + if (!identity?.conversationId) + throw new Error('Conversation identity is unavailable.') + const threadIdentity = { + ...identity, + conversationId: identity.conversationId, + } + // Account style, source context, and optional Kody enrichment are + // independent. Keep their authorization protocols, without serial waits. + const [preferencesRead, threadRead, kodyRead] = await Promise.allSettled([ + task.responsePreferences === null + ? readAccountPreferences(input.userId) + : undefined, + conversationThreadContext(threadIdentity), + Promise.all([ + kodyAvailable && input.bot.workspace_id === `personal:${input.userId}` + ? searchKodyMemory( + this.env, + input.userId, + input.bot.workspace_id, + input.text, + signal, + ).catch(() => []) + : [], + kodyAvailable + ? suggestKodyReferences( + this.env, + { workspaceId: input.bot.workspace_id, userId: input.userId }, + { policy: input.policy, fixture: input.fixture }, + input.text, + ).catch(() => []) + : [], + kodyAvailable && input.bot.workspace_id === `personal:${input.userId}` + ? readKodyGuidance( + this.env, + { workspaceId: input.bot.workspace_id, userId: input.userId }, + { policy: input.policy, fixture: input.fixture }, + AbortSignal.any([signal, AbortSignal.timeout(10000)]), + ).catch(() => undefined) + : undefined, + kodyAvailable + ? suggestKodySkills( + this.env, + { workspaceId: input.bot.workspace_id, userId: input.userId }, + input.text, + ).catch(() => []) + : [], + ]), + ]) if (task.responsePreferences === null) { - // Only new tasks opt in. Never apply today's settings to a legacy - // continuation, and never accept private preferences from request data. + const preferences = + preferencesRead.status === 'fulfilled' + ? preferencesRead.value + : undefined + if (!preferences) + throw new Error( + 'Response preferences could not be loaded for this task. Try again.', + ) + if (signal.aborted || this.state.assistantTask?.id !== task.id) + throw new Error('Stopped') try { - const preferences = await readAccountPreferences(input.userId) - if (signal.aborted || this.state.assistantTask?.id !== task.id) - throw new Error('Stopped') task.responsePreferences = responsePreferencesSnapshotSchema.parse({ revision: preferences.revision, response: preferences.response, @@ -9075,9 +9125,16 @@ export class Conversation extends DurableObject { ) } } + if (threadRead.status === 'rejected') throw threadRead.reason + if (kodyRead.status === 'rejected') throw kodyRead.reason if (signal.aborted) throw new Error('Stopped') - const threadIdentity = await this.authorizeIdentity(this.state.identity!) - const threadSource = await conversationThreadContext(threadIdentity) + const threadSource = threadRead.value + const [ + kodyMemories, + kodySuggestions, + kodyGuidance, + kodySkillSuggestions, + ] = kodyRead.value const selectedReferences = referenceInputsSchema.parse( task.references ?? input.references ?? [], ) @@ -9494,25 +9551,31 @@ export class Conversation extends DurableObject { ) : baseResultStore const storedResults = new StoredResults(resultStore, 12000, 'reject') - // References and plugin versions are authorized above. Inventory and - // skill metadata are independent reads; settle both before leaving their - // shared database context, including when inventory fails. - const [inventory, directory] = await Promise.allSettled([ - connectedMcpServers(this.env, input.userId, input.policy, undefined, { - workspaceId: input.bot.workspace_id, - versions: task.loadedPlugins, - }), - readSkillDirectory((cursor) => - new SkillCatalog(this.env, pluginScope).list( - { cursor }, - task.loadedPlugins, - ), - ), - ]) - if (inventory.status === 'rejected') throw inventory.reason - if (directory.status === 'rejected') throw directory.reason - const connections = inventory.value - const skillDirectory = directory.value + // Plain responses do not need external catalogs. Resolve inventory only + // for an explicit tool reference or when the model asks to discover tools. + const connections: McpConnection[] = [] + let inventoryRead: Promise | undefined + const loadConnections = () => + (inventoryRead ??= (async () => { + const current = await connectedMcpServers( + this.env, + input.userId, + input.policy, + undefined, + { + workspaceId: input.bot.workspace_id, + versions: task.loadedPlugins, + }, + ) + assertCurrentTask() + connections.splice(0, connections.length, ...current) + })()) + if ( + resolvedReferences.references.some( + (reference) => reference.kind === 'tool', + ) + ) + await loadConnections() const currentConnection = async (serverId: string) => { const current = ( await connectedMcpServers( @@ -9577,6 +9640,7 @@ export class Conversation extends DurableObject { ) const connectedTools = assistantMcpTools({ connections, + loadConnections, selectedTools, discoveredEntries: (task.discoveredMcpEntries ??= []), observe: (entry, result) => setupEvidence.register(entry.id, result), @@ -10194,45 +10258,6 @@ export class Conversation extends DurableObject { } }, }) - const [ - kodyMemories, - kodySuggestions, - kodyGuidance, - kodySkillSuggestions, - ] = await Promise.all([ - kodyAvailable && input.bot.workspace_id === `personal:${input.userId}` - ? searchKodyMemory( - this.env, - input.userId, - input.bot.workspace_id, - input.text, - signal, - ).catch(() => []) - : [], - kodyAvailable - ? suggestKodyReferences( - this.env, - { workspaceId: input.bot.workspace_id, userId: input.userId }, - { policy: input.policy, fixture: input.fixture }, - input.text, - ).catch(() => []) - : [], - kodyAvailable && input.bot.workspace_id === `personal:${input.userId}` - ? readKodyGuidance( - this.env, - { workspaceId: input.bot.workspace_id, userId: input.userId }, - { policy: input.policy, fixture: input.fixture }, - AbortSignal.any([signal, AbortSignal.timeout(10000)]), - ).catch(() => undefined) - : undefined, - kodyAvailable - ? suggestKodySkills( - this.env, - { workspaceId: input.bot.workspace_id, userId: input.userId }, - input.text, - ).catch(() => []) - : [], - ]) let instructionKey = '' let cachedInstructions: | ReturnType @@ -10254,7 +10279,6 @@ export class Conversation extends DurableObject { enabledTools: discovery.tools().map((tool) => tool.name), availableTools: discovery.directory, availableToolNames: assistantTools.map((tool) => tool.name), - skillDirectory, workflowInputs: workflowAdmission ? workflowInputReferences(workflowAdmission) : undefined, @@ -10388,17 +10412,26 @@ export class Conversation extends DurableObject { version: skill.version, }) await resolveTaskSkills(this.env, pluginScope, task) - const latest = await connectedMcpServers( - this.env, - input.userId, - input.policy, - undefined, - { - workspaceId: input.bot.workspace_id, - versions: task.loadedPlugins, - }, - ) - connections.splice(0, connections.length, ...latest) + if ( + [ + 'list_connected_tools', + 'inspect_connected_tool', + 'call_connected_tool', + ].includes(tool.toolName) + ) { + const latest = await connectedMcpServers( + this.env, + input.userId, + input.policy, + undefined, + { + workspaceId: input.bot.workspace_id, + versions: task.loadedPlugins, + }, + ) + connections.splice(0, connections.length, ...latest) + inventoryRead = Promise.resolve() + } } catch (error) { return stopModel( error instanceof Error diff --git a/src/chat/server/memory.ts b/src/chat/server/memory.ts index 0d89c4b71..1ca8315de 100644 --- a/src/chat/server/memory.ts +++ b/src/chat/server/memory.ts @@ -81,15 +81,34 @@ export class Memories { } } async preferences() { - await this.authorize() const [row] = await db - .select() - .from(chatMemoryPreferences) + .select({ + enabled: chatMemoryPreferences.enabled, + revision: chatMemoryPreferences.revision, + }) + .from(chatConversations) + .innerJoin(chatBots, eq(chatBots.id, chatConversations.botId)) + .innerJoin( + chatMemberships, + and( + eq(chatMemberships.workspaceId, chatBots.workspaceId), + eq(chatMemberships.userId, this.scope.userId), + ), + ) + .leftJoin( + chatMemoryPreferences, + eq(chatMemoryPreferences.conversationId, chatConversations.id), + ) .where( - eq(chatMemoryPreferences.conversationId, this.scope.conversationId), + and( + eq(chatConversations.id, this.scope.conversationId), + eq(chatConversations.userId, this.scope.userId), + eq(chatBots.workspaceId, this.scope.workspaceId), + isNull(chatBots.deletedAt), + ), ) - await this.authorize() - return { enabled: row?.enabled === true, revision: row?.revision ?? 0 } + if (!row) throw new MemoryError('Conversation access is unavailable.', 403) + return { enabled: row.enabled === true, revision: row.revision ?? 0 } } async setPreferences(input: unknown) { const value = z diff --git a/src/chat/server/run-usage.ts b/src/chat/server/run-usage.ts index 68cb20d72..6e57567f9 100644 --- a/src/chat/server/run-usage.ts +++ b/src/chat/server/run-usage.ts @@ -124,37 +124,39 @@ export async function reserveRunUsage( await tx.execute( sql`SELECT pg_advisory_xact_lock(hashtextextended(${'chat-run-usage:' + day},0))`, ) - const [receipt] = await tx.execute<{ - workspace_id: string - user_id: string - bot_id: string - scheduled: boolean - day: string - }>( - sql`SELECT workspace_id,user_id,bot_id,scheduled,day FROM chat_run_usage_receipts WHERE conversation_id=${identity.conversationId} AND run_id=${runId}`, - ) - if (receipt) { - if ( - receipt.workspace_id !== identity.workspaceId || - receipt.user_id !== identity.userId || - receipt.bot_id !== identity.botId || - receipt.scheduled !== scheduled - ) - throw new RunUsageConflictError() - return { status: 'duplicate' as const, day: receipt.day } - } const [usage] = await tx.execute<{ + receipt: { + workspace_id: string + user_id: string + bot_id: string + scheduled: boolean + day: string + } | null user_turns: number global_turns: number scheduled_turns: number user_spend: number global_spend: number }>(sql`SELECT + (SELECT jsonb_build_object('workspace_id',workspace_id,'user_id',user_id, + 'bot_id',bot_id,'scheduled',scheduled,'day',day) + FROM chat_run_usage_receipts WHERE conversation_id=${identity.conversationId} AND run_id=${runId}) AS receipt, COALESCE((SELECT turns FROM chat_daily_usage WHERE user_id=${identity.userId} AND day=${day}),0)::double precision AS user_turns, COALESCE((SELECT turns FROM chat_daily_usage WHERE user_id='__global' AND day=${day}),0)::double precision AS global_turns, COALESCE((SELECT turns FROM chat_scheduled_daily_usage WHERE user_id=${identity.userId}::uuid AND day=${day}),0)::double precision AS scheduled_turns, COALESCE((SELECT SUM(billed_micros) FROM chat_funded_spend WHERE user_id=${identity.userId}::uuid AND day=${day}),0)::double precision AS user_spend, COALESCE((SELECT SUM(billed_micros) FROM chat_funded_spend WHERE day=${day}),0)::double precision AS global_spend`) + const receipt = usage.receipt + if (receipt) { + if ( + receipt.workspace_id !== identity.workspaceId || + receipt.user_id !== identity.userId || + receipt.bot_id !== identity.botId || + receipt.scheduled !== scheduled + ) + throw new RunUsageConflictError() + return { status: 'duplicate' as const, day: receipt.day } + } if (!lift && usage.user_turns >= dailyLimit) throw new RunUsageAllowanceError('user') if (!lift && usage.global_turns >= 300) @@ -185,20 +187,33 @@ export async function reserveRunUsage( usage.global_spend + spend.reservationMicros > spend.globalCapMicros ) throw new RunUsageAllowanceError('global-spend') - await tx.execute( - sql`INSERT INTO chat_run_usage_receipts(conversation_id,run_id,workspace_id,user_id,bot_id,scheduled,day,created_at,attempt_id) VALUES(${identity.conversationId},${runId},${identity.workspaceId},${identity.userId}::uuid,${identity.botId},${scheduled},${day},${now},${attemptId}::uuid)`, - ) - if (spend) - await tx.execute( - sql`INSERT INTO chat_funded_spend(conversation_id,run_id,user_id,day,reserved_micros,billed_micros) VALUES(${identity.conversationId},${runId},${identity.userId}::uuid,${day},${spend.reservationMicros},${spend.reservationMicros})`, - ) - await tx.execute( - sql`INSERT INTO chat_daily_usage(user_id,day,turns) VALUES(${identity.userId},${day},1),('__global',${day},1) ON CONFLICT(user_id,day) DO UPDATE SET turns=chat_daily_usage.turns+1`, - ) - if (scheduled) - await tx.execute( - sql`INSERT INTO chat_scheduled_daily_usage(user_id,day,turns) VALUES(${identity.userId}::uuid,${day},1) ON CONFLICT(user_id,day) DO UPDATE SET turns=chat_scheduled_daily_usage.turns+1`, - ) + // Publish the receipt, funded reservation, and counters atomically in one + // statement while holding the existing receipt and day locks. + await tx.execute(sql`WITH receipt AS ( + INSERT INTO chat_run_usage_receipts(conversation_id,run_id,workspace_id,user_id,bot_id,scheduled,day,created_at,attempt_id) + VALUES(${identity.conversationId},${runId},${identity.workspaceId},${identity.userId}::uuid,${identity.botId},${scheduled},${day},${now},${attemptId}::uuid) + RETURNING conversation_id + ), turns AS ( + INSERT INTO chat_daily_usage(user_id,day,turns) + SELECT counters.user_id,${day},1 FROM receipt CROSS JOIN (VALUES (${identity.userId}),('__global')) counters(user_id) + ON CONFLICT(user_id,day) DO UPDATE SET turns=chat_daily_usage.turns+1 + )${ + spend + ? sql`, funded AS ( + INSERT INTO chat_funded_spend(conversation_id,run_id,user_id,day,reserved_micros,billed_micros) + SELECT conversation_id,${runId},${identity.userId}::uuid,${day},${spend.reservationMicros},${spend.reservationMicros} FROM receipt + )` + : sql`` + }${ + scheduled + ? sql`, scheduled AS ( + INSERT INTO chat_scheduled_daily_usage(user_id,day,turns) + SELECT ${identity.userId}::uuid,${day},1 FROM receipt + ON CONFLICT(user_id,day) DO UPDATE SET turns=chat_scheduled_daily_usage.turns+1 + )` + : sql`` + } + SELECT conversation_id FROM receipt`) return { status: 'reserved' as const, day } }) } diff --git a/src/db/client.ts b/src/db/client.ts index a9c165e24..9563d0a7a 100644 --- a/src/db/client.ts +++ b/src/db/client.ts @@ -24,7 +24,7 @@ const databaseStorage = new AsyncLocalStorage() function createPostgresClient(connectionString: string) { return postgres(connectionString, { - max: isIsolateRuntime() ? 5 : 1, + max: 5, idle_timeout: 20, connect_timeout: 10, fetch_types: !isIsolateRuntime(), From 613cffd5c6bda90117028d4e3124fc2b4bfe839e Mon Sep 17 00:00:00 2001 From: Tanner Linsley Date: Sun, 4 Oct 2026 23:31:06 -0600 Subject: [PATCH 2/2] perf(chat): configure production Hyperdrive connection pooling --- docs/tanchat-send-latency.md | 4 ++-- vite.config.ts | 1 + wrangler.jsonc | 6 ++++++ 3 files changed, 9 insertions(+), 2 deletions(-) diff --git a/docs/tanchat-send-latency.md b/docs/tanchat-send-latency.md index 929be80f5..b0d62c5ae 100644 --- a/docs/tanchat-send-latency.md +++ b/docs/tanchat-send-latency.md @@ -28,9 +28,9 @@ These numbers measure the beginning of durable admission to provider dispatch. T ## Remaining production work -The checked-in production configuration has no Hyperdrive binding. The host already supports HYPERDRIVE.connectionString and otherwise uses DATABASE_URL. A new pool must target this site's database, not the existing unrelated dashboard demo pool. +The production configuration now binds HYPERDRIVE to the cache-disabled tanstack-com pool, 6ec1513b87bb47e9897e5502727b0388, created after user approval on October 4. The host already supports HYPERDRIVE.connectionString and otherwise uses DATABASE_URL. A new pool must target this site's database, not the existing unrelated dashboard demo pool. -The proposed production pool is named tanstack-com, with binding HYPERDRIVE and query caching disabled. Local development must omit that production binding and continue using its optional DATABASE_URL. Creating the pool, binding it, and deploying are separate production changes awaiting authorization. +Cloudflare confirmed query caching is disabled and the origin connection limit is twenty. Local development omits the production binding and continues using its optional DATABASE_URL. The binding takes effect on the live site when this configuration is deployed. Local startup was verified with CLOUDFLARE_API_TOKEN and CLOUDFLARE_ACCOUNT_ID explicitly empty, and the temporary test server was stopped. No deployment was performed during pool provisioning. Hyperdrive pools connections to reduce TCP, TLS, and authentication setup. It does not remove every SQL round trip or relocate existing Durable Objects. Region placement needs a measurement against the actual database and conversation objects before changing hints. diff --git a/vite.config.ts b/vite.config.ts index a831e80f6..eda40a8d1 100644 --- a/vite.config.ts +++ b/vite.config.ts @@ -355,6 +355,7 @@ export default defineConfig(async ({ command, mode }) => { remoteBindings: useRemoteAi, config: (config) => { if (command !== 'serve') return + config.hyperdrive = [] if (!useRemoteAi) delete config.ai if (localEnv.CLOUDFLARE_ACCOUNT_ID) config.account_id = localEnv.CLOUDFLARE_ACCOUNT_ID diff --git a/wrangler.jsonc b/wrangler.jsonc index 5dedf631e..f5c0642a3 100644 --- a/wrangler.jsonc +++ b/wrangler.jsonc @@ -66,4 +66,10 @@ "bucket_name": "tanstack-notebook-projects", }, ], + "hyperdrive": [ + { + "binding": "HYPERDRIVE", + "id": "6ec1513b87bb47e9897e5502727b0388", + }, + ], }