fix(ai): route missing-workspace stream failures through the standard error path (#22480)

## Rationale When `StreamAgentChatJob` can't find the workspace, it publishes a transient `stream-error` event and **returns before the try/finally exists** (`stream-agent-chat.job.ts`). Consequences on main: - `activeStreamId` is never cleared → every subsequent send in that thread queues behind a dead claim, forever; - no `lastStreamError` is persisted → nothing renders after a reload, and Retry has nothing to retry; - nothing throws → **zero telemetry**. Sentry confirms: the "Workspace not found" issues that exist are all auth/Stripe paths — this path fails in complete silence. ## Why this is the root cause, not a symptom patch The job's catch/finally already implement the correct failure contract for *every other* error: persist a typed `lastStreamError`, publish the typed event, release the claim guarded on the observed streamId. The bug is that one code path bypasses that contract via an early return. The fix removes the bypass — the lookup moves inside the `try` and throws a typed `AiException(WORKSPACE_NOT_FOUND)` — rather than duplicating cleanup in the early-return branch (which would be the symptom patch, and would drift the next time the contract changes). The alternative "prevent the job from existing when the workspace is gone" isn't achievable: workspace deletion between enqueue and pickup is an inherent race, so the job must handle it regardless. ## User impact A workspace deleted/deactivated mid-flight currently bricks the thread silently (the user just sees sends vanish into a queue). With this, the failure is visible (typed error message), recoverable (standard failed-turn state), and observable (real exception in monitoring). ## Stack Based on #22479 (spec harness) — it extends the same spec file with the regression test. `WORKSPACE_NOT_FOUND` is a TypeScript enum member, not a GraphQL schema change: no client-sdk regeneration needed. ## Test plan - [x] Regression test: missing workspace → typed rejection, `lastStreamError` persisted, terminal event published, claim released - [ ] CI green https://claude.ai/code/session_01Lyi6zTema2FMVVh8MD6c38 --- _Generated by [Claude Code](https://claude.ai/code/session_01Lyi6zTema2FMVVh8MD6c38)_  <a href="https://cubic.dev/pr/twentyhq/twenty/pull/22480?utm_source=github" target="_blank" rel="noopener noreferrer" data-no-image-dialog="true"><picture><source media="(prefers-color-scheme: dark)" srcset="https://www.cubic.dev/buttons/review-in-cubic-dark.svg"><source media="(prefers-color-scheme: light)" srcset="https://www.cubic.dev/buttons/review-in-cubic-light.svg"><img alt="Review in cubic" src="https://www.cubic.dev/buttons/review-in-cubic-dark.svg"></picture></a>
2026-07-02 21:11:02 +02:00
parent ca8ab32253
commit 3b76ec528f
4 changed files with 51 additions and 19 deletions
@@ -3,6 +3,8 @@ import { type UIMessageChunk } from 'ai';
 import { type WorkspaceEntity } from 'src/engine/core-modules/workspace/workspace.entity';
 import { StreamAgentChatJob } from 'src/engine/metadata-modules/ai/ai-chat/jobs/stream-agent-chat.job';
 import { type StreamAgentChatJobData } from 'src/engine/metadata-modules/ai/ai-chat/jobs/stream-agent-chat-job.types';
+import { AiExceptionCode } from 'src/engine/metadata-modules/ai/ai.exception';
+

 type PublishedEvent = { type: string } & Record<string, unknown>;

@@ -294,6 +296,36 @@ describe('StreamAgentChatJob', () => {
    );
  });

