[pull] next from novuhq:next

.github/labeler.yml

@@ -12,6 +12,8 @@
   - apps/webhook/**/*
 '@novu/thalamus-observer':
   - enterprise/workers/thalamus-observer/**/*
+'@novu/socket-worker':
+  - enterprise/workers/socket/**/*
 '@novu/dal':
   - libs/dal/**/*
 '@novu/shared':

.github/workflows/deploy.yml

@@ -553,22 +553,26 @@ jobs:
       - name: Checkout
         uses: actions/checkout@93cb6efe18208431cddfb8368fd83d5badbf9bfd # v5
 
+      - name: Install pnpm
+        uses: pnpm/action-setup@0e279bb959325dab635dd2c09392533439d90093 # v6.0.8
+        with:
+          version: 11.0.9
+
       - name: Setup Node
         uses: actions/setup-node@49933ea5288caeca8642d1e84afbd3f7d6820020 # v4
         with:
-          node-version: '20'
+          node-version: '22'
+          cache: 'pnpm'
 
       - name: Install dependencies
-        working-directory: enterprise/workers/thalamus-observer
-        run: npm install
+        run: pnpm install --frozen-lockfile --filter @novu/thalamus-observer-worker...
 
       - name: Deploy with Wrangler
-        working-directory: enterprise/workers/thalamus-observer
         env:
           CLOUDFLARE_API_TOKEN: ${{ secrets.CLOUDFLARE_API_TOKEN }}
           CLOUDFLARE_ACCOUNT_ID: ${{ secrets.CLOUDFLARE_ACCOUNT_ID }}
         run: |
-          npx wrangler deploy \
+          pnpm --filter @novu/thalamus-observer-worker exec wrangler deploy \
             --env "${{ needs.prepare-matrix.outputs.thalamus_wrangler_env }}" \
             --message "GitHub Actions"
 

apps/api/src/app/agents/conversation-runtime/reply/send-agent-welcome-message/send-agent-welcome-message.usecase.spec.ts

