[pull] next from novuhq:next

apps/api/src/app.module.ts

@@ -11,6 +11,7 @@ import { SentryModule } from '@sentry/nestjs/setup';
 import packageJson from '../package.json';
 import { ActivityModule } from './app/activity/activity.module';
 import { AgentsModule } from './app/agents/agents.module';
+import { HumanModule } from './app/human/human.module';
 import { AnalyticsModule } from './app/analytics/analytics.module';
 import { AuthModule } from './app/auth/auth.module';
 import { BlueprintModule } from './app/blueprint/blueprint.module';
@@ -136,6 +137,7 @@ const baseModules: Array<Type | DynamicModule | Promise<DynamicModule> | Forward
   OrganizationModule,
   ActivityModule,
   AgentsModule,
+  HumanModule,
   ConnectModule,
   NovuContextModule,
   DomainsModule.forRoot(),

apps/api/src/app/agents/agent-chat/activity-to-events.spec.ts

@@ -183,6 +183,37 @@ describe('activity-to-events run lifecycle', () => {
     ).to.deep.equal(['approval-activity-1', 'approval-activity-2']);
   });
 
+  it('drops SIGNAL activities from client events', () => {
+    const envelopes = mapNewestFirstEventActivities(
+      [
+        activity({
+          type: ConversationActivityTypeEnum.SIGNAL,
+          identifier: 'workflow-dispatch-origin:wamid.1',
+          sequence: 2,
+          content: 'Workflow origin: order-shipped',
+          signalData: { type: 'workflow_origin', payload: { workflowIdentifier: 'order-shipped' } },
+        }),
+        activity({
+          type: ConversationActivityTypeEnum.SIGNAL,
+          identifier: 'sig-1',
+          sequence: 1,
+          content: 'signal',
+          signalData: { type: 'other' },
+        }),
+        activity({
+          type: ConversationActivityTypeEnum.MESSAGE,
+          identifier: 'msg-1',
+          sequence: 0,
+          content: 'hello',
+        }),
+      ],
+      context
+    );
+
+    expect(envelopes).to.have.lengthOf(1);
+    expect(envelopes[0].event.type).to.equal('message');
+  });
+
   it('derives trust action ids at emit time for managed MCP approvals', () => {
     const envelopes = mapNewestFirstEventActivities(
       [

apps/api/src/app/agents/agent-chat/activity-to-events.ts

@@ -209,8 +209,15 @@ function mapActivityToEvent(activity: ConversationActivityEntity): AgentEvent |
     case ConversationActivityTypeEnum.RUN_ERROR:
       return mapRunLifecycleActivityToEvent(activity);
 
-    default:
+    case ConversationActivityTypeEnum.SIGNAL:
       return null;
+
+    default: {
+      const _exhaustive: never = activity.type;
+      void _exhaustive;
+
+      return null;
+    }
   }
 }
 

apps/api/src/app/agents/agents.module.ts

@@ -18,6 +18,7 @@ import {
   ConversationActivationRepository,
   ConversationActivityRepository,
   ConversationRepository,
+  HumanInteractionRepository,
   IntegrationRepository,
   McpConnectionRepository,
   MessageRepository,
@@ -75,6 +76,8 @@ import { AgentEmailActionsController } from './email/agent-email-actions.control
 import { AgentEmailSender } from './email/agent-email-sender.service';
 import { NovuEmailCleanupService } from './email/novu-email/cleanup-novu-email/cleanup-novu-email.service';
 import { NovuEmailProvisioningService } from './email/novu-email/find-or-create-novu-email/find-or-create-novu-email.service';
+import { HumanInteractionSettlementService } from './human-relay/human-interaction-settlement.service';
+import { HumanRelayRuntime } from './human-relay/human-relay.runtime';
 import { AgentRuntimeDefinitionService } from './managed-runtime/agent-runtime-definition.service';
 import { DemoClaudeQuotaPolicy } from './managed-runtime/demo-claude-quota-policy.service';
 import { ManagedRuntime } from './managed-runtime/managed.runtime';
@@ -161,6 +164,9 @@ import { USE_CASES } from './usecases';
     BridgeExpireSupersededApprovalsService,
     BridgeRuntime,
     ManagedRuntime,
+    HumanRelayRuntime,
+    HumanInteractionSettlementService,
+    HumanInteractionRepository,
     RuntimeResolver,
     ManagedAgentProviderFactory,
     ManagedAgentEventHandler,
@@ -202,6 +208,14 @@ import { USE_CASES } from './usecases';
     AgentConversationEnabledGuard,
     AgentChatEnabledGuard,
   ],
-  exports: [...USE_CASES, ChatInstanceRegistry, InboundDispatcher, OutboundGateway, ConfirmLinkedAuthCards],
+  exports: [
+    ...USE_CASES,
+    ChatInstanceRegistry,
+    InboundDispatcher,
+    OutboundGateway,
+    ConfirmLinkedAuthCards,
+    ConversationActivityLedger,
+    HumanInteractionSettlementService,
+  ],
 })
 export class AgentsModule {}