+  it('persists the error and unblocks the thread when the workspace is missing', async () => {
+    const { job, publishedEvents, threadRepository } = buildJob({
+      workspaceFound: false,
+    });
+
+    await expect(job.handle(jobData)).rejects.toMatchObject({
+      code: AiExceptionCode.WORKSPACE_NOT_FOUND,
+    });
+
+    expect(publishedEvents[publishedEvents.length - 1]).toMatchObject({
+      type: 'stream-error',
+      code: AiExceptionCode.WORKSPACE_NOT_FOUND,
+    });
+    expect(threadRepository.update).toHaveBeenCalledWith(
+      'workspace-id',
+      { id: 'thread-id' },
+      {
+        lastStreamError: expect.objectContaining({
+          code: AiExceptionCode.WORKSPACE_NOT_FOUND,
+        }),
+      },
+    );
+    expect(threadRepository.update).toHaveBeenCalledWith(
+      'workspace-id',
+      { id: 'thread-id', activeStreamId: 'stream-id' },
+      { activeStreamId: null },
+    );
+  });
+
+
  it('resolves without flushing the queue when the stream is cancelled', async () => {
    let triggerCancel: (() => void) | undefined;

@@ -21,6 +21,10 @@ import { AgentMessageRole } from 'src/engine/metadata-modules/ai/ai-agent-execut
 import { computeCostBreakdown } from 'src/engine/metadata-modules/ai/ai-billing/utils/compute-cost-breakdown.util';
 import { convertDollarsToBillingCredits } from 'src/engine/metadata-modules/ai/ai-billing/utils/convert-dollars-to-billing-credits.util';
 import { extractCacheCreationTokens } from 'src/engine/metadata-modules/ai/ai-billing/utils/extract-cache-creation-tokens.util';
+import {
+  AiException,
+  AiExceptionCode,
+} from 'src/engine/metadata-modules/ai/ai.exception';
 import { AgentChatThreadEntity } from 'src/engine/metadata-modules/ai/ai-chat/entities/agent-chat-thread.entity';
 import { AgentChatCancelSubscriberService } from 'src/engine/metadata-modules/ai/ai-chat/services/agent-chat-cancel-subscriber.service';
 import { AgentChatEventPublisherService } from 'src/engine/metadata-modules/ai/ai-chat/services/agent-chat-event-publisher.service';
@@ -64,25 +68,6 @@ export class StreamAgentChatJob {
  async handle(data: StreamAgentChatJobData): Promise<void> {
    await this.eventPublisherService.resetStreamState(data.threadId);

-    const workspace = await this.workspaceRepository.findOne({
-      where: { id: data.workspaceId },
-    });
-
-    if (!workspace) {
-      this.logger.error(`Workspace ${data.workspaceId} not found`);
-      await this.eventPublisherService.publish({
-        threadId: data.threadId,
-        workspaceId: data.workspaceId,
-        event: {
-          type: 'stream-error',
-          code: 'WORKSPACE_NOT_FOUND',
-          message: `Workspace ${data.workspaceId} not found`,
-        },
-      });
-
-      return;
-    }
-
    const abortController = new AbortController();
    const cancelChannel = getCancelChannel(data.threadId);

@@ -91,6 +76,17 @@ export class StreamAgentChatJob {
    });

    try {
+      const workspace = await this.workspaceRepository.findOne({
+        where: { id: data.workspaceId },
+      });
+
+      if (!workspace) {
+        throw new AiException(
+          `Workspace ${data.workspaceId} not found`,
+          AiExceptionCode.WORKSPACE_NOT_FOUND,
+        );
+      }
+
      await this.executeStream(data, workspace, abortController.signal);
    } catch (error) {
      this.logger.error(
@@ -11,6 +11,7 @@ export enum AiExceptionCode {
  AGENT_EXECUTION_FAILED = 'AGENT_EXECUTION_FAILED',
  INVALID_AGENT_INPUT = 'INVALID_AGENT_INPUT',
  THREAD_NOT_FOUND = 'THREAD_NOT_FOUND',
+  WORKSPACE_NOT_FOUND = 'WORKSPACE_NOT_FOUND',
  INVALID_CHAT_THREAD_TITLE = 'INVALID_CHAT_THREAD_TITLE',
  MESSAGE_NOT_FOUND = 'MESSAGE_NOT_FOUND',
  QUESTION_NOT_PENDING = 'QUESTION_NOT_PENDING',
@@ -36,6 +37,8 @@ const getAiExceptionUserFriendlyMessage = (code: AiExceptionCode) => {
      return msg`Invalid agent input.`;
    case AiExceptionCode.THREAD_NOT_FOUND:
      return msg`Chat thread not found.`;
+    case AiExceptionCode.WORKSPACE_NOT_FOUND:
+      return msg`Workspace not found.`;
    case AiExceptionCode.INVALID_CHAT_THREAD_TITLE:
      return msg`Chat thread title cannot be empty.`;
    case AiExceptionCode.MESSAGE_NOT_FOUND:
@@ -23,6 +23,7 @@ export const aiGraphqlApiExceptionHandler = (error: Error) => {
    switch (error.code) {
      case AiExceptionCode.AGENT_NOT_FOUND:
      case AiExceptionCode.THREAD_NOT_FOUND:
+      case AiExceptionCode.WORKSPACE_NOT_FOUND:
      case AiExceptionCode.MESSAGE_NOT_FOUND:
      case AiExceptionCode.ROLE_NOT_FOUND:
        throw new NotFoundError(error);