addMessage

Author: mme

typescript-sdk/packages/client/src/agent/__tests__/subscriber.test.ts

@@ -1,10 +1,9 @@
 import { AbstractAgent } from "../agent";
-import { RunAgentSubscriber, AgentStateMutation, runSubscribersWithMutation } from "../subscriber";
+import { RunAgentSubscriber } from "../subscriber";
 import {
   BaseEvent,
   EventType,
   Message,
-  State,
   RunAgentInput,
   TextMessageStartEvent,
   TextMessageContentEvent,
@@ -17,14 +16,11 @@ import {
   ToolCallEndEvent,
   ToolCallResultEvent,
   CustomEvent,
-  ToolCall,
   StepStartedEvent,
   StepFinishedEvent,
 } from "@ag-ui/core";
 import { Observable, of, throwError, from } from "rxjs";
-import { mergeMap, defaultIfEmpty, startWith } from "rxjs/operators";
-import { v4 as uuidv4 } from "uuid";
-import { defaultApplyEvents } from "@/apply/default";
+import { mergeMap } from "rxjs/operators";
 
 // Mock uuid module
 jest.mock("uuid", () => ({

typescript-sdk/packages/client/src/agent/agent.ts

@@ -1,5 +1,5 @@
 import { defaultApplyEvents } from "@/apply/default";
-import { Message, State, RunAgentInput, BaseEvent } from "@ag-ui/core";
+import { Message, State, RunAgentInput, BaseEvent, ToolCall, AssistantMessage } from "@ag-ui/core";
 
 import { AgentConfig, RunAgentParameters } from "./types";
 import { v4 as uuidv4 } from "uuid";
@@ -283,6 +283,58 @@ export abstract class AbstractAgent {
     return cloned;
   }
 
+  public addMessage(message: Message) {
+    // Add message to the messages array
+    this.messages.push(message);
+
+    // Create minimal input for subscriber context
+    const input: RunAgentInput = {
+      threadId: this.threadId,
+      runId: uuidv4(),
+      tools: [],
+      context: [],
+      forwardedProps: {},
+      state: structuredClone_(this.state),
+      messages: structuredClone_(this.messages),
+    };
+
+    // Fire onMessagesChanged
+    this.subscribers.forEach((subscriber) => {
+      subscriber.onMessagesChanged?.({
+        messages: this.messages,
+        state: this.state,
+        agent: this,
+        input,
+      });
+    });
+
+    // Fire onNewMessage
+    this.subscribers.forEach((subscriber) => {
+      subscriber.onNewMessage?.({
+        message,
+        messages: this.messages,
+        state: this.state,
+        agent: this,
+        input,
+      });
+    });
+
+    // Fire onNewToolCall if the message is from assistant and contains tool calls
+    if (message.role === "assistant" && message.toolCalls) {
+      message.toolCalls.forEach((toolCall: ToolCall) => {
+        this.subscribers.forEach((subscriber) => {
+          subscriber.onNewToolCall?.({
+            toolCall,
+            messages: this.messages,
+            state: this.state,
+            agent: this,
+            input,
+          });
+        });
+      });
+    }
+  }
+
   public legacy_to_be_removed_runAgentBridged(
     config?: RunAgentParameters,
   ): Observable<LegacyRuntimeProtocolEvent> {

typescript-sdk/packages/client/src/agent/subscriber.ts

@@ -132,10 +132,22 @@ export interface RunAgentSubscriber {
   ): MaybePromise<AgentStateMutation | void>;
 
   // State changes
-  onMessagesChanged?(params: RunAgentSubscriberParams): MaybePromise<void>;
-  onStateChanged?(params: RunAgentSubscriberParams): MaybePromise<void>;
-  onNewMessage?(params: { message: Message } & RunAgentSubscriberParams): MaybePromise<void>;
-  onNewToolCall?(params: { toolCall: ToolCall } & RunAgentSubscriberParams): MaybePromise<void>;
+  onMessagesChanged?(
+    params: Omit<RunAgentSubscriberParams, "input"> & { input?: RunAgentInput },
+  ): MaybePromise<void>;
+  onStateChanged?(
+    params: Omit<RunAgentSubscriberParams, "input"> & { input?: RunAgentInput },
+  ): MaybePromise<void>;
+  onNewMessage?(
+    params: { message: Message } & Omit<RunAgentSubscriberParams, "input"> & {
+        input?: RunAgentInput;
+      },
+  ): MaybePromise<void>;
+  onNewToolCall?(
+    params: { toolCall: ToolCall } & Omit<RunAgentSubscriberParams, "input"> & {
+        input?: RunAgentInput;
+      },
+  ): MaybePromise<void>;
 }
 
 export async function runSubscribersWithMutation(
@@ -181,8 +193,15 @@ export async function runSubscribersWithMutation(
         break;
       }
     } catch (error) {
-      // Log subscriber errors but continue processing
-      console.error("Subscriber error:", error);
+      // Log subscriber errors but continue processing (silence during tests)
+      const isTestEnvironment =
+        process.env.NODE_ENV === "test" ||
+        process.env.JEST_WORKER_ID !== undefined ||
+        typeof jest !== "undefined";
+
+      if (!isTestEnvironment) {
+        console.error("Subscriber error:", error);
+      }
       // Continue to next subscriber unless we want to stop propagation
       continue;
     }