Use Redis Streams for active runs (#1778)

Author: neoxelox

.tmuxinator.yml

@@ -10,5 +10,5 @@ windows:
       panes:
         - pnpm build --filter='./packages/**'
         - cd apps/web && pnpm workers:build
-  - docker: docker compose up db redis mailpit --menu=false
+  - docker: docker compose up db redis redisinsight mailpit --menu=false
   - studio: cd packages/core && pnpm db:studio

AGENTS.md

@@ -231,8 +231,6 @@ When implementing new features, follow this order:
 6. **Stores**: Create SWR hooks with optimistic updates
 7. **UI Components**: Build interface following design patterns
 8. **Routes**: Add navigation and routing
-9. **Testing**: Validate with linting and type checking
-10. **Code Formatting**: **ALWAYS run `pnpm prettier`** to standardize patterns
 
 ## Action Organization Patterns
 

apps/gateway/src/routes/api/v3/conversations/attach/attach.handler.ts

@@ -15,7 +15,7 @@ import { AttachRoute } from './attach.route'
 // @ts-expect-error: streamSSE has type issues with zod-openapi
 export const attachHandler: AppRouteHandler<AttachRoute> = async (ctx) => {
   const { conversationUuid } = ctx.req.valid('param')
-  const { stream, interactive } = ctx.req.valid('json')
+  const { stream } = ctx.req.valid('json')
   const workspace = ctx.get('workspace')
 
   const run = await unsafelyFindActiveRun(conversationUuid).then((r) => r.unwrap()) // prettier-ignore
@@ -31,33 +31,27 @@ export const attachHandler: AppRouteHandler<AttachRoute> = async (ctx) => {
     .then((r) => r.unwrap())
 
   const args = { run, project, workspace }
-  if (stream) return handleStreamingMode(ctx, args, interactive)
-  return await handleNonStreamingMode(ctx, args, interactive)
+  if (stream) return handleStreamingMode(ctx, args)
+  return await handleNonStreamingMode(ctx, args)
 }
 
 async function handleStreamingMode(
   ctx: Parameters<AppRouteHandler<AttachRoute>>[0],
   args: Parameters<typeof attachRun>[0],
-  interactive: boolean,
 ) {
   return streamSSE(
     ctx,
     async (stream) => {
-      let abortSignal: AbortSignal | undefined
-      if (interactive) {
-        const abortController = new AbortController()
-        stream.onAbort(() => {
-          abortController.abort()
-          stream.close()
-        })
-        abortSignal = abortController.signal
-      } else {
-        stream.onAbort(() => stream.close())
-      }
+      const abortController = new AbortController()
+      stream.onAbort(() => {
+        abortController.abort()
+        stream.close()
+      })
+      const abortSignal = abortController.signal
 
       try {
         let id = 0
-        const result = await attachRun({
+        await attachRun({
           ...args,
           abortSignal: abortSignal,
           onEvent: ({ event, data }) => {
@@ -68,26 +62,17 @@ async function handleStreamingMode(
             })
           },
         }).then((r) => r.unwrap())
-
-        // Wait for stream to finish
-        const error = await result.error
-        if (error) throw error
       } catch (error) {
         // Handle abort errors gracefully - don't log them as actual errors
-        if (isAbortError(error)) {
-          // Client disconnected, close stream quietly
-          return
-        }
+        if (isAbortError(error)) return
 
         // Re-throw other errors to be handled by the error callback
         throw error
       }
     },
     (error: Error) => {
-      // Don't log abort errors as they are expected when clients disconnect
-      if (isAbortError(error)) {
-        return Promise.resolve()
-      }
+      // Handle abort errors gracefully - don't log them as actual errors
+      if (isAbortError(error)) return Promise.resolve()
 
       const unknownError = getUnknownError(error)
       if (unknownError) captureException(error)
@@ -100,21 +85,15 @@ async function handleStreamingMode(
 async function handleNonStreamingMode(
   ctx: Parameters<AppRouteHandler<AttachRoute>>[0],
   args: Parameters<typeof attachRun>[0],
-  interactive: boolean,
 ) {
-  let abortSignal: AbortSignal | undefined
-  if (interactive) {
-    abortSignal = ctx.req.raw.signal // FIXME: this is not working
-  }
-
+  const abortSignal = ctx.req.raw.signal // FIXME: this is not working
   const result = await attachRun({ ...args, abortSignal }).then((r) => r.unwrap()) // prettier-ignore
 
   // Wait for stream to finish
-  const error = await result.error
+  const error = result.error
   if (error) throw error
 
-  const response = (await result.lastResponse)!
-  const body = runPresenter({ response }).unwrap()
+  const body = runPresenter({ response: result.lastResponse! }).unwrap()
 
   return ctx.json(body)
 }

apps/gateway/src/routes/api/v3/conversations/attach/attach.route.ts

@@ -21,7 +21,6 @@ export const attachRoute = createRoute({
         [http.MediaTypes.JSON]: {
           schema: z.object({
             stream: z.boolean().default(false),
-            interactive: z.boolean().default(false),
           }),
         },
       },

apps/gateway/src/routes/api/v3/conversations/chat/chat.handler.ts

@@ -4,10 +4,7 @@ import { runPresenter } from '$/presenters/runPresenter'
 import { ChatRoute } from '$/routes/api/v3/conversations/chat/chat.route'
 import { Message as LegacyMessage } from '@latitude-data/constants/legacyCompiler'
 import { getUnknownError } from '@latitude-data/core/lib/getUnknownError'
-import {
-  awaitClientToolResult,
-  ToolHandler,
-} from '@latitude-data/core/lib/streamManager/clientTools/handlers'
+import { buildClientToolHandlersMap } from '@latitude-data/core/lib/streamManager/clientTools/handlers'
 import { streamToGenerator } from '@latitude-data/core/lib/streamToGenerator'
 import { addMessages } from '@latitude-data/core/services/documentLogs/addMessages/index'
 import { BACKGROUND, telemetry } from '@latitude-data/core/telemetry'
@@ -82,10 +79,3 @@ export const chatHandler: AppRouteHandler<ChatRoute> = async (c) => {
 
   return c.json(body, 200)
 }
-
-function buildClientToolHandlersMap(tools: string[]) {
-  return tools.reduce((acc: Record<string, ToolHandler>, toolName: string) => {
-    acc[toolName] = awaitClientToolResult
-    return acc
-  }, {})
-}

apps/gateway/src/routes/api/v3/projects/versions/documents/run/run.handler.ts

@@ -9,16 +9,14 @@ import { LogSources } from '@latitude-data/core/constants'
 import { BadRequestError } from '@latitude-data/core/lib/errors'
 import { getUnknownError } from '@latitude-data/core/lib/getUnknownError'
 import { isAbortError } from '@latitude-data/core/lib/isAbortError'
-import {
-  awaitClientToolResult,
-  ToolHandler,
-} from '@latitude-data/core/lib/streamManager/clientTools/handlers'
+import { buildClientToolHandlersMap } from '@latitude-data/core/lib/streamManager/clientTools/handlers'
 import { streamToGenerator } from '@latitude-data/core/lib/streamToGenerator'
 import { runDocumentAtCommit } from '@latitude-data/core/services/commits/runDocumentAtCommit'
 import { enqueueRun } from '@latitude-data/core/services/runs/enqueue'
 import { isFeatureEnabledByName } from '@latitude-data/core/services/workspaceFeatures/isFeatureEnabledByName'
 import { BACKGROUND } from '@latitude-data/core/telemetry'
 import { streamSSE } from 'hono/streaming'
+import type { Context } from 'hono'
 import { RunRoute } from './run.route'
 
 // https://github.com/honojs/middleware/issues/735
@@ -59,34 +57,111 @@ export const runHandler: AppRouteHandler<RunRoute> = async (c) => {
   }
 
   if (background) {
-    if (!runsEnabled) {
-      throw new BadRequestError(
-        'Background runs are not enabled for this workspace',
-      )
-    }
-
-    const { run } = await enqueueRun({
-      document: document,
-      commit: commit,
-      project: project,
-      workspace: workspace,
-      parameters: parameters,
-      customIdentifier: customIdentifier,
-      tools: tools,
-      userMessage: userMessage,
-      source: source,
-    }).then((r) => r.unwrap())
-
-    return c.json({ uuid: run.uuid })
+    return await handleBackgroundRun({
+      c,
+      workspace,
+      document,
+      commit,
+      project,
+      parameters,
+      customIdentifier,
+      tools,
+      userMessage,
+      source,
+      runsEnabled,
+    })
   }
 
-  const result = await runDocumentAtCommit({
+  return await handleForegroundRun({
+    c,
     workspace,
     document,
     commit,
     parameters,
     customIdentifier,
     source: __internal?.source ?? LogSources.API,
+    useSSE,
+    tools,
+    userMessage,
+  })
+}
+
+async function handleBackgroundRun({
+  c,
+  workspace,
+  document,
+  commit,
+  project,
+  parameters,
+  customIdentifier,
+  tools,
+  userMessage,
+  source,
+  runsEnabled,
+}: {
+  c: Context
+  workspace: any
+  document: any
+  commit: any
+  project: any
+  parameters: any
+  customIdentifier: any
+  tools: any
+  userMessage: any
+  source: any
+  runsEnabled: boolean
+}) {
+  if (!runsEnabled) {
+    throw new BadRequestError(
+      'Background runs are not enabled for this workspace',
+    )
+  }
+
+  const { run } = await enqueueRun({
+    document: document,
+    commit: commit,
+    project: project,
+    workspace: workspace,
+    parameters: parameters,
+    customIdentifier: customIdentifier,
+    tools: tools,
+    userMessage: userMessage,
+    source: source,
+  }).then((r) => r.unwrap())
+
+  return c.json({ uuid: run.uuid })
+}
+
+async function handleForegroundRun({
+  c,
+  workspace,
+  document,
+  commit,
+  parameters,
+  customIdentifier,
+  source,
+  useSSE,
+  tools,
+  userMessage,
+}: {
+  c: Context
+  workspace: any
+  document: any
+  commit: any
+  parameters: any
+  customIdentifier: any
+  source: any
+  useSSE: boolean
+  tools: any
+  userMessage: any
+}) {
+  const result = await runDocumentAtCommit({
+    workspace,
+    document,
+    commit,
+    parameters,
+    customIdentifier,
+    source,
     abortSignal: c.req.raw.signal, // FIXME: This does not seem to work
     context: BACKGROUND({ workspaceId: workspace.id }),
     tools: useSSE ? buildClientToolHandlersMap(tools ?? []) : {},
@@ -152,10 +227,3 @@ export const runHandler: AppRouteHandler<RunRoute> = async (c) => {
 
   return c.json(body)
 }
-
-export function buildClientToolHandlersMap(tools: string[]) {
-  return tools.reduce((acc: Record<string, ToolHandler>, toolName: string) => {
-    acc[toolName] = awaitClientToolResult
-    return acc
-  }, {})
-}

apps/gateway/src/server.ts

@@ -53,9 +53,7 @@ if (cluster.isPrimary) {
       hostname: HOSTNAME,
       port: Number(PORT),
       serverOptions: {
-        keepAliveTimeout: process.env.KEEP_ALIVE_TIMEOUT
-          ? Number(process.env.KEEP_ALIVE_TIMEOUT)
-          : 601000,
+        keepAliveTimeout: env.KEEP_ALIVE_TIMEOUT,
       },
     },
     () => {

apps/web/package.json

@@ -25,7 +25,7 @@
     "@latitude-data/constants": "workspace:*",
     "@latitude-data/core": "workspace:*",
     "@latitude-data/env": "workspace:^",
-    "@latitude-data/sdk": "5.2.0-beta.5",
+    "@latitude-data/sdk": "workspace:^",
     "@latitude-data/socket.io-react-hook": "2.4.5",
     "@latitude-data/telemetry": "workspace:*",
     "@latitude-data/web-ui": "workspace:*",

apps/web/src/app/(actions)/actions/[actionType]/_lib/index.tsx

@@ -41,15 +41,8 @@ export function ClientPage<T extends ActionType = ActionType>({
     setTimeout(async () => {
       try {
         const [result, error] = await executeBackendAction({ type, parameters })
-
-        if (error) {
-          setError(error)
-        } else if (result) {
-          await executeFrontendAction({
-            type,
-            parameters: result as ActionFrontendParameters<T>,
-          })
-        }
+        if (error) setError(error)
+        if (result) await executeFrontendAction({ type, parameters: result })
         setTimeout(() => setEnded(true), 5000)
       } catch (error) {
         setError(error as Error)

apps/web/src/app/(private)/projects/[projectId]/versions/[commitUuid]/documents/[documentUuid]/(withTabs)/evaluations/[evaluationUuid]/editor/_components/EvaluationEditor/Playground/index.tsx

@@ -133,6 +133,7 @@ export const Playground = memo(
               runPromptFn={runPromptFn}
               abortCurrentStream={abortCurrentStream}
               hasActiveStream={hasActiveStream}
+              isRunStream={false}
             />
           )}
         </div>