fix(knowledge): record the Slack run's status after its turn is saved

- The Slack Assistant now writes its run's one terminal status after its
  outcome and response are persisted, from the final outcome, so a failed
  save ends the run as an error instead of complete.
- A Stop lookup that fails no longer skips that write: the turn is treated
  as not stopped, logged, and settled as an error.
- The orphaned-run suite deletes the sweep cursor before each test and in
  teardown, so no later suite starts from its leftover position.

Author: waleedlatif1

apps/sim/lib/knowledge/application/slack-search/assistant.integration.ts

@@ -22,6 +22,8 @@ const hoisted = vi.hoisted(() => ({
   /** The turn's own controller, which a Stop aborts through its registered stream. */
   controller: new AbortController(),
   stopped: vi.fn(async () => false),
+  outcome: vi.fn(async () => undefined),
+  finalize: vi.fn(async () => ({ appendedAssistant: true })),
 }))
 vi.mock('@/lib/knowledge/application/slack-search/authorization', () => ({
   authorizeSlackSearchInstallation: async () => ({
@@ -46,7 +48,7 @@ vi.mock('@/lib/knowledge/application/operations', () => ({
   knowledgeOperations: { search: { organizationOperation: { id: 'knowledge.search' } } },
 }))
 vi.mock('@/lib/knowledge/application/slack-search/repository', () => ({
-  recordSlackSearchOutcome: async () => undefined,
+  recordSlackSearchOutcome: hoisted.outcome,
 }))
 vi.mock('@/lib/knowledge/application/slack-search/turns', () => ({
   requireSlackSearchTurnLease: async () => undefined,
@@ -64,7 +66,7 @@ vi.mock('@/lib/mothership/application/load-search-integrations', () => ({
 }))
 vi.mock('@/lib/mothership/chat/payload', () => mothershipChatPayloadMock)
 vi.mock('@/lib/mothership/chat/terminal-state', () => ({
-  finalizeAssistantTurn: async () => ({ appendedAssistant: true }),
+  finalizeAssistantTurn: hoisted.finalize,
 }))
 vi.mock('@/lib/mothership/environment-context', () => mothershipEnvironmentContextMock)
 vi.mock('@/lib/mothership/request/lifecycle/headless', () => ({
@@ -192,8 +194,17 @@ describe('Slack Assistant run record', () => {
 
   beforeEach(() => {
     hoisted.stopped.mockResolvedValue(false)
+    hoisted.outcome.mockResolvedValue(undefined)
+    hoisted.finalize.mockResolvedValue({ appendedAssistant: true })
   })
 
+  const answered = {
+    success: true,
+    content: 'Answer',
+    contentBlocks: [],
+    toolCalls: [],
+  }
+
   it('records a completed turn as complete', async () => {
     hoisted.lifecycle.mockResolvedValueOnce({
       success: true,
@@ -237,4 +248,40 @@ describe('Slack Assistant run record', () => {
     expect(outcome).toBeInstanceOf(Error)
     expect(run.status).toBe('error')
   })
+
+  it('records a failed turn as an error even when its Stop cannot be looked up', async () => {
+    hoisted.stopped.mockRejectedValue(new Error('database unavailable'))
+    hoisted.lifecycle.mockResolvedValueOnce({
+      success: false,
+      error: 'worker failed',
+      content: '',
+      contentBlocks: [],
+      toolCalls: [],
+    })
+
+    const { run, outcome } = await slackTurn()
+
+    expect(outcome).toBeInstanceOf(Error)
+    expect(run.status).toBe('error')
+  })
+
+  it('records an answered turn as an error when its response is not saved', async () => {
+    hoisted.lifecycle.mockResolvedValueOnce(answered)
+    hoisted.finalize.mockResolvedValue({ appendedAssistant: false })
+
+    const { run, outcome } = await slackTurn()
+
+    expect(outcome).toBeInstanceOf(Error)
+    expect(run.status).toBe('error')
+  })
+
+  it('records an answered turn as an error when its outcome is not saved', async () => {
+    hoisted.lifecycle.mockResolvedValueOnce(answered)
+    hoisted.outcome.mockRejectedValueOnce(new Error('outcome write failed'))
+
+    const { run, outcome } = await slackTurn()
+
+    expect(outcome).toBeInstanceOf(Error)
+    expect(run.status).toBe('error')
+  })
 })

apps/sim/lib/knowledge/application/slack-search/assistant.ts

@@ -3,7 +3,7 @@ import type {
   SlackInstallationPrincipal,
 } from '@sim/auth/principal'
 import { createLogger } from '@sim/logger'
-import { toError } from '@sim/utils/errors'
+import { getErrorMessage, toError } from '@sim/utils/errors'
 import { generateId } from '@sim/utils/id'
 import { isRecordLike } from '@sim/utils/object'
 import { resolveOrganizationBillingAttribution } from '@/lib/billing/core/billing-attribution'
@@ -317,20 +317,6 @@ export async function runSlackSearchAssistant(
     await titleTask
     clearInterval(accessPoller)
     clearInterval(abortPoller)
-    try {
-      if (runId) {
-        /** This turn admitted its own run, so it records the terminal status no other path will. */
-        const cancelled =
-          failed &&
-          (isExplicitStopReason(controller.signal.reason) ||
-            (await wasSlackSearchTurnStopped(turnId, leaseId)))
-        await updateRunStatus(runId, failed ? (cancelled ? 'cancelled' : 'error') : 'complete')
-      }
-    } catch (error) {
-      failure = failure
-        ? new AggregateError([failure, error], 'Slack turn run status could not be recorded')
-        : toError(error)
-    }
     try {
       if (questionPersisted) {
         const stopped = failed && (await wasSlackSearchTurnStopped(turnId, leaseId))
@@ -392,6 +378,31 @@ export async function runSlackSearchAssistant(
         ? new AggregateError([failure, error], 'Slack turn and history persistence failed')
         : toError(error)
     } finally {
+      if (runId) {
+        /**
+         * This turn admitted its own run, so it records the one terminal status no other
+         * path will, after its outcome and response were saved: any failure, including
+         * a failed save, ends it as an error unless its user stopped it.
+         */
+        let cancelled = failed && isExplicitStopReason(controller.signal.reason)
+        if (failed && !cancelled) {
+          try {
+            cancelled = await wasSlackSearchTurnStopped(turnId, leaseId)
+          } catch (error) {
+            logger.warn('Slack turn Stop could not be read; recording its run as an error', {
+              turnId,
+              error: getErrorMessage(error),
+            })
+          }
+        }
+        try {
+          await updateRunStatus(runId, !failure ? 'complete' : cancelled ? 'cancelled' : 'error')
+        } catch (error) {
+          failure = failure
+            ? new AggregateError([failure, error], 'Slack turn run status could not be recorded')
+            : toError(error)
+        }
+      }
       unregisterActiveStream(messageId)
       await releasePendingChatStream(chat.id, messageId)
       await cleanupAbortMarker(messageId)

apps/sim/lib/mothership/async-runs/orphaned-runs.integration.ts

@@ -4,7 +4,7 @@
  * use case are production code. A local HTTP server stands in for the worker's abort
  * endpoint.
  */
-import { afterAll, beforeAll, describe, expect, it, vi } from 'vitest'
+import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from 'vitest'
 
 const { redisUrl, inheritedEnv, worker } = await vi.hoisted(async () => {
   const { readTestRedisUrl } = await import('@sim/db/testing/test-infrastructure')
@@ -68,8 +68,16 @@ function redis() {
   return client
 }
 
+/** The sweep's resume point lives in shared Redis; each test and the next suite start fresh. */
+async function resetSweepCursor() {
+  await getRedisClient()?.del('copilot:orphaned-runs:sweep-cursor')
+}
+
+beforeEach(resetSweepCursor)
+
 afterAll(async () => {
   chatPubSub?.dispose()
+  await resetSweepCursor()
   await closeRedisConnection()
   await new Promise<void>((resolve) => worker.server.close(() => resolve()))
   for (const [key, value] of Object.entries(inheritedEnv)) {