fix(mothership): settle chat runs no controller owns (#8457)

* fix(mothership): settle Chat runs no controller will finish

A run whose process died before finalize, whose controller was superseded
with no successor, or that was stopped while no controller existed stayed
unfinished forever: its chat marker kept pointing at it, so the chat read
as busy and a reconnect polled a run nothing would ever end.

- Stop now settles the run as cancelled once no controller of its stream
  holds the chat lock, including after it force-releases a controller that
  did not exit in time.
- The stale-execution cron settles leased runs whose stream holds no chat
  lock, has no replay buffer left, and has been idle past the orchestration
  budget (so a reconnect has nothing left to resume), and runs without a
  lease once idle for 24 hours. A run Stop already closed settles as
  cancelled, any other as error; its chat marker is released.
- Each settle is one conditional update on the run row that requires it to
  be unfinished, idle, and still naming the controller that was observed,
  so a finalizing controller or a successor's claim wins or loses against
  it atomically and the run settles exactly once.

* fix(mothership): lock chats before runs when settling orphans, and label them precisely

- Settling now locks the affected chat rows first, in id order, as a
  controller's claim does. Locking the run and then the chat deadlocked
  against a concurrent reconnect claim.
- Each sweep batch fails on its own, the chat markers of a batch clear in
  one statement, and a sweep settles at most 5k rows with a short pause
  between full batches.
- A run settles as cancelled only when its user pressed Stop. A newer turn
  also closes tool admission on older runs, and those now settle as errors.
- Runs without a controller lease keep their last write as their
  completion and retention time and read "never finalized (no controller
  lease)".
- The leased-run grace no longer derives from the orchestration deadline.
  Liveness comes only from the heartbeat-renewed chat lock; the grace and
  the replay TTL only bound how long a reconnect can resume a dead run.

* fix(mothership): never sweep a current headless run

A headless turn has no chat lease and no heartbeat, so its age says nothing
about whether it is still running once runs have no deadline. The sweep's
lease-less rule now applies only to runs admitted before the current
tool-execution protocol: every run the current code admits records the
current version, so after a deploy no such row can be live. A current
headless run is left to its own lifecycle, which always settles it.

The protocol version moves beside the other async-run constants so the
sweep can read it without importing the repository.

* refactor(mothership): require the recorded Stop inside the stopped-run settle

- Settling a stopped run now passes the Stop-row check as the update's own
  guard, so it cannot cancel a run nobody stopped; the separate stopped
  branch is gone and every settle derives cancelled from the Stop row.
- Chat lock ownership is read through getChatStreamLockOwners and trusted
  only when verified, instead of a second Redis read of the same keys.
- Settle transactions use the shared DbTransaction type.

* fix(mothership): fence orphan settlement on the chat lock and resume sweeps where they stopped

- The sweep takes each unowned leased run's chat lock under the run's own
  stream before settling it and releases it after the commit, so a
  reconnect can no longer lock the chat between the ownership check and
  the settle and then lose its claim; a reconnect that meets the fence
  retries.
- A sweep examines at most 10k candidates and settles at most about 5k,
  resuming from a cursor saved in Redis and wrapping to the first run, so
  runs that cannot be settled yet never starve the ones after them.
- Every settled run whose chat marker was released is announced, legacy
  runs included, so an open client stops showing the chat as busy.

* fix(knowledge): record a terminal status for every Slack Assistant run

The Slack Assistant admits its own run row and only ever marked it as an
error, so every completed or stopped Slack turn stayed active. It now
records the terminal status once, after the turn ends, through the shared
run-status update: complete on success, cancelled when its user stopped
it (in Slack or in Sim), and error otherwise.

Every other run-creating path already settles its run: interactive turns
through their controller's finalize, and headless turns that admit their
own run in the lifecycle's own finally.

* 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/app/api/cron/cleanup-stale-executions/route.ts

