@@ -68,7 +68,7 @@ import {
6868 WorkspaceRepository ,
6969} from "@dx/domain" ;
7070import { D1Client } from "@effect/sql-d1" ;
71- import { dispatch } from "@flue/runtime" ;
71+ import { dispatch , getAgentInstance } from "@flue/runtime" ;
7272import { Effect , Layer , Match , Option , Result , Schema } from "effect" ;
7373import { Hono } from "hono" ;
7474import { DxAgent } from "../agents/dx-agent.js" ;
@@ -341,6 +341,15 @@ type InitialThreadDispatcher = (
341341const dispatchInitialThreadMessage : InitialThreadDispatcher = ( request ) =>
342342 dispatch ( DxAgent , request ) ;
343343
344+ /**
345+ * Starts the Thread's agent Durable Object without changing it, so its cold
346+ * start overlaps model resolution and persistence instead of following them.
347+ */
348+ type InitialThreadAgentWarmer = ( threadId : ThreadId ) => Promise < unknown > ;
349+
350+ const warmInitialThreadAgent : InitialThreadAgentWarmer = ( threadId ) =>
351+ getAgentInstance ( DxAgent , threadId ) ;
352+
344353const resolveCreationResponseDetail = <
345354 Detail ,
346355 DetailError ,
@@ -377,6 +386,7 @@ const resolveCreationResponseDetail = <
377386export const createThreadRoutes = (
378387 dispatchInitialMessage : InitialThreadDispatcher = dispatchInitialThreadMessage ,
379388 titleAgent : DxTitleAgentRunner = runDxTitleAgent ,
389+ warmInitialAgent : InitialThreadAgentWarmer = warmInitialThreadAgent ,
380390) => {
381391 const threadRoutes = new Hono < AppEnv > ( ) ;
382392
@@ -386,6 +396,18 @@ export const createThreadRoutes = (
386396 let requestAdmittedAt = startedAt ;
387397 let sourceAuthorizedAt = startedAt ;
388398 let threadPersistedAt = startedAt ;
399+ let agentWarmedAt : number | undefined ;
400+ let detailResolvedAt : number | undefined ;
401+ let initialMessageAdmittedAt : number | undefined ;
402+ // Creation outlives the request: once it reaches Core, a closed tab or a
403+ // dropped connection cannot stop the Thread or its first message.
404+ const execution = ( ( ) => {
405+ try {
406+ return context . executionCtx ;
407+ } catch {
408+ return undefined ;
409+ }
410+ } ) ( ) ;
389411 let settleTitlePersisted : ( ( persisted : boolean ) => void ) | undefined ;
390412 let recoveredConcurrentCreation = false ;
391413 const operation = Effect . gen ( function * ( ) {
@@ -403,13 +425,7 @@ export const createThreadRoutes = (
403425 const principal = context . get ( "principal" ) ;
404426 const requestedThreadId = input . threadId ;
405427 // DxTitleAgent needs an execution context to outlive the response.
406- const titleExecution = ( ( ) => {
407- try {
408- return context . executionCtx ;
409- } catch {
410- return undefined ;
411- }
412- } ) ( ) ;
428+ const titleExecution = execution ;
413429 const existingThread =
414430 requestedThreadId === undefined
415431 ? undefined
@@ -647,6 +663,23 @@ export const createThreadRoutes = (
647663 threadPersistedAt = Date . now ( ) ;
648664 settleTitlePersisted ?.( ! recoveredConcurrentCreation ) ;
649665 const initialMessage = input . initialMessage ;
666+ // The Thread now exists, so its agent is real: start it while the
667+ // detail and author are read, ahead of the dispatch below.
668+ if ( initialMessage !== undefined )
669+ void warmInitialAgent ( thread . id ) . then (
670+ ( ) => {
671+ agentWarmedAt = Date . now ( ) ;
672+ } ,
673+ ( ) => undefined ,
674+ ) ;
675+ // The author identity is read beside the detail, not after it.
676+ const ownerIdentity =
677+ initialMessage === undefined
678+ ? undefined
679+ : userIdentity ( db , thread . ownerUserId ) . then (
680+ ( identity ) => ( { identity } ) ,
681+ ( ) => ( { identity : undefined } ) ,
682+ ) ;
650683 const detail = yield * resolveCreationResponseDetail (
651684 threadDetailData ( context . env . DB , db , thread ) ,
652685 initialMessage === undefined
@@ -657,13 +690,17 @@ export const createThreadRoutes = (
657690 ) ,
658691 ) ,
659692 ) ;
693+ detailResolvedAt = Date . now ( ) ;
660694 let initialSubmission : { readonly submissionId : string } | undefined ;
661695 const messageSourceThreadId = context . get ( "messageSourceThreadId" ) ;
662- if ( initialMessage !== undefined ) {
663- const owner = yield * Effect . tryPromise ( {
664- try : ( ) => userIdentity ( db , thread . ownerUserId ) ,
665- catch : ( ) => new ThreadInitialAdmissionUnavailable ( ) ,
666- } ) ;
696+ if ( initialMessage !== undefined && ownerIdentity !== undefined ) {
697+ const owner = yield * Effect . flatMap (
698+ Effect . promise ( ( ) => ownerIdentity ) ,
699+ ( { identity } ) =>
700+ identity === undefined
701+ ? Effect . fail ( new ThreadInitialAdmissionUnavailable ( ) )
702+ : Effect . succeed ( identity ) ,
703+ ) ;
667704 const receipt = yield * Effect . tryPromise ( {
668705 try : ( ) =>
669706 dispatchInitialMessage ( {
@@ -693,6 +730,7 @@ export const createThreadRoutes = (
693730 catch : ( ) => new ThreadInitialAdmissionUnavailable ( ) ,
694731 } ) ;
695732 initialSubmission = { submissionId : receipt . submissionId } ;
733+ initialMessageAdmittedAt = Date . now ( ) ;
696734 const observations = submissionStartupObservations (
697735 requestId ,
698736 thread . id ,
@@ -717,16 +755,23 @@ export const createThreadRoutes = (
717755 return { body, threadId : thread . id } ;
718756 } ) ;
719757
720- const result = await Effect . runPromise ( Effect . result ( operation ) ) ;
721- // A creation that failed before persisting leaves no row to title.
722- settleTitlePersisted ?.( false ) ;
758+ const completion = Effect . runPromise ( Effect . result ( operation ) ) . then (
759+ async ( result ) => {
760+ // A creation that failed before persisting leaves no row to title.
761+ settleTitlePersisted ?.( false ) ;
762+ if ( Result . isSuccess ( result ) )
763+ await scheduleRealtimeInvalidation (
764+ ( ) => context . executionCtx ,
765+ context . env ,
766+ result . success . threadId ,
767+ "thread.invalidated" ,
768+ ) ;
769+ return result ;
770+ } ,
771+ ) ;
772+ execution ?. waitUntil ( completion ) ;
773+ const result = await completion ;
723774 if ( Result . isSuccess ( result ) ) {
724- await scheduleRealtimeInvalidation (
725- ( ) => context . executionCtx ,
726- context . env ,
727- result . success . threadId ,
728- "thread.invalidated" ,
729- ) ;
730775 const responseReadyAt = Date . now ( ) ;
731776 const decodedCorrelation = Schema . decodeUnknownResult (
732777 StartupTraceCorrelation ,
@@ -764,7 +809,21 @@ export const createThreadRoutes = (
764809 responseReadyAt ,
765810 ) ,
766811 ] ;
767- context . header ( "Server-Timing" , startupServerTiming ( observations ) ) ;
812+ // Header-only marks between persistence and the response.
813+ const responseMarks = [
814+ [ "agent_warmed" , agentWarmedAt ] ,
815+ [ "detail_resolved" , detailResolvedAt ] ,
816+ [ "initial_message_admitted" , initialMessageAdmittedAt ] ,
817+ ] as const ;
818+ context . header (
819+ "Server-Timing" ,
820+ [
821+ startupServerTiming ( observations ) ,
822+ ...responseMarks . flatMap ( ( [ phase , at ] ) =>
823+ at === undefined ? [ ] : [ `dx_${ phase } ;dur=${ at - startedAt } ` ] ,
824+ ) ,
825+ ] . join ( ", " ) ,
826+ ) ;
768827 await scheduleStartupPersistence (
769828 ( ) => context . executionCtx ,
770829 recordStartupPhases ( context . env . DB , observations ) ,
0 commit comments