From 528150303608efe9fa08a69291ab236d4e752faf Mon Sep 17 00:00:00 2001 From: Brian Love Date: Mon, 7 Sep 2026 13:54:03 -0700 Subject: [PATCH 1/7] feat(langgraph): provide AGENT_LIFECYCLE and warn on ambiguous ref-less injects MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `AGENT_LIFECYCLE` was exported but never provided, so `inject(AGENT_LIFECYCLE)` threw NG0201 unless the app wired the token itself. Both forms of `provideAgent()` now provide it, resolving to the same object as `injectAgent().lifecycle`. The ref form also warns in development mode when several `provideAgent(ref, …)` calls share an injector level and the ambiguous ref-less token is resolved: the message names every competing ref and the one that won. Injecting by ref stays silent, and production builds never log. Co-Authored-By: Claude Fable 5.1 --- libs/langgraph/src/lib/agent.provider.spec.ts | 49 +++++++++++++++- libs/langgraph/src/lib/agent.provider.ts | 54 ++++++++++++++++-- .../langgraph/src/lib/lifecycle-token.spec.ts | 56 +++++++++++++++++++ 3 files changed, 154 insertions(+), 5 deletions(-) create mode 100644 libs/langgraph/src/lib/lifecycle-token.spec.ts diff --git a/libs/langgraph/src/lib/agent.provider.spec.ts b/libs/langgraph/src/lib/agent.provider.spec.ts index cac84b70e..3b88775d8 100644 --- a/libs/langgraph/src/lib/agent.provider.spec.ts +++ b/libs/langgraph/src/lib/agent.provider.spec.ts @@ -1,4 +1,4 @@ -import { describe, it, expect } from 'vitest'; +import { describe, it, expect, vi } from 'vitest'; import { InjectionToken, inject } from '@angular/core'; import { TestBed } from '@angular/core/testing'; import { createAgentRef } from '@threadplane/chat'; @@ -132,6 +132,53 @@ describe('provideAgent', () => { agentA.stop(); }); + it('warns in dev mode when several refs share one injector level, and keeps refs distinct', () => { + const REF_A = createAgentRef('warn-a'); + const REF_B = createAgentRef('warn-b'); + const warn = vi.spyOn(console, 'warn').mockImplementation(() => undefined); + try { + TestBed.configureTestingModule({ + providers: [ + provideAgent(REF_A, { apiUrl: '', assistantId: 'graph-a', transport: new MockAgentTransport() }), + provideAgent(REF_B, { apiUrl: '', assistantId: 'graph-b', transport: new MockAgentTransport() }), + ], + }); + + const agentA = TestBed.runInInjectionContext(() => injectAgent(REF_A)); + const agentB = TestBed.runInInjectionContext(() => injectAgent(REF_B)); + expect(agentA).not.toBe(agentB); + // Injecting by ref alone is unambiguous, so nothing is logged. + expect(warn).not.toHaveBeenCalled(); + + // The ref-less token is the ambiguous one: it warns and resolves the last ref. + const ambiguous = TestBed.runInInjectionContext(() => injectAgent()); + expect(ambiguous).toBe(agentB); + expect(warn).toHaveBeenCalledTimes(1); + const message = String(warn.mock.calls[0][0]); + expect(message).toContain('warn-a'); + expect(message).toContain('warn-b'); + expect(message).toContain('injectAgent()'); + } finally { + warn.mockRestore(); + } + }); + + it('does not warn when a single ref is provided', () => { + const REF = createAgentRef('lonely'); + const warn = vi.spyOn(console, 'warn').mockImplementation(() => undefined); + try { + TestBed.configureTestingModule({ + providers: [ + provideAgent(REF, { apiUrl: '', assistantId: 'graph-a', transport: new MockAgentTransport() }), + ], + }); + TestBed.runInInjectionContext(() => injectAgent()); + expect(warn).not.toHaveBeenCalled(); + } finally { + warn.mockRestore(); + } + }); + it('keeps single-ref behaviour identical: injectAgent() resolves the same instance', () => { const REF = createAgentRef('single'); const transport = new MockAgentTransport(); diff --git a/libs/langgraph/src/lib/agent.provider.ts b/libs/langgraph/src/lib/agent.provider.ts index adda63c06..786902fcb 100644 --- a/libs/langgraph/src/lib/agent.provider.ts +++ b/libs/langgraph/src/lib/agent.provider.ts @@ -1,4 +1,4 @@ -import { InjectionToken, inject, type Provider, type Signal } from '@angular/core'; +import { InjectionToken, inject, isDevMode, type Provider, type Signal } from '@angular/core'; import type { BaseMessage } from '@langchain/core/messages'; import type { BagTemplate } from '@langchain/langgraph-sdk'; import type { AgentRef, AgentRuntimeTelemetrySink } from '@threadplane/chat'; @@ -8,6 +8,7 @@ import type { LangGraphAgent, LangGraphClientOptions, } from './agent.types'; +import { AGENT_LIFECYCLE } from './lifecycle'; /** * Configuration for an agent instance. @@ -76,6 +77,19 @@ export const AGENT_CONFIG = new InjectionToken('AGENT_CONFIG'); */ export const AGENT = new InjectionToken('AGENT'); +/** + * @internal — multi-token collecting the debug name of every `provideAgent(ref, …)` + * call at one injector level, so the ref-less `AGENT` alias can warn in dev mode + * when more than one ref competes for it. Angular does not merge multi-providers + * across parent and child injectors, so this list is exactly "this level". + */ +const AGENT_REF_NAMES = new InjectionToken('AGENT_REF_NAMES'); + +/** @internal — human-readable name of an agent ref for warning copy. */ +function refDebugName(ref: AgentRef): string { + return ref.token.toString().replace(/^InjectionToken\s+/, ''); +} + /** * @internal — builds a LangGraphAgent from an already-resolved config. * Must be called from an injection context (the legacy `agent()` factory calls @@ -140,8 +154,14 @@ function isAgentRef(x: unknown): x is AgentRef { * provided side by side in a single `providers` array and `injectAgent(refA)` * / `injectAgent(refB)` return distinct agents. The ref-less `injectAgent()` * resolves a single shared token, which can only point at one of them: when - * more than one ref is provided at the same level the **last** call wins. - * Always inject by ref when an injector provides more than one agent. + * more than one ref is provided at the same level the **last** call wins, and + * resolving it in dev mode logs a `console.warn` naming every competing ref. + * The `AGENT_LIFECYCLE` token follows the same rule. Always inject by ref when + * an injector provides more than one agent. + * + * **Lifecycle token.** Every form also provides `AGENT_LIFECYCLE`, so + * `inject(AGENT_LIFECYCLE)` returns the same object as `injectAgent().lifecycle` + * without reaching for the agent itself. * * @example Two agents in one providers array * ```ts @@ -199,6 +219,7 @@ export function provideAgent>( // config from here, so the factory is invoked exactly once. { provide: AGENT_CONFIG, useFactory: resolveConfig }, { provide: AGENT, useFactory: agentFactory }, + { provide: AGENT_LIFECYCLE, useFactory: () => inject(AGENT).lifecycle }, ]; } @@ -212,10 +233,35 @@ export function provideAgent>( const refConfig = new InjectionToken>( `AGENT_CONFIG(${ref.token.toString()})`, ); + const thisRefName = refDebugName(ref as AgentRef); return [ { provide: refConfig, useFactory: resolveConfig }, { provide: ref.token, useFactory: () => createAgentFromConfig(inject(refConfig)) }, { provide: AGENT_CONFIG, useExisting: refConfig }, - { provide: AGENT, useExisting: ref.token }, + { provide: AGENT_REF_NAMES, useValue: thisRefName, multi: true }, + { + provide: AGENT, + useFactory: () => { + if (isDevMode()) { + const names = inject(AGENT_REF_NAMES, { optional: true }) ?? []; + const others = names.filter((n) => n !== thisRefName); + if (others.length > 0) { + console.warn( + `[@threadplane/langgraph] provideAgent(): ${names.length} agent refs ` + + `(${names.join(', ')}) are provided at the same injector level. ` + + `The ref-less injectAgent() resolves a single shared token, so it now ` + + `returns the last one provided (${thisRefName}). ` + + `Inject by ref — injectAgent(${others[0]}) / injectAgent(${thisRefName}) — ` + + `to reach each agent unambiguously.`, + ); + } + } + return inject(ref.token) as LangGraphAgent; + }, + }, + { + provide: AGENT_LIFECYCLE, + useFactory: () => (inject(ref.token) as LangGraphAgent).lifecycle, + }, ]; } diff --git a/libs/langgraph/src/lib/lifecycle-token.spec.ts b/libs/langgraph/src/lib/lifecycle-token.spec.ts new file mode 100644 index 000000000..58c0853ad --- /dev/null +++ b/libs/langgraph/src/lib/lifecycle-token.spec.ts @@ -0,0 +1,56 @@ +// @vitest-environment jsdom +import { describe, it, expect, beforeEach } from 'vitest'; +import { TestBed } from '@angular/core/testing'; +import { createAgentRef } from '@threadplane/chat'; +import { provideAgent } from './agent.provider'; +import { injectAgent } from './inject-agent'; +import { AGENT_LIFECYCLE } from './lifecycle'; +import { MockAgentTransport } from './transport/mock-stream.transport'; + +describe('AGENT_LIFECYCLE provider wiring', () => { + beforeEach(() => TestBed.resetTestingModule()); + + it('provideAgent(config) provides AGENT_LIFECYCLE as the agent lifecycle', () => { + TestBed.configureTestingModule({ + providers: [ + provideAgent({ + apiUrl: '', + assistantId: 'a', + transport: new MockAgentTransport(), + threadId: null, + }), + ], + }); + const agent = TestBed.runInInjectionContext(() => injectAgent()); + expect(TestBed.inject(AGENT_LIFECYCLE)).toBe(agent.lifecycle); + }); + + it('provideAgent(ref, config) provides AGENT_LIFECYCLE as that ref agent lifecycle', () => { + const REF = createAgentRef>('lifecycle-ref'); + TestBed.configureTestingModule({ + providers: [ + provideAgent(REF, { + apiUrl: '', + assistantId: 'a', + transport: new MockAgentTransport(), + threadId: null, + }), + ], + }); + const agent = TestBed.runInInjectionContext(() => injectAgent(REF)); + expect(TestBed.inject(AGENT_LIFECYCLE)).toBe(agent.lifecycle); + }); + + it('with several refs at one level AGENT_LIFECYCLE follows the last ref', () => { + const A = createAgentRef>('lifecycle-a'); + const B = createAgentRef>('lifecycle-b'); + TestBed.configureTestingModule({ + providers: [ + provideAgent(A, { apiUrl: '', assistantId: 'a', transport: new MockAgentTransport(), threadId: null }), + provideAgent(B, { apiUrl: '', assistantId: 'b', transport: new MockAgentTransport(), threadId: null }), + ], + }); + const b = TestBed.runInInjectionContext(() => injectAgent(B)); + expect(TestBed.inject(AGENT_LIFECYCLE)).toBe(b.lifecycle); + }); +}); From f8dbc04f6ede1ae82ab5f3f97711c061690ce807 Mon Sep 17 00:00:00 2001 From: Brian Love Date: Mon, 7 Sep 2026 13:54:09 -0700 Subject: [PATCH 2/7] fix(langgraph): carry the AgentErrorKind on streamErrorAt instead of the class name MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `streamErrorAt().classification` stored `error.name`. The bridge normalizes every failure through `toAgentError()` first, so the field was the literal 'AgentError' on every real stream error — useless as a discriminator. The field is renamed to `kind` and now carries the `AgentErrorKind` (`connection` | `auth` | `server` | `interrupted` | `aborted`), the same value `agent.error()?.kind` carries. A failure that slipped past normalization still falls back to a constructor name, so the type is `AgentErrorKind | string`. Co-Authored-By: Claude Fable 5.1 --- .../src/lib/cockpit-telemetry.service.spec.ts | 2 +- libs/langgraph/src/lib/agent.fn.ts | 30 ++++++++++++------- libs/langgraph/src/lib/lifecycle.spec.ts | 15 +++++++++- libs/langgraph/src/lib/lifecycle.ts | 12 ++++++-- .../src/lib/testing/mock-langgraph-agent.ts | 3 +- 5 files changed, 46 insertions(+), 16 deletions(-) diff --git a/libs/cockpit-telemetry/src/lib/cockpit-telemetry.service.spec.ts b/libs/cockpit-telemetry/src/lib/cockpit-telemetry.service.spec.ts index acbca4c7b..2fbc68c83 100644 --- a/libs/cockpit-telemetry/src/lib/cockpit-telemetry.service.spec.ts +++ b/libs/cockpit-telemetry/src/lib/cockpit-telemetry.service.spec.ts @@ -18,7 +18,7 @@ function makeAgentLifecycle(): AgentLifecycle & { const interruptResolvedAt = signal(null); return { streamStartedAt: streamStartedAt.asReadonly(), - streamErrorAt: signal<{ at: number; classification: string } | null>(null).asReadonly(), + streamErrorAt: signal<{ at: number; kind: string } | null>(null).asReadonly(), interruptReceivedAt: signal(null).asReadonly(), interruptResolvedAt: interruptResolvedAt.asReadonly(), threadCreatedAt: signal(null).asReadonly(), diff --git a/libs/langgraph/src/lib/agent.fn.ts b/libs/langgraph/src/lib/agent.fn.ts index a5f5a7f06..3a2805035 100644 --- a/libs/langgraph/src/lib/agent.fn.ts +++ b/libs/langgraph/src/lib/agent.fn.ts @@ -36,7 +36,7 @@ import type { BagTemplate, InferBag } from '@langchain/langgraph-sdk'; import type { AgentEvent, AgentCheckpoint, - AgentError, + AgentErrorKind, AgentInterrupt, AgentStatus, Message, @@ -49,7 +49,7 @@ import type { AgentSubmitOptions, MessageDelivery, } from '@threadplane/chat'; -import { staticDelivery } from '@threadplane/chat'; +import { AgentError, staticDelivery } from '@threadplane/chat'; import { AgentOptions, @@ -246,7 +246,7 @@ export function agent< // Eight signals tracking key transitions for telemetry/observability. // All reset together via resetLifecycle(); see switchThread() below. const lcStreamStartedAt = signal(null); - const lcStreamErrorAt = signal<{ at: number; classification: string } | null>(null); + const lcStreamErrorAt = signal<{ at: number; kind: AgentErrorKind | string } | null>(null); const lcInterruptReceivedAt = signal(null); const lcInterruptResolvedAt = signal(null); const lcThreadCreatedAt = signal(null); @@ -265,11 +265,13 @@ export function agent< toolCallCompletedAt: lcToolCallCompletedAt, }; - // Register with optional lifecycle registry. External instrumentation - // (e.g. cockpit-telemetry) provides AgentLifecycleRegistry to receive - // per-agent lifecycles created within this injection context. - const lifecycleRegistry = inject(AgentLifecycleRegistry, { optional: true }); - lifecycleRegistry?.register(lifecycle); + // Register with the root lifecycle registry. It is `providedIn: 'root'`, so + // every agent registers regardless of which injector built it, and external + // instrumentation (e.g. cockpit-telemetry) sees them all. Registration is + // scoped to this agent's injector lifetime. + const lifecycleRegistry = inject(AgentLifecycleRegistry); + lifecycleRegistry.register(lifecycle); + destroyRef.onDestroy(() => lifecycleRegistry.unregister(lifecycle)); function resetLifecycle(): void { lcStreamStartedAt.set(null); @@ -291,11 +293,17 @@ export function agent< messages$.pipe(takeUntil(destroy$)).subscribe(m => { if (lcStreamStartedAt() === null && m.length > 0) lcStreamStartedAt.set(Date.now()); }); - // Stream error: capture timestamp + classification (Error name or 'unknown'). + // Stream error: capture timestamp + failure class. The bridge normalizes every + // failure through toAgentError(), so `kind` is the actionable AgentErrorKind; + // anything that slipped past normalization falls back to a constructor name. error$.pipe(takeUntil(destroy$)).subscribe(e => { if (e == null) return; - const classification = e instanceof Error ? e.name : typeof e === 'string' ? 'string' : 'unknown'; - lcStreamErrorAt.set({ at: Date.now(), classification }); + const kind = e instanceof AgentError + ? e.kind + : e instanceof Error + ? e.name + : typeof e === 'string' ? 'string' : 'unknown'; + lcStreamErrorAt.set({ at: Date.now(), kind }); }); // First non-null interrupt within this thread. interrupt$.pipe(takeUntil(destroy$)).subscribe(ix => { diff --git a/libs/langgraph/src/lib/lifecycle.spec.ts b/libs/langgraph/src/lib/lifecycle.spec.ts index fe2347c72..007b16264 100644 --- a/libs/langgraph/src/lib/lifecycle.spec.ts +++ b/libs/langgraph/src/lib/lifecycle.spec.ts @@ -162,7 +162,20 @@ describe('AGENT_LIFECYCLE', () => { const err = ref.lifecycle.streamErrorAt(); expect(err).not.toBeNull(); expect(err!.at).toBeGreaterThan(0); - expect(typeof err!.classification).toBe('string'); + // The runtime normalizes every failure to an AgentError, so `kind` carries + // the actionable AgentErrorKind — never the useless constructor name. + expect(err!.kind).toBe('server'); + }); + + it('streamErrorAt kind mirrors the AgentError kind for an HTTP failure', async () => { + const transport = new MockAgentTransport(); + configureAgent({ apiUrl: '', assistantId: 'a', transport }); + const ref = getAgent(); + void ref.submit({ message: 'hi' }).catch(() => undefined); + transport.emitError(new Error('HTTP 401: unauthorized')); + await tick(); + expect(ref.lifecycle.streamErrorAt()!.kind).toBe('auth'); + expect(ref.error()?.kind).toBe('auth'); }); it('all signals reset to null on switchThread(null)', async () => { diff --git a/libs/langgraph/src/lib/lifecycle.ts b/libs/langgraph/src/lib/lifecycle.ts index 6995048bc..7bff838ca 100644 --- a/libs/langgraph/src/lib/lifecycle.ts +++ b/libs/langgraph/src/lib/lifecycle.ts @@ -1,10 +1,18 @@ import { InjectionToken, Signal } from '@angular/core'; +import type { AgentErrorKind } from '@threadplane/chat'; export interface AgentLifecycle { /** Epoch ms of the first stream chunk arrival. Resets on switchThread(). */ readonly streamStartedAt: Signal; - /** Epoch ms + classification of the most recent stream error. Resets on switchThread(). */ - readonly streamErrorAt: Signal<{ at: number; classification: string } | null>; + /** + * Epoch ms + failure class of the most recent stream error. Resets on switchThread(). + * + * `kind` is the {@link AgentErrorKind} of the normalized `AgentError` + * (`connection` | `auth` | `server` | `interrupted` | `aborted`) — the same + * value `agent.error()?.kind` carries. For a failure the runtime could not + * normalize it falls back to the error's constructor name. + */ + readonly streamErrorAt: Signal<{ at: number; kind: AgentErrorKind | string } | null>; /** Epoch ms of the first interrupt$ non-null in this stream. Resets on switchThread(). */ readonly interruptReceivedAt: Signal; /** Epoch ms of the most recent submit({ resume }) call. Resets on switchThread(). */ diff --git a/libs/langgraph/src/lib/testing/mock-langgraph-agent.ts b/libs/langgraph/src/lib/testing/mock-langgraph-agent.ts index 998d72c96..d9a643602 100644 --- a/libs/langgraph/src/lib/testing/mock-langgraph-agent.ts +++ b/libs/langgraph/src/lib/testing/mock-langgraph-agent.ts @@ -16,6 +16,7 @@ import type { MockAgent, MockAgentOptions, AgentError, + AgentErrorKind, AgentInterrupt, AgentCheckpoint, AgentStatus, @@ -172,7 +173,7 @@ export function mockLangGraphAgent( getToolCalls: (_msg: CoreAIMessage): ToolCallWithResult[] => [], lifecycle: { streamStartedAt: signal(null), - streamErrorAt: signal<{ at: number; classification: string } | null>(null), + streamErrorAt: signal<{ at: number; kind: AgentErrorKind | string } | null>(null), interruptReceivedAt: signal(null), interruptResolvedAt: signal(null), threadCreatedAt: signal(null), From 6a0a38a9e94bc42ed162b2fdb7edd2ac0e161313 Mon Sep 17 00:00:00 2001 From: Brian Love Date: Mon, 7 Sep 2026 13:54:17 -0700 Subject: [PATCH 3/7] fix(langgraph): make AgentLifecycleRegistry root-provided so registration is not construction-ordered MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The agent injected the registry optionally at construction, so a registry provided below the agent's injector — or provided after the agent was built — collected nothing. The registry is now `providedIn: 'root'`: every agent registers into the same instance regardless of which injector built it, and an agent created in a route or component injector is visible from the root. Registration is also scoped to the agent's lifetime — an agent unregisters when its injector is destroyed — so `lifecycles()` no longer accumulates dead agents. `provideCockpitTelemetry()` stops re-providing the class, which would have shadowed the instance agents register into. Co-Authored-By: Claude Fable 5.1 --- .../src/lib/provide-cockpit-telemetry.ts | 4 +- .../src/lib/agent-lifecycle-registry.spec.ts | 62 +++++++++++++------ .../src/lib/agent-lifecycle-registry.ts | 24 ++++--- 3 files changed, 61 insertions(+), 29 deletions(-) diff --git a/libs/cockpit-telemetry/src/lib/provide-cockpit-telemetry.ts b/libs/cockpit-telemetry/src/lib/provide-cockpit-telemetry.ts index 5498fffcf..e04e823d4 100644 --- a/libs/cockpit-telemetry/src/lib/provide-cockpit-telemetry.ts +++ b/libs/cockpit-telemetry/src/lib/provide-cockpit-telemetry.ts @@ -7,7 +7,6 @@ import { import { COCKPIT_TELEMETRY_CONFIG, type CockpitTelemetryConfig } from './tokens'; import { CockpitTelemetryService } from './cockpit-telemetry.service'; import { ActivationAggregator } from './activation-aggregator'; -import { AgentLifecycleRegistry } from '@threadplane/langgraph'; export function provideCockpitTelemetry( config: CockpitTelemetryConfig, @@ -15,7 +14,8 @@ export function provideCockpitTelemetry( return makeEnvironmentProviders([ { provide: COCKPIT_TELEMETRY_CONFIG, useValue: config }, ActivationAggregator, - AgentLifecycleRegistry, + // AgentLifecycleRegistry is `providedIn: 'root'` in @threadplane/langgraph: + // re-providing it here would shadow the instance every agent registers into. CockpitTelemetryService, { provide: ENVIRONMENT_INITIALIZER, diff --git a/libs/langgraph/src/lib/agent-lifecycle-registry.spec.ts b/libs/langgraph/src/lib/agent-lifecycle-registry.spec.ts index 11a353789..3a26b4b76 100644 --- a/libs/langgraph/src/lib/agent-lifecycle-registry.spec.ts +++ b/libs/langgraph/src/lib/agent-lifecycle-registry.spec.ts @@ -12,26 +12,50 @@ describe('AgentLifecycleRegistry integration with injectAgent()', () => { TestBed.resetTestingModule(); }); - it('does not error or register when no registry is provided', () => { - TestBed.configureTestingModule({ - providers: [ - provideAgent({ - assistantId: 'a', - apiUrl: 'http://localhost', - transport: new MockAgentTransport(), - threadId: null, - }), - ], - }); - expect(() => - TestBed.runInInjectionContext(() => injectAgent()), - ).not.toThrow(); + it('collects an agent built in a child injector from the root registry', () => { + // The registry is `providedIn: 'root'`, so nothing has to be provided and + // an agent constructed *below* the root still registers upward. + TestBed.configureTestingModule({ providers: [] }); + const registry = TestBed.inject(AgentLifecycleRegistry); + expect(registry.lifecycles()).toEqual([]); + + const child = createEnvironmentInjector( + provideAgent({ + assistantId: 'a', + apiUrl: 'http://localhost', + transport: new MockAgentTransport(), + threadId: null, + }), + TestBed.inject(EnvironmentInjector), + ); + const a = runInInjectionContext(child, () => injectAgent()); + + expect(registry.lifecycles()).toEqual([a.lifecycle]); + }); + + it('unregisters a lifecycle when its injector is destroyed', () => { + TestBed.configureTestingModule({ providers: [] }); + const registry = TestBed.inject(AgentLifecycleRegistry); + + const child = createEnvironmentInjector( + provideAgent({ + assistantId: 'a', + apiUrl: 'http://localhost', + transport: new MockAgentTransport(), + threadId: null, + }), + TestBed.inject(EnvironmentInjector), + ); + const a = runInInjectionContext(child, () => injectAgent()); + expect(registry.lifecycles()).toEqual([a.lifecycle]); + + child.destroy(); + expect(registry.lifecycles()).toEqual([]); }); - it('registers the agent lifecycle when AgentLifecycleRegistry is provided', () => { + it('registers the agent lifecycle without the app providing the registry', () => { TestBed.configureTestingModule({ providers: [ - AgentLifecycleRegistry, provideAgent({ assistantId: 'a', apiUrl: 'http://localhost', @@ -52,10 +76,8 @@ describe('AgentLifecycleRegistry integration with injectAgent()', () => { it('accumulates multiple agent lifecycles in registration order', () => { // Use child environment injectors so two singleton AGENTs can coexist - // while sharing the root-provided AgentLifecycleRegistry. - TestBed.configureTestingModule({ - providers: [AgentLifecycleRegistry], - }); + // while sharing the root AgentLifecycleRegistry. + TestBed.configureTestingModule({ providers: [] }); const registry = TestBed.inject(AgentLifecycleRegistry); const parent = TestBed.inject(EnvironmentInjector); diff --git a/libs/langgraph/src/lib/agent-lifecycle-registry.ts b/libs/langgraph/src/lib/agent-lifecycle-registry.ts index 479b76708..14ea93f22 100644 --- a/libs/langgraph/src/lib/agent-lifecycle-registry.ts +++ b/libs/langgraph/src/lib/agent-lifecycle-registry.ts @@ -2,21 +2,31 @@ import { Injectable, signal, type Signal } from '@angular/core'; import type { AgentLifecycle } from './lifecycle'; /** - * Optional registry that collects per-instance agent lifecycles within - * an Angular injection context. External instrumentation packages - * (e.g. cockpit-telemetry) provide this token and read from it. + * Application-wide registry of every live agent's {@link AgentLifecycle}. * - * `@threadplane/langgraph` does NOT provide this itself — the configured agent - * instance writes to the registry only when an external consumer has provided it. + * It is `providedIn: 'root'`, so it always exists and every agent registers + * into the same instance regardless of which injector built it — an agent + * created in a route or component injector is still visible from the root. + * External instrumentation packages read `lifecycles()` to observe every agent + * in the application without owning the provider graph. + * + * Registration is scoped to the agent's lifetime: an agent unregisters when the + * injector that created it is destroyed, so `lifecycles()` never accumulates + * lifecycles for agents that are gone. */ -@Injectable() +@Injectable({ providedIn: 'root' }) export class AgentLifecycleRegistry { private readonly _lifecycles = signal([]); - /** Reactive list of registered lifecycles. */ + /** Reactive list of the lifecycles of every currently live agent. */ readonly lifecycles: Signal = this._lifecycles.asReadonly(); register(lifecycle: AgentLifecycle): void { this._lifecycles.update((curr) => [...curr, lifecycle]); } + + /** Drop a lifecycle when its agent's injector is destroyed. */ + unregister(lifecycle: AgentLifecycle): void { + this._lifecycles.update((curr) => curr.filter((l) => l !== lifecycle)); + } } From deedc18a84f7606ff50ebd32a597c837cbfd13a1 Mon Sep 17 00:00:00 2001 From: Brian Love Date: Mon, 7 Sep 2026 13:54:17 -0700 Subject: [PATCH 4/7] fix(langgraph): declare interrupt as required on LangGraphAgent `interrupt` is optional on the runtime-neutral `Agent` contract because a runtime without human-in-the-loop support omits it, which meant `injectAgent().interrupt()` did not compile under `strictNullChecks` even though the LangGraph adapter always provides it. `LangGraphAgent` now narrows it to a required `Signal`; the chat contract is untouched. Co-Authored-By: Claude Fable 5.1 --- libs/langgraph/src/lib/agent.types.ts | 11 +++++++++++ libs/langgraph/src/lib/inject-agent.type-spec.ts | 9 +++++++++ 2 files changed, 20 insertions(+) diff --git a/libs/langgraph/src/lib/agent.types.ts b/libs/langgraph/src/lib/agent.types.ts index bada53524..b1477ea54 100644 --- a/libs/langgraph/src/lib/agent.types.ts +++ b/libs/langgraph/src/lib/agent.types.ts @@ -20,6 +20,7 @@ import type { } from '@langchain/langgraph-sdk/ui'; import type { BaseMessage, AIMessage as CoreAIMessage } from '@langchain/core/messages'; import type { + AgentInterrupt, AgentRuntimeTelemetrySink, AgentSubmitInput, AgentSubmitOptions, @@ -348,6 +349,16 @@ export interface LangGraphAgent { // ── Raw LangGraph signals ──────────────────────────────────────────────── + /** + * Current human-in-the-loop pause, or `undefined` when the run is not paused. + * + * Narrowed from the neutral `Agent` contract, where `interrupt` is optional + * because a runtime without human-in-the-loop support omits it. The LangGraph + * adapter always provides it, so `injectAgent().interrupt()` type-checks + * directly under `strictNullChecks` — no `?.()` needed. + */ + interrupt: Signal; + /** Raw LangChain BaseMessage list. Use `messages` for chat rendering. */ langGraphMessages: Signal; diff --git a/libs/langgraph/src/lib/inject-agent.type-spec.ts b/libs/langgraph/src/lib/inject-agent.type-spec.ts index ad83a31e7..0ff3fe569 100644 --- a/libs/langgraph/src/lib/inject-agent.type-spec.ts +++ b/libs/langgraph/src/lib/inject-agent.type-spec.ts @@ -1,4 +1,6 @@ +import type { Signal } from '@angular/core'; import { createAgentRef } from '@threadplane/chat'; +import type { AgentInterrupt } from '@threadplane/chat'; import type { Equal, Expect } from '../testing/type-assert'; import { injectAgent } from './inject-agent'; @@ -15,3 +17,10 @@ type _agentValue = Expect, TripState>>; // no-arg form stays valid (default state). const plain = ctx(() => injectAgent()); type _plainState = Expect, Record>>; + +// `interrupt` is REQUIRED on LangGraphAgent (the adapter always provides it), +// so a plain call compiles under strictNullChecks — no `?.()` needed. +const pending = plain.interrupt(); +type _interruptRequired = Expect>; +type _interruptSignal = Expect>>; +export type { _interruptRequired, _interruptSignal }; From 2950dde5194092236cec4988f05a871841916514 Mon Sep 17 00:00:00 2001 From: Brian Love Date: Mon, 7 Sep 2026 13:54:28 -0700 Subject: [PATCH 5/7] feat(langgraph): make MockAgentTransport emit/emitError/close awaitable, add flush() `stream()` is an async generator, so `emit()` only woke the suspended loop and nothing had reached the signals when it returned. Every spec paid for that with a hand-rolled `await new Promise(r => setTimeout(r, 0))`. `emit()`, `emitError()` and `close()` now return a promise that settles once the generator has drained everything queued at the time of the call (or the run has ended), plus one macrotask so signal writes have landed. `flush()` waits the same way without emitting. An emit after the run finished resolves instead of hanging. The langgraph specs that hand-rolled the macrotask flush after an emit are converted to `await transport.emit(...)`; removing the await makes them fail, so the await is load-bearing. Throttle waits (16 ms and up) are left alone. Co-Authored-By: Claude Fable 5.1 --- libs/langgraph/src/lib/agent.fn.spec.ts | 9 +- .../internals/stream-manager.bridge.spec.ts | 39 +++------ .../transport/mock-stream.transport.spec.ts | 87 ++++++++++++++++++- .../lib/transport/mock-stream.transport.ts | 86 +++++++++++++++--- 4 files changed, 177 insertions(+), 44 deletions(-) diff --git a/libs/langgraph/src/lib/agent.fn.spec.ts b/libs/langgraph/src/lib/agent.fn.spec.ts index 98e7fd02c..aceab5dbc 100644 --- a/libs/langgraph/src/lib/agent.fn.spec.ts +++ b/libs/langgraph/src/lib/agent.fn.spec.ts @@ -133,12 +133,11 @@ describe('agent', () => { ); const submitted = ref.submit({ message: 'hello' }); - transport.emit([{ + await transport.emit([{ type: 'messages', messages: [{ id: 'ai-live', type: 'ai', content: 'answer' }], messageMetadata: { langgraph_node: 'model' }, }]); - await new Promise(resolve => setTimeout(resolve, 0)); const streaming = ref.messages().find(message => message.id === 'ai-live')?.delivery; expect(streaming).toEqual({ generation: expect.any(String), phase: 'streaming' }); @@ -206,7 +205,7 @@ describe('agent', () => { ], }], }]); - transport.emit([{ + await transport.emit([{ type: 'messages|tools:call-success', namespace: ['tools:call-success'], messages: [{ id: 'sub-success', type: 'ai', content: 'result' }], messageMetadata: { checkpoint_ns: 'tools:call-success|model' }, @@ -215,7 +214,6 @@ describe('agent', () => { messages: [{ id: 'sub-error', type: 'ai', content: 'partial' }], messageMetadata: { checkpoint_ns: 'tools:call-error|model' }, }]); - await new Promise(resolve => setTimeout(resolve, 0)); const successStreaming = ref.subagents().get('call-success')?.messages()[0].delivery; const errorStreaming = ref.subagents().get('call-error')?.messages()[0].delivery; @@ -223,14 +221,13 @@ describe('agent', () => { expect(errorStreaming).toMatchObject({ phase: 'streaming' }); expect(successStreaming?.generation).not.toBe(errorStreaming?.generation); - transport.emit([{ + await transport.emit([{ type: 'messages', messages: [ { id: 'tool-success', type: 'tool', tool_call_id: 'call-success', content: 'done', status: 'success' }, { id: 'tool-error', type: 'tool', tool_call_id: 'call-error', content: 'failed', status: 'error' }, ], }]); - await new Promise(resolve => setTimeout(resolve, 0)); expect(ref.subagents().get('call-success')?.messages()[0].delivery).toEqual({ generation: successStreaming?.generation, diff --git a/libs/langgraph/src/lib/internals/stream-manager.bridge.spec.ts b/libs/langgraph/src/lib/internals/stream-manager.bridge.spec.ts index 6cb1bdc57..b086c5bd9 100644 --- a/libs/langgraph/src/lib/internals/stream-manager.bridge.spec.ts +++ b/libs/langgraph/src/lib/internals/stream-manager.bridge.spec.ts @@ -374,12 +374,11 @@ describe('createStreamManagerBridge', () => { }); const submitted = bridge.submit({}); - transport.emit([{ + await transport.emit([{ type: 'messages', messages: [{ id: 'ai-1', type: 'ai', content: 'hel' }], messageMetadata: { langgraph_node: 'model' }, }]); - await new Promise(resolve => setTimeout(resolve, 0)); const streaming = bridge.getMessageDelivery('ai-1'); expect(streaming).toEqual({ @@ -411,20 +410,18 @@ describe('createStreamManagerBridge', () => { }); void bridge.submit({}); - transport.emit([{ + await transport.emit([{ type: 'messages', messages: [{ id: 'revision-ai', type: 'ai', content: 'a' }], messageMetadata: { langgraph_node: 'model' }, }]); - await new Promise(resolve => setTimeout(resolve, 0)); const afterFirstChunk = bridge.deliveryRevision(); - transport.emit([{ + await transport.emit([{ type: 'messages', messages: [{ id: 'revision-ai', type: 'ai', content: 'b' }], messageMetadata: { langgraph_node: 'model' }, }]); - await new Promise(resolve => setTimeout(resolve, 0)); expect(bridge.deliveryRevision()).toBe(afterFirstChunk); await bridge.stop(); @@ -444,12 +441,11 @@ describe('createStreamManagerBridge', () => { await new Promise(resolve => setTimeout(resolve, 0)); const submitted = bridge.submit({}); - transport.emit([{ + await transport.emit([{ type: 'messages', messages: [{ id: 'streamed-id', type: 'ai', content: 'final answer' }], messageMetadata: { langgraph_node: 'model' }, }]); - await new Promise(resolve => setTimeout(resolve, 0)); const streaming = bridge.getMessageDelivery('streamed-id'); transport.history = [{ @@ -998,12 +994,11 @@ describe('createStreamManagerBridge', () => { }); const submitted = bridge.submit({}); - transport.emit([{ + await transport.emit([{ type: 'messages', messages: [{ id: 'ai-aborted', type: 'ai', content: 'partial' }], messageMetadata: { langgraph_node: 'model' }, }]); - await new Promise(resolve => setTimeout(resolve, 0)); await bridge.stop(); transport.close(); @@ -1397,21 +1392,19 @@ describe('createStreamManagerBridge', () => { }); void bridge.submit({}); - transport.emit([{ + await transport.emit([{ type: 'messages', messages: [{ id: 'ai-tool-call', type: 'ai', content: 'search', tool_calls: [{ id: 'call-1', name: 'search', args: {} }] }], messageMetadata: { langgraph_node: 'model' }, }]); - await new Promise(resolve => setTimeout(resolve, 0)); expect(bridge.getMessageDelivery('ai-tool-call').phase).toBe('streaming'); transport.emit([{ type: 'values', values: { toolStepComplete: true } }]); - transport.emit([{ + await transport.emit([{ type: 'messages', messages: [{ id: 'ai-final', type: 'ai', content: 'search complete' }], messageMetadata: { langgraph_node: 'model' }, }]); - await new Promise(resolve => setTimeout(resolve, 0)); expect(bridge.getMessageDelivery('ai-tool-call')).toMatchObject({ phase: 'complete', @@ -1434,7 +1427,7 @@ describe('createStreamManagerBridge', () => { }); void bridge.submit({}); - transport.emit([{ + await transport.emit([{ type: 'messages', messages: [{ id: 'tool-chunk-a', type: 'ai', content: 'hel', @@ -1449,7 +1442,6 @@ describe('createStreamManagerBridge', () => { }], messageMetadata: { langgraph_node: 'model' }, }]); - await new Promise(resolve => setTimeout(resolve, 0)); expect(subjects.messages$.value).toEqual([ expect.objectContaining({ @@ -1484,7 +1476,7 @@ describe('createStreamManagerBridge', () => { }); const submitted = bridge.submit({}); - transport.emit([{ + await transport.emit([{ type: 'messages', messages: [{ id: 'chunk-event-1', type: 'ai', content: 'hel' }], messageMetadata: { langgraph_node: 'model' }, @@ -1493,7 +1485,6 @@ describe('createStreamManagerBridge', () => { messages: [{ id: 'chunk-event-2', type: 'ai', content: 'lo' }], messageMetadata: { langgraph_node: 'model' }, }]); - await new Promise(resolve => setTimeout(resolve, 0)); expect(subjects.messages$.value).toEqual([ expect.objectContaining({ id: 'chunk-event-1', content: 'hello' }), @@ -1535,7 +1526,7 @@ describe('createStreamManagerBridge', () => { }); const submitted = bridge.submit({}); - transport.emit([{ + await transport.emit([{ type: 'messages', messages: [{ id: 'ai-earlier', type: 'ai', content: '', @@ -1553,7 +1544,6 @@ describe('createStreamManagerBridge', () => { messages: [{ id: 'ai-active', type: 'ai', content: 'partial' }], messageMetadata: { langgraph_node: 'model' }, }]); - await new Promise(resolve => setTimeout(resolve, 0)); if (outcome === 'error') { transport.emit([{ type: 'error', error: new Error('failed') }]); @@ -2508,7 +2498,7 @@ describe('createStreamManagerBridge', () => { }); void bridge.submit({}); - transport.emit([{ + await transport.emit([{ type, messages: [ { id: 'historical-ai', type: 'ai', content: 'old answer' }, @@ -2516,7 +2506,6 @@ describe('createStreamManagerBridge', () => { { id: 'active-ai', type: 'ai', content: 'new answer' }, ], }]); - await new Promise(resolve => setTimeout(resolve, 0)); expect(bridge.getMessageDelivery('historical-ai')).toEqual({ generation: 'historical-ai', @@ -2551,7 +2540,7 @@ describe('createStreamManagerBridge', () => { const historicalDelivery = bridge.getMessageDelivery('historical-enriched-ai'); void bridge.submit({}); - transport.emit([{ + await transport.emit([{ type, messages: [ { @@ -2565,7 +2554,6 @@ describe('createStreamManagerBridge', () => { { id: 'active-enriched-ai', type: 'ai', content: 'new answer' }, ], }]); - await new Promise(resolve => setTimeout(resolve, 0)); expect(subjects.messages$.value.find(message => (message as unknown as { id?: string }).id === 'historical-enriched-ai' @@ -3525,13 +3513,12 @@ describe('createStreamManagerBridge', () => { }); const done = bridge.submit({}); - transport.emit([{ + await transport.emit([{ type: 'messages|research:abc123' as StreamEvent['type'], namespace: ['research:abc123'], messages: [{ id: 'child-ai', type: 'ai', content: 'brief' }], messageMetadata: { checkpoint_ns: 'research:abc123' }, } satisfies StreamEvent]); - await new Promise(r => setTimeout(r, 0)); expect(subjects.subagents$.value.get('research:abc123')?.status()).toBe('running'); transport.emit([{ type: 'values', data: { done: true } } as StreamEvent]); diff --git a/libs/langgraph/src/lib/transport/mock-stream.transport.spec.ts b/libs/langgraph/src/lib/transport/mock-stream.transport.spec.ts index 1348b1bd4..9729a289e 100644 --- a/libs/langgraph/src/lib/transport/mock-stream.transport.spec.ts +++ b/libs/langgraph/src/lib/transport/mock-stream.transport.spec.ts @@ -1,4 +1,4 @@ -import { describe, it, expect, beforeEach } from 'vitest'; +import { describe, it, expect } from 'vitest'; import { MockAgentTransport } from './mock-stream.transport'; describe('MockAgentTransport', () => { @@ -39,6 +39,91 @@ describe('MockAgentTransport', () => { expect(events).toHaveLength(1); }); + describe('awaitable emit()', () => { + it('emit() resolves only after the consumer has pulled the batch', async () => { + const t = new MockAgentTransport(); + const events: unknown[] = []; + const ac = new AbortController(); + const collecting = (async () => { + for await (const e of t.stream('agent', null, {}, ac.signal)) { events.push(e); } + })(); + + await t.emit([ + { type: 'values', values: { foo: 1 } }, + { type: 'values', values: { foo: 2 } }, + ]); + // No bare setTimeout: awaiting emit() is enough for the batch to land. + expect(events).toHaveLength(2); + + await t.close(); + await collecting; + }); + + it('emit() before the stream starts resolves once the stream drains it', async () => { + const t = new MockAgentTransport(); + const events: unknown[] = []; + const ac = new AbortController(); + const emitted = t.emit([{ type: 'values', values: { foo: 1 } }]); + const collecting = (async () => { + for await (const e of t.stream('agent', null, {}, ac.signal)) { events.push(e); } + })(); + await emitted; + expect(events).toHaveLength(1); + await t.close(); + await collecting; + }); + + it('flush() resolves without emitting anything', async () => { + const t = new MockAgentTransport(); + const ac = new AbortController(); + const collecting = (async () => { + for await (const _ of t.stream('agent', null, {}, ac.signal)) { /* noop */ } + })(); + await t.emit([{ type: 'values', values: { foo: 1 } }]); + await t.flush(); + expect(t.isStreaming()).toBe(true); + await t.close(); + await collecting; + }); + + it('close() resolves once the run has finished', async () => { + const t = new MockAgentTransport(); + const ac = new AbortController(); + const collecting = (async () => { + for await (const _ of t.stream('agent', null, {}, ac.signal)) { /* noop */ } + })(); + await t.close(); + expect(t.isStreaming()).toBe(false); + await collecting; + }); + + it('emitError() resolves once the stream has thrown', async () => { + const t = new MockAgentTransport(); + const ac = new AbortController(); + let thrown: unknown; + const collecting = (async () => { + try { + for await (const _ of t.stream('agent', null, {}, ac.signal)) { /* noop */ } + } catch (e) { thrown = e; } + })(); + await t.emitError(new Error('transport error')); + expect(thrown).toBeInstanceOf(Error); + await collecting; + }); + + it('emit() after the run has finished resolves instead of hanging', async () => { + const t = new MockAgentTransport(); + const ac = new AbortController(); + const collecting = (async () => { + for await (const _ of t.stream('agent', null, {}, ac.signal)) { /* noop */ } + })(); + await t.close(); + await collecting; + await expect(t.emit([{ type: 'values', values: { foo: 1 } }])).resolves.toBeUndefined(); + await expect(t.flush()).resolves.toBeUndefined(); + }); + }); + it('emitError() causes stream to throw', async () => { const t = new MockAgentTransport(); const ac = new AbortController(); diff --git a/libs/langgraph/src/lib/transport/mock-stream.transport.ts b/libs/langgraph/src/lib/transport/mock-stream.transport.ts index 4fc6438e4..0eabd5661 100644 --- a/libs/langgraph/src/lib/transport/mock-stream.transport.ts +++ b/libs/langgraph/src/lib/transport/mock-stream.transport.ts @@ -7,12 +7,19 @@ import type { ThreadState } from '@langchain/langgraph-sdk'; * Script event batches upfront, then emit them manually or step through them * in your test specs. Supports error injection and close control. * + * `emit()`, `emitError()`, `close()` and `flush()` are awaitable: the returned + * promise settles once the adapter has consumed everything queued so far (and + * one macrotask later, so throttled signal writes have landed), which removes + * the hand-rolled `await new Promise(resolve => setTimeout(resolve, 0))` flush + * from specs. + * * @example * ```typescript * const transport = new MockAgentTransport([ * [{ type: 'values', messages: [aiMsg('Hello')] }], * [{ type: 'values', messages: [aiMsg('Done')] }], * ]); + * await transport.emit(transport.nextBatch()); * ``` */ export class MockAgentTransport implements AgentTransport { @@ -28,8 +35,15 @@ export class MockAgentTransport implements AgentTransport { private eventQueue: StreamEvent[] = []; // Each resolver simply wakes the stream loop to re-check state. private resolvers: Array<() => void> = []; + // Awaiters registered by emit()/emitError()/close()/flush(), settled once the + // stream loop has drained everything queued at the time they were registered. + private consumers: Array<() => void> = []; private closed = false; private pendingError: Error | null = null; + /** True while the stream loop is suspended with an empty queue. */ + private idle = false; + /** True once a stream run has ended (or before any run has started). */ + private finished = true; /** @param script - Array of event batches. Each batch is emitted as a group. */ constructor(script: StreamEvent[][] = []) { @@ -42,22 +56,43 @@ export class MockAgentTransport implements AgentTransport { return this.script[this.scriptIndex++]; } - /** Manually emit events into the stream. */ - emit(events: StreamEvent[]): void { + /** + * Manually emit events into the stream. + * + * Await the returned promise: it resolves once the adapter has pulled this + * batch out of the stream (or the run has ended), so signals are settled and + * assertions read live state rather than the value from before the emit. + */ + emit(events: StreamEvent[]): Promise { this.eventQueue.push(...events); - this.flush(); + this.wake(); + return this.consumed(); } - /** Inject an error into the stream. */ - emitError(err: Error): void { + /** Inject an error into the stream. Resolves once the stream has thrown. */ + emitError(err: Error): Promise { this.pendingError = err; - this.flush(); + this.wake(); + return this.consumed(); } - /** Close the stream. Remaining queued events are drained before completion. */ - close(): void { + /** + * Close the stream. Remaining queued events are drained before completion. + * Resolves once the run has finished. + */ + close(): Promise { this.closed = true; - this.flush(); + this.wake(); + return this.consumed(); + } + + /** + * Resolve once everything emitted so far has been consumed, without emitting + * anything new. Useful after driving the agent by some other route (a + * `submit()`, a `switchThread()`) that has to reach the transport first. + */ + flush(): Promise { + return this.consumed(); } /** Returns true if a stream is currently active. */ @@ -74,6 +109,7 @@ export class MockAgentTransport implements AgentTransport { ): AsyncIterable { this.streams.push({ threadId: _threadId, payload: _payload, options }); this.streaming = true; + this.finished = false; try { while (!this.closed && !signal.aborted) { if (this.pendingError) throw this.pendingError; @@ -81,11 +117,15 @@ export class MockAgentTransport implements AgentTransport { const event = this.eventQueue.shift(); if (event) yield event; } else { - // Wait until flush() wakes us, then loop again to check state. + // The queue is drained: everything awaited so far has been consumed. + this.settleConsumers(); + this.idle = true; + // Wait until wake() rouses us, then loop again to check state. await new Promise((resolve) => { if (signal.aborted) { resolve(); return; } this.resolvers.push(resolve); }); + this.idle = false; } } if (signal.aborted) return; @@ -96,6 +136,11 @@ export class MockAgentTransport implements AgentTransport { } } finally { this.streaming = false; + this.idle = false; + this.finished = true; + // The run is over: nothing queued will ever be consumed, so release + // every awaiter rather than leaving a spec hanging. + this.settleConsumers(); } } @@ -141,8 +186,27 @@ export class MockAgentTransport implements AgentTransport { yield { type: 'values', values: { queued: true } }; } - private flush(): void { + /** Rouse the suspended stream loop so it re-checks queue/error/closed state. */ + private wake(): void { const resolve = this.resolvers.shift(); if (resolve) resolve(); } + + /** + * A promise for "everything queued right now has been consumed". Resolved on + * a macrotask so throttled signal writes inside the adapter have landed. + */ + private consumed(): Promise { + const alreadySettled = + this.finished || + (this.idle && this.eventQueue.length === 0 && this.pendingError === null && !this.closed); + if (alreadySettled) return new Promise((resolve) => setTimeout(resolve, 0)); + return new Promise((resolve) => { this.consumers.push(resolve); }); + } + + private settleConsumers(): void { + const pending = this.consumers; + this.consumers = []; + for (const resolve of pending) setTimeout(resolve, 0); + } } From 7b37209a2a71ce64e0c7b613fb812df6b7e198d9 Mon Sep 17 00:00:00 2001 From: Brian Love Date: Mon, 7 Sep 2026 13:54:28 -0700 Subject: [PATCH 6/7] docs(langgraph): describe the lifecycle token, root registry, error kind, required interrupt, and awaitable transport - lifecycle: `kind` replaces `classification` and holds the `AgentErrorKind`; the registry is root-provided and unregisters on destroy; `AGENT_LIFECYCLE` comes from `provideAgent()` and follows the last-ref-wins rule. - provide-agent: documents the `AGENT_LIFECYCLE` token and the dev-mode warning on an ambiguous ref-less inject. - testing and mock-stream-transport: every emit is awaited rather than chased with a macrotask flush; `flush()` is documented; `chat.interrupt()` drops the `?.` now that `LangGraphAgent` requires it. Both pages' spec fences were executed verbatim against the adapter and pass; the transport page's first fence was additionally missing the optimistic user message in its assertion. - agent-contract and introduction: the narrowed `interrupt`, and the corrected lifecycle/registry facts. Co-Authored-By: Claude Fable 5.1 --- .../content/docs/langgraph/api/api-docs.json | 49 ++++++++++----- .../langgraph/api/mock-stream-transport.mdx | 25 ++++++-- .../docs/langgraph/api/provide-agent.mdx | 19 +++++- .../langgraph/concepts/agent-contract.mdx | 4 +- .../getting-started/introduction.mdx | 2 +- .../docs/langgraph/guides/lifecycle.mdx | 38 ++++++------ .../content/docs/langgraph/guides/testing.mdx | 61 +++++++------------ 7 files changed, 116 insertions(+), 82 deletions(-) diff --git a/apps/website/content/docs/langgraph/api/api-docs.json b/apps/website/content/docs/langgraph/api/api-docs.json index 8b09bb050..7732d3c42 100644 --- a/apps/website/content/docs/langgraph/api/api-docs.json +++ b/apps/website/content/docs/langgraph/api/api-docs.json @@ -2,14 +2,14 @@ { "name": "AgentLifecycleRegistry", "kind": "class", - "description": "Optional registry that collects per-instance agent lifecycles within\nan Angular injection context. External instrumentation packages\n(e.g. cockpit-telemetry) provide this token and read from it.\n\n`@threadplane/langgraph` does NOT provide this itself — the configured agent\ninstance writes to the registry only when an external consumer has provided it.", + "description": "Application-wide registry of every live agent's AgentLifecycle.\n\nIt is `providedIn: 'root'`, so it always exists and every agent registers\ninto the same instance regardless of which injector built it — an agent\ncreated in a route or component injector is still visible from the root.\nExternal instrumentation packages read `lifecycles()` to observe every agent\nin the application without owning the provider graph.\n\nRegistration is scoped to the agent's lifetime: an agent unregisters when the\ninjector that created it is destroyed, so `lifecycles()` never accumulates\nlifecycles for agents that are gone.", "params": [], "examples": [], "properties": [ { "name": "lifecycles", "type": "Signal", - "description": "Reactive list of registered lifecycles.", + "description": "Reactive list of the lifecycles of every currently live agent.", "optional": false } ], @@ -26,6 +26,19 @@ "optional": false } ] + }, + { + "name": "unregister", + "signature": "unregister(lifecycle: AgentLifecycle): void", + "description": "Drop a lifecycle when its agent's injector is destroyed.", + "params": [ + { + "name": "lifecycle", + "type": "AgentLifecycle", + "description": "", + "optional": false + } + ] } ] }, @@ -563,7 +576,7 @@ { "name": "MockAgentTransport", "kind": "class", - "description": "Test transport for deterministic agent testing without a real LangGraph server.\n\nScript event batches upfront, then emit them manually or step through them\nin your test specs. Supports error injection and close control.", + "description": "Test transport for deterministic agent testing without a real LangGraph server.\n\nScript event batches upfront, then emit them manually or step through them\nin your test specs. Supports error injection and close control.\n\n`emit()`, `emitError()`, `close()` and `flush()` are awaitable: the returned\npromise settles once the adapter has consumed everything queued so far (and\none macrotask later, so throttled signal writes have landed), which removes\nthe hand-rolled `await new Promise(resolve => setTimeout(resolve, 0))` flush\nfrom specs.", "params": [ { "name": "script", @@ -573,7 +586,7 @@ } ], "examples": [ - "```typescript\nconst transport = new MockAgentTransport([\n [{ type: 'values', messages: [aiMsg('Hello')] }],\n [{ type: 'values', messages: [aiMsg('Done')] }],\n]);\n```" + "```typescript\nconst transport = new MockAgentTransport([\n [{ type: 'values', messages: [aiMsg('Hello')] }],\n [{ type: 'values', messages: [aiMsg('Done')] }],\n]);\nawait transport.emit(transport.nextBatch());\n```" ], "properties": [ { @@ -641,8 +654,8 @@ }, { "name": "close", - "signature": "close(): void", - "description": "Close the stream. Remaining queued events are drained before completion.", + "signature": "close(): Promise", + "description": "Close the stream. Remaining queued events are drained before completion.\nResolves once the run has finished.", "params": [] }, { @@ -684,8 +697,8 @@ }, { "name": "emit", - "signature": "emit(events: StreamEvent[]): void", - "description": "Manually emit events into the stream.", + "signature": "emit(events: StreamEvent[]): Promise", + "description": "Manually emit events into the stream.\n\nAwait the returned promise: it resolves once the adapter has pulled this\nbatch out of the stream (or the run has ended), so signals are settled and\nassertions read live state rather than the value from before the emit.", "params": [ { "name": "events", @@ -697,8 +710,8 @@ }, { "name": "emitError", - "signature": "emitError(err: Error): void", - "description": "Inject an error into the stream.", + "signature": "emitError(err: Error): Promise", + "description": "Inject an error into the stream. Resolves once the stream has thrown.", "params": [ { "name": "err", @@ -708,6 +721,12 @@ } ] }, + { + "name": "flush", + "signature": "flush(): Promise", + "description": "Resolve once everything emitted so far has been consumed, without emitting\nanything new. Useful after driving the agent by some other route (a\n`submit()`, a `switchThread()`) that has to reach the transport first.", + "params": [] + }, { "name": "getHistory", "signature": "getHistory(threadId: string, signal: AbortSignal): Promise[]>", @@ -975,7 +994,7 @@ { "name": "streamErrorAt", "type": "Signal", - "description": "Epoch ms + classification of the most recent stream error. Resets on switchThread().", + "description": "Epoch ms + failure class of the most recent stream error. Resets on switchThread().\n\n`kind` is the AgentErrorKind of the normalized `AgentError`\n(`connection` | `auth` | `server` | `interrupted` | `aborted`) — the same\nvalue `agent.error()?.kind` carries. For a failure the runtime could not\nnormalize it falls back to the error's constructor name.", "optional": false }, { @@ -1512,8 +1531,8 @@ { "name": "interrupt", "type": "Signal", - "description": "", - "optional": true + "description": "Current human-in-the-loop pause, or `undefined` when the run is not paused.\n\nNarrowed from the neutral `Agent` contract, where `interrupt` is optional\nbecause a runtime without human-in-the-loop support omits it. The LangGraph\nadapter always provides it, so `injectAgent().interrupt()` type-checks\ndirectly under `strictNullChecks` — no `?.()` needed.", + "optional": false }, { "name": "isLoading", @@ -1946,7 +1965,7 @@ { "name": "interrupt", "type": "WritableSignal", - "description": "", + "description": "Current human-in-the-loop pause, or `undefined` when the run is not paused.\n\nNarrowed from the neutral `Agent` contract, where `interrupt` is optional\nbecause a runtime without human-in-the-loop support omits it. The LangGraph\nadapter always provides it, so `injectAgent().interrupt()` type-checks\ndirectly under `strictNullChecks` — no `?.()` needed.", "optional": false }, { @@ -2499,7 +2518,7 @@ { "name": "provideAgent", "kind": "function", - "description": "Wire the LangGraph adapter into Angular's dependency injection.\n\nRegisters a singleton `LangGraphAgent` constructed from `config`. Retrieve it\nin any component with `injectAgent()`. Provide this at the application root\n(`app.config.ts`) for an app-wide agent.\n\nTo use a different agent in a component subtree, re-provide\n`provideAgent({...})` in that component's `providers: []` array —\nAngular's hierarchical DI scopes the singleton accordingly.\n\n**Static vs factory config.** Pass a plain `AgentConfig` object when the\nconfig is known up front. Pass a `() => AgentConfig` factory when the config\ndepends on runtime/DI state — the factory runs inside an Angular injection\ncontext, so it may call `inject()` to read services, route params, or\ncomponent-scoped signals.\n\n**Typed state via AgentRef.** Pass a typed ref as the first argument to flow\nthe state shape from `provideAgent` to `injectAgent` without repeating the\ngeneric at every call site.\n\n**Several agents at one injector level.** Each `provideAgent(ref, …)` call\nbuilds its own agent from its own config, so two (or more) refs may be\nprovided side by side in a single `providers` array and `injectAgent(refA)`\n/ `injectAgent(refB)` return distinct agents. The ref-less `injectAgent()`\nresolves a single shared token, which can only point at one of them: when\nmore than one ref is provided at the same level the **last** call wins.\nAlways inject by ref when an injector provides more than one agent.", + "description": "Wire the LangGraph adapter into Angular's dependency injection.\n\nRegisters a singleton `LangGraphAgent` constructed from `config`. Retrieve it\nin any component with `injectAgent()`. Provide this at the application root\n(`app.config.ts`) for an app-wide agent.\n\nTo use a different agent in a component subtree, re-provide\n`provideAgent({...})` in that component's `providers: []` array —\nAngular's hierarchical DI scopes the singleton accordingly.\n\n**Static vs factory config.** Pass a plain `AgentConfig` object when the\nconfig is known up front. Pass a `() => AgentConfig` factory when the config\ndepends on runtime/DI state — the factory runs inside an Angular injection\ncontext, so it may call `inject()` to read services, route params, or\ncomponent-scoped signals.\n\n**Typed state via AgentRef.** Pass a typed ref as the first argument to flow\nthe state shape from `provideAgent` to `injectAgent` without repeating the\ngeneric at every call site.\n\n**Several agents at one injector level.** Each `provideAgent(ref, …)` call\nbuilds its own agent from its own config, so two (or more) refs may be\nprovided side by side in a single `providers` array and `injectAgent(refA)`\n/ `injectAgent(refB)` return distinct agents. The ref-less `injectAgent()`\nresolves a single shared token, which can only point at one of them: when\nmore than one ref is provided at the same level the **last** call wins, and\nresolving it in dev mode logs a `console.warn` naming every competing ref.\nThe `AGENT_LIFECYCLE` token follows the same rule. Always inject by ref when\nan injector provides more than one agent.\n\n**Lifecycle token.** Every form also provides `AGENT_LIFECYCLE`, so\n`inject(AGENT_LIFECYCLE)` returns the same object as `injectAgent().lifecycle`\nwithout reaching for the agent itself.", "signature": "provideAgent(ref: AgentRef, configOrFactory: AgentConfig | () => AgentConfig): Provider[]", "params": [ { diff --git a/apps/website/content/docs/langgraph/api/mock-stream-transport.mdx b/apps/website/content/docs/langgraph/api/mock-stream-transport.mdx index 6382287ca..b90c4b8bc 100644 --- a/apps/website/content/docs/langgraph/api/mock-stream-transport.mdx +++ b/apps/website/content/docs/langgraph/api/mock-stream-transport.mdx @@ -48,17 +48,20 @@ describe('ChatComponent', () => { fixture.detectChanges(); const stream = fixture.componentInstance.chat.submit({ message: 'Hello' }); - transport.emit([ + await transport.emit([ { type: 'values', messages: [{ type: 'ai', content: 'Hi there' }], }, ]); - transport.close(); + await transport.close(); await stream; fixture.detectChanges(); + // submit() appends the user message optimistically, so the streamed + // assistant reply is the last entry, not the only one. expect(fixture.componentInstance.chat.messages()).toEqual([ + expect.objectContaining({ role: 'user', content: 'Hello' }), expect.objectContaining({ role: 'assistant', content: 'Hi there' }), ]); expect(fixture.componentInstance.chat.status()).toBe('idle'); @@ -69,7 +72,7 @@ describe('ChatComponent', () => { fixture.detectChanges(); const stream = fixture.componentInstance.chat.submit({ message: 'Hello' }); - transport.emitError(new Error('not found')); + await transport.emitError(new Error('not found')); await stream; fixture.detectChanges(); @@ -85,9 +88,10 @@ describe('ChatComponent', () => { |--------|-------------| | `constructor(script?: StreamEvent[][])` | Optionally seeds scripted event batches for manual stepping (defaults to `[]`). | | `nextBatch()` | Returns the next scripted event batch. | -| `emit(events)` | Pushes one or more `StreamEvent` objects into the active stream. | -| `emitError(err)` | Makes the active stream reject with `err`. The runtime catches it, so `submit()` still resolves — assert on `status()` and `error()` instead of on a rejected promise. | -| `close()` | Closes the active stream after queued events drain. | +| `emit(events)` | Pushes one or more `StreamEvent` objects into the active stream. Returns a promise that resolves once the adapter has consumed the batch. | +| `emitError(err)` | Makes the active stream reject with `err`. The runtime catches it, so `submit()` still resolves — assert on `status()` and `error()` instead of on a rejected promise. Returns a promise that resolves once the stream has thrown. | +| `close()` | Closes the active stream after queued events drain. Returns a promise that resolves once the run has finished. | +| `flush()` | Resolves once everything emitted so far has been consumed, without emitting anything new. | | `isStreaming()` | Returns whether a stream is currently active. | Messages inside a `values` event are raw LangGraph messages, so their role comes from `type` (`'human'`, `'ai'`, `'tool'`, `'system'`), not from a `role` field. The projector reads `_getType()` or the raw `type` and falls back to `'ai'`; a `role` key is never read. @@ -99,6 +103,15 @@ The transport also records calls in `streams`, `createdQueuedRuns`, `cancelledRu or `close()`. This makes stream state and payload assertions deterministic. + + `emit()`, `emitError()`, `close()` and `flush()` are asynchronous. `stream()` + is an async generator, so nothing has reached the signals at the moment + `emit()` returns. Await the promise before asserting — a spec that reads + signals synchronously after an emit reads stale state and passes or fails for + the wrong reason. `flush()` covers the same wait when the agent was driven by + something other than an emit. + + ## What's Next diff --git a/apps/website/content/docs/langgraph/api/provide-agent.mdx b/apps/website/content/docs/langgraph/api/provide-agent.mdx index 1050e9670..5abfef6bd 100644 --- a/apps/website/content/docs/langgraph/api/provide-agent.mdx +++ b/apps/website/content/docs/langgraph/api/provide-agent.mdx @@ -82,7 +82,7 @@ provideAgent({ const chat = injectAgent(); ``` -Several agents may also coexist at one injector level. Each `provideAgent(ref, …)` call builds its own agent from its own config, so `injectAgent(refA)` and `injectAgent(refB)` return distinct instances. The ref-less `injectAgent()` resolves a single shared token, which can only point at one of them — the last ref-form call at that level wins. Always inject by ref when an injector provides more than one agent. +Several agents may also coexist at one injector level. Each `provideAgent(ref, …)` call builds its own agent from its own config, so `injectAgent(refA)` and `injectAgent(refB)` return distinct instances. The ref-less `injectAgent()` resolves a single shared token, which can only point at one of them — the last ref-form call at that level wins, and resolving it in development mode logs a warning naming every competing ref. Always inject by ref when an injector provides more than one agent. ```ts export const LIVE = createAgentRef('live'); @@ -95,6 +95,23 @@ providers: [ // injectAgent(LIVE) !== injectAgent(REPLAY) ``` + +In development mode, resolving the ref-less token while several refs share the injector level logs a `console.warn` naming every competing ref and the one that won. It fires only on the ambiguous path: `injectAgent(LIVE)` and `injectAgent(REPLAY)` are unambiguous and stay silent. Production builds never log it. + + +## The `AGENT_LIFECYCLE` token + +Every form of `provideAgent()` also provides `AGENT_LIFECYCLE`, so instrumentation can read an agent's [lifecycle signals](/docs/langgraph/guides/lifecycle) without depending on the agent itself: + +```ts +import { inject } from '@angular/core'; +import { AGENT_LIFECYCLE } from '@threadplane/langgraph'; + +const lifecycle = inject(AGENT_LIFECYCLE); // === injectAgent().lifecycle +``` + +The token follows the same last-one-wins rule as the ref-less `injectAgent()`: with several refs at one injector level it resolves the last ref's lifecycle. Read `injectAgent(ref).lifecycle` instead when that is the case. + ## Transcript node filtering LangGraph streams `messages-tuple` chunks for every LLM node in a run. If your graph has side-effect LLM nodes, such as a title generator or evaluator, set `transcriptNodeNames` so only your conversational node updates `messages()`. diff --git a/apps/website/content/docs/langgraph/concepts/agent-contract.mdx b/apps/website/content/docs/langgraph/concepts/agent-contract.mdx index 92d42fd84..106645cb8 100644 --- a/apps/website/content/docs/langgraph/concepts/agent-contract.mdx +++ b/apps/website/content/docs/langgraph/concepts/agent-contract.mdx @@ -136,7 +136,9 @@ The UI lifecycle is intentionally boring — and boring is the goal. 5. `stop()` aborts the active run when supported, and `retry()` clears `error()` and re-runs the last submitted input after a failure. 6. `regenerate(index)` rolls back from an assistant message and reruns from the preceding user message. -LangGraph adds deeper lifecycle and history surfaces. `@threadplane/langgraph` exposes `injectAgent().lifecycle` and exports `AgentLifecycle`, `AgentLifecycleRegistry`, and the low-level `AGENT_LIFECYCLE` token. Those are useful for telemetry, debugging, persistence, and time-travel UI. They are not required by `@threadplane/chat`. +LangGraph adds deeper lifecycle and history surfaces. `@threadplane/langgraph` exposes `injectAgent().lifecycle`, provides the `AGENT_LIFECYCLE` token from every `provideAgent()` call, and ships a root-provided `AgentLifecycleRegistry` that collects the lifecycle of every live agent. Those are useful for telemetry, debugging, persistence, and time-travel UI. They are not required by `@threadplane/chat`. See [Agent lifecycle signals](/docs/langgraph/guides/lifecycle). + +`LangGraphAgent` also narrows one optional member of the neutral contract: `interrupt` is required, because the LangGraph adapter always provides it. Code written against `LangGraphAgent` calls `agent.interrupt()` directly; code written against the neutral `Agent` still needs `interrupt?.()`. ## Testing And Mocks diff --git a/apps/website/content/docs/langgraph/getting-started/introduction.mdx b/apps/website/content/docs/langgraph/getting-started/introduction.mdx index 702f6fe6d..a499acf47 100644 --- a/apps/website/content/docs/langgraph/getting-started/introduction.mdx +++ b/apps/website/content/docs/langgraph/getting-started/introduction.mdx @@ -40,7 +40,7 @@ chat.messages(); // Message[] chat.status(); // 'idle' | 'running' | 'error' chat.isLoading(); // boolean chat.error(); // AgentError | undefined -chat.interrupt?.(); // AgentInterrupt | undefined +chat.interrupt(); // AgentInterrupt | undefined chat.history(); // AgentCheckpoint[] chat.langGraphHistory(); // ThreadState[] ``` diff --git a/apps/website/content/docs/langgraph/guides/lifecycle.mdx b/apps/website/content/docs/langgraph/guides/lifecycle.mdx index 5266c3be5..00e405310 100644 --- a/apps/website/content/docs/langgraph/guides/lifecycle.mdx +++ b/apps/website/content/docs/langgraph/guides/lifecycle.mdx @@ -10,12 +10,13 @@ The `@threadplane/langgraph` library exposes per-agent lifecycle signals on ever ```typescript import { InjectionToken, Signal } from '@angular/core'; +import type { AgentErrorKind } from '@threadplane/chat'; export interface AgentLifecycle { /** Epoch ms of the first stream chunk arrival. Resets on switchThread(). */ readonly streamStartedAt: Signal; - /** Epoch ms + classification of the most recent stream error. Resets on switchThread(). */ - readonly streamErrorAt: Signal<{ at: number; classification: string } | null>; + /** Epoch ms + failure class of the most recent stream error. Resets on switchThread(). */ + readonly streamErrorAt: Signal<{ at: number; kind: AgentErrorKind | string } | null>; /** Epoch ms of the first interrupt$ non-null in this stream. Resets on switchThread(). */ readonly interruptReceivedAt: Signal; /** Epoch ms of the most recent submit({ resume }) call. Resets on switchThread(). */ @@ -40,7 +41,7 @@ Six of the eight signals derive directly from existing stream subjects on the ag | Signal | Source | |--------|--------| | `streamStartedAt` | first non-empty `values$` or `messages$` emission | -| `streamErrorAt` | `error$` emission, classified | +| `streamErrorAt` | `error$` emission, classified by `AgentErrorKind` | | `interruptReceivedAt` | first non-null `interrupt$` value | | `threadPersistedAt` | first non-empty `history$` emission | | `toolCallStartedAt` | first tool-call append in `toolCalls$` | @@ -53,8 +54,8 @@ Two signals require explicit hook points that the agent already invokes: | `interruptResolvedAt` | `submit({ resume })` | | `threadCreatedAt` | the first `submit()` on an agent that has no thread yet | - -`streamErrorAt().classification` is the error's constructor name, taken from `error.name` (or `'string'` / `'unknown'` for a non-`Error` value). Because the runtime normalizes every failure into an `AgentError`, that value is almost always the literal `'AgentError'`, which makes it a poor discriminator. The actionable one is `agent.error()?.kind` — `connection`, `auth`, `server`, `interrupted`, or `aborted`. See [Error handling](/docs/chat/guides/error-handling). + +`streamErrorAt().kind` is the [`AgentErrorKind`](/docs/chat/guides/error-handling) of the failure — `connection`, `auth`, `server`, `interrupted`, or `aborted` — the same value `agent.error()?.kind` carries. The runtime normalizes every failure through `toAgentError()` before it reaches the lifecycle, so the discriminator is actionable. A value the runtime could not normalize falls back to the error's constructor name, which is why the type is `AgentErrorKind | string`. ## Subscribing @@ -71,39 +72,38 @@ export class MyComponent { effect(() => { const err = this.chat.lifecycle.streamErrorAt(); if (err) { - console.log('Stream error at', err.at, 'classification:', err.classification); + console.log('Stream error at', err.at, 'kind:', err.kind); } }); } } ``` -For application-wide instrumentation, provide `AgentLifecycleRegistry` and read back the lifecycles registered by agents created in that injection context. An agent registers itself at construction, so the registry has to be provided at or above the injector that calls `provideAgent()` — an agent constructed before the registry exists registers nowhere: +For application-wide instrumentation, inject `AgentLifecycleRegistry` and read back the lifecycles of every live agent. It is `providedIn: 'root'`, so nothing has to be provided and every agent registers into the same instance no matter which injector built it — a route-scoped or component-scoped `provideAgent()` is still visible from the root: ```typescript -import { ApplicationConfig, inject } from '@angular/core'; +import { inject } from '@angular/core'; import { AgentLifecycleRegistry } from '@threadplane/langgraph'; -export const appConfig: ApplicationConfig = { - providers: [AgentLifecycleRegistry], -}; -``` - -```typescript const registry = inject(AgentLifecycleRegistry); const lifecycles = registry.lifecycles(); ``` -The exported `AGENT_LIFECYCLE` token is a low-level token for custom integrations. The library never provides it — it is an empty token you provide yourself when you want a lifecycle reachable by injection: +Registration is scoped to the agent's lifetime: an agent unregisters when the injector that created it is destroyed, so `lifecycles()` never accumulates lifecycles for agents that are gone. + +`provideAgent()` also provides the `AGENT_LIFECYCLE` token, so a service that needs only the timings can inject the lifecycle without reaching for the agent: ```typescript -import { AGENT_LIFECYCLE, injectAgent } from '@threadplane/langgraph'; +import { inject } from '@angular/core'; +import { AGENT_LIFECYCLE } from '@threadplane/langgraph'; -providers: [ - { provide: AGENT_LIFECYCLE, useFactory: () => injectAgent().lifecycle }, -]; +const lifecycle = inject(AGENT_LIFECYCLE); // === injectAgent().lifecycle ``` + +`AGENT_LIFECYCLE` follows the same last-one-wins rule as the ref-less `injectAgent()`: when several `provideAgent(ref, …)` calls share an injector level, the token resolves the lifecycle of the **last** ref provided. Read `injectAgent(ref).lifecycle` when an injector provides more than one agent. See [provideAgent()](/docs/langgraph/api/provide-agent). + + ## Reset semantics All eight signals reset on `switchThread()`. This keeps lifecycle observations scoped to the current thread. diff --git a/apps/website/content/docs/langgraph/guides/testing.mdx b/apps/website/content/docs/langgraph/guides/testing.mdx index 67afe7e48..f2a29f3ff 100644 --- a/apps/website/content/docs/langgraph/guides/testing.mdx +++ b/apps/website/content/docs/langgraph/guides/testing.mdx @@ -171,11 +171,11 @@ Lead with `provideFakeAgent()` for the simple case. Reach for `MockAgentTranspor Register it through `provideAgent({ transport })` in the TestBed `providers` array, then resolve the agent with `TestBed.runInInjectionContext(() => injectAgent())`. - -`MockAgentTransport.stream()` is an async generator: `emit()` only wakes the suspended stream loop, so nothing has reached the signals when `emit()` returns. Await one macrotask (`await new Promise(resolve => setTimeout(resolve, 0))`) after every `emit()`, `emitError()`, and `close()` before asserting, and keep the promise `submit()` returns so the spec can await the end of the run. Specs that assert synchronously read stale signals and pass or fail for the wrong reason. + +`MockAgentTransport.stream()` is an async generator: `emit()` only wakes the suspended stream loop, so nothing has reached the signals at the moment it is called. `emit()`, `emitError()` and `close()` therefore return a promise that resolves once the adapter has consumed the batch — `await` each one before asserting, and keep the promise `submit()` returns so the spec can await the end of the run. Specs that assert synchronously read stale signals and pass or fail for the wrong reason. `flush()` waits the same way without emitting anything, for the cases where something other than an emit drove the agent. -These examples also pass `throttle: false`. The adapter throttles signal writes by 16 ms by default, which one macrotask does not cover. +These examples also pass `throttle: false`. The adapter throttles signal writes by 16 ms by default, and the awaited emit does not cover that window. ### Basic setup @@ -186,8 +186,6 @@ import { describe, it, expect } from 'vitest'; import { TestBed } from '@angular/core/testing'; import { MockAgentTransport, injectAgent, provideAgent } from '@threadplane/langgraph'; -const flush = () => new Promise((resolve) => setTimeout(resolve, 0)); - describe('basic transport', () => { it('projects an emitted assistant chunk into messages()', async () => { const transport = new MockAgentTransport(); @@ -199,14 +197,13 @@ describe('basic transport', () => { const chat = TestBed.runInInjectionContext(() => injectAgent()); const submitted = chat.submit({ message: 'Hi' }); - transport.emit([ + await transport.emit([ { type: 'messages', messages: [{ id: 'ai-1', type: 'ai', content: 'Hello!' }] }, ]); - await flush(); expect(chat.messages().at(-1)?.content).toBe('Hello!'); - transport.close(); + await transport.close(); await submitted; expect(chat.status()).toBe('idle'); }); @@ -222,8 +219,6 @@ import { describe, it, expect } from 'vitest'; import { TestBed } from '@angular/core/testing'; import { MockAgentTransport, injectAgent, provideAgent } from '@threadplane/langgraph'; -const flush = () => new Promise((resolve) => setTimeout(resolve, 0)); - describe('streaming lifecycle', () => { it('runs from submit through two batches back to idle', async () => { const transport = new MockAgentTransport([ @@ -241,15 +236,13 @@ describe('streaming lifecycle', () => { expect(chat.status()).toBe('running'); expect(chat.isLoading()).toBe(true); - transport.emit(transport.nextBatch()); - await flush(); + await transport.emit(transport.nextBatch()); expect(chat.messages().at(-1)?.content).toBe('Analyzing'); - transport.emit(transport.nextBatch()); - await flush(); + await transport.emit(transport.nextBatch()); expect(chat.messages().at(-1)?.content).toBe('Analyzing... done.'); - transport.close(); + await transport.close(); await submitted; expect(chat.status()).toBe('idle'); expect(chat.isLoading()).toBe(false); @@ -266,8 +259,6 @@ import { describe, it, expect } from 'vitest'; import { TestBed } from '@angular/core/testing'; import { MockAgentTransport, injectAgent, provideAgent } from '@threadplane/langgraph'; -const flush = () => new Promise((resolve) => setTimeout(resolve, 0)); - describe('interrupt handling', () => { it('surfaces the interrupt payload', async () => { const transport = new MockAgentTransport(); @@ -279,17 +270,16 @@ describe('interrupt handling', () => { const chat = TestBed.runInInjectionContext(() => injectAgent()); const submitted = chat.submit({ message: 'Delete my account' }); - transport.emit([ + await transport.emit([ { type: 'interrupt', interrupt: { id: 'approval', value: { action: 'delete_account', risk: 'high' } }, }, ]); - await flush(); - transport.close(); + await transport.close(); await submitted; - const pending = chat.interrupt?.(); + const pending = chat.interrupt(); expect(pending).toBeDefined(); const value = pending?.value as { action: string; risk: string }; expect(value.action).toBe('delete_account'); @@ -298,8 +288,8 @@ describe('interrupt handling', () => { }); ``` - -`interrupt` is optional on the neutral `Agent` contract, because a runtime without human-in-the-loop support omits it. Under `strictNullChecks` the call site therefore needs `?.()`. The LangGraph adapter always provides it, and `langGraphInterrupts()` is a required signal carrying the raw array. + +`interrupt` is optional on the runtime-neutral `Agent` contract, because a runtime without human-in-the-loop support omits it — code written against that contract needs `interrupt?.()` under `strictNullChecks`. `LangGraphAgent` narrows it to required, because the LangGraph adapter always provides it, so `injectAgent().interrupt()` type-checks as a plain call. `langGraphInterrupts()` carries the raw LangGraph array alongside it. Resume the paused run with `chat.submit({ resume: { approved: true } })`, exactly as a component would. @@ -313,8 +303,6 @@ import { describe, it, expect } from 'vitest'; import { TestBed } from '@angular/core/testing'; import { MockAgentTransport, injectAgent, provideAgent } from '@threadplane/langgraph'; -const flush = () => new Promise((resolve) => setTimeout(resolve, 0)); - describe('error handling', () => { it('classifies a transport failure', async () => { const transport = new MockAgentTransport(); @@ -326,9 +314,8 @@ describe('error handling', () => { const chat = TestBed.runInInjectionContext(() => injectAgent()); const submitted = chat.submit({ message: 'Hello' }); - transport.emitError(new Error('HTTP 500: connection lost')); + await transport.emitError(new Error('HTTP 500: connection lost')); await submitted.catch(() => undefined); - await flush(); const err = chat.error(); expect(err?.kind).toBe('server'); @@ -347,16 +334,15 @@ describe('error handling', () => { const chat = TestBed.runInInjectionContext(() => injectAgent()); const submitted = chat.submit({ message: 'Hello' }); - transport.emitError(new Error('HTTP 503: service unavailable')); + await transport.emitError(new Error('HTTP 503: service unavailable')); await submitted.catch(() => undefined); - await flush(); expect(chat.status()).toBe('error'); const retried = chat.retry(); expect(chat.error()).toBeUndefined(); expect(chat.isLoading()).toBe(true); - transport.close(); + await transport.close(); await retried; }); }); @@ -373,8 +359,6 @@ import { describe, it, expect } from 'vitest'; import { TestBed } from '@angular/core/testing'; import { MockAgentTransport, injectAgent, provideAgent } from '@threadplane/langgraph'; -const flush = () => new Promise((resolve) => setTimeout(resolve, 0)); - describe('thread switching', () => { it('clears the transcript when switching to a new thread', async () => { const transport = new MockAgentTransport(); @@ -386,16 +370,15 @@ describe('thread switching', () => { const chat = TestBed.runInInjectionContext(() => injectAgent()); const submitted = chat.submit({ message: 'Hi' }); - transport.emit([ + await transport.emit([ { type: 'messages', messages: [{ id: 'ai-1', type: 'ai', content: 'Thread A response' }] }, ]); - await flush(); - transport.close(); + await transport.close(); await submitted; expect(chat.messages().at(-1)?.content).toBe('Thread A response'); chat.switchThread(null); - await flush(); + await transport.flush(); expect(chat.messages()).toEqual([]); }); }); @@ -416,11 +399,11 @@ Pass the transport into `provideAgent({ ..., transport })` in the TestBed `provi Call `TestBed.runInInjectionContext(() => injectAgent())` so the adapter can reach Angular's injector for signal creation and cleanup. - -Use `emit()` for ad-hoc events, `nextBatch()` for pre-scripted sequences, and `emitError()` for failures — awaiting one macrotask after each call. + +Use `emit()` for ad-hoc events, `nextBatch()` for pre-scripted sequences, and `emitError()` for failures — awaiting the promise each one returns before asserting. -Read `chat.messages()`, `chat.status()`, `chat.interrupt?.()`, and `chat.error()` to verify the agent reacted, then `close()` the stream and await the submit promise. +Read `chat.messages()`, `chat.status()`, `chat.interrupt()`, and `chat.error()` to verify the agent reacted, then `close()` the stream and await the submit promise. From 249bf5ebdb7ad914b81a4705755473274dc0ab45 Mon Sep 17 00:00:00 2001 From: Brian Love Date: Mon, 7 Sep 2026 19:15:05 -0700 Subject: [PATCH 7/7] fix(examples): drop the dead interrupt existence guard in the demo shell LangGraphAgent.interrupt is now a required member, so `agent.interrupt &&` is always true and the packaged-consumer build rejects it with TS2774. The call itself still returns AgentInterrupt | undefined, so the remaining `agent.interrupt()` test is the real condition. Co-Authored-By: Claude Opus 5 --- examples/chat/angular/src/app/shell/demo-shell.component.html | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/examples/chat/angular/src/app/shell/demo-shell.component.html b/examples/chat/angular/src/app/shell/demo-shell.component.html index a44f6e31f..b3aae4d57 100644 --- a/examples/chat/angular/src/app/shell/demo-shell.component.html +++ b/examples/chat/angular/src/app/shell/demo-shell.component.html @@ -135,7 +135,7 @@
- @if (agent.interrupt && agent.interrupt()) { + @if (agent.interrupt()) {
@@ -148,7 +148,7 @@ [attr.data-sidenav-mode]="sidenavMode() !== 'drawer' ? sidenavMode() : null" > - @if (agent.interrupt && agent.interrupt()) { + @if (agent.interrupt()) {