@@ -32,6 +32,7 @@ import {
   STALE_SWEEPABLE_EXECUTION_STATUSES,
   type StaleSweepableExecutionStatus,
 } from '@/lib/logs/types'
+import { sweepOrphanedRuns } from '@/lib/mothership/async-runs/orphaned-runs'
 import { cancelStaleDispatches } from '@/lib/table/dispatcher'
 import { deleteFile } from '@/lib/uploads/core/storage-service'
 import {
@@ -738,6 +739,20 @@ export const GET = withRouteHandler(async (request: NextRequest) => {
       })
     }
 
+    /**
+     * Settle Chat runs no controller will finish: their process died, their
+     * controller was superseded without a successor, or Stop found none. Without
+     * this they stay unfinished forever and keep their chat marked as busy.
+     */
+    let orphanedRunsSettled = 0
+    try {
+      orphanedRunsSettled = (await sweepOrphanedRuns()).settledRunIds.length
+    } catch (error) {
+      logger.error('Failed to settle orphaned Chat runs:', {
+        error: toError(error).message,
+      })
+    }
+
     return NextResponse.json({
       success: true,
       executions: {
@@ -768,6 +783,9 @@ export const GET = withRouteHandler(async (request: NextRequest) => {
         pruned: deploymentOperationsPruned,
         retentionDays: DEPLOYMENT_OPERATION_RETENTION_DAYS,
       },
+      chatRuns: {
+        orphanedSettled: orphanedRunsSettled,
+      },
     })
   } catch (error) {
     logger.error('Error in stale execution cleanup job:', error)

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

@@ -0,0 +1,287 @@
+/**
+ * The Slack Assistant's run record against real PostgreSQL: the run row it admits and
+ * the terminal status it records are production code. Slack delivery, the worker
+ * lifecycle, identity, and chat locking are stood in, since only the record's outcome
+ * is under test.
+ */
+import { billingAttributionMock } from '@sim/testing/mocks/billing-attribution.mock'
+import {
+  mothershipChatPayloadMock,
+  mothershipChatPayloadMockFns,
+} from '@sim/testing/mocks/mothership-chat-payload.mock'
+import { mothershipEnvironmentContextMock } from '@sim/testing/mocks/mothership-environment-context.mock'
+import { organizationAuthorizationMock } from '@sim/testing/mocks/organization-authorization.mock'
+import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from 'vitest'
+
+const hoisted = vi.hoisted(() => ({
+  chat: { id: '', model: 'default' },
+  turnId: '',
+  userId: '',
+  organizationId: '',
+  lifecycle: vi.fn(),
+  /** 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 () => ({
+    installation: { id: 'i1', organizationId: hoisted.organizationId, teamId: 'T1' },
+    secret: { botToken: 'token' },
+  }),
+}))
+vi.mock('@/lib/internal/slack/search-client', () => ({
+  getSlackSearchSender: async () => ({ email: 'member@example.com' }),
+}))
+vi.mock('@/lib/knowledge/application/slack-search/identity', () => ({
+  resolveSlackSearchMember: async () => hoisted.userId,
+  SlackSearchIdentityError: class extends Error {},
+}))
+vi.mock('@/lib/knowledge/application/slack-search/chat', () => ({
+  resolveSlackSearchChat: async () => hoisted.chat,
+  persistSlackSearchQuestion: async () => undefined,
+  slackSearchChatOperation: { id: 'organization.chats.slack' },
+}))
+vi.mock('@/lib/core/application/organization-authorization', () => organizationAuthorizationMock)
+vi.mock('@/lib/knowledge/application/operations', () => ({
+  knowledgeOperations: { search: { organizationOperation: { id: 'knowledge.search' } } },
+}))
+vi.mock('@/lib/knowledge/application/slack-search/repository', () => ({
+  recordSlackSearchOutcome: hoisted.outcome,
+}))
+vi.mock('@/lib/knowledge/application/slack-search/turns', () => ({
+  requireSlackSearchTurnLease: async () => undefined,
+  wasSlackSearchTurnStopped: hoisted.stopped,
+}))
+vi.mock('@/lib/knowledge/application/slack-search/onboarding', () => ({
+  sendSlackSearchOnboarding: vi.fn(),
+}))
+vi.mock('@/lib/knowledge/application/slack-search/title', () => ({
+  generateSlackSearchChatTitle: async () => undefined,
+}))
+vi.mock('@/lib/billing/core/billing-attribution', () => billingAttributionMock)
+vi.mock('@/lib/mothership/application/load-search-integrations', () => ({
+  loadCopilotSearchIntegrations: async () => '{"connections":[],"available":[]}',
+}))
+vi.mock('@/lib/mothership/chat/payload', () => mothershipChatPayloadMock)
+vi.mock('@/lib/mothership/chat/terminal-state', () => ({
+  finalizeAssistantTurn: hoisted.finalize,
+}))
+vi.mock('@/lib/mothership/environment-context', () => mothershipEnvironmentContextMock)
+vi.mock('@/lib/mothership/request/lifecycle/headless', () => ({
+  runHeadlessCopilotLifecycle: hoisted.lifecycle,
+}))
+vi.mock('@/lib/mothership/request/session/abort', () => ({
+  acquirePendingChatStream: async () => true,
+  cleanupAbortMarker: async () => undefined,
+  getChatStreamLockOwners: async () => ({
+    status: 'verified',
+    ownersByChatId: new Map([[hoisted.chat.id, hoisted.turnId]]),
+  }),
+  registerActiveStream: vi.fn(),
+  releasePendingChatStream: async () => undefined,
+  startAbortPoller: () => 0,
+  unregisterActiveStream: vi.fn(),
+}))
+vi.mock('@/executor/utils/resolved-secret-content-projection', () => ({
+  projectResolvedSecretDiagnosticContent: (value: unknown) => ({ safe: true, value }),
+}))
+vi.mock('@/lib/slack-search/connections', () => ({ deliverSlackSearchConnections: vi.fn() }))
+vi.mock('@/lib/slack-search/assistant-stream', () => ({
+  SlackSearchAssistantStream: class {
+    start = async () => undefined
+    finish = async () => undefined
+    finishWithError = async () => undefined
+    terminateAfterFailure = async () => undefined
+    onEvent = async () => undefined
+    assertHealthy = () => undefined
+  },
+}))
+
+import { db } from '@sim/db'
+import { copilotChats, copilotRuns, organization, user } from '@sim/db/schema'
+import { generateId } from '@sim/utils/id'
+import { eq } from 'drizzle-orm'
+import { runSlackSearchAssistant } from '@/lib/knowledge/application/slack-search/assistant'
+import { AbortReason } from '@/lib/mothership/request/session/abort-reason'
+
+const principal = {
+  kind: 'slack_installation',
+  credentialId: 'c1',
+  credentialVersion: 'v1',
+  appId: 'A1',
+  teamId: 'T1',
+  eventId: 'Ev1',
+  receivedAt: new Date(),
+} as const
+
+function job() {
+  return {
+    installationId: 'i1',
+    revision: 'r1',
+    credentialId: 'c1',
+    credentialVersion: 'v1',
+    receivedAt: Date.now(),
+    message: {
+      appId: 'A1',
+      teamId: 'T1',
+      eventId: 'Ev1',
+      channelId: 'D1',
+      userId: 'U1',
+      messageTs: '1800000000.000001',
+      query: 'release notes',
+      queryTooLong: false,
+    },
+  }
+}
+
+/** Runs one Slack turn in a fresh private chat and returns the run it recorded. */
+async function slackTurn() {
+  const chatId = generateId()
+  const turnId = generateId()
+  hoisted.chat = { id: chatId, model: 'default' }
+  hoisted.turnId = turnId
+  await db.insert(copilotChats).values({
+    id: chatId,
+    userId: hoisted.userId,
+    organizationId: hoisted.organizationId,
+    type: 'mothership',
+  })
+  const controller = new AbortController()
+  hoisted.controller = controller
+  const outcome = await runSlackSearchAssistant(principal, {
+    job: job(),
+    turnId,
+    leaseId: generateId(),
+    controller,
+  }).then(
+    () => undefined,
+    (error: unknown) => error
+  )
+  const [run] = await db.select().from(copilotRuns).where(eq(copilotRuns.chatId, chatId))
+  return { run, outcome }
+}
+
+describe('Slack Assistant run record', () => {
+  beforeAll(async () => {
+    hoisted.userId = generateId()
+    hoisted.organizationId = generateId()
+    const now = new Date()
+    await db.insert(user).values({
+      id: hoisted.userId,
+      name: 'Slack Assistant fixture',
+      email: `${hoisted.userId}@slack-assistant.test`,
+      emailVerified: true,
+      createdAt: now,
+      updatedAt: now,
+    })
+    await db.insert(organization).values({
+      id: hoisted.organizationId,
+      name: 'Slack Assistant fixture',
+      slug: `slack-assistant-${hoisted.organizationId}`,
+    })
+    mothershipChatPayloadMockFns.mockBuildCopilotRequestPayload.mockResolvedValue({
+      mode: 'assistant',
+    })
+  })
+
+  afterAll(async () => {
+    await db.delete(copilotChats).where(eq(copilotChats.userId, hoisted.userId))
+    await db.delete(organization).where(eq(organization.id, hoisted.organizationId))
+    await db.delete(user).where(eq(user.id, hoisted.userId))
+  })
+
+  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,
+      content: 'Answer',
+      contentBlocks: [],
+      toolCalls: [],
+    })
+
+    const { run, outcome } = await slackTurn()
+
+    expect(outcome).toBeUndefined()
+    expect(run.status).toBe('complete')
+    expect(run.completedAt).not.toBeNull()
+  })
+
+  it('records a turn its user stopped as cancelled', async () => {
+    hoisted.lifecycle.mockImplementationOnce(async () => {
+      /** A Slack Stop marks the turn stopped, then aborts its registered stream. */
+      hoisted.stopped.mockResolvedValue(true)
+      hoisted.controller.abort(AbortReason.UserStop)
+      return { success: false, cancelled: true, content: '', contentBlocks: [], toolCalls: [] }
+    })
+
+    const { run } = await slackTurn()
+
+    expect(run.status).toBe('cancelled')
+    expect(run.completedAt).not.toBeNull()
+  })
+
+  it('records a failed turn as an error', async () => {
+    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 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'
@@ -49,6 +49,7 @@ import {
   startAbortPoller,
   unregisterActiveStream,
 } from '@/lib/mothership/request/session/abort'
+import { isExplicitStopReason } from '@/lib/mothership/request/session/abort-reason'
 import type { OrchestratorResult } from '@/lib/mothership/request/types'
 import { organizationRoutes } from '@/lib/navigation/paths'
 import { SlackSearchAssistantStream } from '@/lib/slack-search/assistant-stream'
@@ -303,7 +304,6 @@ export async function runSlackSearchAssistant(
           ]
         : []),
       recordSlackSearchOutcome(installation, 'assistant_or_delivery_failed'),
-      ...(runId ? [updateRunStatus(runId, 'error')] : []),
     ])
     const errors = outcomes.flatMap((outcome) =>
       outcome.status === 'rejected' ? [outcome.reason] : []
@@ -378,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/lifecycle.ts

@@ -4,6 +4,9 @@ import {
   MothershipStreamV1ToolOutcome,
 } from '@/lib/mothership/generated/mothership-stream-v1'
 
+/** Recorded on every run the current code admits; older values mark runs from earlier protocols. */
+export const SIM_TOOL_EXECUTION_VERSION = 2
+
 export const ASYNC_TOOL_STATUS = MothershipStreamV1AsyncToolRecordStatus
 
 export const EXECUTABLE_TOOL_PERMISSION_DECISIONS = [