Add client-hosted agent registration

ts/packages/agentServer/client/package.json

@@ -32,6 +32,7 @@
   },
   "dependencies": {
     "@typeagent/agent-rpc": "workspace:*",
+    "@typeagent/agent-sdk": "workspace:*",
     "@typeagent/agent-server-protocol": "workspace:*",
     "@typeagent/dispatcher-rpc": "workspace:*",
     "debug": "^4.4.0",

ts/packages/agentServer/client/src/agentServerClient.ts

@@ -4,6 +4,8 @@
 import { createChannelProviderAdapter } from "@typeagent/agent-rpc/channel";
 import type { ChannelProviderAdapter } from "@typeagent/agent-rpc/channel";
 import { createRpc } from "@typeagent/agent-rpc/rpc";
+import { createAgentRpcServer } from "@typeagent/agent-rpc/server";
+import type { AppAgent, AppAgentManifest } from "@typeagent/agent-sdk";
 import { createClientIORpcServer } from "@typeagent/dispatcher-rpc/clientio/server";
 import {
     createDispatcherRpcClient,
@@ -132,6 +134,23 @@ export type AgentServerConnection = {
      * isn't configured, so callers can hide/disable the mic affordance.
      */
     getSpeechToken(): Promise<SpeechToken | undefined>;
+    /**
+     * Register a client-hosted app agent with a joined conversation on the
+     * server. The agent's handlers run in this process; the server installs an
+     * rpc proxy that forwards calls back over the connection. Pass
+     * `conversationId` to target a specific joined conversation, or omit it when
+     * exactly one conversation is joined. Re-registering the same `name` (e.g.
+     * after {@link reconnect}) replaces the previous registration. Rejects if
+     * another client already registered `name` on the target conversation.
+     */
+    registerClientAgent(
+        name: string,
+        manifest: AppAgentManifest,
+        agent: AppAgent,
+        conversationId?: string,
+    ): Promise<void>;
+    /** Unregister an agent previously registered via registerClientAgent. */
+    unregisterClientAgent(name: string, conversationId?: string): Promise<void>;
     /**
      * Reopen the underlying transport and rebind the control rpc onto it,
      * reusing this connection object instead of building a new one. Returns
@@ -203,6 +222,11 @@ export function createAgentServerConnection(
         { dispatcher: Dispatcher; connectionId: string }
     >();
 
+    // Client-hosted agents registered on the server, name → agent-rpc server
+    // closeFn. Used to tear down the local rpc server when unregistering,
+    // re-registering, or closing the connection.
+    const clientAgentServers = new Map<string, () => void>();
+
     let closed = false;
 
     const connection: AgentServerConnection = {
@@ -338,6 +362,51 @@ export function createAgentServerConnection(
             return rpc.invoke("getSpeechToken");
         },
 
+        async registerClientAgent(
+            name: string,
+            manifest: AppAgentManifest,
+            agent: AppAgent,
+            conversationId?: string,
+        ): Promise<void> {
+            // Drop any previous rpc server for this name (e.g. re-registering
+            // after a reconnect, where the old server sat on a stale channel).
+            clientAgentServers.get(name)?.();
+            clientAgentServers.delete(name);
+
+            const { closeFn, agentInterface } = createAgentRpcServer(
+                name,
+                agent,
+                currentChannel,
+            );
+            try {
+                await rpc.invoke("registerClientAgent", {
+                    name,
+                    manifest,
+                    agentInterface,
+                    ...(conversationId !== undefined ? { conversationId } : {}),
+                });
+            } catch (e) {
+                closeFn();
+                throw e;
+            }
+            clientAgentServers.set(name, closeFn);
+        },
+
+        async unregisterClientAgent(
+            name: string,
+            conversationId?: string,
+        ): Promise<void> {
+            try {
+                await rpc.invoke("unregisterClientAgent", {
+                    name,
+                    ...(conversationId !== undefined ? { conversationId } : {}),
+                });
+            } finally {
+                clientAgentServers.get(name)?.();
+                clientAgentServers.delete(name);
+            }
+        },
+
         async reconnect(): Promise<boolean> {
             if (closed || reopenTransport === undefined) {
                 return false;
@@ -351,6 +420,12 @@ export function createAgentServerConnection(
             // The prior transport's per-conversation channels are gone; drop
             // our local bookkeeping so the caller re-joins cleanly.
             joinedConversations.clear();
+            // Client-agent rpc servers were bound to the old channel; drop them
+            // so the caller re-registers them on the new channel after re-join.
+            for (const closeFn of clientAgentServers.values()) {
+                closeFn();
+            }
+            clientAgentServers.clear();
             return true;
         },
 
@@ -360,6 +435,10 @@ export function createAgentServerConnection(
             }
             closed = true;
             debug("Closing agent server connection");
+            for (const closeFn of clientAgentServers.values()) {
+                closeFn();
+            }
+            clientAgentServers.clear();
             closeTransport();
         },
     };

ts/packages/agentServer/client/test/conversation-stubConnection.ts

@@ -206,6 +206,8 @@ export function makeStubConnection(
         async getSpeechToken() {
             return undefined;
         },
+        async registerClientAgent() {},
+        async unregisterClientAgent() {},
         async reconnect() {
             return true;
         },

ts/packages/agentServer/protocol/package.json

@@ -26,6 +26,8 @@
     "tsc": "tsc -b"
   },
   "dependencies": {
+    "@typeagent/agent-rpc": "workspace:*",
+    "@typeagent/agent-sdk": "workspace:*",
     "@typeagent/dispatcher-rpc": "workspace:*",
     "@typeagent/dispatcher-types": "workspace:*"
   },

ts/packages/agentServer/protocol/src/index.ts

@@ -18,6 +18,8 @@ export {
     ConversationNameCollisionBehavior,
     JoinConversationResult,
     RenameConversationOptions,
+    RegisterClientAgentParams,
+    UnregisterClientAgentParams,
     UserIdentity,
     DefaultUserIdentity,
     SpeechToken,

ts/packages/agentServer/protocol/src/protocol.ts

@@ -3,6 +3,8 @@
 
 import type { PendingInteractionRequest } from "@typeagent/dispatcher-types";
 import type { QueueSnapshot } from "@typeagent/dispatcher-types";
+import type { AppAgentManifest } from "@typeagent/agent-sdk";
+import type { AgentInterfaceFunctionName } from "@typeagent/agent-rpc/server";
 
 export type DispatcherConnectOptions = {
     filter?: boolean; // filter to message for own request. Default is false (no filtering)
@@ -124,6 +126,42 @@ export type AgentServerInvokeFunctions = {
      * mic affordance.
      */
     getSpeechToken: () => Promise<SpeechToken | undefined>;
+    /**
+     * Register a client-hosted app agent with a joined conversation's
+     * dispatcher. The agent's handlers run in the connecting client's process
+     * (over agent-rpc on the connection's channel provider); the server builds
+     * an rpc proxy from `agentInterface` and installs it as a dynamic agent.
+     * The agent is removed automatically when the connection drops or leaves
+     * the conversation.
+     *
+     * The client must create its agent-rpc server on the `agent:<name>`
+     * channel (via createAgentRpcServer over the connection channel provider)
+     * before calling this. Rejects if an agent with `name` is already
+     * registered on the target conversation (e.g. a second client trying to
+     * register the same singleton agent).
+     */
+    registerClientAgent: (param: RegisterClientAgentParams) => Promise<void>;
+    /** Unregister a previously registered client-hosted agent. */
+    unregisterClientAgent: (
+        param: UnregisterClientAgentParams,
+    ) => Promise<void>;
+};
+
+export type RegisterClientAgentParams = {
+    name: string;
+    manifest: AppAgentManifest;
+    agentInterface: AgentInterfaceFunctionName[];
+    /**
+     * Target conversation. If omitted, the server uses the connection's single
+     * joined conversation (and errors if the connection has joined none or
+     * more than one).
+     */
+    conversationId?: string;
+};
+
+export type UnregisterClientAgentParams = {
+    name: string;
+    conversationId?: string;
 };
 
 /**

ts/packages/agentServer/server/package.json

@@ -41,6 +41,7 @@
   "dependencies": {
     "@azure/identity": "^4.10.0",
     "@typeagent/agent-rpc": "workspace:*",
+    "@typeagent/agent-sdk": "workspace:*",
     "@typeagent/agent-server-client": "workspace:*",
     "@typeagent/agent-server-protocol": "workspace:*",
     "@typeagent/common-utils": "workspace:*",

ts/packages/agentServer/server/src/connectionHandler.ts

@@ -4,6 +4,7 @@
 import { createDispatcherRpcServer } from "@typeagent/dispatcher-rpc/dispatcher/server";
 import { createClientIORpcClient } from "@typeagent/dispatcher-rpc/clientio/client";
 import { createRpc } from "@typeagent/agent-rpc/rpc";
+import { createAgentRpcClient } from "@typeagent/agent-rpc/client";
 import type { ChannelProvider } from "@typeagent/agent-rpc/channel";
 import {
     AgentServerInvokeFunctions,
@@ -103,6 +104,39 @@ export function createAgentServerConnectionHandler(
             { dispatcher: Dispatcher; connectionId: string }
         >();
 
+        // Client-hosted agents this connection registered, per conversation.
+        // conversationId → set of agent names. Used to tear them down when the
+        // connection drops so they don't linger on the (longer-lived) shared
+        // dispatcher.
+        const clientAgents = new Map<string, Set<string>>();
+
+        // Resolve the conversation a client-agent operation targets. When no id
+        // is given, use the single joined conversation; error if there are zero
+        // or many so the caller must disambiguate.
+        const resolveClientAgentConversation = (
+            conversationId?: string,
+        ): string => {
+            if (conversationId !== undefined) {
+                if (!joinedConversations.has(conversationId)) {
+                    throw new Error(
+                        `Not joined to conversation: ${conversationId}`,
+                    );
+                }
+                return conversationId;
+            }
+            if (joinedConversations.size === 1) {
+                return joinedConversations.keys().next().value as string;
+            }
+            if (joinedConversations.size === 0) {
+                throw new Error(
+                    "Cannot register client agent: no conversation joined",
+                );
+            }
+            throw new Error(
+                "Cannot register client agent: multiple conversations joined; specify conversationId",
+            );
+        };
+
         // Warn this connection about a stale server build at most once, even
         // if it joins several conversations.
         let notifiedStale = false;
@@ -288,11 +322,71 @@ export function createAgentServerConnectionHandler(
             },
             getUserIdentity: async () => getUserIdentity(),
             getSpeechToken: async () => getSpeechToken(),
+            registerClientAgent: async (param) => {
+                const conversationId = resolveClientAgentConversation(
+                    param.conversationId,
+                );
+                const { name, manifest, agentInterface } = param;
+                if (clientAgents.get(conversationId)?.has(name)) {
+                    throw new Error(
+                        `Client agent '${name}' is already registered on conversation '${conversationId}'`,
+                    );
+                }
+                // Build the rpc proxy on the connection's own channel provider
+                // (the client hosts the real agent via createAgentRpcServer on
+                // the matching agent:<name> channel).
+                const appAgent = await createAgentRpcClient(
+                    name,
+                    channelProvider,
+                    agentInterface,
+                );
+                try {
+                    await conversationManager.addClientAgent(
+                        conversationId,
+                        name,
+                        manifest,
+                        appAgent,
+                    );
+                } catch (e) {
+                    channelProvider.deleteChannel(`agent:${name}`);
+                    throw e;
+                }
+                let set = clientAgents.get(conversationId);
+                if (set === undefined) {
+                    set = new Set();
+                    clientAgents.set(conversationId, set);
+                }
+                set.add(name);
+            },
+            unregisterClientAgent: async (param) => {
+                const conversationId = resolveClientAgentConversation(
+                    param.conversationId,
+                );
+                const { name } = param;
+                await conversationManager.removeClientAgent(
+                    conversationId,
+                    name,
+                );
+                channelProvider.deleteChannel(`agent:${name}`);
+                clientAgents.get(conversationId)?.delete(name);
+            },
         };
 
         // Clean up all conversations on disconnect
         channelProvider.on("disconnect", () => {
             onDisconnect?.();
+            // Remove client-hosted agents first so they don't linger on the
+            // shared dispatcher after this connection's socket is gone.
+            for (const [conversationId, names] of clientAgents.entries()) {
+                for (const name of names) {
+                    conversationManager
+                        .removeClientAgent(conversationId, name)
+                        .catch(() => {
+                            // Best effort on disconnect
+                        });
+                }
+            }
+            clientAgents.clear();
             for (const [
                 conversationId,
                 { connectionId },