feat(langgraph): translate CustomStreamEvents into structured AgentEvent

toAgent(ref) now emits events$: Observable<AgentEvent>. Translates
state_update events into AgentStateUpdateEvent (with data: Record),
all others into the structured AgentCustomEvent escape hatch.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

Author: blove

libs/langgraph/src/lib/to-agent.spec.ts

@@ -2,7 +2,7 @@
 import { signal } from '@angular/core';
 import { TestBed } from '@angular/core/testing';
 import { HumanMessage, AIMessage } from '@langchain/core/messages';
-import type { Agent, AgentCustomEvent } from '@cacheplane/chat';
+import type { Agent, AgentEvent } from '@cacheplane/chat';
 import type { AgentRef, CustomStreamEvent } from './agent.types';
 import { ResourceStatus } from './agent.types';
 import { toAgent } from './to-agent';
@@ -115,7 +115,39 @@ describe('toAgent (LangGraph adapter)', () => {
     });
   });
 
-  it('exposes customEvents$ that emits newly-appended events with type aliased from name', () => {
+  it('translates a state_update CustomStreamEvent into AgentStateUpdateEvent', () => {
+    TestBed.runInInjectionContext(() => {
+      const customEvents = signal<any[]>([]);
+      const ref = stubAgentRef({ customEvents } as any);
+      const chat = toAgent(ref);
+
+      const received: any[] = [];
+      chat.events$.subscribe((e) => received.push(e));
+
+      customEvents.set([{ name: 'state_update', data: { count: 1 } }]);
+      TestBed.flushEffects();
+
+      expect(received).toEqual([{ type: 'state_update', data: { count: 1 } }]);
+    });
+  });
+
+  it('wraps non-state_update CustomStreamEvent as AgentCustomEvent', () => {
+    TestBed.runInInjectionContext(() => {
+      const customEvents = signal<any[]>([]);
+      const ref = stubAgentRef({ customEvents } as any);
+      const chat = toAgent(ref);
+
+      const received: any[] = [];
+      chat.events$.subscribe((e) => received.push(e));
+
+      customEvents.set([{ name: 'tick', data: 42 }]);
+      TestBed.flushEffects();
+
+      expect(received).toEqual([{ type: 'custom', name: 'tick', data: 42 }]);
+    });
+  });
+
+  it('exposes events$ that emits newly-appended events as structured AgentEvent', () => {
     const customSig = signal<CustomStreamEvent[]>([]);
     const ref = stubAgentRef({ customEvents: customSig });
 
@@ -124,8 +156,8 @@ describe('toAgent (LangGraph adapter)', () => {
       adapter = toAgent(ref);
     });
 
-    const received: AgentCustomEvent[] = [];
-    adapter.customEvents$!.subscribe((e) => received.push(e));
+    const received: AgentEvent[] = [];
+    adapter.events$.subscribe((e) => received.push(e));
 
     customSig.set([{ name: 'state_update', data: { counter: 1 } }]);
     TestBed.flushEffects();
@@ -142,7 +174,7 @@ describe('toAgent (LangGraph adapter)', () => {
 
     expect(received).toEqual([
       { type: 'state_update', data: { counter: 1 } },
-      { type: 'a2ui.surface', data: { surfaceId: 'main' } },
+      { type: 'custom', name: 'a2ui.surface', data: { surfaceId: 'main' } },
     ]);
   });
 });

libs/langgraph/src/lib/to-agent.ts

@@ -4,18 +4,9 @@ import { Subject, type Observable } from 'rxjs';
 import type { BaseMessage } from '@langchain/core/messages';
 import type { ToolCallWithResult, Interrupt } from '@langchain/langgraph-sdk';
 import type {
-  AgentWithHistory,
-  AgentCheckpoint,
-  AgentCustomEvent,
-  Message,
-  Role,
-  AgentStatus,
-  ToolCall,
-  ToolCallStatus,
-  AgentInterrupt,
-  Subagent,
-  AgentSubmitInput,
-  AgentSubmitOptions,
+  AgentWithHistory, AgentCheckpoint, AgentEvent,
+  Message, Role, ToolCall, ToolCallStatus, AgentStatus,
+  AgentInterrupt, Subagent, AgentSubmitInput, AgentSubmitOptions,
 } from '@cacheplane/chat';
 import type { AgentRef, CustomStreamEvent, SubagentStreamRef, ThreadState } from './agent.types';
 import { ResourceStatus } from './agent.types';
@@ -56,7 +47,7 @@ export function toAgent<T>(ref: AgentRef<T, any>): AgentWithHistory {
     return out;
   });
 
-  const customEvents$ = buildCustomEvents$(ref);
+  const events$ = buildEvents$(ref);
 
   const history = computed<AgentCheckpoint[]>(() =>
     ref.history().map(toCheckpoint),
@@ -71,7 +62,7 @@ export function toAgent<T>(ref: AgentRef<T, any>): AgentWithHistory {
     state,
     interrupt,
     subagents,
-    customEvents$,
+    events$,
     history,
     submit: (input: AgentSubmitInput, opts?: AgentSubmitOptions) =>
       ref.submit(buildSubmitPayload(input), opts ? { signal: opts.signal } as never : undefined),
@@ -80,15 +71,15 @@ export function toAgent<T>(ref: AgentRef<T, any>): AgentWithHistory {
 }
 
 /**
- * Build an Observable<AgentCustomEvent> that bridges LangGraph's
+ * Build an Observable<AgentEvent> that bridges LangGraph's
  * `Signal<CustomStreamEvent[]>` (append-only array) into a stream of newly
  * emitted events. Each effect firing compares against a cursor tracking the
  * previously-seen length and emits only the tail slice.
  */
-function buildCustomEvents$(
+function buildEvents$(
   ref: AgentRef<unknown, any>,
-): Observable<AgentCustomEvent> {
-  const subject = new Subject<AgentCustomEvent>();
+): Observable<AgentEvent> {
+  const subject = new Subject<AgentEvent>();
   let seen = 0;
   effect(() => {
     const all = ref.customEvents();
@@ -97,15 +88,18 @@ function buildCustomEvents$(
       seen = 0;
     }
     for (let i = seen; i < all.length; i++) {
-      subject.next(toCustomEvent(all[i]));
+      subject.next(toAgentEvent(all[i]));
     }
     seen = all.length;
   });
   return subject.asObservable();
 }
 
-function toCustomEvent(e: CustomStreamEvent): AgentCustomEvent {
-  return { type: e.name, data: e.data };
+function toAgentEvent(e: CustomStreamEvent): AgentEvent {
+  if (e.name === 'state_update' && isRecord(e.data)) {
+    return { type: 'state_update', data: e.data };
+  }
+  return { type: 'custom', name: e.name, data: e.data };
 }
 
 function mapStatus(s: ResourceStatus): AgentStatus {