mirror of
https://github.com/CopilotKit/CopilotKit.git
synced 2026-09-14 16:26:20 +08:00
feat(angular): active-thread connector wiring config to the live agent
connectActiveThread reactively pins the resolved thread onto agent.threadId. On an explicit switch it connects the agent, owning the loading-cursor + abort + detach lifecycle of the standalone connectToAgent path (per-run AbortController on HttpAgents, a single caught connect chain so a rejecting connect never leaks an unhandled rejection, and a staleness-guarded cursor settle). On a fresh / non-explicit switch it clears messages only on a genuine new-thread transition, never on mount or a same-thread agent swap. Injection-context-only; not exported on the public surface.
This commit is contained in:
Benjamin Taylor
@@ -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<Promise<void>> = [
|
||||
new Promise<void>((resolve) => {
|
||||
resolveFirst = resolve;
|
||||
}),
|
||||
new Promise<void>((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);
|
||||
});
|
||||
@@ -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 `<copilot-chat>` `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 `<copilot-chat>` `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<AgentStore>,
|
||||
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 `<copilot-chat>` 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;
|
||||
});
|
||||
});
|
||||
}
|
||||
Reference in New Issue
Block a user