diff --git a/packages/angular/src/lib/active-thread-connector.spec.ts b/packages/angular/src/lib/active-thread-connector.spec.ts new file mode 100644 index 0000000000..1c352e49ef --- /dev/null +++ b/packages/angular/src/lib/active-thread-connector.spec.ts @@ -0,0 +1,309 @@ +import { + EnvironmentInjector, + createEnvironmentInjector, + runInInjectionContext, + signal, +} from "@angular/core"; +import { TestBed } from "@angular/core/testing"; +import { test, expect, vi } from "vitest"; +import { HttpAgent } from "@ag-ui/client"; +import { + injectChatConfiguration, + provideCopilotChatConfiguration, +} from "./chat-configuration"; +import type { AgentStore } from "./agent"; +import { connectActiveThread } from "./active-thread-connector"; + +/** + * Builds a fake agent + agent-store signal and wires the connector under an + * injection context with a real {@link CopilotChatConfiguration}. + * + * @returns The config service, the fake agent, and the connect spy. + */ +function setup() { + const fake = { + agent: { + threadId: "t0", + messages: [{ id: "m1" }] as { id: string }[], + setMessages: vi.fn((arr: { id: string }[]) => { + fake.agent.messages = arr; + }), + detachActiveRun: vi.fn(() => Promise.resolve()), + }, + }; + + const connect = vi.fn(); + + TestBed.configureTestingModule({ + providers: [provideCopilotChatConfiguration()], + }); + + const config = TestBed.runInInjectionContext(() => { + const cfg = injectChatConfiguration(); + const agentStore = signal(fake as never) as unknown as () => AgentStore; + connectActiveThread(cfg, agentStore, connect); + return cfg; + }); + + return { config, fake, connect }; +} + +test("explicit switch connects the agent to the picked thread", async () => { + const { config, fake, connect } = setup(); + + config.setActiveThreadId("picked-1"); + TestBed.flushEffects(); + await Promise.resolve(); + + expect(fake.agent.threadId).toBe("picked-1"); + expect(connect).toHaveBeenCalledWith( + expect.objectContaining({ agent: fake.agent }), + ); +}); + +test("initial mount does not clear messages", async () => { + const { fake, connect } = setup(); + + TestBed.flushEffects(); + await Promise.resolve(); + + expect(fake.agent.setMessages).not.toHaveBeenCalled(); + expect(fake.agent.messages).toEqual([{ id: "m1" }]); + expect(connect).not.toHaveBeenCalled(); +}); + +test("an explicit switch detaches the prior in-flight run on re-run", async () => { + const { config, fake } = setup(); + + config.setActiveThreadId("picked-1"); + TestBed.flushEffects(); + await Promise.resolve(); + + expect(fake.agent.detachActiveRun).not.toHaveBeenCalled(); + + config.setActiveThreadId("picked-2"); + TestBed.flushEffects(); + await Promise.resolve(); + + expect(fake.agent.detachActiveRun).toHaveBeenCalledTimes(1); +}); + +test("destroying the injector detaches the connected run", async () => { + const fake = { + agent: { + threadId: "t0", + messages: [{ id: "m1" }] as { id: string }[], + setMessages: vi.fn(), + detachActiveRun: vi.fn(() => Promise.resolve()), + }, + }; + const connect = vi.fn(); + TestBed.configureTestingModule({ + providers: [provideCopilotChatConfiguration()], + }); + const parent = TestBed.inject(EnvironmentInjector); + const childInjector = createEnvironmentInjector([], parent); + const config = runInInjectionContext(childInjector, () => { + const cfg = injectChatConfiguration(); + const agentStore = signal(fake as never) as unknown as () => AgentStore; + connectActiveThread(cfg, agentStore, connect); + return cfg; + }); + + config.setActiveThreadId("picked-1"); + TestBed.flushEffects(); + await Promise.resolve(); + + childInjector.destroy(); + + expect(fake.agent.detachActiveRun).toHaveBeenCalledTimes(1); +}); + +test("a genuine new-thread transition clears messages and skips connect", async () => { + const { config, fake } = setup(); + + config.setActiveThreadId("picked"); + TestBed.flushEffects(); + await Promise.resolve(); + + config.startNewThread(); + TestBed.flushEffects(); + await Promise.resolve(); + + expect(fake.agent.setMessages).toHaveBeenCalledWith([]); + expect(fake.agent.messages).toEqual([]); +}); + +/** + * Builds a fake agent + agent-store signal and wires the connector with + * explicit cursor hooks and a caller-supplied connect implementation. + * + * @param connect - The connect implementation under test. + * @returns The config service, the fake agent, the connect spy, and the + * `onConnectStart`/`onConnectSettle` hook spies. + */ +function setupWithHooks(connect: (params: { agent: unknown }) => unknown) { + const fake = { + agent: { + threadId: "t0", + messages: [{ id: "m1" }] as { id: string }[], + setMessages: vi.fn((arr: { id: string }[]) => { + fake.agent.messages = arr; + }), + detachActiveRun: vi.fn(() => Promise.resolve()), + }, + }; + + const onConnectStart = vi.fn(); + const onConnectSettle = vi.fn(); + + TestBed.configureTestingModule({ + providers: [provideCopilotChatConfiguration()], + }); + + const config = TestBed.runInInjectionContext(() => { + const cfg = injectChatConfiguration(); + const agentStore = signal(fake as never) as unknown as () => AgentStore; + connectActiveThread(cfg, agentStore, connect as never, { + onConnectStart, + onConnectSettle, + }); + return cfg; + }); + + return { config, fake, onConnectStart, onConnectSettle }; +} + +test("N1: a rejecting connect does not raise an unhandled rejection and still settles the cursor", async () => { + const unhandled: unknown[] = []; + const onUnhandled = (event: PromiseRejectionEvent) => { + unhandled.push(event.reason); + }; + globalThis.addEventListener?.("unhandledrejection", onUnhandled); + + const { config, onConnectStart, onConnectSettle } = setupWithHooks(() => + Promise.reject(new Error("connect failed")), + ); + + config.setActiveThreadId("picked-1", { explicit: true }); + TestBed.flushEffects(); + + // Flush microtasks so the connect promise and its caught/finally chain run. + await Promise.resolve(); + await Promise.resolve(); + await Promise.resolve(); + + expect(onConnectStart).toHaveBeenCalledTimes(1); + expect(onConnectSettle).toHaveBeenCalledTimes(1); + expect(unhandled).toEqual([]); + + globalThis.removeEventListener?.("unhandledrejection", onUnhandled); +}); + +test("N2: a superseded connect does not settle the cursor; the live connect does", async () => { + let resolveFirst: (() => void) | undefined; + let resolveSecond: (() => void) | undefined; + const deferreds: Array> = [ + new Promise((resolve) => { + resolveFirst = resolve; + }), + new Promise((resolve) => { + resolveSecond = resolve; + }), + ]; + let call = 0; + const connect = vi.fn(() => deferreds[call++]); + + const { config, onConnectSettle } = setupWithHooks(connect); + + config.setActiveThreadId("picked-1", { explicit: true }); + TestBed.flushEffects(); + await Promise.resolve(); + + // Second explicit thread supersedes the first connect before it settles. + config.setActiveThreadId("picked-2", { explicit: true }); + TestBed.flushEffects(); + await Promise.resolve(); + + // Settle the FIRST (now superseded) connect: its settle must be suppressed. + resolveFirst?.(); + await Promise.resolve(); + await Promise.resolve(); + expect(onConnectSettle).not.toHaveBeenCalled(); + + // Settle the SECOND (live) connect: its settle fires. + resolveSecond?.(); + await Promise.resolve(); + await Promise.resolve(); + expect(onConnectSettle).toHaveBeenCalledTimes(1); +}); + +test("N3: cleanup detaches the connected run", async () => { + const fake = { + agent: { + threadId: "t0", + messages: [{ id: "m1" }] as { id: string }[], + setMessages: vi.fn(), + detachActiveRun: vi.fn(() => Promise.resolve()), + }, + }; + const connect = vi.fn(() => Promise.resolve()); + + TestBed.configureTestingModule({ + providers: [provideCopilotChatConfiguration()], + }); + const parent = TestBed.inject(EnvironmentInjector); + const childInjector = createEnvironmentInjector([], parent); + const config = runInInjectionContext(childInjector, () => { + const cfg = injectChatConfiguration(); + const agentStore = signal(fake as never) as unknown as () => AgentStore; + connectActiveThread(cfg, agentStore, connect as never); + return cfg; + }); + + config.setActiveThreadId("picked-1", { explicit: true }); + TestBed.flushEffects(); + await Promise.resolve(); + + childInjector.destroy(); + + expect(fake.agent.detachActiveRun).toHaveBeenCalledTimes(1); +}); + +test("N3: cleanup aborts the agent's AbortController for an HttpAgent", async () => { + const abort = vi.fn(); + // A minimal HttpAgent: the connector sets `agent.abortController` only when + // `agent instanceof HttpAgent`, so use a real instance with a stubbed + // controller to assert the abort fires on teardown. + const agent = new HttpAgent({ url: "http://localhost/agent" }); + agent.abortController = { abort } as never; + const detachSpy = vi + .spyOn(agent, "detachActiveRun") + .mockResolvedValue(undefined); + + const fake = { agent }; + const connect = vi.fn(() => Promise.resolve()); + + TestBed.configureTestingModule({ + providers: [provideCopilotChatConfiguration()], + }); + const parent = TestBed.inject(EnvironmentInjector); + const childInjector = createEnvironmentInjector([], parent); + const config = runInInjectionContext(childInjector, () => { + const cfg = injectChatConfiguration(); + const agentStore = signal(fake as never) as unknown as () => AgentStore; + connectActiveThread(cfg, agentStore, connect as never); + return cfg; + }); + + config.setActiveThreadId("picked-1", { explicit: true }); + TestBed.flushEffects(); + await Promise.resolve(); + + childInjector.destroy(); + + // The connector replaced `agent.abortController` with its own per-run + // controller before connecting; assert it aborted on teardown and detached. + expect(detachSpy).toHaveBeenCalledTimes(1); + expect(agent.abortController?.signal.aborted).toBe(true); +}); diff --git a/packages/angular/src/lib/active-thread-connector.ts b/packages/angular/src/lib/active-thread-connector.ts new file mode 100644 index 0000000000..088dd14a28 --- /dev/null +++ b/packages/angular/src/lib/active-thread-connector.ts @@ -0,0 +1,132 @@ +import { effect, untracked, type Signal } from "@angular/core"; +import { HttpAgent, type AbstractAgent } from "@ag-ui/client"; +import type { AgentStore } from "./agent"; +import type { CopilotChatConfiguration } from "./chat-configuration"; + +/** + * Signature of `CopilotKitCore.connectAgent`, injected for testability. + * + * The active thread is carried on `agent.threadId`, which the connector sets + * before invoking this function; there is no separate thread parameter. The + * connector receives the raw `core.connectAgent`, not a cursor-wrapping fn — + * the cursor lifecycle is owned by the connector via {@link ConnectActiveThreadCursorHooks}. + */ +export type ConnectAgentFn = (params: { agent: AbstractAgent }) => unknown; + +/** + * Loading-cursor hooks the connector drives around an explicit connect, with + * the same timing as the standalone `` `connectToAgent` path: + * cursor on at connect start, off when that connect settles — but only if its + * run was not superseded by a newer connect (the staleness guard). + */ +export interface ConnectActiveThreadCursorHooks { + /** Called synchronously when an explicit connect begins (cursor on). */ + onConnectStart?: () => void; + /** Called when that connect settles, ONLY if its run was not superseded (cursor off). */ + onConnectSettle?: () => void; +} + +/** + * Wires the active chat thread to the live agent. + * + * Reactively observes the resolved thread id (and whether it was chosen + * explicitly) from {@link CopilotChatConfiguration} and the current agent from + * the agent-store signal. On every change it pins the thread onto + * `agent.threadId`, then: + * + * - **Explicit switch** (user picked a thread): connects the agent to that + * thread via {@link ConnectAgentFn}, owning the loading-cursor + abort + + * detach lifecycle of the standalone `` `connectToAgent` path. + * (Connect-error *logging* is the one intentional divergence — see below.) + * Before connecting it installs a per-run `AbortController` on the + * agent (only when the agent is an {@link HttpAgent}) and calls + * {@link ConnectActiveThreadCursorHooks.onConnectStart}. When that connect + * settles it calls {@link ConnectActiveThreadCursorHooks.onConnectSettle} — + * but only if the run was not superseded (the `detached` staleness guard), so + * a stale connect cannot clear the cursor out from under a newer one. The + * connect's promise is caught before its `finally` so a rejecting connect + * never produces an unhandled rejection. Connect errors surface via the + * AgentStore's run/error subscription and are intentionally NOT re-logged here + * — the deliberate divergence from `connectToAgent`, which additionally + * `console.error`s unexpected (non-`AGUIConnectNotImplementedError`) failures. + * On the next effect re-run + * (thread/agent switch) and on destroy, the cleanup aborts the in-flight + * request via the per-run controller AND detaches the run via + * `agent.detachActiveRun()`, mirroring the standalone teardown so a rapid + * switch or component destroy does not leak a prior run. + * - **Fresh / non-explicit switch** (e.g. {@link CopilotChatConfiguration.startNewThread}): + * clears the agent's messages via `agent.setMessages([])` and skips the + * connect — the runtime assigns the server thread id on first send. The clear + * fires only on an actual transition to a *new* fresh thread id, never on the + * initial mount nor on an agent-store swap that leaves the thread id unchanged + * (which would otherwise wipe a resumed/shared agent's existing history). + * + * Tracked reads happen in the effect's reactive scope; all mutation and the + * connect call run inside `untracked()` so they do not register as + * dependencies (mirrors the effect/untracked idiom in `threads.ts`). + * + * @param config - The chat configuration exposing the resolved thread signals. + * @param agentStore - Signal yielding the current {@link AgentStore}. + * @param connectAgent - The raw `CopilotKitCore.connectAgent` implementation. + * @param hooks - Optional loading-cursor hooks driven around the explicit connect. + */ +export function connectActiveThread( + config: CopilotChatConfiguration, + agentStore: Signal, + connectAgent: ConnectAgentFn, + hooks?: ConnectActiveThreadCursorHooks, +): void { + // Tracks the thread id observed on the previous effect run so the + // non-explicit branch can distinguish a genuine new-thread transition from + // the initial mount (`undefined`) or an agent-store swap that left the thread + // unchanged. Clearing only on a real transition prevents wiping a + // resumed/shared agent's existing message history. + let lastThreadId: string | undefined; + effect((onCleanup) => { + const threadId = config.threadId(); + const explicit = config.hasExplicitThreadId(); + const store = agentStore(); + untracked(() => { + const agent = store.agent; + agent.threadId = threadId; + if (explicit) { + // Mirror the standalone `connectToAgent` cursor/abort/detach lifecycle: + // a per-run staleness flag, a per-run AbortController installed on + // HttpAgents, cursor-on, a single caught connect chain, and an + // abort+detach cleanup. (Connect-error logging is intentionally omitted + // here — errors surface via the AgentStore run/error subscription.) + let detached = false; + const abortController = new AbortController(); + if (agent instanceof HttpAgent) { + agent.abortController = abortController; + } + + hooks?.onConnectStart?.(); + + const result = connectAgent({ agent }); + Promise.resolve(result) + .catch(() => { + // connect errors surface via the AgentStore's run/error + // subscription, not here. + }) + .finally(() => { + if (!detached) hooks?.onConnectSettle?.(); + }); + + // Mirror the standalone `` teardown: mark the run stale, + // abort the in-flight request, and detach the run for the agent + // connected this run. Fires on the next effect re-run (thread/agent + // switch) and on destroy so a prior run does not leak. + onCleanup(() => { + detached = true; + abortController.abort(); + agent.detachActiveRun().catch(() => {}); + }); + } else if (lastThreadId !== undefined && threadId !== lastThreadId) { + // Real switch to a new fresh thread; not mount and not a same-thread swap. + agent.setMessages([]); + } + lastThreadId = threadId; + }); + }); +}