feat(framework): self-hosted brain-controlled typing indicator fixes NV-8128 (#11671)

Author: ChmaraX

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

@@ -36,6 +36,7 @@ export class AgentReplyController {
         resolve: body.resolve,
         signals: body.signals as Signal[],
         addReactions: body.addReactions,
+        typing: body.typing,
       })
     );
   }

apps/api/src/app/agents/conversation-runtime/reply/handle-agent-reply/handle-agent-reply.command.ts

@@ -48,6 +48,9 @@ export class HandleAgentReplyCommand extends EnvironmentWithUserCommand {
   @IsObject()
   plan?: { model: PlanModel; phase: PlanPhase; messageId?: string };
 
+  @IsOptional()
+  typing?: { status?: string } | 'stop';
+
   @IsOptional()
   slackNative?: SlackNativeDelivery;
 }

apps/api/src/app/agents/conversation-runtime/reply/handle-agent-reply/handle-agent-reply.usecase.ts

@@ -57,10 +57,11 @@ export class HandleAgentReply {
       !command.resolve &&
       !command.signals?.length &&
       !command.addReactions?.length &&
-      !command.plan
+      !command.plan &&
+      !command.typing
     ) {
       throw new BadRequestException(
-        'At least one of reply, edit, resolve, signals, addReactions, or plan must be provided'
+        'At least one of reply, edit, resolve, signals, addReactions, plan, or typing must be provided'
       );
     }
 
