wip

Author: mme

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

@@ -316,6 +316,7 @@ describe("RunAgentSubscriber", () => {
       const mutatingSubscriber: RunAgentSubscriber = {
         onStateSnapshotEvent: jest.fn().mockReturnValue({
           state: { modifiedBySubscriber: true },
+          stopPropagation: true, // Prevent the event from applying its snapshot
         }),
         onStateChanged: jest.fn(),
       };
@@ -618,8 +619,8 @@ describe("RunAgentSubscriber", () => {
       expect(specificSubscriber.onTextMessageStartEvent).toHaveBeenCalledWith(
         expect.objectContaining({
           event: textStartEvent,
-          messages: agent.messages,
-          state: agent.state,
+          messages: [{ content: "Hello", id: "msg-1", role: "user" }], // Pre-mutation state
+          state: { counter: 0 }, // Pre-mutation state
           agent,
         }),
       );
@@ -628,8 +629,8 @@ describe("RunAgentSubscriber", () => {
         expect.objectContaining({
           event: textContentEvent,
           textMessageBuffer: "Hello",
-          messages: agent.messages,
-          state: agent.state,
+          messages: agent.messages, // Can check final state for content events
+          state: agent.state, // Can check final state for content events
           agent,
         }),
       );

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

@@ -154,31 +154,38 @@ export async function runSubscribersWithMutation(
   let stopPropagation: boolean | undefined = undefined;
 
   for (const subscriber of subscribers) {
-    const mutation = await executor(
-      subscriber,
-      structuredClone_(messages),
-      structuredClone_(state),
-    );
-
-    if (mutation === undefined) {
-      // Nothing returned – keep going
+    try {
+      const mutation = await executor(
+        subscriber,
+        structuredClone_(messages),
+        structuredClone_(state),
+      );
+
+      if (mutation === undefined) {
+        // Nothing returned – keep going
+        continue;
+      }
+
+      // Merge messages/state so next subscriber sees latest view
+      if (mutation.messages !== undefined) {
+        messages = mutation.messages;
+      }
+
+      if (mutation.state !== undefined) {
+        state = mutation.state;
+      }
+
+      stopPropagation = mutation.stopPropagation;
+
+      if (stopPropagation === true) {
+        break;
+      }
+    } catch (error) {
+      // Log subscriber errors but continue processing
+      console.error("Subscriber error:", error);
+      // Continue to next subscriber unless we want to stop propagation
       continue;
     }
-
-    // Merge messages/state so next subscriber sees latest view
-    if (mutation.messages !== undefined) {
-      messages = mutation.messages;
-    }
-
-    if (mutation.state !== undefined) {
-      state = mutation.state;
-    }
-
-    stopPropagation = mutation.stopPropagation;
-
-    if (stopPropagation === true) {
-      break;
-    }
   }
 
   return {

typescript-sdk/packages/client/src/apply/default.ts

@@ -68,6 +68,19 @@ export const defaultApplyEvents = (
 
   return events$.pipe(
     concatMap(async (event) => {
+      const mutation = await runSubscribersWithMutation(
+        subscribers,
+        messages,
+        state,
+        (subscriber, messages, state) =>
+          subscriber.onEvent?.({ event, agent, input, messages, state }),
+      );
+      applyMutation(mutation);
+
+      if (mutation.stopPropagation === true) {
+        return emitUpdates();
+      }
+
       switch (event.type) {
         case EventType.TEXT_MESSAGE_START: {
           const mutation = await runSubscribersWithMutation(