apps/api/src/app/agents/channels/agent-config-resolver.service.ts

@@ -316,7 +316,10 @@ export class AgentConfigResolver {
       integrationIdentifier,
       integrationId: integration._id,
       providerId: integration.providerId,
-      removeNovuBranding: await this.resolveRemoveNovuBranding(organizationId),
+      // Human-relay messages are utility traffic between a person and their own
+      // agents — never consumer-facing agent chat — so they always ship unbranded.
+      removeNovuBranding:
+        agent.runtime === 'human_relay' ? true : await this.resolveRemoveNovuBranding(organizationId),
       acknowledgeOnReceived: agent.behavior?.acknowledgeOnReceived !== false,
       reactionOnResolved: await resolveReaction(
         agent.behavior?.reactionOnResolved,

apps/api/src/app/agents/conversation-runtime/conversation/agent-conversation.service.spec.ts

@@ -1,9 +1,4 @@
-import {
-  ConversationActivityTypeEnum,
-  ConversationParticipantTypeEnum,
-  ConversationRepository,
-  ConversationStatusEnum,
-} from '@novu/dal';
+import { ConversationParticipantTypeEnum, ConversationRepository, ConversationStatusEnum } from '@novu/dal';
 import { expect } from 'chai';
 import sinon from 'sinon';
 import {
@@ -43,15 +38,19 @@ describe('AgentConversationService', () => {
 
   function makeLedger(overrides: Partial<Record<keyof ConversationActivityLedger, sinon.SinonStub>> = {}) {
     return {
-      persistAgentMessage: overrides.persistAgentMessage ?? sinon.stub().resolves({ activity: {}, created: true }),
-      persistWorkflowOriginHydration: overrides.persistWorkflowOriginHydration ?? sinon.stub().resolves(undefined),
-      isWorkflowOriginHydrated: overrides.isWorkflowOriginHydrated ?? sinon.stub().resolves(false),
-      persistMcpConnectionRequest: overrides.persistMcpConnectionRequest ?? sinon.stub().resolves({}),
-      persistMcpConnectionResult: overrides.persistMcpConnectionResult ?? sinon.stub().resolves({}),
-      persistToolResult: overrides.persistToolResult ?? sinon.stub().resolves(undefined),
-      persistInboundMessage: overrides.persistInboundMessage ?? sinon.stub().resolves({}),
-      listForView: overrides.listForView ?? sinon.stub().resolves({ data: [], hasMore: false }),
-      mint: overrides.mint ?? sinon.stub().resolves(1),
+      persistAgentMessage: sinon.stub().resolves({ activity: {}, created: true }),
+      persistWorkflowOriginHydration: sinon.stub().resolves(undefined),
+      isWorkflowOriginHydrated: sinon.stub().resolves(false),
+      persistMcpConnectionRequest: sinon.stub().resolves({}),
+      persistMcpConnectionResult: sinon.stub().resolves({}),
+      persistToolResult: sinon.stub().resolves(undefined),
+      persistInboundMessage: sinon.stub().resolves({}),
+      persistResolveSignal: sinon.stub().resolves(undefined),
+      persistTriggerSignal: sinon.stub().resolves(undefined),
+      persistRunLifecycle: sinon.stub().resolves(null),
+      listForView: sinon.stub().resolves({ data: [], hasMore: false }),
+      mint: sinon.stub().resolves(1),
+      ...overrides,
     } as unknown as ConversationActivityLedger;
   }
 

apps/api/src/app/agents/conversation-runtime/conversation/agent-conversation.service.ts

@@ -369,6 +369,15 @@ export class AgentConversationService {
     return this.ledger.isWorkflowOriginHydrated(environmentId, conversationId, platformMessageId);
   }
 
+  async setNotificationId(
+    environmentId: string,
+    organizationId: string,
+    conversationId: string,
+    notificationId: string
+  ): Promise<void> {
+    await this.conversationRepository.setNotificationId(environmentId, organizationId, conversationId, notificationId);
+  }
+
   async listForView(params: {
     view: ActivityView;
     environmentId: string;

apps/api/src/app/agents/conversation-runtime/conversation/agent-conversation.types.ts

@@ -84,8 +84,8 @@ export interface PersistTriggerSignalParams extends ConversationActivityContext
 export interface PersistWorkflowOriginHydrationParams extends ConversationActivityContext {
   platformMessageId: string;
   platformThreadId: string;
-  messageContent: string;
   signalData: Record<string, unknown>;
+  messageBody?: string;
 }
 
 export interface PersistToolApprovalDecisionParams extends ConversationActivityContext {

apps/api/src/app/agents/conversation-runtime/conversation/conversation-activity-ledger.spec.ts

@@ -200,8 +200,12 @@ describe('ConversationActivityLedger', () => {
         organizationId: 'org-1',
         platformMessageId: 'wamid.abc',
         platformThreadId: 'whatsapp:15551234567',
-        messageContent: 'Your order shipped',
-        signalData: { workflowIdentifier: 'order-alerts' },
+        signalData: {
+          notificationId: 'notif-1',
+          workflowIdentifier: 'order-alerts',
+          messageId: 'msg-1',
+          payload: { orderId: 'ORD-1' },
+        },
       };
     }
 
@@ -233,6 +237,86 @@ describe('ConversationActivityLedger', () => {
         expect((err as Error).message).to.equal('mongo timeout');
       }
     });
+
+    it('writes a single SIGNAL activity and no MESSAGE row when messageBody is omitted', async () => {
+      const activityRepository = makeActivityRepository({
+        createSignalActivity: sinon.stub().resolves({ _id: 'signal-1' }),
+        createAgentActivity: sinon.stub().resolves({ _id: 'should-not-run' }),
+      });
+      const ledger = makeLedger(activityRepository);
+
+      await ledger.persistWorkflowOriginHydration(makeHydrationParams());
+
+      expect(activityRepository.createSignalActivity.calledOnce).to.equal(true);
+      expect(activityRepository.createAgentActivity.called).to.equal(false);
+      const args = activityRepository.createSignalActivity.firstCall.args[0];
+      expect(args.identifier).to.equal('workflow-dispatch-origin:wamid.abc');
+      expect(args.content).to.equal('Workflow origin: order-alerts');
+      expect(args.signalData).to.deep.equal({
+        type: 'workflow_origin',
+        payload: {
+          notificationId: 'notif-1',
+          workflowIdentifier: 'order-alerts',
+          messageId: 'msg-1',
+          payload: { orderId: 'ORD-1' },
+        },
+      });
+    });
+
+    it('skips the MESSAGE row when messageBody is empty', async () => {
+      const activityRepository = makeActivityRepository({
+        createSignalActivity: sinon.stub().resolves({ _id: 'signal-1' }),
+        createAgentActivity: sinon.stub().resolves({ _id: 'should-not-run' }),
+      });
+      const ledger = makeLedger(activityRepository);
+
+      await ledger.persistWorkflowOriginHydration({ ...makeHydrationParams(), messageBody: '   ' });
+
+      expect(activityRepository.createAgentActivity.called).to.equal(false);
+      expect(activityRepository.createSignalActivity.calledOnce).to.equal(true);
+    });
+
+    it('writes the MESSAGE before the SIGNAL when messageBody is present', async () => {
+      const activityRepository = makeActivityRepository({
+        createSignalActivity: sinon.stub().resolves({ _id: 'signal-1' }),
+        createAgentActivity: sinon.stub().resolves({ _id: 'message-1' }),
+      });
+      const ledger = makeLedger(activityRepository);
+
+      await ledger.persistWorkflowOriginHydration({
+        ...makeHydrationParams(),
+        messageBody: 'Your order shipped',
+      });
+
+      expect(activityRepository.createAgentActivity.calledOnce).to.equal(true);
+      expect(activityRepository.createSignalActivity.calledOnce).to.equal(true);
+      expect(activityRepository.createAgentActivity.calledBefore(activityRepository.createSignalActivity)).to.equal(
+        true
+      );
+      expect(activityRepository.createAgentActivity.firstCall.args[0]).to.deep.include({
+        identifier: 'workflow-origin-message:wamid.abc',
+        content: 'Your order shipped',
+        type: ConversationActivityTypeEnum.MESSAGE,
+      });
+      expect(activityRepository.createAgentActivity.firstCall.args[0].platformMessageId).to.equal(undefined);
+    });
+
+    it('still writes the SIGNAL when the MESSAGE identifier already exists', async () => {
+      const duplicateError = Object.assign(new Error('duplicate key'), { code: 11000 });
+      const activityRepository = makeActivityRepository({
+        createAgentActivity: sinon.stub().rejects(duplicateError),
+        findOne: sinon.stub().resolves({ _id: 'existing-message', identifier: 'workflow-origin-message:wamid.abc' }),
+        createSignalActivity: sinon.stub().resolves({ _id: 'signal-1' }),
+      });
+      const ledger = makeLedger(activityRepository);
+
+      await ledger.persistWorkflowOriginHydration({
+        ...makeHydrationParams(),
+        messageBody: 'Your order shipped',
+      });
+
+      expect(activityRepository.createSignalActivity.calledOnce).to.equal(true);
+    });
   });
 
   describe('isWorkflowOriginHydrated', () => {

apps/api/src/app/agents/conversation-runtime/conversation/conversation-activity-ledger.ts

@@ -43,11 +43,14 @@ export interface ListActivityViewParams {
   before?: string;
 }
 
-/** Stable per-origin identifier for the workflow-origin signal — see `persistWorkflowOriginHydration`. */
 function workflowOriginSignalIdentifier(platformMessageId: string): string {
   return `workflow-dispatch-origin:${platformMessageId}`;
 }
 
+function workflowOriginMessageIdentifier(platformMessageId: string): string {
+  return `workflow-origin-message:${platformMessageId}`;
+}
+
 @Injectable()
 export class ConversationActivityLedger {
   constructor(
@@ -459,18 +462,22 @@ export class ConversationActivityLedger {
     return count > 0;
   }
 
+  /** Persist a logging-only SIGNAL for the workflow origin, and an agent MESSAGE when a body exists. */
   async persistWorkflowOriginHydration(params: PersistWorkflowOriginHydrationParams): Promise<void> {
-    await this.persistAgentMessage({
-      conversationId: params.conversationId,
-      channel: params.channel,
-      agentIdentifier: params.agentIdentifier,
-      environmentId: params.environmentId,
-      organizationId: params.organizationId,
-      platformMessageId: params.platformMessageId,
-      platformThreadId: params.platformThreadId,
-      identifier: `workflow-dispatch-msg:${params.platformMessageId}`,
-      content: params.messageContent,
-    });
+    const messageBody = params.messageBody?.trim() ?? '';
+
+    if (messageBody.length > 0) {
+      await this.persistAgentMessage({
+        conversationId: params.conversationId,
+        channel: params.channel,
+        agentIdentifier: params.agentIdentifier,
+        environmentId: params.environmentId,
+        organizationId: params.organizationId,
+        identifier: workflowOriginMessageIdentifier(params.platformMessageId),
+        platformThreadId: params.platformThreadId,
+        content: messageBody,
+      });
+    }
 
     try {
       await this.persistSignal({