@@ -76,6 +77,10 @@ export class HandleAgentReply {
     const channel = this.conversationService.getPrimaryChannel(conversation);
     const agentName = await this.resolveValidatedAgentNameForDelivery(command, conversation);
 
+    if (command.typing) {
+      await this.deliverTyping(command, conversation, channel, command.typing);
+    }
+
     if (command.edit) {
       return this.deliverEdit(command, conversation, channel, command.edit, agentName);
     }
@@ -149,6 +154,7 @@ export class HandleAgentReply {
     if (triggerSignalCount > 0) actions.push('trigger_signals');
     if (metadataSignalCount > 0) actions.push('metadata_signals');
     if (reactionCount > 0) actions.push('add_reactions');
+    if (command.typing) actions.push('typing');
 
     trackAgentReplyProcessed(this.analyticsService, {
       userId: command.userId,
@@ -302,6 +308,26 @@ export class HandleAgentReply {
     );
   }
 
+  private async deliverTyping(
+    command: HandleAgentReplyCommand,
+    conversation: ConversationEntity,
+    channel: ConversationChannel,
+    typing: NonNullable<HandleAgentReplyCommand['typing']>
+  ): Promise<void> {
+    const status = typing === 'stop' ? '' : (typing.status ?? 'Thinking...');
+
+    try {
+      await this.outboundGateway.startTypingInConversation(
+        conversation._agentId,
+        command.integrationIdentifier,
+        channel.platformThreadId,
+        status
+      );
+    } catch (err) {
+      this.logger.warn(err, `[agent:${command.agentIdentifier}] Failed to set typing status`);
+    }
+  }
+
   private async executeSignals(
     command: HandleAgentReplyCommand,
     conversation: ConversationEntity,

apps/api/src/app/agents/e2e/agent-reply.e2e.ts

@@ -38,6 +38,7 @@ describe('Agent Reply - /agents/:agentId/reply #novu-v2', () => {
       .resolves({ messageId: 'platform-msg-1', platformThreadId: 'platform-thread-1' });
     sinon.stub(outboundGateway, 'reactToMessage').resolves();
     sinon.stub(outboundGateway, 'removeReaction').resolves();
+    sinon.stub(outboundGateway, 'startTypingInConversation').resolves();
   });
 
   function postReply(body: Record<string, unknown>) {
@@ -325,6 +326,84 @@ describe('Agent Reply - /agents/:agentId/reply #novu-v2', () => {
     });
   });
 
+  describe('Typing', () => {
+    it('should set a typing status from a typing-only request', async () => {
+      const conversationId = await seedConversation(ctx);
+      const outboundGateway = testServer.getService(OutboundGateway);
+
+      const res = await postReply({
+        conversationId,
+        integrationIdentifier: ctx.integrationIdentifier,
+        typing: { status: 'Searching the docs…' },
+      });
+
+      expect(res.status).to.equal(200);
+      expect(res.body.data).to.be.null;
+
+      const stub = outboundGateway.startTypingInConversation as sinon.SinonStub;
+      expect(stub.callCount).to.equal(1);
+      expect(stub.getCall(0).args[3]).to.equal('Searching the docs…');
+    });
+
+    it('should default the status text when typing has no status', async () => {
+      const conversationId = await seedConversation(ctx);
+      const outboundGateway = testServer.getService(OutboundGateway);
+
+      const res = await postReply({
+        conversationId,
+        integrationIdentifier: ctx.integrationIdentifier,
+        typing: {},
+      });
+
+      expect(res.status).to.equal(200);
+
+      const stub = outboundGateway.startTypingInConversation as sinon.SinonStub;
+      expect(stub.getCall(0).args[3]).to.equal('Thinking...');
+    });
+
+    it('should clear the status with an empty string for typing "stop"', async () => {
+      const conversationId = await seedConversation(ctx);
+      const outboundGateway = testServer.getService(OutboundGateway);
+
+      const res = await postReply({
+        conversationId,
+        integrationIdentifier: ctx.integrationIdentifier,
+        typing: 'stop',
+      });
+
+      expect(res.status).to.equal(200);
+
+      const stub = outboundGateway.startTypingInConversation as sinon.SinonStub;
+      expect(stub.getCall(0).args[3]).to.equal('');
+    });
+
+    it('should not fail the turn when the typing call throws', async () => {
+      const conversationId = await seedConversation(ctx);
+      const outboundGateway = testServer.getService(OutboundGateway);
+      (outboundGateway.startTypingInConversation as sinon.SinonStub).rejects(new Error('platform down'));
+
+      const res = await postReply({
+        conversationId,
+        integrationIdentifier: ctx.integrationIdentifier,
+        typing: { status: 'Working…' },
+      });
+
+      expect(res.status).to.equal(200);
+    });
+
+    it('should reject an invalid typing op', async () => {
+      const conversationId = await seedConversation(ctx);
+
+      const res = await postReply({
+        conversationId,
+        integrationIdentifier: ctx.integrationIdentifier,
+        typing: 'go',
+      });
+
+      expect(res.status).to.equal(422);
+    });
+  });
+
   describe('Inactive agent', () => {
     it('should return 422 when agent is inactive', async () => {
       const conversationId = await seedConversation(ctx);

apps/api/src/app/agents/shared/dtos/agent-reply-payload.dto.ts

@@ -149,6 +149,26 @@ export class IsValidReplyContent implements ValidatorConstraintInterface {
   }
 }
 
+@ValidatorConstraint({ name: 'isValidTypingOp', async: false })
+export class IsValidTypingOp implements ValidatorConstraintInterface {
+  validate(value: unknown): boolean {
+    if (value === undefined) return true;
+    if (value === 'stop') return true;
+
+    if (typeof value === 'object' && value !== null) {
+      const status = (value as { status?: unknown }).status;
+
+      return status === undefined || typeof status === 'string';
+    }
+
+    return false;
+  }
+
+  defaultMessage(): string {
+    return "typing must be 'stop' or an object with an optional string status.";
+  }
+}
+
 export class ReplyContentDto {
   @ApiPropertyOptional()
   @IsOptional()
@@ -283,4 +303,13 @@ export class AgentReplyPayloadDto {
   @ValidateNested({ each: true })
   @Type(() => AddReactionPayloadDto)
   addReactions?: AddReactionPayloadDto[];
+
+  @ApiPropertyOptional({
+    description:
+      'Per-turn typing/status control. `{ status?: string }` sets the status text ' +
+      '(omit for the default "Thinking…"); `"stop"` clears it. Best-effort per platform.',
+  })
+  @IsOptional()
+  @Validate(IsValidTypingOp)
+  typing?: { status?: string } | 'stop';
 }

packages/framework/src/resources/agent/agent.context.ts

@@ -18,6 +18,8 @@ import type {
   SentMessageInfo,
   Signal,
   TriggerRecipientsPayload,
+  TypingControl,
+  TypingOp,
 } from './agent.types';
 import { AgentEventEnum } from './agent.types';
 
@@ -258,6 +260,7 @@ export class AgentContextImpl {
   readonly history: AgentHistoryEntry[];
   readonly platform: string;
   readonly platformContext: AgentPlatformContext;
+  readonly typing: TypingControl;
 
   readonly metadata: {
     get(key: string): unknown;
@@ -317,6 +320,17 @@ export class AgentContextImpl {
         return { ...self._metadataState } as Readonly<Record<string, unknown>>;
       },
     };
+
+    const postTyping = (op: TypingOp): Promise<void> =>
+      this._post({
+        conversationId: this._conversationId,
+        integrationIdentifier: this._integrationIdentifier,
+        typing: op,
+      }).then(() => undefined);
+
+    const typing = ((status?: string) => postTyping(status === undefined ? {} : { status })) as TypingControl;
+    typing.stop = () => postTyping('stop');
+    this.typing = typing;
   }
 
   async reply(content: MessageContent, options?: { files?: FileRef[] }): Promise<ReplyHandle> {

packages/framework/src/resources/agent/agent.test.ts

@@ -1868,4 +1868,101 @@ describe('agent dispatch via NovuRequestHandler', () => {
     expect(JSON.parse(replyCalls[0][1].body).reply.markdown).toBe('Thinking…');
     expect(JSON.parse(replyCalls[1][1].body).reply.markdown).toBe('Final answer');
   });
+
+  it('should post a typing status op for ctx.typing(text)', async () => {
+    const testBot = agent('test-bot', {
+      onMessage: async (_message, ctx) => {
+        await ctx.typing('Searching the docs…');
+      },
+    });
+
+    const handler = new NovuRequestHandler({
+      frameworkName: 'test',
+      agents: [testBot],
+      client,
+      handler: () => ({
+        body: () => createMockBridgeRequest(),
+        headers: () => null,
+        method: () => 'POST',
+        url: () => new URL(`http://localhost?action=${PostActionEnum.AGENT_EVENT}&agentId=test-bot&event=onMessage`),
+        transformResponse: (res: any) => res,
+      }),
+    });
+
+    await handler.createHandler()();
+    await vi.waitFor(() => expect(fetchMock).toHaveBeenCalled());
+
+    const replyCall = fetchMock.mock.calls.find(
+      (call: any[]) => call[0] === 'https://api.novu.co/v1/agents/test-bot/reply'
+    );
+    const body = JSON.parse(replyCall![1].body);
+
+    expect(body.typing).toEqual({ status: 'Searching the docs…' });
+    expect(body.reply).toBeUndefined();
+    expect(body.conversationId).toBe('conv-456');
+    expect(body.integrationIdentifier).toBe('slack-main');
+  });
+
+  it('should post an empty status op for ctx.typing() with no text', async () => {
+    const testBot = agent('test-bot', {
+      onMessage: async (_message, ctx) => {
+        await ctx.typing();
+      },
+    });
+
+    const handler = new NovuRequestHandler({
+      frameworkName: 'test',
+      agents: [testBot],
+      client,
+      handler: () => ({
+        body: () => createMockBridgeRequest(),
+        headers: () => null,
+        method: () => 'POST',
+        url: () => new URL(`http://localhost?action=${PostActionEnum.AGENT_EVENT}&agentId=test-bot&event=onMessage`),
+        transformResponse: (res: any) => res,
+      }),
+    });
+
+    await handler.createHandler()();
+    await vi.waitFor(() => expect(fetchMock).toHaveBeenCalled());
+
+    const replyCall = fetchMock.mock.calls.find(
+      (call: any[]) => call[0] === 'https://api.novu.co/v1/agents/test-bot/reply'
+    );
+    const body = JSON.parse(replyCall![1].body);
+
+    expect(body.typing).toEqual({});
+  });
+
+  it('should post a stop op for ctx.typing.stop()', async () => {
+    const testBot = agent('test-bot', {
+      onMessage: async (_message, ctx) => {
+        await ctx.typing.stop();
+      },
+    });
+
+    const handler = new NovuRequestHandler({
+      frameworkName: 'test',
+      agents: [testBot],
+      client,
+      handler: () => ({
+        body: () => createMockBridgeRequest(),
+        headers: () => null,
+        method: () => 'POST',
+        url: () => new URL(`http://localhost?action=${PostActionEnum.AGENT_EVENT}&agentId=test-bot&event=onMessage`),
+        transformResponse: (res: any) => res,
+      }),
+    });
+
+    await handler.createHandler()();
+    await vi.waitFor(() => expect(fetchMock).toHaveBeenCalled());
+
+    const replyCall = fetchMock.mock.calls.find(
+      (call: any[]) => call[0] === 'https://api.novu.co/v1/agents/test-bot/reply'
+    );
+    const body = JSON.parse(replyCall![1].body);
+
+    expect(body.typing).toBe('stop');
+    expect(body.reply).toBeUndefined();
+  });
 });

packages/framework/src/resources/agent/agent.types.ts

@@ -302,6 +302,20 @@ export interface AgentContextBase {
    *   await ctx.reply('Done!');
    */
   addReaction(messageId: string, emojiName: Emoji): void;
+  /**
+   * Control the typing / "Thinking…" status for the current turn.
+   * Posts immediately (like `reply()`), updating the indicator Novu already shows on inbound.
+   *
+   * @example
+   *   await ctx.typing('Searching the docs…'); // set / replace the status text
+   *   await ctx.typing();                       // reset to the default "Thinking…"
+   *   await ctx.typing.stop();                  // clear it for this turn
+   *
+   * Behaviour is best-effort per platform: custom text shows on Slack-like platforms,
+   * a generic typing bubble on others, and is a no-op where there is no typing channel
+   * (e.g. email). A normal turn that ends with `ctx.reply()` clears the status automatically.
+   */
+  typing: TypingControl;
 }
 
 /** Context passed to the `onMessage` handler. */
@@ -434,6 +448,20 @@ export interface AddReactionPayload {
   emojiName: Emoji;
 }
 
+/**
+ * Per-turn typing/status control op sent on the reply contract.
+ * - `{ status?: string }` — set/replace the status; omit `status` for the default "Thinking…".
+ * - `'stop'` — clear the status for this turn.
+ */
+export type TypingOp = { status?: string } | 'stop';
+
+/**
+ * `ctx.typing` surface: a callable that sets/updates the status, plus `.stop()` to clear it.
+ */
+export type TypingControl = ((status?: string) => Promise<void>) & {
+  stop: () => Promise<void>;
+};
+
 export interface AgentReplyPayload {
   conversationId: string;
   integrationIdentifier: string;
@@ -442,6 +470,7 @@ export interface AgentReplyPayload {
   resolve?: { summary?: string };
   signals?: Signal[];
   addReactions?: AddReactionPayload[];
+  typing?: TypingOp;
 }
 
 /** Shape returned by /agents/:id/reply when a reply or edit was delivered. */