@@ -113,7 +113,7 @@ describe('SendAgentWelcomeMessage usecase', () => {
   it('sends a welcome email to the dashboard subscriber address', async () => {
     const result = await buildUsecase().execute(buildCommand());
 
-    expect(result).to.deep.equal({ sent: true, conversationId: 'conversation-id', claimToken: undefined });
+    expect(result).to.deep.equal({ sent: true, conversationId: 'conversation-id' });
     expect(subscriberRepository.findBySubscriberId.calledOnceWith(ENV_ID, USER_ID)).to.equal(true);
     expect(channelEndpointRepository.findOne.called).to.equal(false);
     expect(
@@ -184,7 +184,7 @@ describe('SendAgentWelcomeMessage usecase', () => {
 
     const result = await buildUsecase().execute(buildCommand({ integrationIdentifier: 'slack-integration' }));
 
-    expect(result).to.deep.equal({ sent: true, conversationId: 'conversation-id', claimToken: undefined });
+    expect(result).to.deep.equal({ sent: true, conversationId: 'conversation-id' });
     expect(
       channelConnectionRepository.findOne.calledOnceWith(
         sinon.match({
@@ -278,6 +278,54 @@ describe('SendAgentWelcomeMessage usecase', () => {
     expect(outboundGateway.sendDirectMessage.calledOnce).to.equal(true);
   });
 
+  it('returns a claim token for a keyless org even when a welcome conversation already exists', async () => {
+    const originalKeylessOrgId = process.env.KEYLESS_ORGANIZATION_ID;
+    process.env.KEYLESS_ORGANIZATION_ID = ORG_ID;
+    conversationService.findByAgentIntegrationParticipant.resolves({
+      _id: 'existing-conversation-id',
+      title: 'Connected! Reply to this email to try it out.',
+    });
+
+    try {
+      const result = await buildUsecase().execute(buildCommand());
+
+      expect(result).to.deep.equal({
+        sent: true,
+        conversationId: 'existing-conversation-id',
+        claimToken: 'claim-token',
+      });
+      expect(connectClaimTokenService.issueOrGetForEnvironment.calledOnceWith({ env: ENV_ID, org: ORG_ID })).to.equal(
+        true
+      );
+      expect(outboundGateway.sendDirectMessage.called).to.equal(false);
+    } finally {
+      if (originalKeylessOrgId === undefined) {
+        delete process.env.KEYLESS_ORGANIZATION_ID;
+      } else {
+        process.env.KEYLESS_ORGANIZATION_ID = originalKeylessOrgId;
+      }
+    }
+  });
+
+  it('includes a claim token and signup card when sending a keyless welcome', async () => {
+    const originalKeylessOrgId = process.env.KEYLESS_ORGANIZATION_ID;
+    process.env.KEYLESS_ORGANIZATION_ID = ORG_ID;
+
+    try {
+      const result = await buildUsecase().execute(buildCommand());
+
+      expect(result).to.deep.equal({ sent: true, conversationId: 'conversation-id', claimToken: 'claim-token' });
+      expect(outboundGateway.sendDirectMessage.calledOnce).to.equal(true);
+      expect(outboundGateway.sendDirectMessage.firstCall.args[3]).to.have.property('card');
+    } finally {
+      if (originalKeylessOrgId === undefined) {
+        delete process.env.KEYLESS_ORGANIZATION_ID;
+      } else {
+        process.env.KEYLESS_ORGANIZATION_ID = originalKeylessOrgId;
+      }
+    }
+  });
+
   it('returns sent:false when the dashboard subscriber has no email', async () => {
     subscriberRepository.findBySubscriberId.resolves({ subscriberId: USER_ID, email: '' });
 

apps/api/src/app/agents/conversation-runtime/reply/send-agent-welcome-message/send-agent-welcome-message.usecase.ts

@@ -9,7 +9,6 @@ import {
   SubscriberRepository,
 } from '@novu/dal';
 import { SLACK_AGENT_WELCOME_SUGGESTED_PROMPTS, SLACK_AGENT_WELCOME_SUGGESTED_PROMPTS_TITLE } from '@novu/shared';
-import type { CardElement } from 'chat';
 import { ConnectClaimTokenService } from '../../../../connect/services/connect-claim-token.service';
 import { isKeylessOrganization } from '../../../../keyless/keyless-organization.helpers';
 import { buildConnectClaimUrl, buildKeylessWelcomeCard } from '../../../../keyless/keyless-signup.helpers';
@@ -93,6 +92,7 @@ export class SendAgentWelcomeMessage {
 
     const { platformUserId, workspaceId } = recipient;
     const welcomeText = getWelcomeText(platform);
+    const claimToken = await this.resolveKeylessClaimToken(command);
     const existingWelcomeConversation = await this.findExistingWelcomeConversation({
       environmentId: command.environmentId,
       organizationId: command.organizationId,
@@ -105,12 +105,13 @@ export class SendAgentWelcomeMessage {
     });
 
     if (existingWelcomeConversation) {
-      return { sent: true, conversationId: existingWelcomeConversation._id };
+      return { sent: true, conversationId: existingWelcomeConversation._id, ...withClaimToken(claimToken) };
     }
 
     try {
-      const keylessWelcome = await this.resolveKeylessWelcomeCard(command, welcomeText);
-      const welcomeReplyCard = keylessWelcome?.card;
+      const welcomeReplyCard = claimToken
+        ? buildKeylessWelcomeCard(welcomeText, buildConnectClaimUrl(claimToken))
+        : undefined;
       const welcomeContent = welcomeReplyCard ? { card: welcomeReplyCard } : { markdown: welcomeText };
       const sent = await this.outboundGateway.sendDirectMessage(
         agent._id,
@@ -168,11 +169,11 @@ export class SendAgentWelcomeMessage {
         platform,
       });
 
-      return { sent: true, conversationId: conversation._id, claimToken: keylessWelcome?.claimToken };
+      return { sent: true, conversationId: conversation._id, ...withClaimToken(claimToken) };
     } catch (err) {
       this.logger.warn(err, `Failed to send welcome message for agent "${command.agentIdentifier}"`);
 
-      return { sent: false };
+      return { sent: false, ...withClaimToken(claimToken) };
     }
   }
 
@@ -299,29 +300,25 @@ export class SendAgentWelcomeMessage {
     });
   }
 
-  private async resolveKeylessWelcomeCard(
-    command: SendAgentWelcomeMessageCommand,
-    welcomeText: string
-  ): Promise<{ card: CardElement; claimToken: string } | null> {
+  private async resolveKeylessClaimToken(command: SendAgentWelcomeMessageCommand): Promise<string | undefined> {
     if (!isKeylessOrganization(command.organizationId)) {
-      return null;
+      return undefined;
     }
 
     try {
       const { token } = await this.connectClaimTokenService.issueOrGetForEnvironment({
         env: command.environmentId,
         org: command.organizationId,
       });
-      const claimUrl = buildConnectClaimUrl(token);
 
-      return { card: buildKeylessWelcomeCard(welcomeText, claimUrl), claimToken: token };
+      return token;
     } catch (err) {
       this.logger.warn(
         err,
-        `Failed to build keyless welcome signup link for agent "${command.agentIdentifier}" — sending plain welcome`
+        `Failed to issue keyless claim token for agent "${command.agentIdentifier}" — sending welcome without a signup link`
       );
 
-      return null;
+      return undefined;
     }
   }
 
@@ -388,3 +385,11 @@ export class SendAgentWelcomeMessage {
     }
   }
 }
+
+function withClaimToken(claimToken: string | undefined): { claimToken?: string } {
+  if (!claimToken) {
+    return {};
+  }
+
+  return { claimToken };
+}

apps/api/src/app/channel-endpoints/channel-endpoints.controller.ts

@@ -26,6 +26,7 @@ import {
 import { RequireAuthentication } from '../auth/framework/auth.decorator';
 import { ThrottlerCategory } from '../rate-limiting/guards/throttler.decorator';
 import { ApiCommonResponses, ApiResponse } from '../shared/framework/response.decorator';
+import { KeylessAccessible } from '../shared/framework/swagger/keyless.security';
 import { SdkGroupName, SdkMethodName } from '../shared/framework/swagger/sdk.decorators';
 import { UserSession } from '../shared/framework/user.decorator';
 import { CreateChannelEndpointRequest } from './dtos/create-channel-endpoint-request.dto';
@@ -131,6 +132,7 @@ export class ChannelEndpointsController {
   })
   @ApiResponse(ListChannelEndpointsResponseDto, 200)
   @ExternalApiAccessible()
+  @KeylessAccessible()
   @SdkMethodName('list')
   @RequirePermissions(PermissionsEnum.INTEGRATION_READ)
   async listChannelEndpoints(

enterprise/workers/socket/README.md

@@ -4,20 +4,38 @@ Cloudflare Worker + Durable Object for Novu Cloud WebSockets (PartySocket).
 
 ## Local development
 
-This package is standalone (not in the pnpm workspace). From this directory:
+This package is part of the pnpm workspace (see `pnpm-workspace.yaml`).
+
+Install dependencies from the repo root:
 
 ```bash
-npm install
-cp .dev.vars.example .dev.vars
-# Fill JWT_SECRET and INTERNAL_API_KEY from apps/api/src/.env
-# (INTERNAL_API_KEY must match INTERNAL_SERVICES_API_KEY)
-npm run dev
+pnpm install
+```
+
+Run from the repo root:
+
+```bash
+pnpm dev:socket-worker
 ```
 
-`npm run dev` runs `wrangler dev --env local` (usually `http://127.0.0.1:8787`).
+Or run directly from this folder:
+
+```bash
+pnpm run dev
+```
+
+`pnpm run dev` runs `wrangler dev --env local` on **`http://127.0.0.1:8787`**.
 
 Local `thalamus-observer` uses **8788** so both workers can run at once; keep socket on 8787.
 
+First-time setup in this folder:
+
+```bash
+cp .dev.vars.example .dev.vars
+# Fill JWT_SECRET and INTERNAL_API_KEY from apps/api/src/.env
+# (INTERNAL_API_KEY must match INTERNAL_SERVICES_API_KEY)
+```
+
 `.dev.vars` is gitignored. The `local` wrangler env sets `API_URL` to `http://127.0.0.1:3000`.
 
 ### Wire API / worker / playground

enterprise/workers/socket/package.json

@@ -15,7 +15,7 @@
   "devDependencies": {
     "@cloudflare/workers-types": "^4.20250620.0",
     "typescript": "^5.5.2",
-    "wrangler": "^4.20.5"
+    "wrangler": "^4.49.0"
   },
   "dependencies": {
     "@tsndr/cloudflare-worker-jwt": "^3.2.0",

enterprise/workers/thalamus-observer/README.md

@@ -12,14 +12,27 @@ Companion to `@novu/thalamus`. The API points at this Worker via `THALAMUS_CF_UR
 
 ## Local development
 
-This package is standalone (not in the pnpm workspace). From this directory:
+This package is part of the pnpm workspace (see `pnpm-workspace.yaml`).
+
+Install dependencies from the repo root:
+
+```bash
+pnpm install
+```
+
+Run from the repo root:
+
+```bash
+pnpm dev:thalamus-observer
+```
+
+Or run directly from this folder:
 
 ```bash
-npm install
-npm run dev
+pnpm run dev
 ```
 
-`npm run dev` listens on **`http://127.0.0.1:8788`** (local wrangler env `dev.port`, not the Wrangler default 8787). That leaves **8787** free for `@novu/socket-worker`.
+`pnpm run dev` listens on **`http://127.0.0.1:8788`** (Wrangler `--port 8788`, not the default 8787). That leaves **8787** free for `@novu/socket-worker`.
 
 Point the API at it:
 
@@ -52,13 +65,13 @@ Required secrets on those GitHub Environments: `CLOUDFLARE_API_TOKEN`, `CLOUDFLA
 Emergency local deploy (break-glass):
 
 ```bash
-npm run deploy:staging
-npm run deploy:production
+pnpm run deploy:staging
+pnpm run deploy:production
 ```
 
 Worker-bound secrets (one-time, not in CI):
 
 ```bash
-npx wrangler secret put API_KEY --env staging
-npx wrangler secret put API_KEY --env production
+pnpm exec wrangler secret put API_KEY --env staging
+pnpm exec wrangler secret put API_KEY --env production
 ```

enterprise/workers/thalamus-observer/package.json

@@ -14,12 +14,12 @@
   },
   "dependencies": {
     "@novu/thalamus": "0.1.0-alpha.18",
-    "agents": "^0.12.4",
+    "agents": "^0.21.0",
     "eventsource-parser": "^3.0.8"
   },
   "devDependencies": {
     "@cloudflare/workers-types": "^4.20250424.0",
     "typescript": "^5.5.2",
-    "wrangler": "^4.20.5"
+    "wrangler": "^4.49.0"
   }
 }

enterprise/workers/thalamus-observer/src/worker.ts

@@ -1,3 +1,4 @@
+import { getAgentByName } from 'agents';
 import type { SessionObserver } from './session-observer';
 import type { Env } from './types';
 import { validateEnqueueParams, validateObservationParams } from './validation';
@@ -52,7 +53,7 @@ export default {
             { status: 400 }
           );
         }
-        const stub = env.SESSION_OBSERVER.getByName(body.sessionId) as DurableObjectStub<SessionObserver>;
+        const stub = await getAgentByName<Env, SessionObserver>(env.SESSION_OBSERVER, body.sessionId);
         const result = await stub.handleEnqueue(body);
 
         return Response.json(result, { status: 200 });
@@ -72,15 +73,15 @@ export default {
             { status: 400 }
           );
         }
-        const stub = env.SESSION_OBSERVER.getByName(body.sessionId) as DurableObjectStub<SessionObserver>;
+        const stub = await getAgentByName<Env, SessionObserver>(env.SESSION_OBSERVER, body.sessionId);
         await stub.startObserving(body);
 
         return new Response(null, { status: 204 });
       }
 
       if (request.method === 'DELETE' && path.startsWith('/observe/')) {
         const sessionId = decodeURIComponent(path.slice('/observe/'.length));
-        const stub = env.SESSION_OBSERVER.getByName(sessionId) as DurableObjectStub<SessionObserver>;
+        const stub = await getAgentByName<Env, SessionObserver>(env.SESSION_OBSERVER, sessionId);
         await stub.stopObserving();
 
         return new Response(null, { status: 204 });

package.json

@@ -21,6 +21,8 @@
     "dev-environment-setup": "./scripts/dev-environment-setup.sh",
     "dev:portless": "node scripts/mprocs-dev.mjs",
     "dev:config": "node scripts/novu-dev-config.mjs",
+    "dev:thalamus-observer": "pnpm --filter @novu/thalamus-observer-worker dev",
+    "dev:socket-worker": "pnpm --filter @novu/socket-worker dev",
     "docker:build": "pnpm -r --if-present --parallel docker:build",
     "g:module": "hygen module new --name=$pnpm_config_name",
     "g:usecase": "hygen usecase new --name=$pnpm_config_name --module=$pnpm_config_module",