perf(chat): unblock response delivery and measure send phases

docs/tanchat-send-latency.md

@@ -41,3 +41,13 @@ Sources: [Hyperdrive connection pooling](https://developers.cloudflare.com/hyper
 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.
+
+## Follow-up phase diagnosis, October 5
+
+A warm live Assistant reply was observed 5.284 seconds after sending, with a 1.35-second recorded model call. Browser automation and observation are included in that 5.284-second result. The difference is not a database measurement.
+
+Live Worker tracing on the next short reply measured 1,398 ms of wall time and 59 ms of CPU time for the Conversation.begin RPC. The existing publisher also coupled transcript delivery to sidebar activity publication: after both initial flushes started, another transcript flush could not begin until the sidebar flush completed. The two outboxes now have independent, coalesced publishers. Their durable ordering, retry alarms, producer receipts, and copy activation checks remain unchanged. A runtime regression holds sidebar publication open and verifies that later transcript updates still publish.
+
+The HTTP send path now overlaps workflow ownership, lifecycle metadata, and run context reads after resolving the authenticated conversation identity. All three reads settle before the request database context is released. Authorization, lifecycle rejection, workflow ownership rejection, and post-preparation checks remain in place.
+
+Structured chat_send_timing, chat_admission_timing, and chat_run_timing logs record cumulative milliseconds within each operation. They separate HTTP authentication and metadata, admission guards and preparation, model credentials and selection, usage reservation, enrichment, context preparation, first visible text, its durable save, and response completion. They do not log message text, credentials, or account identity. Admission and run logs use the message receipt ID for correlation. The first-text save is not a measurement of browser paint or stream delivery.

harness-tests/pending-runtime/conversation-runs-runtime.test.ts

@@ -1,6 +1,8 @@
 import type { SqlStorage } from '@cloudflare/workers-types'
-import { expect, it } from 'vitest'
+import { expect, it, vi } from 'vitest'
 import { conversationHarness } from './fixtures/conversation-runtime'
+import { BotActivityOutbox } from '../../src/chat/server/bot-activity'
+import { ConversationStream } from '../../src/chat/server/conversation-stream'
 const identity = {
   workspaceId: 'w',
   userId: '00000000-0000-4000-8000-000000000001',
@@ -322,3 +324,27 @@ it.each(['assistant', 'system-one', 'tools'] as const)(
     expect((await restored.snapshot()).queue?.paused).toBe(true)
   },
 )
+
+it('publishes new transcript updates while sidebar activity publication is blocked', async () => {
+  const h = await conversationHarness()
+  let releaseActivity = () => {}
+  const activityBlocked = new Promise<void>((resolve) => {
+    releaseActivity = resolve
+  })
+  const activity = vi
+    .spyOn(BotActivityOutbox.prototype, 'flush')
+    .mockImplementationOnce(() => activityBlocked)
+  const stream = vi.spyOn(ConversationStream.prototype, 'flush')
+  try {
+    await h.c.updateQueue({ type: 'pause', version: 0 })
+    const firstPublication = stream.mock.calls.length
+    expect(firstPublication).toBeGreaterThan(0)
+    await h.c.updateQueue({ type: 'resume', version: 1 })
+    expect(stream.mock.calls.length).toBeGreaterThan(firstPublication)
+  } finally {
+    releaseActivity()
+    await h.settle()
+    activity.mockRestore()
+    stream.mockRestore()
+  }
+})

src/chat/server/conversation-http.server.ts

@@ -364,10 +364,17 @@ export async function handleConversationSend(
     return jsonError('Not found.', 404)
   const originError = validateSameOriginRequest(request)
   if (originError) return jsonError(originError.message, originError.status)
+  const startedAt = Date.now()
+  const timings: Record<string, number> = {}
+  const mark = (phase: string) => {
+    timings[phase] = Date.now() - startedAt
+  }
   const user = await getAuthService().getCurrentUser(request)
+  mark('authentication')
   if (!user) return jsonError('Sign in to continue.', 401)
   if (!(await hasChatAccess(user)))
     return jsonError('Chat access is unavailable.', 403)
+  mark('access')
   const url = new URL(request.url)
   const workspace = z
     .string()
@@ -382,6 +389,7 @@ export async function handleConversationSend(
       workspaceId: workspace.data,
       ...(target === 'bot' ? { botId: conversationId } : { conversationId }),
     })
+    mark('identity')
     if (operation === 'read') {
       const { version } = z
         .object({
@@ -404,15 +412,22 @@ export async function handleConversationSend(
         ),
       })
     }
-    if (await readConversationWorkflowOwner(identity.conversationId))
+    // These reads share the freshly authorized identity but do not depend on
+    // each other. Settle them before releasing the request's database context.
+    const [ownerRead, lifecycleRead, contextRead] = await Promise.allSettled([
+      readConversationWorkflowOwner(identity.conversationId),
+      readConversationLifecycle(identity.botId, identity.conversationId),
+      operation === 'send' ? readConversationRunContext(identity) : undefined,
+    ])
+    mark('metadata')
+    if (ownerRead.status === 'rejected') throw ownerRead.reason
+    if (ownerRead.value)
       return jsonError(
         'Manage this workflow step from its owning conversation.',
         409,
       )
-    const lifecycle = await readConversationLifecycle(
-      identity.botId,
-      identity.conversationId,
-    )
+    if (lifecycleRead.status === 'rejected') throw lifecycleRead.reason
+    const lifecycle = lifecycleRead.value
     if (!lifecycle.bot) throw new ConversationIdentityError()
     if (
       operation === 'send' &&
@@ -465,12 +480,15 @@ export async function handleConversationSend(
         ? json(result.snapshot)
         : json({ error: result.error }, result.status)
     }
-    const loadContext = async () => ({
-      ...(await readConversationRunContext(identity)),
-      conversationId: identity.conversationId,
-      fixture: false,
-      appOrigin: url.origin,
-    })
+    const loadContext = async () => {
+      if (contextRead.status === 'rejected') throw contextRead.reason
+      return {
+        ...(contextRead.value ?? (await readConversationRunContext(identity))),
+        conversationId: identity.conversationId,
+        fixture: false,
+        appOrigin: url.origin,
+      }
+    }
     if (operation !== 'send')
       return (
         (await conversationControlApi(
@@ -486,7 +504,11 @@ export async function handleConversationSend(
         )) ?? jsonError('Not found.', 404)
       )
 
-    return await conversationSendApi(request, stub, await loadContext())
+    const context = await loadContext()
+    mark('context')
+    const response = await conversationSendApi(request, stub, context)
+    mark('admission')
+    return response
   } catch (error) {
     if (
       error instanceof ConversationIdentityError ||
@@ -496,5 +518,14 @@ export async function handleConversationSend(
     if (error instanceof z.ZodError)
       return jsonError('Check the form fields and try again.', 400)
     throw error
+  } finally {
+    if (operation === 'send')
+      console.info(
+        JSON.stringify({
+          event: 'chat_send_timing',
+          timings,
+          totalMs: Date.now() - startedAt,
+        }),
+      )
   }
 }

src/chat/server/conversation.ts

@@ -549,6 +549,8 @@ export class Conversation extends DurableObject<ConversationEnvironment> {
   >
   private activityOutbox!: BotActivityOutbox
   private publishing = false
+  private activityPublicationPending = false
+  private activityPublishing = false
   private publicationPending = false
   private ledger: UsageLedger
   private usageContext!: UsageContext
@@ -925,20 +927,35 @@ export class Conversation extends DurableObject<ConversationEnvironment> {
     if (due !== undefined) await this.scheduleWake(due)
     // The state and publication outboxes have committed. Publishing must not
     // hold up admission or model execution when a subscriber is slow.
+    // Transcript updates must not wait for the sidebar's PostgreSQL projection.
+    // Each outbox serializes its own durable publication and retains its retries.
     this.publicationPending = true
     if (!this.publishing) this.waitUntil(this.publishUpdates())
+    this.activityPublicationPending = true
+    if (!this.activityPublishing) this.waitUntil(this.publishActivityUpdates())
   }
   private async publishUpdates() {
     this.publishing = true
     try {
       while (this.publicationPending) {
         this.publicationPending = false
-        await Promise.all([this.flushStream(), this.flushActivity()])
+        await this.flushStream()
       }
     } finally {
       this.publishing = false
     }
   }
+  private async publishActivityUpdates() {
+    this.activityPublishing = true
+    try {
+      while (this.activityPublicationPending) {
+        this.activityPublicationPending = false
+        await this.flushActivity()
+      }
+    } finally {
+      this.activityPublishing = false
+    }
+  }
   private restoreCommittedState() {
     const saved = this.ctx.storage.sql
       .exec<{ json: string }>('SELECT json FROM state WHERE id=1')
@@ -6582,8 +6599,14 @@ export class Conversation extends DurableObject<ConversationEnvironment> {
   }
   async begin(input: RunInput) {
     return this.runWithRuntime(async () => {
+      const startedAt = Date.now()
+      const timings: Record<string, number> = {}
+      const mark = (phase: string) => {
+        timings[phase] = Date.now() - startedAt
+      }
       const admissionEpoch = this.state.transcriptEpoch
-      return this.withActiveBot(input, async () => {
+      const result = await this.withActiveBot(input, async () => {
+        mark('activeGuard')
         input = {
           ...input,
           retry: retrySendBindingSchema.optional().parse(input.retry),
@@ -6672,6 +6695,7 @@ export class Conversation extends DurableObject<ConversationEnvironment> {
                 .parse(this.state.identity!.conversationId),
             }).preferences(),
           ])
+        mark('preparation')
         if (filesRead.status === 'rejected') throw filesRead.reason
         if (referencesRead.status === 'rejected') throw referencesRead.reason
         if (modelRead.status === 'rejected') throw modelRead.reason
@@ -6697,6 +6721,7 @@ export class Conversation extends DurableObject<ConversationEnvironment> {
           selected?.connection,
         )
         await this.guardActive(input)
+        mark('activeRecheck')
         if (this.state.transcriptEpoch !== admissionEpoch)
           throw new ReferenceError(
             'This conversation was reset while preparing the request. Review it before sending again.',
@@ -6781,6 +6806,15 @@ export class Conversation extends DurableObject<ConversationEnvironment> {
           resolvedReferences.references,
         )
       })
+      mark('accepted')
+      console.info(
+        JSON.stringify({
+          event: 'chat_admission_timing',
+          messageId: input.messageId,
+          timings,
+        }),
+      )
+      return result
     })
   }
   async sendReceipt(messageId: string) {
@@ -8292,6 +8326,11 @@ export class Conversation extends DurableObject<ConversationEnvironment> {
     })
   }
   private async run(input: RunInput) {
+    const startedAt = Date.now()
+    const timings: Record<string, number> = {}
+    const mark = (phase: string) => {
+      timings[phase] = Date.now() - startedAt
+    }
     const executionAbort = this.abort!
     const signal = executionAbort.signal
     const contextScope = {
@@ -8806,6 +8845,7 @@ export class Conversation extends DurableObject<ConversationEnvironment> {
       const credentials = input.fixture
         ? null
         : await readCredentials(this.env, input.userId)
+      mark('credentials')
       const kodyAvailable = input.policy.allowKody && !!credentials?.kody
       const selected =
         input.runModel || input.policy.allowChatModels
@@ -8820,6 +8860,7 @@ export class Conversation extends DurableObject<ConversationEnvironment> {
               credentials,
             )
           : undefined
+      mark('modelSelection')
       const connection: Connection = selected?.connection ??
         credentials?.connection ?? {
           provider: 'included',
@@ -8876,6 +8917,7 @@ export class Conversation extends DurableObject<ConversationEnvironment> {
           input.policy,
           connection.provider === 'included' || input.policy.allowJev,
         )
+      mark('usageReservation')
       if (!input.fixture && noChatModels && route.type !== 'recipe') {
         modelAllowed = true
         try {
@@ -9101,6 +9143,7 @@ export class Conversation extends DurableObject<ConversationEnvironment> {
             : [],
         ]),
       ])
+      mark('enrichment')
       if (task.responsePreferences === null) {
         const preferences =
           preferencesRead.status === 'fulfilled'
@@ -10342,6 +10385,7 @@ export class Conversation extends DurableObject<ConversationEnvironment> {
           await this.save()
         },
       })
+      mark('toolsAndInstructions')
       const stream = chat({
         threadId: this.state.identity!.conversationId,
         runId: this.state.activeRun!,
@@ -10535,6 +10579,7 @@ export class Conversation extends DurableObject<ConversationEnvironment> {
         modelOptions: selected?.modelOptions,
         abortController: this.abort,
       })
+      mark('modelContext')
       let lastSave = 0
       const visibleResponses = new Set<string>()
       const reasoningFilters = new Map<string, ReasoningTextFilter>()
@@ -10576,7 +10621,17 @@ export class Conversation extends DurableObject<ConversationEnvironment> {
             ? chunk.messageId
             : undefined
         if (firstTextMessageId !== undefined || Date.now() - lastSave > 250) {
+          if (
+            firstTextMessageId !== undefined &&
+            timings.firstText === undefined
+          )
+            mark('firstText')
           await this.save()
+          if (
+            firstTextMessageId !== undefined &&
+            timings.firstTextSaved === undefined
+          )
+            mark('firstTextSaved')
           if (firstTextMessageId !== undefined)
             visibleResponses.add(firstTextMessageId)
           lastSave = Date.now()
@@ -10637,6 +10692,14 @@ export class Conversation extends DurableObject<ConversationEnvironment> {
       this.trace('error', this.state.error)
       this.state.status = 'error'
     } finally {
+      mark('responseComplete')
+      console.info(
+        JSON.stringify({
+          event: 'chat_run_timing',
+          messageId: input.messageId,
+          timings,
+        }),
+      )
       this.state.resumingTask = undefined
       clearTimeout(timeout)
       if (!input.fixture && this.state.identity?.conversationId) {