diff --git a/apps/api/src/app.module.ts b/apps/api/src/app.module.ts index a18ef75eecd..f17f319f084 100644 --- a/apps/api/src/app.module.ts +++ b/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 | Forward OrganizationModule, ActivityModule, AgentsModule, + HumanModule, ConnectModule, NovuContextModule, DomainsModule.forRoot(), diff --git a/apps/api/src/app/agents/agent-chat/activity-to-events.spec.ts b/apps/api/src/app/agents/agent-chat/activity-to-events.spec.ts index abf6b693b69..5c6661b95e1 100644 --- a/apps/api/src/app/agents/agent-chat/activity-to-events.spec.ts +++ b/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( [ diff --git a/apps/api/src/app/agents/agent-chat/activity-to-events.ts b/apps/api/src/app/agents/agent-chat/activity-to-events.ts index dc42a702f84..081045193b8 100644 --- a/apps/api/src/app/agents/agent-chat/activity-to-events.ts +++ b/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; + } } } diff --git a/apps/api/src/app/agents/agents.module.ts b/apps/api/src/app/agents/agents.module.ts index 35c39a18667..92605c90d0f 100644 --- a/apps/api/src/app/agents/agents.module.ts +++ b/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 {} diff --git a/apps/api/src/app/agents/channels/agent-config-resolver.service.ts b/apps/api/src/app/agents/channels/agent-config-resolver.service.ts index 4f66e1d2603..b96c621d7a0 100644 --- a/apps/api/src/app/agents/channels/agent-config-resolver.service.ts +++ b/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, diff --git a/apps/api/src/app/agents/conversation-runtime/conversation/agent-conversation.service.spec.ts b/apps/api/src/app/agents/conversation-runtime/conversation/agent-conversation.service.spec.ts index 4e754c0b951..618f480a365 100644 --- a/apps/api/src/app/agents/conversation-runtime/conversation/agent-conversation.service.spec.ts +++ b/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> = {}) { 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; } diff --git a/apps/api/src/app/agents/conversation-runtime/conversation/agent-conversation.service.ts b/apps/api/src/app/agents/conversation-runtime/conversation/agent-conversation.service.ts index 65cced8ebef..6e5e6fb8afe 100644 --- a/apps/api/src/app/agents/conversation-runtime/conversation/agent-conversation.service.ts +++ b/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 { + await this.conversationRepository.setNotificationId(environmentId, organizationId, conversationId, notificationId); + } + async listForView(params: { view: ActivityView; environmentId: string; diff --git a/apps/api/src/app/agents/conversation-runtime/conversation/agent-conversation.types.ts b/apps/api/src/app/agents/conversation-runtime/conversation/agent-conversation.types.ts index c71fdf5a35c..5c11dfbca95 100644 --- a/apps/api/src/app/agents/conversation-runtime/conversation/agent-conversation.types.ts +++ b/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; + messageBody?: string; } export interface PersistToolApprovalDecisionParams extends ConversationActivityContext { diff --git a/apps/api/src/app/agents/conversation-runtime/conversation/conversation-activity-ledger.spec.ts b/apps/api/src/app/agents/conversation-runtime/conversation/conversation-activity-ledger.spec.ts index e95b1e9b7e4..084eb836fd1 100644 --- a/apps/api/src/app/agents/conversation-runtime/conversation/conversation-activity-ledger.spec.ts +++ b/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', () => { diff --git a/apps/api/src/app/agents/conversation-runtime/conversation/conversation-activity-ledger.ts b/apps/api/src/app/agents/conversation-runtime/conversation/conversation-activity-ledger.ts index 5ef9b80cebc..774b565fb2d 100644 --- a/apps/api/src/app/agents/conversation-runtime/conversation/conversation-activity-ledger.ts +++ b/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 { - 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({ diff --git a/apps/api/src/app/agents/conversation-runtime/ingress/inbound-turn.handler.spec.ts b/apps/api/src/app/agents/conversation-runtime/ingress/inbound-turn.handler.spec.ts index 486885fc13f..4ffcd365140 100644 --- a/apps/api/src/app/agents/conversation-runtime/ingress/inbound-turn.handler.spec.ts +++ b/apps/api/src/app/agents/conversation-runtime/ingress/inbound-turn.handler.spec.ts @@ -45,6 +45,28 @@ describe('AgentInboundHandler', () => { }; } + function makeOriginSnapshot( + overrides: { + body?: string; + platformMessageId?: string; + payload?: Record; + source?: 'hydrated' | 'existing'; + } = {} + ) { + return { + data: { + notificationId: 'notif1', + workflowIdentifier: 'order-alerts', + messageId: 'msg1', + platformMessageId: overrides.platformMessageId ?? '1777837477.371619', + sentAt: '2026-01-01T00:00:00.000Z', + body: overrides.body ?? 'Order alerts', + payload: overrides.payload ?? {}, + }, + source: overrides.source ?? ('hydrated' as const), + }; + } + function makeResolvedSubscriberOverrides(subscriberId = 'sub1', internalSubscriberId = 'subscriber-mongo-1') { return { subscriberResolve: sinon.stub().resolves(subscriberId), @@ -210,6 +232,7 @@ describe('AgentInboundHandler', () => { }; const workflowOriginService = { resolve: sinon.stub().resolves(null), + resolveForTurn: sinon.stub().resolves(null), hydrate: sinon.stub().resolves(null), }; const handler = new AgentInboundHandler( @@ -433,9 +456,11 @@ describe('AgentInboundHandler', () => { _notificationId: 'notif1', identifier: 'D123:1777837477.371619', }; + const snapshot = makeOriginSnapshot({ body: 'Order shipped' }); conversationService.findByPlatformThread.resolves(null); workflowOriginService.resolve.resolves({ origin, notificationId: 'notif1' }); + workflowOriginService.resolveForTurn.resolves(snapshot); const thread = makeSlackDmThread(); const message = { @@ -454,8 +479,8 @@ describe('AgentInboundHandler', () => { existingConversation: null, }); expect(conversationService.createOrGetConversation.firstCall.args[0].notificationId).to.equal('notif1'); - expect(workflowOriginService.hydrate.calledOnce).to.equal(true); - expect(workflowOriginService.hydrate.firstCall.args[0].origin).to.equal(origin); + expect(workflowOriginService.resolveForTurn.calledOnce).to.equal(true); + expect(workflowOriginService.resolveForTurn.firstCall.args[0].resolution.origin).to.equal(origin); }); it('should forward the hydrated origin to a managed dispatch on an existing Telegram conversation', async () => { @@ -472,6 +497,7 @@ describe('AgentInboundHandler', () => { ...makeResolvedSubscriberOverrides('sub-tg', 'sub-mongo'), agentFindOne: sinon.stub().resolves(makeManagedAgentStub()), }); + const snapshot = makeOriginSnapshot({ platformMessageId: '42' }); conversationService.findByPlatformThread.resolves({ _id: 'conv1', @@ -480,7 +506,7 @@ describe('AgentInboundHandler', () => { participants: [], }); workflowOriginService.resolve.resolves({ origin: { _id: 'msg1', _notificationId: 'notif1', identifier: '42' } }); - workflowOriginService.hydrate.resolves('Your order shipped'); + workflowOriginService.resolveForTurn.resolves(snapshot); const thread = { id: 'telegram:42', @@ -502,10 +528,67 @@ describe('AgentInboundHandler', () => { await handler.handle('agent1', telegramConfig as any, thread as any, message as any, AgentEventEnum.ON_MESSAGE); expect(managedAgentService.dispatch.calledOnce).to.equal(true); - expect(managedAgentService.dispatch.firstCall.args[0].workflowOriginContent).to.equal('Your order shipped'); + expect(managedAgentService.dispatch.firstCall.args[0].workflowOrigin).to.deep.equal(snapshot); }); - it('should leave workflowOriginContent unset when nothing was hydrated', async () => { + it('should read the latest persisted origin on later turns when nothing new hydrates', async () => { + const telegramConfig = { + ...config, + platform: AgentPlatformEnum.TELEGRAM, + integrationIdentifier: 'telegram-main', + isManaged: true, + subscriberAccess: AgentSubscriberAccessEnum.OPEN, + }; + const existingConversation = { + _id: 'conv1', + externalSessionId: 'ses_live', + channels: [{ platform: AgentPlatformEnum.TELEGRAM, _integrationId: 'int1', platformThreadId: 'telegram:42' }], + participants: [], + }; + const snapshot = makeOriginSnapshot({ + body: 'Your order shipped', + platformMessageId: '42', + payload: { orderId: 'ORD-9' }, + source: 'existing', + }); + const { handler, conversationService, workflowOriginService, managedAgentService } = makeHandler({ + ...makeResolvedSubscriberOverrides('sub-tg', 'sub-mongo'), + agentFindOne: sinon.stub().resolves(makeManagedAgentStub()), + }); + + conversationService.findByPlatformThread.resolves(existingConversation); + conversationService.createOrGetConversation.resolves(existingConversation); + workflowOriginService.resolve.resolves(null); + workflowOriginService.resolveForTurn.resolves(snapshot); + + const thread = { + id: 'telegram:42', + channelId: '42', + isDM: true, + toJSON: () => ({ id: 'telegram:42', channelId: '42', isDM: true }), + startTyping: sinon.stub().resolves(undefined), + post: sinon.stub().resolves({ id: 'reply-1', threadId: 'telegram:42' }), + }; + const message = { + id: 'msg-3', + threadId: 'telegram:42', + text: 'and the eta?', + author: { userId: '42', fullName: 'TG User', userName: 'tguser', isBot: false }, + raw: {}, + attachments: [], + }; + + await handler.handle('agent1', telegramConfig as any, thread as any, message as any, AgentEventEnum.ON_MESSAGE); + + expect(workflowOriginService.resolveForTurn.calledOnce).to.equal(true); + expect(workflowOriginService.resolveForTurn.firstCall.args[0]).to.include({ + subscriberId: 'sub-tg', + resolution: null, + }); + expect(managedAgentService.dispatch.firstCall.args[0].workflowOrigin).to.deep.equal(snapshot); + }); + + it('should leave workflowOrigin unset when nothing was hydrated', async () => { const telegramConfig = { ...config, platform: AgentPlatformEnum.TELEGRAM, @@ -541,7 +624,7 @@ describe('AgentInboundHandler', () => { await handler.handle('agent1', telegramConfig as any, thread as any, message as any, AgentEventEnum.ON_MESSAGE); expect(managedAgentService.dispatch.calledOnce).to.equal(true); - expect(managedAgentService.dispatch.firstCall.args[0].workflowOriginContent).to.equal(undefined); + expect(managedAgentService.dispatch.firstCall.args[0].workflowOrigin).to.equal(undefined); }); it('should not hydrate when WorkflowOriginService.resolve returns null', async () => { @@ -559,7 +642,8 @@ describe('AgentInboundHandler', () => { ); expect(workflowOriginService.resolve.calledOnce).to.equal(true); - expect(workflowOriginService.hydrate.called).to.equal(false); + expect(workflowOriginService.resolveForTurn.calledOnce).to.equal(true); + expect(workflowOriginService.resolveForTurn.firstCall.args[0].resolution).to.equal(null); expect(conversationService.createOrGetConversation.firstCall.args[0].notificationId).to.equal(undefined); }); @@ -1589,9 +1673,11 @@ describe('AgentInboundHandler', () => { templateIdentifier: 'order-alerts', identifier: 'thread1:1777837477.371619', }; + const snapshot = makeOriginSnapshot(); conversationService.findByPlatformThread.resolves(null); workflowOriginService.resolve.resolves({ origin, notificationId: 'notif1' }); + workflowOriginService.resolveForTurn.resolves(snapshot); await handler.handleAction( 'agent1', @@ -1601,10 +1687,11 @@ describe('AgentInboundHandler', () => { 'user1' ); - expect(workflowOriginService.hydrate.calledOnce).to.equal(true); - expect(workflowOriginService.hydrate.firstCall.args[0].origin).to.equal(origin); + expect(workflowOriginService.resolveForTurn.calledOnce).to.equal(true); + expect(workflowOriginService.resolveForTurn.firstCall.args[0].resolution.origin).to.equal(origin); // The origin must reach history before the runtime reads the conversation. - expect(workflowOriginService.hydrate.calledBefore(bridgeExecutor.execute)).to.equal(true); + expect(workflowOriginService.resolveForTurn.calledBefore(bridgeExecutor.execute)).to.equal(true); + expect(bridgeExecutor.execute.firstCall.args[0].workflowOrigin).to.deep.equal(snapshot); }); it('should use the clicked Slack message timestamp when resolving an action-only thread', async () => { @@ -1620,6 +1707,7 @@ describe('AgentInboundHandler', () => { conversationService.findByPlatformThread.resolves(null); workflowOriginService.resolve.resolves({ origin, notificationId: 'notif1' }); + workflowOriginService.resolveForTurn.resolves(makeOriginSnapshot()); await handler.handleAction( 'agent1', @@ -1632,7 +1720,7 @@ describe('AgentInboundHandler', () => { expect(conversationService.findByPlatformThread.firstCall.args[4]).to.equal(platformThreadId); expect(workflowOriginService.resolve.firstCall.args[0].platformThreadId).to.equal(platformThreadId); expect(conversationService.createOrGetConversation.firstCall.args[0].platformThreadId).to.equal(platformThreadId); - expect(workflowOriginService.hydrate.firstCall.args[0].platformThreadId).to.equal(platformThreadId); + expect(workflowOriginService.resolveForTurn.firstCall.args[0].platformThreadId).to.equal(platformThreadId); expect(bridgeExecutor.execute.firstCall.args[0].platformContext.threadId).to.equal(platformThreadId); }); @@ -1647,7 +1735,8 @@ describe('AgentInboundHandler', () => { 'user1' ); - expect(workflowOriginService.hydrate.called).to.equal(false); + expect(workflowOriginService.resolveForTurn.calledOnce).to.equal(true); + expect(workflowOriginService.resolveForTurn.firstCall.args[0].resolution).to.equal(null); }); it('should still hydrate workflow origin when a link-button click is the first-ever interaction on a seeded thread', async () => { @@ -1662,6 +1751,7 @@ describe('AgentInboundHandler', () => { origin: { _id: 'msg1', _notificationId: 'notif1' }, notificationId: 'notif1', }); + workflowOriginService.resolveForTurn.resolves(makeOriginSnapshot({ platformMessageId: 'p1' })); await handler.handleAction( 'agent1', @@ -1672,7 +1762,10 @@ describe('AgentInboundHandler', () => { ); expect(conversationService.createOrGetConversation.firstCall.args[0].notificationId).to.equal('notif1'); - expect(workflowOriginService.hydrate.calledOnce).to.equal(true); + expect(workflowOriginService.resolveForTurn.calledOnce).to.equal(true); + expect(workflowOriginService.resolveForTurn.firstCall.args[0].resolution).to.include({ + notificationId: 'notif1', + }); }); it('should still dispatch the action when workflow origin resolve returns null', async () => { @@ -1691,7 +1784,8 @@ describe('AgentInboundHandler', () => { 'user1' ); - expect(workflowOriginService.hydrate.called).to.equal(false); + expect(workflowOriginService.resolveForTurn.calledOnce).to.equal(true); + expect(workflowOriginService.resolveForTurn.firstCall.args[0].resolution).to.equal(null); expect(bridgeExecutor.execute.calledOnce).to.equal(true); }); }); @@ -1754,5 +1848,20 @@ describe('AgentInboundHandler', () => { const params = bridgeExecutor.execute.firstCall.args[0]; expect(params.reaction.sourceMessageStoredAttachments).to.deep.equal(storedAttachments); }); + + it('attaches the workflow origin to the ON_REACTION turn', async () => { + const snapshot = makeOriginSnapshot(); + const { handler, bridgeExecutor, workflowOriginService } = makeHandler(); + workflowOriginService.resolveForTurn.resolves(snapshot); + + await handler.handleReaction('agent1', config as any, makeReactionEvent() as any); + + expect(workflowOriginService.resolve.calledOnce).to.equal(true); + expect(workflowOriginService.resolveForTurn.calledOnce).to.equal(true); + expect(workflowOriginService.resolveForTurn.firstCall.args[0].resolution).to.equal(null); + const params = bridgeExecutor.execute.firstCall.args[0]; + expect(params.event).to.equal(AgentEventEnum.ON_REACTION); + expect(params.workflowOrigin).to.deep.equal(snapshot); + }); }); }); diff --git a/apps/api/src/app/agents/conversation-runtime/ingress/inbound-turn.handler.ts b/apps/api/src/app/agents/conversation-runtime/ingress/inbound-turn.handler.ts index 6c7e24e5884..5d8388ea57c 100644 --- a/apps/api/src/app/agents/conversation-runtime/ingress/inbound-turn.handler.ts +++ b/apps/api/src/app/agents/conversation-runtime/ingress/inbound-turn.handler.ts @@ -429,7 +429,7 @@ export class AgentInboundHandler implements OnModuleInit { return; } - const workflowOrigin = await this.workflowOriginService.resolve({ + const workflowOriginResolution = await this.workflowOriginService.resolve({ agentId, config, platformThreadId, @@ -455,22 +455,21 @@ export class AgentInboundHandler implements OnModuleInit { isDirectMessage: thread.isDM, workspaceId: extractWorkspaceId(config.platform, message.raw) ?? undefined, identifier: this.agentChatConversationIdentifier(config.platform, platformThreadId), - notificationId: workflowOrigin?.notificationId, + notificationId: workflowOriginResolution?.notificationId, contextKeys: config.platform === AgentPlatformEnum.AGENT_CHAT ? ((message.raw as AgentChatRawMessage | undefined)?.contextKeys ?? []) : undefined, }); - const workflowOriginContent = workflowOrigin - ? await this.workflowOriginService.hydrate({ - agentId, - config, - conversation, - platformThreadId, - origin: workflowOrigin.origin, - }) - : null; + const workflowOrigin = await this.workflowOriginService.resolveForTurn({ + agentId, + config, + conversation, + platformThreadId, + subscriberId, + resolution: workflowOriginResolution, + }); if (config.isKeyless) { const aiEnabled = await this.keylessAbuseGuard.isKeylessAgentAiEnabled(config.organizationId); @@ -545,7 +544,7 @@ export class AgentInboundHandler implements OnModuleInit { thread, platformThreadId, storedAttachments: message.attachments?.length ? storedAttachments : undefined, - workflowOriginContent: workflowOriginContent ?? undefined, + workflowOrigin: workflowOrigin ?? undefined, }; // On buttonless platforms (iMessage/SMS) a pending tool approval is @@ -1085,6 +1084,25 @@ export class AgentInboundHandler implements OnModuleInit { platformUserId ); const runtime = this.runtimeResolver.resolve(agent); + + const workflowOriginResolution = await this.workflowOriginService.resolve({ + agentId, + config, + platformThreadId: threadId, + subscriberId, + message: event.message ?? null, + existingConversation: conversation, + isDirectMessage: event.thread?.isDM, + }); + const workflowOrigin = await this.workflowOriginService.resolveForTurn({ + agentId, + config, + conversation, + platformThreadId: threadId, + subscriberId, + resolution: workflowOriginResolution, + }); + const turn: ConversationTurn = { agentId, agent: agent ?? { _id: agentId }, @@ -1099,6 +1117,7 @@ export class AgentInboundHandler implements OnModuleInit { thread: event.thread ?? ({ id: threadId, channelId: '', isDM: false } as Thread), platformThreadId: threadId, reaction: reactionPayload, + workflowOrigin: workflowOrigin ?? undefined, }; // On buttonless platforms (iMessage/SMS) a pending tool approval can be @@ -1148,7 +1167,7 @@ export class AgentInboundHandler implements OnModuleInit { platformThreadId ); - const workflowOrigin = await this.workflowOriginService.resolve({ + const workflowOriginResolution = await this.workflowOriginService.resolve({ agentId, config, platformThreadId, @@ -1171,22 +1190,21 @@ export class AgentInboundHandler implements OnModuleInit { firstMessageText: `[action:${action.id}]`, isDirectMessage: thread.isDM, workspaceId: extractWorkspaceId(config.platform, rawEvent) ?? undefined, - notificationId: workflowOrigin?.notificationId, + notificationId: workflowOriginResolution?.notificationId, contextKeys: config.platform === AgentPlatformEnum.AGENT_CHAT ? ((rawEvent as AgentChatRawMessage | undefined)?.contextKeys ?? []) : undefined, }); - const workflowOriginContent = workflowOrigin - ? await this.workflowOriginService.hydrate({ - agentId, - config, - conversation, - platformThreadId, - origin: workflowOrigin.origin, - }) - : null; + const workflowOrigin = await this.workflowOriginService.resolveForTurn({ + agentId, + config, + conversation, + platformThreadId, + subscriberId, + resolution: workflowOriginResolution, + }); trackAgentInboundAction(this.analyticsService, { organizationId: config.organizationId, @@ -1245,7 +1263,7 @@ export class AgentInboundHandler implements OnModuleInit { thread, platformThreadId, action, - workflowOriginContent: workflowOriginContent ?? undefined, + workflowOrigin: workflowOrigin ?? undefined, }; await runtime.dispatch(turn); diff --git a/apps/api/src/app/agents/conversation-runtime/ingress/workflow-origin.helpers.ts b/apps/api/src/app/agents/conversation-runtime/ingress/workflow-origin.helpers.ts index 6bc8c98dfff..1aec71e4dbf 100644 --- a/apps/api/src/app/agents/conversation-runtime/ingress/workflow-origin.helpers.ts +++ b/apps/api/src/app/agents/conversation-runtime/ingress/workflow-origin.helpers.ts @@ -3,7 +3,25 @@ import type { Message } from 'chat'; import { AgentPlatformEnum } from '../../shared/enums/agent-platform.enum'; import { asRecord } from '../../shared/util/raw-record'; -export const WORKFLOW_ORIGIN_CONTENT_MAX_CHARS = 2_000; +export interface WorkflowOriginData { + notificationId: string; + workflowIdentifier: string; + messageId: string; + platformMessageId: string; + sentAt: string; + body: string; + payload: Record; + jobId?: string; + stepId?: string; + transactionId?: string; + subscriberId?: string; +} + +export interface WorkflowOriginSnapshot { + data: WorkflowOriginData; + source: 'hydrated' | 'existing'; +} + export const WORKFLOW_ORIGIN_LOOKBACK_MS = 7 * 24 * 60 * 60 * 1000; /** Platforms that reuse one conversation indefinitely — origin is re-checked on later turns, not just at open. */ @@ -14,19 +32,6 @@ export const RECHECK_WORKFLOW_ORIGIN_PLATFORMS: ReadonlySet = AgentPlatformEnum.TEAMS, ]); -export function buildWorkflowOriginSummary( - workflowIdentifier: string, - messageContent: string, - payload: Record -): string { - const message = - messageContent.length > 0 ? messageContent : `A notification was sent by the ${workflowIdentifier} workflow.`; - const additionalData = - Object.keys(payload).length > 0 ? `\n\nAdditional data for this message:\n${JSON.stringify(payload, null, 2)}` : ''; - - return `${message}${additionalData}`.slice(0, WORKFLOW_ORIGIN_CONTENT_MAX_CHARS); -} - /** Conversation uses `slack:{channel}:{ts}`; Message.identifier stores bare `{channel}:{ts}`. */ export function toProviderMessageLookupKey(platformThreadId: string): string { return platformThreadId.startsWith('slack:') ? platformThreadId.slice('slack:'.length) : platformThreadId; @@ -170,3 +175,48 @@ export function resolvePlatformMessageId( return originMessage.identifier.slice(colon + 1); } + +const NAMED_ENTITIES: Record = { + amp: '&', + lt: '<', + gt: '>', + quot: '"', + apos: "'", + nbsp: ' ', +}; + +function decodeEntities(text: string): string { + return text.replace(/&(?:#(\d+)|#x([0-9a-f]+)|([a-z]+));/gi, (match, dec, hex, name) => { + if (dec) { + return String.fromCodePoint(Number(dec)); + } + if (hex) { + return String.fromCodePoint(parseInt(hex, 16)); + } + + return NAMED_ENTITIES[name.toLowerCase()] ?? match; + }); +} + +/** Same algorithm as `@novu/chat-adapter-email` `stripHtml` (ESM-only; cannot import from apps/api). */ +export function stripHtml(html: string): string { + const chars: string[] = []; + let depth = 0; + for (const ch of html) { + if (ch === '<') { + depth++; + continue; + } + if (ch === '>') { + if (depth > 0) { + depth--; + } + continue; + } + if (depth === 0) { + chars.push(ch); + } + } + + return decodeEntities(chars.join('').replace(/\s+/g, ' ')).trim(); +} diff --git a/apps/api/src/app/agents/conversation-runtime/ingress/workflow-origin.service.spec.ts b/apps/api/src/app/agents/conversation-runtime/ingress/workflow-origin.service.spec.ts index 10b348257af..3cf0d8853b0 100644 --- a/apps/api/src/app/agents/conversation-runtime/ingress/workflow-origin.service.spec.ts +++ b/apps/api/src/app/agents/conversation-runtime/ingress/workflow-origin.service.spec.ts @@ -12,6 +12,10 @@ describe('WorkflowOriginService', () => { participants: [{ type: 'subscriber', id: 'sub1' }], }; + afterEach(() => { + delete (conversation as { _notificationId?: string })._notificationId; + }); + function makeLogger() { return { warn: sinon.stub(), @@ -37,6 +41,7 @@ describe('WorkflowOriginService', () => { getPrimaryChannel: sinon.stub().callsFake((conv) => conv.channels[0]), persistWorkflowOriginHydration: overrides.persistWorkflowOriginHydration ?? sinon.stub().resolves(undefined), isWorkflowOriginHydrated: overrides.isWorkflowOriginHydrated ?? sinon.stub().resolves(false), + setNotificationId: sinon.stub().resolves(undefined), }; const subscriberRepository = { findBySubscriberId: overrides.findBySubscriberId ?? sinon.stub().resolves({ _id: 'subscriber-mongo-1' }), @@ -147,12 +152,14 @@ describe('WorkflowOriginService', () => { origin: { _id: 'msg1', _notificationId: 'notif1', + _templateId: 'template-1', _jobId: 'job1', transactionId: 'txn1', templateIdentifier: 'order-alerts', stepId: 'chat-1', content: 'Order ORD-1 shipped', identifier: 'D123:1777837477.371619', + createdAt: '2026-01-01T00:00:00.000Z', } as any, }); @@ -160,10 +167,203 @@ describe('WorkflowOriginService', () => { expect(conversationService.persistWorkflowOriginHydration.calledOnce).to.equal(true); const hydrateArgs = conversationService.persistWorkflowOriginHydration.firstCall.args[0]; expect(hydrateArgs.platformMessageId).to.equal('1777837477.371619'); - expect(hydrateArgs.messageContent).to.equal( - 'Order ORD-1 shipped\n\nAdditional data for this message:\n{\n "orderId": "ORD-1"\n}' - ); + expect(hydrateArgs.messageBody).to.equal('Order ORD-1 shipped'); expect(hydrateArgs.signalData.workflowIdentifier).to.equal('order-alerts'); + expect(hydrateArgs.signalData.payload).to.deep.equal({ orderId: 'ORD-1' }); + expect(conversationService.setNotificationId.calledOnce).to.equal(true); + expect( + conversationService.setNotificationId.calledBefore(conversationService.persistWorkflowOriginHydration) + ).to.equal(true); + }); + + it('omits messageBody when the origin content is empty', async () => { + const { service, conversationService } = makeService({ + notificationFindOne: sinon.stub().resolves({ payload: {} }), + }); + + await service.hydrate({ + agentId: 'agent1', + config: config as any, + conversation: conversation as any, + platformThreadId: 'slack:D123:1777837477.371619', + origin: { + _id: 'msg1', + _notificationId: 'notif1', + templateIdentifier: 'order-alerts', + content: ' ', + identifier: 'D123:1777837477.371619', + } as any, + }); + + const hydrateArgs = conversationService.persistWorkflowOriginHydration.firstCall.args[0]; + expect(hydrateArgs).to.not.have.property('messageBody'); + }); + + it('keeps the origin re-derivable when the hydration marker write fails', async () => { + const { service, conversationService } = makeService({ + notificationFindOne: sinon.stub().resolves({ payload: { orderId: 'ORD-1' } }), + persistWorkflowOriginHydration: sinon.stub().rejects(new Error('mongo timeout')), + }); + const target = { ...conversation } as any; + + const snapshot = await service.hydrate({ + agentId: 'agent1', + config: config as any, + conversation: target, + platformThreadId: 'slack:D123:1777837477.371619', + origin: { + _id: 'msg1', + _notificationId: 'notif1', + templateIdentifier: 'order-alerts', + content: 'Order ORD-1 shipped', + identifier: 'D123:1777837477.371619', + } as any, + }); + + expect(snapshot).to.equal(null); + expect(conversationService.setNotificationId.calledOnce).to.equal(true); + expect(target._notificationId).to.equal('notif1'); + }); + }); + + describe('resolveForTurn', () => { + const config = { + environmentId: 'env1', + organizationId: 'org1', + platform: AgentPlatformEnum.SLACK, + agentIdentifier: 'support-agent', + providerId: 'slack', + }; + const conversation = { + _id: 'conv1', + channels: [{ platform: AgentPlatformEnum.SLACK, _integrationId: 'int1', platformThreadId: 'slack:D123:' }], + participants: [{ type: 'subscriber', id: 'sub1' }], + }; + + afterEach(() => { + delete (conversation as { _notificationId?: string })._notificationId; + }); + + it('hydrates when a resolution is present and returns a hydrated snapshot', async () => { + const { service, conversationService } = makeService({ + notificationFindOne: sinon.stub().resolves({ payload: { orderId: 'ORD-1' } }), + }); + const origin = { + _id: 'msg1', + _notificationId: 'notif1', + _templateId: 'template-1', + templateIdentifier: 'order-alerts', + content: 'Order ORD-1 shipped', + identifier: 'D123:1777837477.371619', + channel: 'chat', + createdAt: '2026-01-01T00:00:00.000Z', + }; + + const snapshot = await service.resolveForTurn({ + agentId: 'agent1', + config: config as any, + conversation: conversation as any, + platformThreadId: 'slack:D123:1777837477.371619', + subscriberId: 'sub1', + resolution: { origin: origin as any, notificationId: 'notif1' }, + }); + + expect(conversationService.persistWorkflowOriginHydration.calledOnce).to.equal(true); + expect(snapshot?.source).to.equal('hydrated'); + expect(snapshot?.data.body).to.equal('Order ORD-1 shipped'); + expect(snapshot?.data.workflowIdentifier).to.equal('order-alerts'); + }); + + it('re-derives from conversation._notificationId on later turns', async () => { + const origin = { + _id: 'msg1', + _notificationId: 'notif1', + templateIdentifier: 'order-alerts', + content: 'Your order shipped', + identifier: 'D123:1777837477.371619', + channel: 'chat', + createdAt: '2026-01-01T00:00:00.000Z', + }; + const { service, conversationService, messageRepository } = makeService({ + find: sinon.stub().resolves([origin]), + notificationFindOne: sinon.stub().resolves({ payload: { orderId: 'ORD-9' } }), + }); + + const snapshot = await service.resolveForTurn({ + agentId: 'agent1', + config: config as any, + conversation: { ...conversation, _notificationId: 'notif1' } as any, + platformThreadId: 'slack:D123:1777837477.371619', + subscriberId: 'sub1', + resolution: null, + }); + + expect(conversationService.persistWorkflowOriginHydration.called).to.equal(false); + expect(messageRepository.find.calledOnce).to.equal(true); + expect(messageRepository.find.firstCall.args[0]).to.deep.equal({ + _environmentId: 'env1', + _agentId: 'agent1', + _subscriberId: 'subscriber-mongo-1', + _notificationId: 'notif1', + }); + expect(snapshot?.source).to.equal('existing'); + expect(snapshot?.data.body).to.equal('Your order shipped'); + expect(snapshot?.data.payload).to.deep.equal({ orderId: 'ORD-9' }); + }); + + it('does not re-derive the origin for another participant in the same thread', async () => { + const { service, messageRepository } = makeService({ + findBySubscriberId: sinon.stub().resolves({ _id: 'other-participant-mongo' }), + find: sinon.stub().resolves([]), + notificationFindOne: sinon.stub().resolves({ payload: { orderId: 'ORD-9' } }), + }); + + const snapshot = await service.resolveForTurn({ + agentId: 'agent1', + config: config as any, + conversation: { ...conversation, _notificationId: 'notif1' } as any, + platformThreadId: 'slack:D123:1777837477.371619', + subscriberId: 'other-participant', + resolution: null, + }); + + expect(messageRepository.find.firstCall.args[0]._subscriberId).to.equal('other-participant-mongo'); + expect(snapshot).to.equal(null); + }); + + it('skips the re-derive when the turn has no resolved subscriber', async () => { + const { service, messageRepository, subscriberRepository } = makeService({ + find: sinon.stub().resolves([{ _id: 'msg1', identifier: 'D123:1777837477.371619' }]), + }); + + const snapshot = await service.resolveForTurn({ + agentId: 'agent1', + config: config as any, + conversation: { ...conversation, _notificationId: 'notif1' } as any, + platformThreadId: 'slack:D123:1777837477.371619', + subscriberId: null, + resolution: null, + }); + + expect(snapshot).to.equal(null); + expect(subscriberRepository.findBySubscriberId.called).to.equal(false); + expect(messageRepository.find.called).to.equal(false); + }); + + it('skips the re-derive on a conversation that was never opened from a workflow send', async () => { + const { service, messageRepository } = makeService(); + + const snapshot = await service.resolveForTurn({ + agentId: 'agent1', + config: config as any, + conversation: conversation as any, + platformThreadId: 'slack:D123:1777837477.371619', + subscriberId: 'sub1', + resolution: null, + }); + + expect(snapshot).to.equal(null); + expect(messageRepository.find.called).to.equal(false); }); }); @@ -266,6 +466,52 @@ describe('WorkflowOriginService', () => { ORIGIN_MESSAGE_ID ); }); + + it('strips HTML from the origin body before passing messageBody', async () => { + const { service, conversationService } = makeService({ + notificationFindOne: sinon.stub().resolves({ payload: { orderId: 'ORD-1' } }), + }); + + await service.hydrate({ + agentId: 'agent1', + config: config as any, + conversation: conversation as any, + platformThreadId: 'email:thread1:', + origin: { + _id: ORIGIN_MESSAGE_ID, + _notificationId: 'notif1', + templateIdentifier: 'order-alerts', + content: '

Order ORD-1 shipped

', + identifier: 'provider-send-id', + } as any, + }); + + expect(conversationService.persistWorkflowOriginHydration.firstCall.args[0].messageBody).to.equal( + 'Order ORD-1 shipped' + ); + }); + + it('omits messageBody when stripped HTML is empty', async () => { + const { service, conversationService } = makeService({ + notificationFindOne: sinon.stub().resolves({ payload: {} }), + }); + + await service.hydrate({ + agentId: 'agent1', + config: config as any, + conversation: conversation as any, + platformThreadId: 'email:thread1:', + origin: { + _id: ORIGIN_MESSAGE_ID, + _notificationId: 'notif1', + templateIdentifier: 'order-alerts', + content: '

', + identifier: 'provider-send-id', + } as any, + }); + + expect(conversationService.persistWorkflowOriginHydration.firstCall.args[0]).to.not.have.property('messageBody'); + }); }); describe('WhatsApp', () => { @@ -514,6 +760,7 @@ describe('WorkflowOriginService', () => { const telegramOrigin = { _id: 'tg-msg1', _notificationId: 'tg-notif1', + _templateId: 'tg-template1', _jobId: 'tg-job1', transactionId: 'tg-txn1', templateIdentifier: 'order-alerts', @@ -763,7 +1010,7 @@ describe('WorkflowOriginService', () => { notificationFindOne: sinon.stub().resolves({ payload: { orderId: 'ORD-9' } }), }); - const content = await service.hydrate({ + const originData = await service.hydrate({ agentId: 'agent1', config: telegramConfig as any, conversation: conversation as any, @@ -777,8 +1024,9 @@ describe('WorkflowOriginService', () => { expect( conversationService.persistWorkflowOriginHydration.firstCall.args[0].signalData.workflowIdentifier ).to.equal('order-alerts'); - // Returned so a live managed session receives the origin it can no longer read from the transcript. - expect(content).to.equal('Your order shipped\n\nAdditional data for this message:\n{\n "orderId": "ORD-9"\n}'); + expect(originData?.data.workflowIdentifier).to.equal('order-alerts'); + expect(originData?.data.payload).to.deep.equal({ orderId: 'ORD-9' }); + expect(originData?.source).to.equal('hydrated'); }); it('no-ops hydrate when the thread id is unparseable', async () => { diff --git a/apps/api/src/app/agents/conversation-runtime/ingress/workflow-origin.service.ts b/apps/api/src/app/agents/conversation-runtime/ingress/workflow-origin.service.ts index 997bb1b64d4..5f2ce298e3f 100644 --- a/apps/api/src/app/agents/conversation-runtime/ingress/workflow-origin.service.ts +++ b/apps/api/src/app/agents/conversation-runtime/ingress/workflow-origin.service.ts @@ -8,6 +8,7 @@ import { NotificationRepository, SubscriberRepository, } from '@novu/dal'; +import { buildWorkflowOriginLine } from '@novu/framework/internal'; import { ChannelTypeEnum, ENDPOINT_TYPES } from '@novu/shared'; import type { Message } from 'chat'; import { ResolvedAgentConfig } from '../../channels/agent-config-resolver.service'; @@ -15,7 +16,6 @@ import { AgentPlatformEnum } from '../../shared/enums/agent-platform.enum'; import { captureAgentWarning } from '../../shared/errors/capture-agent-sentry'; import { AgentConversationService } from '../conversation/agent-conversation.service'; import { - buildWorkflowOriginSummary, extractAgentEmailOriginToken, extractTeamsQuotedActivityId, extractTelegramChatIdFromThreadId, @@ -24,8 +24,11 @@ import { isSendblueDirectThreadId, RECHECK_WORKFLOW_ORIGIN_PLATFORMS, resolvePlatformMessageId, + stripHtml, toProviderMessageLookupKey, WORKFLOW_ORIGIN_LOOKBACK_MS, + type WorkflowOriginData, + type WorkflowOriginSnapshot, } from './workflow-origin.helpers'; export interface WorkflowOriginResolution { @@ -166,21 +169,52 @@ export class WorkflowOriginService { } } - /** - * Returns the origin summary written to the transcript, or null when nothing was - * hydrated. Runtimes that keep history server-side (managed) only ever receive the - * new turn, so they need this value to see an origin attached mid-conversation. - */ + async resolveForTurn(params: { + agentId: string; + config: ResolvedAgentConfig; + conversation: ConversationEntity; + platformThreadId: string; + subscriberId: string | null; + resolution: WorkflowOriginResolution | null; + }): Promise { + const { agentId, config, conversation, platformThreadId, subscriberId, resolution } = params; + + if (resolution) { + return this.hydrate({ + agentId, + config, + conversation, + platformThreadId, + origin: resolution.origin, + }); + } + + const notificationId = conversation._notificationId; + if (!notificationId) { + return null; + } + + return this.rederiveFromNotificationId({ + agentId, + config, + conversation, + platformThreadId, + subscriberId, + notificationId, + }); + } + async hydrate(params: { agentId: string; config: ResolvedAgentConfig; conversation: ConversationEntity; platformThreadId: string; origin: MessageEntity; - }): Promise { + }): Promise { const { agentId, config, conversation, platformThreadId, origin } = params; - if (!origin._notificationId) { + const notificationId = origin._notificationId; + if (!notificationId) { return null; } @@ -190,13 +224,28 @@ export class WorkflowOriginService { } try { - const { messageContent, signalData } = await this.buildWorkflowOriginContext( + const data = await this.buildWorkflowOriginData( origin, conversation, config.environmentId, - config.organizationId + config.organizationId, + platformMessageId, + notificationId ); + // The notification id has to land first: it is what later turns re-derive from, while the + // hydration marker suppresses re-hydration. Marking first and failing here would strand the + // conversation without any way back to its origin. + await this.conversationService.setNotificationId( + config.environmentId, + config.organizationId, + conversation._id, + notificationId + ); + conversation._notificationId = notificationId; + + const messageBody = this.extractHydrationMessageBody(config.platform, origin); + await this.conversationService.persistWorkflowOriginHydration({ conversationId: conversation._id, channel: this.conversationService.getPrimaryChannel(conversation), @@ -205,11 +254,20 @@ export class WorkflowOriginService { organizationId: config.organizationId, platformMessageId, platformThreadId, - messageContent, - signalData, + signalData: { + notificationId: data.notificationId, + jobId: data.jobId, + messageId: data.messageId, + transactionId: data.transactionId, + workflowIdentifier: data.workflowIdentifier, + stepId: data.stepId, + subscriberId: data.subscriberId, + payload: data.payload, + }, + ...(messageBody ? { messageBody } : {}), }); - return messageContent; + return { data, source: 'hydrated' }; } catch (err) { captureAgentWarning(err, { component: 'workflow-origin-service', @@ -218,7 +276,77 @@ export class WorkflowOriginService { }); this.logger.warn( { err, agentId, platformThreadId, messageId: origin._id, notificationId: origin._notificationId }, - 'Failed to hydrate workflow origin into conversation history' + 'Failed to hydrate workflow origin into conversation' + ); + + return null; + } + } + + private async rederiveFromNotificationId(params: { + agentId: string; + config: ResolvedAgentConfig; + conversation: ConversationEntity; + platformThreadId: string; + subscriberId: string | null; + notificationId: string; + }): Promise { + const { agentId, config, conversation, platformThreadId, subscriberId, notificationId } = params; + + if (!subscriberId) { + return null; + } + + try { + // A thread can carry turns from other participants (shared Slack channel, forwarded email), + // so the origin payload is only re-derived for the subscriber the notification was sent to. + const subscriber = await this.subscriberRepository.findBySubscriberId(config.environmentId, subscriberId); + if (!subscriber) { + return null; + } + + const [origin] = await this.messageRepository.find( + { + _environmentId: config.environmentId, + _agentId: agentId, + _subscriberId: subscriber._id, + _notificationId: notificationId, + }, + '', + { + sort: { createdAt: -1 }, + limit: 1, + } + ); + + if (!origin) { + return null; + } + + const platformMessageId = resolvePlatformMessageId(config.platform, origin, platformThreadId); + if (!platformMessageId) { + return null; + } + + const data = await this.buildWorkflowOriginData( + origin, + conversation, + config.environmentId, + config.organizationId, + platformMessageId, + notificationId + ); + + return { data, source: 'existing' }; + } catch (err) { + captureAgentWarning(err, { + component: 'workflow-origin-service', + operation: 'rederive-workflow-origin', + agentId, + }); + this.logger.warn( + { err, agentId, conversationId: conversation._id, notificationId }, + 'Failed to re-derive workflow origin from conversation notification id' ); return null; @@ -298,18 +426,30 @@ export class WorkflowOriginService { return this.conversationService.isWorkflowOriginHydrated(config.environmentId, conversationId, platformMessageId); } - private async buildWorkflowOriginContext( + private extractHydrationMessageBody(platform: AgentPlatformEnum, origin: MessageEntity): string { + const storedContent = typeof origin.content === 'string' ? origin.content.trim() : ''; + if (!storedContent) { + return ''; + } + + if (platform === AgentPlatformEnum.EMAIL) { + return stripHtml(storedContent); + } + + return storedContent; + } + + private async buildWorkflowOriginData( originMessage: MessageEntity, conversation: ConversationEntity, environmentId: string, - organizationId: string - ): Promise<{ - messageContent: string; - signalData: Record; - }> { + organizationId: string, + platformMessageId: string, + notificationId: string + ): Promise { const notification = await this.notificationRepository.findOne( { - _id: originMessage._notificationId, + _id: notificationId, _environmentId: environmentId, _organizationId: organizationId, }, @@ -323,24 +463,24 @@ export class WorkflowOriginService { const storedContent = typeof originMessage.content === 'string' ? originMessage.content.trim() : ''; const workflowIdentifier = originMessage.templateIdentifier || 'unknown'; - const messageContent = buildWorkflowOriginSummary(workflowIdentifier, storedContent, payload); + const body = buildWorkflowOriginLine(workflowIdentifier, storedContent); const subscriberId = conversation.participants.find( (p) => p.type === ConversationParticipantTypeEnum.SUBSCRIBER )?.id; return { - messageContent, - signalData: { - notificationId: originMessage._notificationId, - jobId: originMessage._jobId, - messageId: originMessage._id, - transactionId: originMessage.transactionId, - workflowIdentifier, - stepId: originMessage.stepId, - subscriberId, - payload, - }, + notificationId, + workflowIdentifier, + messageId: originMessage._id, + platformMessageId, + sentAt: originMessage.createdAt, + body, + payload, + ...(originMessage._jobId ? { jobId: originMessage._jobId } : {}), + ...(originMessage.stepId ? { stepId: originMessage.stepId } : {}), + ...(originMessage.transactionId ? { transactionId: originMessage.transactionId } : {}), + ...(subscriberId ? { subscriberId } : {}), }; } } diff --git a/apps/api/src/app/agents/conversation-runtime/runtime/bridge-executor.service.spec.ts b/apps/api/src/app/agents/conversation-runtime/runtime/bridge-executor.service.spec.ts index 090ef1749ae..fa4d4ecb24f 100644 --- a/apps/api/src/app/agents/conversation-runtime/runtime/bridge-executor.service.spec.ts +++ b/apps/api/src/app/agents/conversation-runtime/runtime/bridge-executor.service.spec.ts @@ -185,6 +185,46 @@ describe('BridgeExecutorService', () => { expect(payload).to.not.have.property('eventsUrl'); expect(payload.replyUrl).to.match(/\/v1\/agents\/agent-1\/reply$/); }); + + it('should map workflowOrigin onto notification for the bridge wire', async () => { + const { service } = makeService(); + const workflowOrigin = { + data: { + notificationId: 'notif-1', + workflowIdentifier: 'order-shipped', + messageId: 'msg-1', + platformMessageId: 'wamid.abc', + sentAt: '2026-01-01T00:00:00.000Z', + body: 'Your order ORD-1 shipped', + payload: { orderId: 'ORD-1' }, + jobId: 'job-1', + }, + source: 'existing' as const, + }; + + const payload = await (service as any).buildPayload({ + ...makeExecutionParams(), + workflowOrigin, + }); + + expect(payload.notification).to.deep.equal({ + id: 'notif-1', + workflowId: 'order-shipped', + messageId: 'msg-1', + platformMessageId: 'wamid.abc', + sentAt: '2026-01-01T00:00:00.000Z', + body: 'Your order ORD-1 shipped', + payload: { orderId: 'ORD-1' }, + }); + }); + + it('should send notification null when no workflow origin is present', async () => { + const { service } = makeService(); + + const payload = await (service as any).buildPayload(makeExecutionParams()); + + expect(payload.notification).to.equal(null); + }); }); describe('mapRichContentForBridge', () => { diff --git a/apps/api/src/app/agents/conversation-runtime/runtime/bridge-executor.service.ts b/apps/api/src/app/agents/conversation-runtime/runtime/bridge-executor.service.ts index a18734d0032..3a60700e3ce 100644 --- a/apps/api/src/app/agents/conversation-runtime/runtime/bridge-executor.service.ts +++ b/apps/api/src/app/agents/conversation-runtime/runtime/bridge-executor.service.ts @@ -18,24 +18,21 @@ import type { AgentConversation, AgentHistoryEntry, AgentMessage, + AgentNotification, AgentPlatformContext, AgentReaction, AgentSubscriber, } from '@novu/framework'; import type { AgentBridgeRequest } from '@novu/framework/internal'; import { AgentEventEnum, HttpHeaderKeysEnum } from '@novu/framework/internal'; -import { - AGENT_PLATFORM_PROVISION_SOURCE, - AGENT_PROVISION_DATA_KEYS, - AgentSubscriberAccessEnum, - FeatureFlagsKeysEnum, -} from '@novu/shared'; +import { FeatureFlagsKeysEnum } from '@novu/shared'; import type { Message } from 'chat'; import { ResolvedAgentConfig } from '../../channels/agent-config-resolver.service'; import { captureAgentException, captureAgentWarning } from '../../shared/errors/capture-agent-sentry'; import { buildAgentApiRootUrl } from '../../shared/util/agent-api-root-url'; import { AgentAttachmentStorage, type StoredAttachment } from '../conversation/agent-attachment-storage.service'; import { AgentConversationService } from '../conversation/agent-conversation.service'; +import type { WorkflowOriginData, WorkflowOriginSnapshot } from '../ingress/workflow-origin.helpers'; const MAX_RETRIES = 2; @@ -145,6 +142,7 @@ export interface AgentExecutionParams { platformContext: AgentPlatformContext; /** Trusted connect-time context resolved from the inbound channel connection; forwarded as `ctx.context`. */ context?: AgentContextPayload | null; + workflowOrigin?: WorkflowOriginSnapshot | null; /** * Per-context bridge URL override resolved from the connect-time context. Takes precedence over the * agent's default `bridgeUrl` (but not the active dev bridge). Re-validated by the SSRF guard on @@ -399,6 +397,7 @@ export class BridgeExecutorService { subscriber: this.mapSubscriber(subscriber), subscriberAccess: config.subscriberAccess, context: params.context ?? null, + notification: params.workflowOrigin ? mapWorkflowOriginToNotification(params.workflowOrigin.data) : null, history: await this.mapHistory(history), platform: config.platform, platformContext, @@ -693,3 +692,15 @@ export class BridgeExecutorService { return `${context.organizationId}/${context.environmentId}/${AGENTS_STORAGE_FOLDER}/${context.conversationId}/`; } } + +function mapWorkflowOriginToNotification(origin: WorkflowOriginData): AgentNotification { + return { + id: origin.notificationId, + workflowId: origin.workflowIdentifier, + messageId: origin.messageId, + platformMessageId: origin.platformMessageId, + sentAt: origin.sentAt, + body: origin.body, + payload: origin.payload, + }; +} diff --git a/apps/api/src/app/agents/conversation-runtime/runtime/bridge.runtime.ts b/apps/api/src/app/agents/conversation-runtime/runtime/bridge.runtime.ts index cd696c8f9e6..26936276c3e 100644 --- a/apps/api/src/app/agents/conversation-runtime/runtime/bridge.runtime.ts +++ b/apps/api/src/app/agents/conversation-runtime/runtime/bridge.runtime.ts @@ -65,6 +65,7 @@ export class BridgeRuntime implements AgentRuntime { conversation: turn.conversation, subscriber: turn.subscriber, context: turn.context ?? null, + workflowOrigin: turn.workflowOrigin ?? null, bridgeUrlOverride: turn.bridgeUrlOverride, message: turn.message, platformContext: buildAgentPlatformContext({ diff --git a/apps/api/src/app/agents/conversation-runtime/runtime/conversation-turn.ts b/apps/api/src/app/agents/conversation-runtime/runtime/conversation-turn.ts index 9d2e35e33e0..ba517aa62af 100644 --- a/apps/api/src/app/agents/conversation-runtime/runtime/conversation-turn.ts +++ b/apps/api/src/app/agents/conversation-runtime/runtime/conversation-turn.ts @@ -5,6 +5,7 @@ import type { ResolvedAgentConfig } from '../../channels/agent-config-resolver.s import type { AgentEventEnum } from '../../shared/enums/agent-event.enum'; import type { SubscriberResolution } from '../../shared/types/subscriber-resolution'; import type { StoredAttachment } from '../conversation/agent-attachment-storage.service'; +import type { WorkflowOriginSnapshot } from '../ingress/workflow-origin.helpers'; import type { BridgeReaction } from './bridge-executor.service'; export interface ConversationTurn { @@ -32,5 +33,5 @@ export interface ConversationTurn { storedAttachments?: StoredAttachment[]; action?: AgentAction; reaction?: BridgeReaction; - workflowOriginContent?: string; + workflowOrigin?: WorkflowOriginSnapshot | null; } diff --git a/apps/api/src/app/agents/conversation-runtime/runtime/runtime-resolver.service.ts b/apps/api/src/app/agents/conversation-runtime/runtime/runtime-resolver.service.ts index 4c91dfcf673..7a3bc18d0c0 100644 --- a/apps/api/src/app/agents/conversation-runtime/runtime/runtime-resolver.service.ts +++ b/apps/api/src/app/agents/conversation-runtime/runtime/runtime-resolver.service.ts @@ -1,5 +1,6 @@ import { Injectable } from '@nestjs/common'; import type { AgentEntity } from '@novu/dal'; +import { HumanRelayRuntime } from '../../human-relay/human-relay.runtime'; import { ManagedRuntime } from '../../managed-runtime/managed.runtime'; import type { AgentRuntime } from './agent-runtime.port'; import { BridgeRuntime } from './bridge.runtime'; @@ -8,10 +9,15 @@ import { BridgeRuntime } from './bridge.runtime'; export class RuntimeResolver { constructor( private readonly bridgeRuntime: BridgeRuntime, - private readonly managedRuntime: ManagedRuntime + private readonly managedRuntime: ManagedRuntime, + private readonly humanRelayRuntime: HumanRelayRuntime ) {} resolve(agent: Pick | null): AgentRuntime { + if (agent?.runtime === 'human_relay') { + return this.humanRelayRuntime; + } + if (agent?.runtime === 'managed' && agent.managedRuntime) { return this.managedRuntime; } diff --git a/apps/api/src/app/agents/e2e/agent-webhook.e2e.ts b/apps/api/src/app/agents/e2e/agent-webhook.e2e.ts index 3c7d4822af0..428767481e9 100644 --- a/apps/api/src/app/agents/e2e/agent-webhook.e2e.ts +++ b/apps/api/src/app/agents/e2e/agent-webhook.e2e.ts @@ -6,12 +6,15 @@ import { ConversationActivitySenderTypeEnum, ConversationParticipantTypeEnum, ConversationStatusEnum, + MessageRepository, + NotificationRepository, SubscriberRepository, } from '@novu/dal'; -import { ENDPOINT_TYPES } from '@novu/shared'; +import { ChannelTypeEnum, ENDPOINT_TYPES } from '@novu/shared'; import { testServer } from '@novu/testing'; import { expect } from 'chai'; import type { EmojiValue } from 'chat'; +import { Types } from 'mongoose'; import sinon from 'sinon'; import { AgentConfigResolver } from '../channels/agent-config-resolver.service'; import { ChatInstanceRegistry } from '../conversation-runtime/ingress/chat-instance.registry'; @@ -386,6 +389,82 @@ describe('Agent Webhook - inbound flow #novu-v2', () => { expect(call.platformContext.threadId).to.equal(threadId); expect(call.platformContext.channelId).to.equal('C_TEST'); expect(call.platformContext.isDM).to.equal(false); + expect(call.workflowOrigin, 'no origin seeded on first touch').to.equal(null); + }); + + it('forwards a re-derived workflow origin on later turns via workflowOrigin', async () => { + const subscriberRepository = new SubscriberRepository(); + const messageRepository = new MessageRepository(); + const notificationRepository = new NotificationRepository(); + const subscriber = await subscriberRepository.create({ + subscriberId: `sub-origin-${Date.now()}`, + firstName: 'Origin', + lastName: 'Test', + email: 'origin@test.com', + _environmentId: ctx.session.environment._id, + _organizationId: ctx.session.organization._id, + }); + + await seedChannelEndpoint(ctx, 'U_ORIGIN', subscriber.subscriberId); + + const threadId = `T_ORIGIN_${Date.now()}`; + await invokeInbound(threadId, mockMessage({ userId: 'U_ORIGIN', text: 'first' })); + await waitForBridgeCallCount(1); + + const conversation = await conversationRepository.findByPlatformThread( + ctx.session.environment._id, + ctx.session.organization._id, + ctx.agentId, + ctx.integrationId, + threadId + ); + expect(conversation).to.exist; + + const notification = await notificationRepository.create({ + _environmentId: ctx.session.environment._id, + _organizationId: ctx.session.organization._id, + _subscriberId: subscriber._id, + _templateId: new Types.ObjectId().toString(), + transactionId: `txn-origin-${Date.now()}`, + channels: [ChannelTypeEnum.CHAT], + payload: { orderId: 'ORD-42' }, + to: { subscriberId: subscriber.subscriberId }, + }); + + await messageRepository.create({ + _environmentId: ctx.session.environment._id, + _organizationId: ctx.session.organization._id, + _subscriberId: subscriber._id, + _agentId: ctx.agentId, + _notificationId: notification._id, + _templateId: notification._templateId, + content: 'Your order ORD-42 shipped', + templateIdentifier: 'order-shipped', + identifier: `C_TEST:1777837477.371619`, + channel: ChannelTypeEnum.CHAT, + transactionId: notification.transactionId, + }); + + await conversationRepository.update( + { + _id: conversation!._id, + _environmentId: ctx.session.environment._id, + _organizationId: ctx.session.organization._id, + }, + { $set: { _notificationId: notification._id } } + ); + + bridgeCalls = []; + await invokeInbound(threadId, mockMessage({ userId: 'U_ORIGIN', text: 'where is my order?' })); + await waitForBridgeCallCount(1); + + expect(bridgeCalls[0].workflowOrigin?.data).to.deep.include({ + workflowIdentifier: 'order-shipped', + notificationId: notification._id, + payload: { orderId: 'ORD-42' }, + body: 'Your order ORD-42 shipped', + }); + expect(bridgeCalls[0].workflowOrigin?.source).to.equal('existing'); }); it('Passes null subscriber in the bridge payload for first-time Slack senders on open custom-code', async () => { diff --git a/apps/api/src/app/agents/e2e/helpers/telegram-api-stub.ts b/apps/api/src/app/agents/e2e/helpers/telegram-api-stub.ts index dee412aee55..3261cbcc258 100644 --- a/apps/api/src/app/agents/e2e/helpers/telegram-api-stub.ts +++ b/apps/api/src/app/agents/e2e/helpers/telegram-api-stub.ts @@ -82,6 +82,16 @@ export async function startTelegramApiStub(): Promise { const payload = await readJsonBody(req); calls.push({ method, payload }); + // The adapter probes the non-standard `sendRichMessage*` extension before + // falling back to `sendMessage`. Real Telegram answers 404 — mirror that so + // markdown deliveries exercise the production fallback path. + if (method.startsWith('sendRichMessage')) { + res.writeHead(404, { 'content-type': 'application/json' }); + res.end(JSON.stringify({ ok: false, error_code: 404, description: 'Not Found: method not found' })); + + return; + } + res.writeHead(200, { 'content-type': 'application/json' }); res.end(JSON.stringify(buildResponse(method, payload))); }); diff --git a/apps/api/src/app/agents/human-relay/human-action-id.spec.ts b/apps/api/src/app/agents/human-relay/human-action-id.spec.ts new file mode 100644 index 00000000000..c4e409f26f2 --- /dev/null +++ b/apps/api/src/app/agents/human-relay/human-action-id.spec.ts @@ -0,0 +1,51 @@ +import { expect } from 'chai'; +import { + buildHumanApproveActionId, + buildHumanDenyActionId, + buildHumanDisambiguationActionId, + buildHumanOptionActionId, + parseHumanActionId, +} from './human-action-id'; + +describe('human action id grammar', () => { + it('round-trips approve and deny', () => { + expect(parseHumanActionId(buildHumanApproveActionId('hi_abc123'))).to.deep.equal({ + type: 'approve', + identifier: 'hi_abc123', + }); + expect(parseHumanActionId(buildHumanDenyActionId('hi_abc123'))).to.deep.equal({ + type: 'deny', + identifier: 'hi_abc123', + }); + }); + + it('round-trips choose options (option ids may contain separators)', () => { + expect(parseHumanActionId(buildHumanOptionActionId('hi_abc123', 'opt_2'))).to.deep.equal({ + type: 'option', + identifier: 'hi_abc123', + optionId: 'opt_2', + }); + expect(parseHumanActionId('human:hi_x:opt:a:b')).to.deep.equal({ + type: 'option', + identifier: 'hi_x', + optionId: 'a:b', + }); + }); + + it('round-trips disambiguation picks', () => { + expect(parseHumanActionId(buildHumanDisambiguationActionId('hi_abc123'))).to.deep.equal({ + type: 'disambiguation-pick', + identifier: 'hi_abc123', + }); + }); + + it('ignores foreign and malformed ids', () => { + expect(parseHumanActionId(undefined)).to.equal(null); + expect(parseHumanActionId('tool-approval:x:approve')).to.equal(null); + expect(parseHumanActionId('human:')).to.equal(null); + expect(parseHumanActionId('human:hi_x')).to.equal(null); + expect(parseHumanActionId('human:hi_x:unknown')).to.equal(null); + expect(parseHumanActionId('human:hi_x:opt:')).to.equal(null); + expect(parseHumanActionId('human:pick:')).to.equal(null); + }); +}); diff --git a/apps/api/src/app/agents/human-relay/human-action-id.ts b/apps/api/src/app/agents/human-relay/human-action-id.ts new file mode 100644 index 00000000000..49150f6cc61 --- /dev/null +++ b/apps/api/src/app/agents/human-relay/human-action-id.ts @@ -0,0 +1,67 @@ +/** + * Action-id grammar for human-interaction cards. Kept distinct from + * `tool-approval:*` / `mcp-approval:*` so `parseToolApprovalActionId` and the + * human parser can never claim each other's clicks. + * + * human::approve approve verdict + * human::deny deny verdict + * human::opt: choose pick + * human:pick: disambiguation pick ("which question?") + */ + +const PREFIX = 'human:'; +const PICK_PREFIX = 'human:pick:'; + +export type HumanActionParsed = + | { type: 'approve' | 'deny'; identifier: string } + | { type: 'option'; identifier: string; optionId: string } + | { type: 'disambiguation-pick'; identifier: string }; + +export function buildHumanApproveActionId(identifier: string): string { + return `${PREFIX}${identifier}:approve`; +} + +export function buildHumanDenyActionId(identifier: string): string { + return `${PREFIX}${identifier}:deny`; +} + +export function buildHumanOptionActionId(identifier: string, optionId: string): string { + return `${PREFIX}${identifier}:opt:${optionId}`; +} + +export function buildHumanDisambiguationActionId(identifier: string): string { + return `${PICK_PREFIX}${identifier}`; +} + +export function parseHumanActionId(actionId: string | undefined): HumanActionParsed | null { + if (!actionId?.startsWith(PREFIX)) { + return null; + } + + if (actionId.startsWith(PICK_PREFIX)) { + const identifier = actionId.slice(PICK_PREFIX.length); + + return identifier ? { type: 'disambiguation-pick', identifier } : null; + } + + const rest = actionId.slice(PREFIX.length); + const firstColon = rest.indexOf(':'); + if (firstColon <= 0) { + return null; + } + + const identifier = rest.slice(0, firstColon); + const verb = rest.slice(firstColon + 1); + + if (verb === 'approve' || verb === 'deny') { + return { type: verb, identifier }; + } + + if (verb.startsWith('opt:')) { + const optionId = verb.slice('opt:'.length); + + return optionId ? { type: 'option', identifier, optionId } : null; + } + + return null; +} diff --git a/apps/api/src/app/agents/human-relay/human-card.builder.ts b/apps/api/src/app/agents/human-relay/human-card.builder.ts new file mode 100644 index 00000000000..4dc3800a430 --- /dev/null +++ b/apps/api/src/app/agents/human-relay/human-card.builder.ts @@ -0,0 +1,194 @@ +import type { HumanInteractionEntity } from '@novu/dal'; +import { HumanInteractionKindEnum, HumanInteractionStatusEnum, type HumanInteractionOption } from '@novu/shared'; +import type { ActionsElement, ButtonElement, CardElement, TextElement } from 'chat'; +import { + buildHumanApproveActionId, + buildHumanDenyActionId, + buildHumanDisambiguationActionId, + buildHumanOptionActionId, +} from './human-action-id'; +import type { ReplyContentDto } from '../shared/dtos/agent-reply-payload.dto'; + +const LISTED_OPTION_LABEL_MAX = 200; + +/** + * Letters, not full labels, on `choose`/disambiguation buttons. Chat platform + * button UIs (Telegram inline keyboards especially) truncate or wrap long + * label text badly; the full text is always listed in the message body + * instead, and the button just needs to be tappable and short. + */ +const OPTION_LETTERS = 'ABCDEFGHIJ'; + +function optionLetter(index: number): string { + return OPTION_LETTERS[index] ?? String(index + 1); +} + +function button(id: string, label: string, style: 'default' | 'primary'): ButtonElement { + return { type: 'button', id, label, style }; +} + +function textChild(content: string): TextElement { + return { type: 'text', content }; +} + +/** + * The portable-card renderer joins title/subtitle/body with a single `\n` + * (no blank line), which reads as one cramped block. A leading `\n` on the + * first body text child turns that single join newline into a real blank + * line — use this for whichever text child comes right after title/subtitle. + */ +function bodyText(content: string): TextElement { + return textChild(`\n${content}`); +} + +/** "**A.**