feat(agent): add server-scoped session services

packages/agent/docs/plugins.md

@@ -1,6 +1,6 @@
 # Coding-Agent Application Hosts and Plugin Facets
 
-> **Status:** Tentative design input, not a normative contract or implementation handoff. The transport-neutral service token, provider, singleton `use()`, keyed `observe()`, and `RemoteState` substrate now have an experimental implementation; the application host contexts, plugin kernel, events, references, telemetry propagation, and most example facets remain illustrative. Reconcile this design with `rpc.md`, `telemetry.md`, and the final harness contract before adding it to `harness.md` or creating a work package.
+> **Status:** Tentative design input, not a normative contract or implementation handoff. The transport-neutral service token, provider, singleton `use()`, keyed `observe()`, and `RemoteState` substrate now have an experimental implementation, including `Models`, `SessionDirectory`, and `SessionManagement` vertical slices. The application host contexts, plugin kernel, events, references, telemetry propagation, and most other example facets remain illustrative. Reconcile this design with `rpc.md`, `telemetry.md`, and the final harness contract before adding it to `harness.md` or creating a work package.
 
 This document assumes you already understand `AgentHarness`, `AgentLane`, `Session`, `SessionTree`, `SessionRepo`, invocation `Context`, and telemetry. Read `rpc.md` for wire frames, remote references, subscriptions, and trace carriers, and `telemetry.md` for context propagation and cancellation semantics.
 

packages/agent/docs/rpc.md

@@ -65,7 +65,7 @@ The client maps `context.abortSignal` to cancellation of that one request. Disco
 
 ## Replicated state, events, and keyed instances
 
-`RemoteState` is authoritative latest-value replication, not event history, durable storage, a CRDT, or multi-writer state. A cold replica has `value === undefined`; subscribing before hydration records a listener without invoking it. Hydration installs a complete snapshot atomically before later updates are delivered, so there is no snapshot/update gap. Once hydrated, `value` is synchronous and subscribing reports the current value followed by later updates. The first snapshot callback uses a fresh hydration context; an already hydrated replica uses a fresh local delivery context rather than retaining the original write context. State values are detached JSON values.
+`RemoteState` is authoritative latest-value replication, not event history, durable storage, a CRDT, or multi-writer state. A cold replica has `value === undefined`; subscribing before hydration records a listener without invoking it. Hydration installs a complete snapshot atomically before later updates are delivered, so there is no snapshot/update gap. Once hydrated, `value` is synchronous and subscribing reports the current value followed by later updates. The first snapshot callback uses a fresh hydration context; an already hydrated replica uses a fresh local delivery context rather than retaining the original write context. State values are borrowed immutable JSON and are not defensively cloned; callers must not mutate or retain them.
 
 The providing host attaches source trace metadata to updates. The consumer reconstructs fresh delivery contexts; hydration uses a fresh context parented to the subscription. Disconnect may retain a value as stale display data. Reconnect replaces it from a complete snapshot, while a switch to a different provider clears readiness and disposes binding-specific resources. The precise subscription, buffering, sequence, acknowledgement, and flow-control frames are transport mechanics.
 

packages/client/README.md

@@ -33,11 +33,11 @@ await watch.dispose();
 
 The client verifies that the physical endpoint reports the expected logical `serverId`. Server-wide requests carry that ID, and every Session request carries the full live target `{ serverId, sessionId, attachmentId }`. The combined durable address prevents cross-server or cross-session misrouting; the server-generated attachment ID rejects delayed frames after switching or reattaching.
 
-`createSession()` creates a presentation-safe Session summary without attaching or opening a worker; pass `id` to request an exact ID or omit it to generate one. The server derives private fields such as the working directory from its authenticated workspace context. `attachSession()` returns `{ sessionId, attachmentId }` and updates the target used by subsequent Session operations. Multiple clients may attach to one Session. `promptSession()` accepts the protocol's serializable Harness prompt overloads and returns a structural `RunResult`.
+`createSession()` creates a presentation-safe Session summary without attaching or opening a worker; pass `id` to request an exact ID or omit it to generate one. The server derives private fields such as the working directory from its authenticated workspace context. `attachSession()` is a compatibility helper: the management call returns no routing data, while an ordered out-of-band attachment message updates the target and lets the helper return `{ sessionId, attachmentId }`. Multiple clients may attach to one Session. `promptSession()` accepts the protocol's serializable Harness prompt overloads and returns a structural `RunResult`.
 
 `watchSession()` creates a main-lane watch and returns its authoritative snapshot without starting event delivery. Install the listener with `await watch.start(listener)`; this then flushes events buffered after the snapshot and continues with live events while `promptSession()` is pending. `watch.dispose()` stops server delivery and waits for already-received listener work. A disconnected watch is stale and cannot be reused after reconnection.
 
-Plugin presentation hosts use `request()` and `subscribeService()` as low-level transport adapters. A service subscription returns a complete provider snapshot; the host installs it and then calls `start()` to release updates buffered during hydration. `PiClient` deliberately does not construct typed service proxies or interpret plugin contracts.
+Plugin presentation hosts use `request()` and `subscribeService()` as low-level transport adapters for both the connected server and selected Session namespaces. A service subscription returns a complete provider snapshot; the host installs it and then calls `start()` to release updates buffered during hydration. `PiClient` applies out-of-band attachment changes but deliberately does not construct typed service proxies or interpret plugin contracts.
 
 On disconnect or disposal, pending prompts reject locally, but accepted work may still complete remotely before the attachment is released. The client clears its live attachment route. It never reconnects or replays requests automatically. After disconnection, call `reconnect()`, attach again, and explicitly repeat only operations known to be safe.
 

packages/client/src/client.ts

@@ -1,4 +1,5 @@
 import {
+	type AttachmentEnvelope,
 	createRpcClient,
 	createServiceSubscribeCall,
 	createServiceUnsubscribeCall,
@@ -109,6 +110,10 @@ export class PiClient {
 		return this.#connection.state === "connected";
 	}
 
+	get serverId(): string {
+		return this.#options.serverId;
+	}
+
 	get hello(): ServerHello | undefined {
 		return this.#hello;
 	}
@@ -172,9 +177,31 @@ export class PiClient {
 		if (typeof session !== "string" && session.serverId !== this.#options.serverId) {
 			throw new PiServerError({ code: "wrong_server", message: "Session belongs to another server" });
 		}
-		const attached = await this.#rpc.attach(sessionId);
-		this.#setAttachment({ serverId: this.#options.serverId, ...attached });
-		return attached;
+		const result = await this.#request(
+			{ serverId: this.#options.serverId },
+			{ serviceId: "session-management", member: "attach", args: [sessionId] },
+		);
+		if (
+			typeof result === "object" &&
+			result !== null &&
+			"sessionId" in result &&
+			"attachmentId" in result &&
+			typeof result.sessionId === "string" &&
+			typeof result.attachmentId === "string"
+		) {
+			this.#setAttachment({
+				serverId: this.#options.serverId,
+				sessionId: result.sessionId,
+				attachmentId: result.attachmentId,
+			});
+		}
+		const attachment = this.#attachment;
+		if (attachment?.sessionId !== sessionId) {
+			const error = new ProtocolValidationError("Session attach completed without a matching attachment route");
+			this.#connection.fail(error);
+			throw error;
+		}
+		return { sessionId, attachmentId: attachment.attachmentId };
 	}
 
 	promptSession(sessionId: string, text: string, images?: PromptImage[]): Promise<ServiceRpcResult<"prompt">>;
@@ -350,7 +377,15 @@ export class PiClient {
 		return promise;
 	}
 
-	#handleMessage(message: ResponseEnvelope | ServerEventEnvelope): void {
+	#handleMessage(message: ResponseEnvelope | ServerEventEnvelope | AttachmentEnvelope): void {
+		if (message.type === "attachment") {
+			if (message.attachment !== null && message.attachment.serverId !== this.#options.serverId) {
+				this.#connection.fail(new ProtocolValidationError("Attachment update belongs to another server"));
+				return;
+			}
+			this.#setAttachment(message.attachment ?? undefined);
+			return;
+		}
 		if (message.type === "event") {
 			const active = this.#watchListeners.get(message.watchId);
 			if (active === undefined) return;

packages/client/test/client.test.ts

@@ -46,6 +46,32 @@ describe("PiClient service operations", () => {
 		await client.dispose();
 	});
 
+	test("updates attachment state from out-of-band server routing", async () => {
+		const server = new MemoryByteServer();
+		const client = await connectClient(server);
+		const changes: Array<string | undefined> = [];
+		client.onAttachmentChange((attachment) => changes.push(attachment?.sessionId));
+
+		const attaching = client.attachSession("session-1");
+		await server.waitForMessages(2);
+		server.send({
+			type: "attachment",
+			attachment: {
+				serverId: "00000000-0000-4000-8000-000000000001",
+				sessionId: "session-1",
+				attachmentId: "attachment-1",
+			},
+		});
+		server.send({ type: "response", id: "request-1", ok: true });
+		await expect(attaching).resolves.toEqual({ sessionId: "session-1", attachmentId: "attachment-1" });
+		expect(client.attachment).toMatchObject({ sessionId: "session-1", attachmentId: "attachment-1" });
+
+		server.send({ type: "attachment", attachment: null });
+		expect(client.attachment).toBeUndefined();
+		expect(changes).toEqual(["session-1", undefined]);
+		await client.dispose();
+	});
+
 	test("addresses Session operations to the configured server", async () => {
 		const server = new MemoryByteServer();
 		const client = await connectClient(server);

packages/coding-agent/src/experimental/client.ts

@@ -1,9 +1,12 @@
 import { basename } from "node:path";
+import { BACKGROUND_CONTEXT, type RemoteServiceNamespace, type RemoteState } from "@earendil-works/pi-agent-core";
 import { PiClient } from "@earendil-works/pi-client";
 import { createUnixTransportFactory, discoverUnixServers, type UnixServerRoute } from "@earendil-works/pi-client/unix";
 import { isServerId, type LaneEvent, type SessionAddress } from "@earendil-works/pi-protocol";
 import type { ClientCommand } from "../cli/experimental/commands/client.ts";
 import { activateServer, ENV_SERVER_ID, resolveServerDirectory, resolveSessionDirectory } from "./server.ts";
+import { createPiServerServiceNamespace } from "./services/connection.ts";
+import { SessionDirectory, SessionManagement, type SessionManagementService } from "./services/sessions.ts";
 
 export type ClientResult =
 	| {
@@ -53,8 +56,14 @@ export async function runClient(command: ClientCommand, options: RunClientOption
 		}
 	}
 	const openedClients = new Set<PiClient>();
+	const serviceNamespaces = new Set<RemoteServiceNamespace>();
 	if (activatedClient) openedClients.add(activatedClient);
-	const discovered: { route: UnixServerRoute; sessionIds: string[]; client: PiClient }[] = [];
+	const discovered: {
+		route: UnixServerRoute;
+		sessionIds: string[];
+		client: PiClient;
+		management: SessionManagementService;
+	}[] = [];
 
 	try {
 		for (const route of routes) {
@@ -66,8 +75,14 @@ export async function runClient(command: ClientCommand, options: RunClientOption
 				}));
 			activatedClient = undefined;
 			openedClients.add(client);
-			const sessions = await client.listSessions();
-			discovered.push({ route, sessionIds: sessions.map(({ sessionId }) => sessionId), client });
+			const services = createPiServerServiceNamespace(client, {
+				services: [SessionDirectory, SessionManagement],
+			});
+			serviceNamespaces.add(services);
+			const directory = services.use(SessionDirectory);
+			const management = services.use(SessionManagement);
+			const state = await waitForState(directory.state);
+			discovered.push({ route, sessionIds: state.sessions.map(({ sessionId }) => sessionId), client, management });
 		}
 
 		let sessionId = command.sessionId;
@@ -91,7 +106,7 @@ export async function runClient(command: ClientCommand, options: RunClientOption
 				throw new Error("Client prompt requires exactly one discovered server to create a Session");
 			}
 			match = discovered[0]!;
-			sessionId = (await match.client.createSession({})).sessionId;
+			sessionId = (await match.management.create({}, BACKGROUND_CONTEXT)).sessionId;
 		} else {
 			const selectedSessionId = sessionId;
 			const matches = discovered.filter((candidate) => candidate.sessionIds.includes(selectedSessionId));
@@ -106,10 +121,10 @@ export async function runClient(command: ClientCommand, options: RunClientOption
 					throw new Error(`No discovered server contains session ${selectedSessionId}`);
 				}
 				match = discovered[0]!;
-				await match.client.createSession({ id: selectedSessionId });
+				await match.management.create({ id: selectedSessionId }, BACKGROUND_CONTEXT);
 			}
 		}
-		await match.client.attachSession({ serverId: match.route.serverId, sessionId });
+		await match.management.attach(sessionId, BACKGROUND_CONTEXT);
 		if (command.prompt === undefined) {
 			return { kind: "attached", serverId: match.route.serverId, sessionId };
 		}
@@ -133,10 +148,22 @@ export async function runClient(command: ClientCommand, options: RunClientOption
 			await watch?.dispose();
 		}
 	} finally {
+		await Promise.all([...serviceNamespaces].map((services) => services.dispose(BACKGROUND_CONTEXT)));
 		await Promise.all([...openedClients].map((client) => client.dispose()));
 	}
 }
 
+function waitForState<T>(state: RemoteState<T>): Promise<T> {
+	if (state.value !== undefined) return Promise.resolve(state.value);
+	return new Promise((resolve) => {
+		let unsubscribe: (() => void) | undefined;
+		unsubscribe = state.subscribe((value) => {
+			queueMicrotask(() => unsubscribe?.());
+			resolve(value);
+		});
+	});
+}
+
 function routeFromExplicitPath(path: string): UnixServerRoute {
 	const name = basename(path);
 	const serverId = name.endsWith(".sock") ? name.slice(0, -".sock".length) : "";

packages/coding-agent/src/experimental/server.ts

@@ -3,18 +3,29 @@ import { chmod, lstat, mkdir, readFile, writeFile } from "node:fs/promises";
 import { homedir } from "node:os";
 import { isAbsolute, join, resolve } from "node:path";
 import { fileURLToPath } from "node:url";
-import { type JsonlSessionMetadata, JsonlSessionRepo, TODO_CONTEXT } from "@earendil-works/pi-agent-core";
+import {
+	BACKGROUND_CONTEXT,
+	type JsonlSessionMetadata,
+	JsonlSessionRepo,
+	TODO_CONTEXT,
+} from "@earendil-works/pi-agent-core";
 import { NodeExecutionEnv } from "@earendil-works/pi-agent-core/node";
 import { PiClient, PiDisconnectedError, PiServerError } from "@earendil-works/pi-client";
 import { createUnixTransportFactory, type UnixServerRoute } from "@earendil-works/pi-client/unix";
-import { isServerId, type ServerId } from "@earendil-works/pi-protocol";
-import type { PiServer, PiServerHost } from "@earendil-works/pi-server";
+import { isServerId, type ServerId, type SessionSummary } from "@earendil-works/pi-protocol";
+import {
+	type PiServer,
+	type PiServerHost,
+	SessionAmbiguousError,
+	SessionNotFoundError,
+} from "@earendil-works/pi-server";
 import { createUnixServer, getUnixSocketPath } from "@earendil-works/pi-server/unix";
 import lockfile from "proper-lockfile";
 import { getAgentDir } from "../config.ts";
 import { resolvePath } from "../utils/paths.ts";
 import { CoordinatorConnection, type CoordinatorStartupLease, ensureCoordinator } from "./coordinator.ts";
 import { consumeInternalProcessRole, spawnInternalProcess, terminateInternalProcess } from "./process.ts";
+import { createExperimentalServerServices } from "./services/server.ts";
 import { SessionWorkerManager } from "./session-worker-manager.ts";
 
 export const ENV_SERVER_DIR = "PI_SERVER_DIR";
@@ -320,38 +331,72 @@ interface StartServerBackendOptions {
 	readonly sessionDir?: string;
 }
 
+interface RunningServerBackend extends RunningServer {
+	refreshSessions(): Promise<void>;
+}
+
 async function startServerBackend(
 	options: StartServerBackendOptions,
 	workers: SessionWorkerManager,
 	onConnectionCountChanged?: (count: number) => void,
-): Promise<RunningServer> {
+): Promise<RunningServerBackend> {
 	const serverId = options.serverId;
 	const sessionDir = resolveSessionDirectory(options.sessionDir);
 	const executionEnv = new NodeExecutionEnv({ cwd: process.cwd() });
 	const repo = new JsonlSessionRepo({ fileSystem: executionEnv, sessionsRoot: sessionDir });
-	const host: PiServerHost<JsonlSessionMetadata> = {
-		sessions: {
-			list: async (context) => {
-				const sessions = new Map(
-					(await repo.list(undefined, context)).map((metadata) => [metadata.path, metadata]),
-				);
-				for (const metadata of workers.trackedSessions) sessions.set(metadata.path, metadata);
-				return [...sessions.values()];
-			},
-			create: async (createOptions, context) => {
-				const session = await repo.create({ ...createOptions, cwd: process.cwd() }, context);
-				try {
-					return session.metadata;
-				} finally {
-					await session.close(context);
-				}
-			},
+	const listSessions: PiServerHost<JsonlSessionMetadata>["sessions"]["list"] = async (context) => {
+		const sessions = new Map((await repo.list(undefined, context)).map((metadata) => [metadata.path, metadata]));
+		for (const metadata of workers.trackedSessions) sessions.set(metadata.path, metadata);
+		return [...sessions.values()];
+	};
+	const createSession: PiServerHost<JsonlSessionMetadata>["sessions"]["create"] = async (createOptions, context) => {
+		const session = await repo.create({ ...createOptions, cwd: process.cwd() }, context);
+		try {
+			return session.metadata;
+		} finally {
+			await session.close(context);
+		}
+	};
+	const summarize = (metadata: JsonlSessionMetadata): SessionSummary => ({
+		serverId,
+		sessionId: metadata.id,
+		createdAt: metadata.createdAt,
+	});
+	const closeStorage = async (): Promise<void> => {
+		const cleanup = await Promise.allSettled([repo.close(TODO_CONTEXT), executionEnv.cleanup(TODO_CONTEXT)]);
+		const errors = cleanup.flatMap((result) => (result.status === "rejected" ? [result.reason] : []));
+		if (errors.length === 1) throw errors[0];
+		if (errors.length > 1) throw new AggregateError(errors, "Experimental session storage cleanup failed");
+	};
+	const serverServices = await createExperimentalServerServices({
+		list: async (context) =>
+			(await listSessions(context))
+				.map(summarize)
+				.sort((left, right) => left.sessionId.localeCompare(right.sessionId) || left.createdAt - right.createdAt),
+		create: async (createOptions, context) => summarize(await createSession(createOptions, context)),
+		remove: async (sessionId, context) => {
+			const matches = (await listSessions(context)).filter((metadata) => metadata.id === sessionId);
+			if (matches.length === 0) throw new SessionNotFoundError(`Unknown session: ${sessionId}`);
+			if (matches.length > 1) throw new SessionAmbiguousError();
+			await workers.closeSession(matches[0]!, context);
+			await repo.delete(matches[0]!, context);
 		},
+	}).catch(async (error: unknown) => {
+		try {
+			await closeStorage();
+		} catch (cleanupError) {
+			throw new AggregateError([error, cleanupError], "Server service startup and cleanup failed");
+		}
+		throw error;
+	});
+	const host: PiServerHost<JsonlSessionMetadata> = {
+		serverServices: serverServices.host,
+		sessions: { list: listSessions, create: createSession },
 		openSession: (metadata, context) => workers.openSession(metadata, context),
 	};
 	const socketPath = options.path;
 	const closeCatalog = async (): Promise<void> => {
-		const cleanup = await Promise.allSettled([repo.close(TODO_CONTEXT), executionEnv.cleanup(TODO_CONTEXT)]);
+		const cleanup = await Promise.allSettled([serverServices.dispose(), closeStorage()]);
 		const errors = cleanup.flatMap((result) => (result.status === "rejected" ? [result.reason] : []));
 		if (errors.length === 1) throw errors[0];
 		if (errors.length > 1) throw new AggregateError(errors, "Experimental session catalog cleanup failed");
@@ -392,6 +437,7 @@ async function startServerBackend(
 		server,
 		workerPids: workers.workerPids,
 		closed,
+		refreshSessions: () => serverServices.refresh(BACKGROUND_CONTEXT),
 		close() {
 			closePromise ??= server.close().then(
 				() => closed,
@@ -414,7 +460,7 @@ export async function startServer(options: StartServerOptions = {}): Promise<Run
 	const directory = resolveServerDirectory(options.directory);
 	const { serverId, release } = await acquireServerProfile(directory, options.serverId ?? process.env[ENV_SERVER_ID]);
 	const lifetime = new ServerLifetime(options.keepAlive ?? true);
-	let backend: RunningServer | undefined;
+	let backend: RunningServerBackend | undefined;
 	let coordinator: CoordinatorConnection | undefined;
 	let startupLease: CoordinatorStartupLease | undefined;
 	let workers: SessionWorkerManager | undefined;
@@ -440,6 +486,7 @@ export async function startServer(options: StartServerOptions = {}): Promise<Run
 		startupLease.close();
 		startupLease = undefined;
 		await workers.discover(coordinator.peerIds);
+		await backend.refreshSessions();
 
 		const activeBackend = backend;
 		const activeCoordinator = coordinator;

packages/coding-agent/src/experimental/services/connection.ts

@@ -8,19 +8,46 @@ import {
 	RemoteServiceNamespace,
 } from "@earendil-works/pi-agent-core";
 import type { PiClient } from "@earendil-works/pi-client";
-import type { ProtocolRpcCall, ServiceProviderUpdate } from "@earendil-works/pi-protocol";
+import type { ProtocolRpcCall, RpcTarget, ServiceProviderUpdate } from "@earendil-works/pi-protocol";
 
-export interface PiSessionServiceNamespaceOptions {
+export interface PiServiceNamespaceOptions {
 	readonly services: readonly { readonly id: string }[];
 	readonly onError?: (error: Error) => void;
 }
 
+class PiServerServiceNamespace extends RemoteServiceNamespace {
+	readonly #removeConnectionListener: () => void;
+	readonly #onError: ((error: Error) => void) | undefined;
+	#transition = Promise.resolve();
+
+	constructor(client: PiClient, connection: RemoteServiceConnection, options: PiServiceNamespaceOptions) {
+		super({
+			services: options.services,
+			connection,
+			bound: client.connected,
+			...(options.onError === undefined ? {} : { onError: options.onError }),
+		});
+		this.#onError = options.onError;
+		this.#removeConnectionListener = client.onConnectionStateChange(({ state }) => {
+			this.#transition = this.#transition
+				.then(() => this.rebind(state === "connected", BACKGROUND_CONTEXT))
+				.catch((error: unknown) => this.#onError?.(toError(error)));
+		});
+	}
+
+	override async dispose(context: Context): Promise<void> {
+		this.#removeConnectionListener();
+		await this.#transition;
+		await super.dispose(context);
+	}
+}
+
 class PiSessionServiceNamespace extends RemoteServiceNamespace {
 	readonly #removeAttachmentListener: () => void;
 	readonly #onError: ((error: Error) => void) | undefined;
 	#transition = Promise.resolve();
 
-	constructor(client: PiClient, connection: RemoteServiceConnection, options: PiSessionServiceNamespaceOptions) {
+	constructor(client: PiClient, connection: RemoteServiceConnection, options: PiServiceNamespaceOptions) {
 		super({
 			services: options.services,
 			connection,
@@ -42,15 +69,29 @@ class PiSessionServiceNamespace extends RemoteServiceNamespace {
 	}
 }
 
+/** Bind plugin service facades to the connected server route. */
+export function createPiServerServiceNamespace(
+	client: PiClient,
+	options: PiServiceNamespaceOptions,
+): RemoteServiceNamespace {
+	const connection = createServiceConnection(client, () => ({ serverId: client.serverId }));
+	return new PiServerServiceNamespace(client, connection, options);
+}
+
 /** Bind plugin service facades to the Pi client's selected Session route. */
 export function createPiSessionServiceNamespace(
 	client: PiClient,
-	options: PiSessionServiceNamespaceOptions,
+	options: PiServiceNamespaceOptions,
 ): RemoteServiceNamespace {
-	const connection: RemoteServiceConnection = {
+	const connection = createServiceConnection(client, () => client.attachment);
+	return new PiSessionServiceNamespace(client, connection, options);
+}
+
+function createServiceConnection(client: PiClient, getTarget: () => RpcTarget | undefined): RemoteServiceConnection {
+	return {
 		invoke: async (call, context) => {
-			const target = client.attachment;
-			if (target === undefined) throw new Error("No Session is attached");
+			const target = getTarget();
+			if (target === undefined) throw new Error("Service namespace is not routed");
 			const wireCall: ProtocolRpcCall = {
 				serviceId: call.serviceId,
 				...(call.instance === undefined ? {} : { instance: call.instance }),
@@ -62,8 +103,8 @@ export function createPiSessionServiceNamespace(
 			return result;
 		},
 		subscribe: async (serviceId, mode, listener, context) => {
-			const target = client.attachment;
-			if (target === undefined) throw new Error("No Session is attached");
+			const target = getTarget();
+			if (target === undefined) throw new Error("Service namespace is not routed");
 			const subscription = await client.subscribeService(
 				target,
 				serviceId,
@@ -79,7 +120,6 @@ export function createPiSessionServiceNamespace(
 			};
 		},
 	};
-	return new PiSessionServiceNamespace(client, connection, options);
 }
 
 function toError(error: unknown): Error {

packages/coding-agent/src/experimental/services/server.ts

@@ -0,0 +1,152 @@
+import {
+	BACKGROUND_CONTEXT,
+	type Context,
+	type ServiceProviderUpdate as CoreServiceProviderUpdate,
+	RemoteServiceProvider,
+	remoteState,
+	type ServiceProviderSubscription,
+} from "@earendil-works/pi-agent-core";
+import {
+	decodeServiceControlCall,
+	type JsonValue,
+	JsonValueSchema,
+	type ServiceProviderUpdate as ProtocolServiceProviderUpdate,
+	ServiceProviderUpdateSchema,
+	type SessionCreateOptions,
+	type SessionSummary,
+} from "@earendil-works/pi-protocol";
+import type { RoutedServerServiceAttachment, RoutedServerServiceHost } from "@earendil-works/pi-server";
+import { Check } from "typebox/value";
+import { SessionDirectory, type SessionDirectoryState, SessionManagement } from "./sessions.ts";
+
+export interface ExperimentalServerServices {
+	readonly host: RoutedServerServiceHost;
+	refresh(context?: Context): Promise<void>;
+	dispose(): Promise<void>;
+}
+
+export async function createExperimentalServerServices(options: {
+	list(context: Context): Promise<SessionSummary[]>;
+	create(createOptions: SessionCreateOptions, context: Context): Promise<SessionSummary>;
+	remove(sessionId: string, context: Context): Promise<void>;
+}): Promise<ExperimentalServerServices> {
+	let revision = 1;
+	const directory = remoteState<SessionDirectoryState>({
+		revision,
+		sessions: await options.list(BACKGROUND_CONTEXT),
+	});
+	const attachments = new Set<RoutedServerServiceAttachment>();
+	let mutationTail = Promise.resolve();
+
+	const refreshNow = async (context: Context): Promise<void> => {
+		revision += 1;
+		directory.set({ revision, sessions: await options.list(context) }, context);
+	};
+	const serialize = <T>(operation: () => Promise<T>): Promise<T> => {
+		const result = mutationTail.catch(() => {}).then(operation);
+		mutationTail = result.then(
+			() => undefined,
+			() => undefined,
+		);
+		return result;
+	};
+
+	return {
+		host: {
+			attachClient(presentation) {
+				const provider = new RemoteServiceProvider([SessionDirectory, SessionManagement]);
+				provider.provide(SessionDirectory, { state: directory });
+				provider.provide(SessionManagement, {
+					create: (createOptions, context) =>
+						serialize(async () => {
+							const created = await options.create(createOptions, context);
+							await refreshNow(context);
+							return created;
+						}),
+					remove: (sessionId, context) =>
+						serialize(async () => {
+							await presentation.prepareSessionRemoval(sessionId, context);
+							await options.remove(sessionId, context);
+							await refreshNow(context);
+						}),
+					attach: (sessionId, context) =>
+						serialize(async () => {
+							await presentation.attachSession(sessionId, context);
+						}),
+					detach: (context) =>
+						serialize(async () => {
+							await presentation.detachSession(context);
+						}),
+				});
+				const attachment = createProviderAttachment(provider, () => attachments.delete(attachment));
+				attachments.add(attachment);
+				return attachment;
+			},
+		},
+		refresh: (context = BACKGROUND_CONTEXT) => serialize(() => refreshNow(context)),
+		async dispose() {
+			const releases = await Promise.allSettled(
+				[...attachments].map((attachment) => attachment.release(BACKGROUND_CONTEXT)),
+			);
+			attachments.clear();
+			await mutationTail;
+			const errors = releases.flatMap((result) => (result.status === "rejected" ? [result.reason] : []));
+			if (errors.length === 1) throw errors[0];
+			if (errors.length > 1) throw new AggregateError(errors, "Failed to release server service attachments");
+		},
+	};
+}
+
+function createProviderAttachment(
+	provider: RemoteServiceProvider,
+	onRelease: () => void,
+): RoutedServerServiceAttachment {
+	const subscriptions = new Map<string, ServiceProviderSubscription>();
+	let released = false;
+	return {
+		async invokeService(call, publish, context) {
+			if (released) throw new Error("Server service attachment is released");
+			const control = decodeServiceControlCall(call);
+			if (control?.type === "subscribe") {
+				if (subscriptions.has(control.subscriptionId)) {
+					throw new Error("Service subscription ID is already active");
+				}
+				const subscription = provider.subscribe(control.serviceId, control.mode, (update, updateContext) => {
+					void Promise.resolve(
+						publish(control.subscriptionId, toProtocolServiceUpdate(update), updateContext),
+					).catch(() => {});
+				});
+				subscriptions.set(control.subscriptionId, subscription);
+				subscription.activate();
+				return toProtocolJson(subscription.snapshot);
+			}
+			if (control?.type === "unsubscribe") {
+				const subscription = subscriptions.get(control.subscriptionId);
+				if (subscription === undefined) throw new Error("Service subscription was not found");
+				subscription.close();
+				subscriptions.delete(control.subscriptionId);
+				return undefined;
+			}
+			return provider.invoke(call, context);
+		},
+		release() {
+			if (released) return;
+			released = true;
+			for (const subscription of subscriptions.values()) subscription.close();
+			subscriptions.clear();
+			provider.dispose();
+			onRelease();
+		},
+	};
+}
+
+function toProtocolJson(value: unknown): JsonValue {
+	if (!Check(JsonValueSchema, value)) throw new Error("Service control value is not strict JSON");
+	return value;
+}
+
+function toProtocolServiceUpdate(update: CoreServiceProviderUpdate): ProtocolServiceProviderUpdate {
+	const candidate: unknown = update;
+	if (!Check(ServiceProviderUpdateSchema, candidate)) throw new Error("Service produced an invalid update");
+	return candidate;
+}

packages/coding-agent/src/experimental/services/sessions.ts

@@ -0,0 +1,22 @@
+import { type Context, defineRemoteService, type RemoteState } from "@earendil-works/pi-agent-core";
+import type { SessionCreateOptions, SessionSummary } from "@earendil-works/pi-protocol";
+
+export interface SessionDirectoryState {
+	revision: number;
+	sessions: SessionSummary[];
+}
+
+export interface SessionDirectoryService {
+	readonly state: RemoteState<SessionDirectoryState>;
+}
+
+export const SessionDirectory = defineRemoteService<SessionDirectoryService>("session-directory");
+
+export interface SessionManagementService {
+	create(options: SessionCreateOptions, context: Context): Promise<SessionSummary>;
+	remove(sessionId: string, context: Context): Promise<void>;
+	attach(sessionId: string, context: Context): Promise<void>;
+	detach(context: Context): Promise<void>;
+}
+
+export const SessionManagement = defineRemoteService<SessionManagementService>("session-management");

packages/coding-agent/src/experimental/session-worker-manager.ts

@@ -177,6 +177,11 @@ export class SessionWorkerManager {
 		return this.#routedHandle(await this.#launch(metadata, context));
 	}
 
+	async closeSession(metadata: JsonlSessionMetadata, context: Context): Promise<void> {
+		const worker = this.#workersBySession.get(metadata.path) ?? (await this.#pending.get(metadata.path)?.promise);
+		if (worker !== undefined) await this.#stopWorker(worker, context);
+	}
+
 	#routedHandle(worker: WorkerRecord): RoutedSessionHandle {
 		return {
 			terminated: worker.terminated,

packages/coding-agent/test/experimental-remote-runtime.test.ts

@@ -7,8 +7,12 @@ import { afterEach, beforeEach, describe, expect, test, vi } from "vitest";
 import { runClient } from "../src/experimental/client.ts";
 import * as processRuntime from "../src/experimental/process.ts";
 import { activateServer, type RunningServer, startServer } from "../src/experimental/server.ts";
-import { createPiSessionServiceNamespace } from "../src/experimental/services/connection.ts";
+import {
+	createPiServerServiceNamespace,
+	createPiSessionServiceNamespace,
+} from "../src/experimental/services/connection.ts";
 import { Models } from "../src/experimental/services/models.ts";
+import { SessionDirectory, SessionManagement } from "../src/experimental/services/sessions.ts";
 import {
 	configureExperimentalWorkerModel,
 	createExperimentalSessions,
@@ -277,6 +281,61 @@ describe("experimental durable server composition", () => {
 		expect(socket.mode & 0o777).toBe(0o600);
 	});
 
+	test("hydrates and mutates server Session services across framed clients", async () => {
+		const { runtime } = await makeServer();
+		const firstClient = await PiClient.connect({
+			serverId: runtime.serverId,
+			transportFactory: createUnixTransportFactory({ path: runtime.socketPath }),
+		});
+		const secondClient = await PiClient.connect({
+			serverId: runtime.serverId,
+			transportFactory: createUnixTransportFactory({ path: runtime.socketPath }),
+		});
+		clients.add(firstClient);
+		clients.add(secondClient);
+		const errors: Error[] = [];
+		const firstServices = createPiServerServiceNamespace(firstClient, {
+			services: [SessionDirectory, SessionManagement],
+			onError: (error) => errors.push(error),
+		});
+		const secondServices = createPiServerServiceNamespace(secondClient, {
+			services: [SessionDirectory, SessionManagement],
+			onError: (error) => errors.push(error),
+		});
+		const firstDirectory = firstServices.use(SessionDirectory);
+		const secondDirectory = secondServices.use(SessionDirectory);
+		const firstManagement = firstServices.use(SessionManagement);
+		const secondManagement = secondServices.use(SessionManagement);
+
+		await vi.waitFor(() => {
+			expect(firstDirectory.state.value?.sessions.map(({ sessionId }) => sessionId)).toEqual(["demo-1", "demo-2"]);
+			expect(secondDirectory.state.value).toEqual(firstDirectory.state.value);
+		});
+		await firstManagement.create({ id: "demo-3" }, BACKGROUND_CONTEXT);
+		await vi.waitFor(() => {
+			expect(firstDirectory.state.value?.sessions.map(({ sessionId }) => sessionId)).toContain("demo-3");
+			expect(secondDirectory.state.value).toEqual(firstDirectory.state.value);
+		});
+
+		await Promise.all([
+			firstManagement.attach("demo-1", BACKGROUND_CONTEXT),
+			secondManagement.attach("demo-1", BACKGROUND_CONTEXT),
+		]);
+		expect(firstClient.attachment?.sessionId).toBe("demo-1");
+		expect(secondClient.attachment?.sessionId).toBe("demo-1");
+		await firstManagement.remove("demo-1", BACKGROUND_CONTEXT);
+		await vi.waitFor(() => {
+			expect(firstClient.attachment).toBeUndefined();
+			expect(secondClient.attachment).toBeUndefined();
+			expect(firstDirectory.state.value?.sessions.map(({ sessionId }) => sessionId)).not.toContain("demo-1");
+			expect(secondDirectory.state.value).toEqual(firstDirectory.state.value);
+		});
+		await secondManagement.detach(BACKGROUND_CONTEXT);
+		expect(errors).toEqual([]);
+
+		await Promise.all([firstServices.dispose(BACKGROUND_CONTEXT), secondServices.dispose(BACKGROUND_CONTEXT)]);
+	});
+
 	test("shares one worker across concurrent Session attachments", async () => {
 		const { runtime } = await makeServer();
 		const client = await attachClient(runtime, "demo-1");

packages/protocol/README.md

@@ -2,19 +2,19 @@
 
 Runtime-neutral schemas, types, CBOR encoding, and byte-stream framing for the experimental Pi protocol.
 
-Protocol version `3` adds the first plugin-service slice:
+Protocol version `3` carries the current server- and Session-scoped plugin-service slice:
 
 - a version handshake that identifies the logical `serverId`;
 - explicit server and Session request targets;
 - contract-agnostic service calls with optional keyed-instance identity;
 - singleton and keyed-service subscription snapshots plus ordered state, spawn, and close updates;
-- correlated responses, request cancellation, attachment-scoped events, and bounded protocol errors.
+- correlated responses, request cancellation, out-of-band attachment updates, attachment-scoped events, and bounded protocol errors.
 
-The transport carries `{ serviceId, instance?, member, args }` calls. A server target contains `{ serverId }`; a Session target contains `{ serverId, sessionId, attachmentId }`. Keyed addresses contain a plugin key and provider-owned generation so delayed calls cannot reach a replacement instance. Service subscriptions return a complete member/state snapshot before update delivery starts. `list()` and `create()` return presentation-safe `SessionSummary` values containing no working directory, owner, storage, or parent metadata. The server derives private creation fields from authenticated workspace context. `attach()` creates or selects one presentation attachment and returns its Session and attachment IDs. `prompt()` targets that exact live attachment. The real `Session` and `AgentHarness` remain process-local. Disconnecting releases only that presentation's attachment after admitted work settles.
+The transport carries `{ serviceId, instance?, member, args }` calls. A server target contains `{ serverId }`; a Session target contains `{ serverId, sessionId, attachmentId }`. Keyed addresses contain a plugin key and provider-owned generation so delayed calls cannot reach a replacement instance. Service subscriptions return a complete member/state snapshot before update delivery starts. Session-directory state and management results contain no working directory, owner, storage, or parent metadata. The server derives private creation fields from authenticated workspace context. Management `attach()` and `detach()` return no routing identifiers; the server publishes the selected live route in an out-of-band `attachment` message. `prompt()` targets that exact route. The real `Session` and `AgentHarness` remain process-local. Disconnecting releases only that presentation's attachment after admitted work settles.
 
 `PromptArguments` contains one serializable `AgentLane.prompt()` overload. `PromptMessage` is the protocol's closed set of built-in message DTOs; application-defined `AgentMessage` extensions are not accepted implicitly. `RunResult` is the wire-safe structural equivalent of the Harness result and contains no JavaScript `Error` instances.
 
-The temporary `ServiceRpc` manifest still backs discovery, attachment, prompting, and transcript compatibility methods while those features migrate to plugin services. Session plugin calls outside that facade are routed opaquely to the attached worker, where the provider validates and invokes them.
+The temporary `ServiceRpc` manifest still backs client compatibility methods and transcript operations while those features migrate to plugin services. Experimental presentations consume server-owned `SessionDirectory` and `SessionManagement` services; Session plugin calls route opaquely to the attached worker, where the provider validates and invokes them.
 
 A lane watch uses three RPCs so snapshot ordering does not depend on response/event scheduling: `watch()` creates a buffering watch and returns its authoritative snapshot, `startWatch()` flushes buffered events and begins live delivery, and `stopWatch()` releases it. Events carry their watch ID. Streaming message updates carry compact assistant-message frames rather than cumulative partial messages. Reconnection creates a new watch and snapshot rather than replaying old events.
 

packages/protocol/src/protocol.ts

@@ -345,17 +345,24 @@ export const ServiceEventEnvelopeSchema = StrictObject({
 	subscriptionId: IdSchema,
 	update: ServiceProviderUpdateSchema,
 });
+/** Out-of-band update to this presentation's selected Session route. */
+export const AttachmentEnvelopeSchema = StrictObject({
+	type: Type.Literal("attachment"),
+	attachment: Type.Union([SessionTargetSchema, Type.Null()]),
+});
 export const ServerMessageSchema = Type.Union([
 	ServerHelloSchema,
 	ServerHelloErrorSchema,
 	ResponseEnvelopeSchema,
 	EventEnvelopeSchema,
 	ServiceEventEnvelopeSchema,
+	AttachmentEnvelopeSchema,
 ]);
 export type ServerHello = Static<typeof ServerHelloSchema>;
 export type ServerHelloError = Static<typeof ServerHelloErrorSchema>;
 export type ResponseEnvelope = Static<typeof ResponseEnvelopeSchema>;
 export type EventEnvelope = Static<typeof EventEnvelopeSchema>;
 export type ServiceEventEnvelope = Static<typeof ServiceEventEnvelopeSchema>;
+export type AttachmentEnvelope = Static<typeof AttachmentEnvelopeSchema>;
 export type ServerEventEnvelope = EventEnvelope | ServiceEventEnvelope;
 export type ServerMessage = Static<typeof ServerMessageSchema>;

packages/protocol/test/protocol.test.ts

@@ -429,14 +429,21 @@ describe("protocol validation", () => {
 		).toThrow(ProtocolValidationError);
 	});
 
-	test("validates attach results", () => {
-		const message: ServerMessage = {
-			type: "response",
-			id: "request-1",
-			ok: true,
-			result: { sessionId: "session-1" },
+	test("validates attachment route updates", () => {
+		const attached: ServerMessage = {
+			type: "attachment",
+			attachment: {
+				serverId: "00000000-0000-4000-8000-000000000001",
+				sessionId: "session-1",
+				attachmentId: "attachment-1",
+			},
 		};
-		expect(parseServerMessage(message)).toEqual(message);
+		const detached: ServerMessage = { type: "attachment", attachment: null };
+		expect(parseServerMessage(attached)).toEqual(attached);
+		expect(parseServerMessage(detached)).toEqual(detached);
+		expect(() => parseServerMessage({ ...attached, attachment: { sessionId: "session-1" } })).toThrow(
+			ProtocolValidationError,
+		);
 	});
 
 	test.each([

packages/server/README.md

@@ -2,17 +2,19 @@
 
 Experimental local server for the new durable Session and Agent Harness interfaces.
 
-The current slice supports Session discovery, creation, multi-presentation attachment, plugin-service routing, prompting, and optional main-lane observation. `RoutedSessionHandle.attachClient()` returns a presentation-scoped capability. Its optional `invokeService()` forwards an opaque service/member envelope to the selected Session endpoint; the server validates the attachment route but does not load the plugin contract. Its optional `watch()` supplies an authoritative snapshot plus buffered events.
+The current slice supports server- and Session-scoped plugin-service routing, multi-presentation attachment, prompting, and optional main-lane observation. `RoutedServerServiceHost.attachClient()` creates one connection-scoped server service endpoint with narrow attachment-management capabilities. `RoutedSessionHandle.attachClient()` returns a presentation-scoped Session capability. Its optional `invokeService()` forwards an opaque service/member envelope to the selected Session endpoint; the server validates the attachment route but does not load the plugin contract. Its optional `watch()` supplies an authoritative snapshot plus buffered events.
 
-- `list` calls the host's private Session catalog without opening Sessions and projects presentation-safe summaries.
-- `create` asks the host to persist an optional Session ID; private workspace and working-directory data are server-derived.
-- `attach` finds the requested metadata, passes it to the host, and retains the returned routed Session handle in the server.
+- server service calls and subscriptions route opaquely through the connection's `RoutedServerServiceAttachment`;
+- the application-owned `SessionDirectory` projects the private catalog into replicated presentation-safe state;
+- the application-owned `SessionManagement` creates, removes, attaches, and detaches Sessions without exposing route IDs in business results;
+- attachment changes are published out of band after the router installs or clears the live route;
+- the compatibility `list`, `create`, and `attach` operations remain available to low-level clients during migration;
 - `prompt` executes one serializable prompt through the requesting presentation's attachment capability.
 - unknown Session service calls route through `invokeService` without server-side business-payload decoding;
 - service subscription updates remain scoped to the requesting attachment;
 - `watch`, `startWatch`, and `stopWatch` provide snapshot-first lane observation when supported by the host.
 
-A Session may have multiple presentation attachments. Repeating `attach` from one connection is idempotent; every successful attachment has a server-generated `attachmentId`. Session requests carry `{ serverId, sessionId, attachmentId }`, and the server rejects stale or mismatched routes. Losing a connection rejects its local responses but releases its attachment only after admitted prompts settle. The host decides when zero presentation demand and worker-local Harness activity permit worker retirement. Server shutdown closes every routed Session handle, releasing its worker and Session writer ownership.
+A Session may have multiple presentation attachments. Repeating `attach` from one connection is idempotent; every successful attachment has a server-generated `attachmentId` delivered only as routing control data. Session requests carry `{ serverId, sessionId, attachmentId }`, and the server rejects stale or mismatched routes. Losing a connection rejects its local responses but releases its attachment only after admitted prompts settle. The host decides when zero presentation demand and worker-local Harness activity permit worker retirement. Server shutdown closes every routed Session handle, releasing its worker and Session writer ownership.
 
 ```ts
 import { randomUUID } from "node:crypto";

packages/server/src/connection.ts

@@ -1,6 +1,6 @@
 import type { ClientMessageDecoder, RpcTarget } from "@earendil-works/pi-protocol";
 
-import type { MaybePromise } from "./types.ts";
+import type { MaybePromise, RoutedServerServiceAttachment } from "./types.ts";
 
 /** An established, authorized ordered byte connection. */
 export interface ByteConnection {
@@ -26,6 +26,7 @@ export interface ConnectionState {
 	disconnected: boolean;
 	handshake?: Promise<void>;
 	handshakeTimeout: NodeJS.Timeout;
+	serverServices?: RoutedServerServiceAttachment;
 	activeRequests: Map<string, { controller: AbortController; target: RpcTarget }>;
 }
 

packages/server/src/server.ts

@@ -1,4 +1,10 @@
-import { BACKGROUND_CONTEXT, type SessionMetadata, TODO_CONTEXT, withAbortSignal } from "@earendil-works/pi-agent-core";
+import {
+	BACKGROUND_CONTEXT,
+	RemoteServiceError,
+	type SessionMetadata,
+	TODO_CONTEXT,
+	withAbortSignal,
+} from "@earendil-works/pi-agent-core";
 import {
 	type CancelEnvelope,
 	type ClientHello,
@@ -40,6 +46,7 @@ export class PiServer<TMetadata extends SessionMetadata = SessionMetadata> {
 	/** Resolves after shutdown, or rejects when listener or routed-Session cleanup fails. */
 	readonly closed: Promise<void>;
 
+	private readonly host: PiServerHost<TMetadata>;
 	private readonly listeners: readonly PiServerListener[];
 	private readonly maxFrameLength: number;
 	private readonly handshakeTimeoutMs: number;
@@ -57,6 +64,7 @@ export class PiServer<TMetadata extends SessionMetadata = SessionMetadata> {
 
 	constructor(host: PiServerHost<TMetadata>, options: PiServerOptions) {
 		const resolved = resolveOptions(options);
+		this.host = host;
 		this.listeners = options.listeners;
 		this.serverId = options.serverId;
 		this.maxFrameLength = resolved.maxFrameLength;
@@ -65,7 +73,14 @@ export class PiServer<TMetadata extends SessionMetadata = SessionMetadata> {
 		this.onError = options.onError;
 		this.sessions = new SessionRouter({
 			host,
+			serverId: this.serverId,
 			isClosing: () => this.closing,
+			publishAttachment: async (client, attachment) => {
+				await this.sendMessage(client as ConnectionState, {
+					type: "attachment",
+					attachment: attachment ?? null,
+				});
+			},
 			reportError: (error) => this.reportError(error),
 		});
 		this.closed = new Promise((resolve, reject) => {
@@ -251,6 +266,24 @@ export class PiServer<TMetadata extends SessionMetadata = SessionMetadata> {
 		}
 
 		if (this.closing || state.disconnected || state.stage !== "handshaking" || state.connection.closed) return;
+		const serviceHost = this.host.serverServices;
+		if (serviceHost !== undefined) {
+			const services = await serviceHost.attachClient(
+				{
+					attachSession: async (sessionId, context) => {
+						await this.sessions.attachClient(state, sessionId, context);
+					},
+					detachSession: (context) => this.sessions.detachClient(state, context),
+					prepareSessionRemoval: (sessionId, context) => this.sessions.removeSession(sessionId, context),
+				},
+				TODO_CONTEXT,
+			);
+			if (this.closing || state.disconnected || state.stage !== "handshaking" || state.connection.closed) {
+				await services.release(TODO_CONTEXT);
+				return;
+			}
+			state.serverServices = services;
+		}
 		const sent = await this.sendMessage(state, {
 			type: "hello",
 			version: PROTOCOL_VERSION,
@@ -288,7 +321,15 @@ export class PiServer<TMetadata extends SessionMetadata = SessionMetadata> {
 			if (envelope.target.serverId !== this.serverId) throw new WrongServerError();
 			let result: ProtocolRpcResult;
 			const call = decodeServiceRpcCall(envelope.call);
-			if (call !== undefined) {
+			if (!("sessionId" in envelope.target) && state.serverServices !== undefined && call?.method !== "list") {
+				result = await state.serverServices.invokeService(
+					envelope.call,
+					async (subscriptionId, update) => {
+						await this.sendMessage(state, { type: "service_event", subscriptionId, update });
+					},
+					context,
+				);
+			} else if (call !== undefined) {
 				result = await this.sessions.executeCall(
 					call,
 					envelope.target,
@@ -354,7 +395,12 @@ export class PiServer<TMetadata extends SessionMetadata = SessionMetadata> {
 		}
 		connection.activeRequests.clear();
 		if (this.connections.delete(connection)) this.notifyConnectionCountChanged();
-		void Promise.allSettled([this.sessions.disconnect(connection, TODO_CONTEXT)]).then((results) => {
+		const serverServices = connection.serverServices;
+		delete connection.serverServices;
+		void Promise.allSettled([
+			this.sessions.disconnect(connection, TODO_CONTEXT),
+			serverServices?.release(TODO_CONTEXT),
+		]).then((results) => {
 			for (const result of results) if (result.status === "rejected") this.reportError(result.reason);
 		});
 	}
@@ -420,7 +466,7 @@ export class PiServer<TMetadata extends SessionMetadata = SessionMetadata> {
 	}
 
 	private toProtocolError(error: unknown): ProtocolError {
-		if (error instanceof PiServerError) {
+		if (error instanceof PiServerError || error instanceof RemoteServiceError) {
 			return { code: error.code, message: error.message };
 		}
 		if (error instanceof ProtocolValidationError) {

packages/server/src/session-router.ts

@@ -14,6 +14,7 @@ import {
 	type ServiceRpcCall,
 	type ServiceRpcResultUnion,
 	type SessionSummary,
+	type SessionTarget,
 } from "@earendil-works/pi-protocol";
 import {
 	NotSupportedError,
@@ -64,7 +65,9 @@ interface HostedSession {
 
 interface SessionRouterOptions<TMetadata extends SessionMetadata> {
 	host: PiServerHost<TMetadata>;
+	serverId: string;
 	isClosing: () => boolean;
+	publishAttachment(client: object, attachment: SessionTarget | undefined, context: Context): Promise<void>;
 	reportError: (error: unknown) => void;
 }
 
@@ -102,7 +105,7 @@ export class SessionRouter<TMetadata extends SessionMetadata = SessionMetadata>
 				attach: async ({ client, context, target }, sessionId) => {
 					this.requireServerTarget(target);
 					if (this.options.isClosing()) throw new ServerDrainingError();
-					return this.runForClient(client, () => this.attachClient(client, sessionId, context));
+					return this.attachClient(client, sessionId, context);
 				},
 				prompt: async ({ client, context, target }, prompt) => {
 					const admitted = await this.runForClient(client, () =>
@@ -144,12 +147,47 @@ export class SessionRouter<TMetadata extends SessionMetadata = SessionMetadata>
 		return admitted.result;
 	}
 
+	attachClient(
+		client: object,
+		sessionId: string,
+		context: Context,
+	): Promise<{ sessionId: string; attachmentId: string }> {
+		if (this.options.isClosing()) return Promise.reject(new ServerDrainingError());
+		return this.runForClient(client, () => this.attachClientNow(client, sessionId, context));
+	}
+
+	detachClient(client: object, context: Context): Promise<void> {
+		return this.runForClient(client, async () => {
+			const attachment = this.attachmentsByClient.get(client);
+			if (attachment) await this.releaseAttachment(attachment, context);
+		});
+	}
+
+	async removeSession(sessionId: string, context: Context): Promise<void> {
+		if (this.options.isClosing()) throw new ServerDrainingError();
+		const hosted = this.hostedSessions.get(sessionId);
+		if (hosted === undefined) return;
+		const errors: unknown[] = [];
+		const releases = await Promise.allSettled(
+			[...hosted.attachments].map((attachment) => this.releaseAttachment(attachment, context)),
+		);
+		for (const result of releases) if (result.status === "rejected") errors.push(result.reason);
+		try {
+			await hosted.handle.close(context);
+		} catch (error) {
+			errors.push(error);
+		}
+		if (this.hostedSessions.get(sessionId) === hosted) this.hostedSessions.delete(sessionId);
+		if (errors.length === 1) throw errors[0];
+		if (errors.length > 1) throw new AggregateError(errors, `Failed to close Session ${sessionId}`);
+	}
+
 	async disconnect(client: object, context: Context): Promise<void> {
 		this.disconnectedClients.add(client);
 		try {
 			await this.runForClient(client, async () => {
 				const attachment = this.attachmentsByClient.get(client);
-				if (attachment) await this.releaseAttachment(attachment, context);
+				if (attachment) await this.releaseAttachment(attachment, context, false);
 			});
 		} finally {
 			this.disconnectedClients.delete(client);
@@ -213,7 +251,7 @@ export class SessionRouter<TMetadata extends SessionMetadata = SessionMetadata>
 		return result;
 	}
 
-	private async attachClient(
+	private async attachClientNow(
 		client: object,
 		sessionId: string,
 		context: Context,
@@ -223,7 +261,7 @@ export class SessionRouter<TMetadata extends SessionMetadata = SessionMetadata>
 		if (current?.session.id === sessionId) return { sessionId, attachmentId: current.id };
 		const hosted = await this.acquire(sessionId, context);
 		if (this.options.isClosing() || this.disconnectedClients.has(client)) throw new ServerDrainingError();
-		if (current) await this.releaseAttachment(current, context);
+		if (current) await this.releaseAttachment(current, context, false);
 		const attachment: ClientAttachment = {
 			id: randomUUID(),
 			client,
@@ -249,6 +287,11 @@ export class SessionRouter<TMetadata extends SessionMetadata = SessionMetadata>
 			throw new ServerDrainingError();
 		}
 		this.attachmentsByClient.set(client, attachment);
+		await this.options.publishAttachment(
+			client,
+			{ serverId: this.options.serverId, sessionId, attachmentId: attachment.id },
+			context,
+		);
 		return { sessionId, attachmentId: attachment.id };
 	}
 
@@ -368,7 +411,7 @@ export class SessionRouter<TMetadata extends SessionMetadata = SessionMetadata>
 		if ("sessionId" in target) throw new ProtocolValidationError("Server operation cannot target a Session");
 	}
 
-	private releaseAttachment(attachment: ClientAttachment, context: Context): Promise<void> {
+	private releaseAttachment(attachment: ClientAttachment, context: Context, publish = true): Promise<void> {
 		attachment.releasing ??= (async () => {
 			const errors: unknown[] = [];
 			try {
@@ -389,16 +432,17 @@ export class SessionRouter<TMetadata extends SessionMetadata = SessionMetadata>
 				if (errors.length === 1) throw errors[0];
 				if (errors.length > 1) throw new AggregateError(errors, "Failed to release Session attachment");
 			} finally {
-				this.clearAttachment(attachment);
+				await this.clearAttachment(attachment, context, publish);
 			}
 		})();
 		return attachment.releasing;
 	}
 
-	private clearAttachment(attachment: ClientAttachment): void {
+	private async clearAttachment(attachment: ClientAttachment, context: Context, publish: boolean): Promise<void> {
 		attachment.session.attachments.delete(attachment);
 		if (this.attachmentsByClient.get(attachment.client) === attachment) {
 			this.attachmentsByClient.delete(attachment.client);
+			if (publish) await this.options.publishAttachment(attachment.client, undefined, context);
 		}
 	}
 

packages/server/src/types.ts

@@ -38,6 +38,28 @@ export interface RoutedSessionAttachment {
 	release(context: Context): MaybePromise<void>;
 }
 
+/** Presentation-scoped routing capabilities available to server service implementations. */
+export interface RoutedServerPresentation {
+	attachSession(sessionId: string, context: Context): Promise<void>;
+	detachSession(context: Context): Promise<void>;
+	/** Release routed attachments and handles before the application deletes durable metadata. */
+	prepareSessionRemoval(sessionId: string, context: Context): Promise<void>;
+}
+
+/** One connection's server-scoped service endpoint. */
+export interface RoutedServerServiceAttachment {
+	invokeService(
+		call: ProtocolRpcCall,
+		publish: (subscriptionId: string, update: ServiceProviderUpdate, context: Context) => MaybePromise<void>,
+		context: Context,
+	): Promise<ProtocolRpcResult>;
+	release(context: Context): MaybePromise<void>;
+}
+
+export interface RoutedServerServiceHost {
+	attachClient(presentation: RoutedServerPresentation, context: Context): MaybePromise<RoutedServerServiceAttachment>;
+}
+
 /** Snapshot-first observation handle supplied by a routed Session attachment. */
 export interface RoutedSessionWatch {
 	readonly snapshot: LaneSnapshot;
@@ -55,6 +77,7 @@ export interface RoutedSessionHandle {
 
 /** Application capabilities used by server-wide management and Session routing. */
 export interface PiServerHost<TMetadata extends SessionMetadata = SessionMetadata> {
+	readonly serverServices?: RoutedServerServiceHost;
 	readonly sessions: {
 		list(context: Context): Promise<TMetadata[]>;
 		create(options: SessionCreateOptions, context: Context): Promise<TMetadata>;

packages/server/test/conformance.test.ts

@@ -7,7 +7,7 @@ import type { PiServerHost } from "../src/types.ts";
 
 const servers = new Set<PiServer>();
 
-function createServer(host: TestServerHost, serverId = "00000000-0000-4000-8000-000000000001"): PiServer {
+function createServer(host: PiServerHost, serverId = "00000000-0000-4000-8000-000000000001"): PiServer {
 	const server = new PiServer(host, { listeners: [], serverId });
 	servers.add(server);
 	return server;
@@ -170,6 +170,76 @@ describe("Session protocol", () => {
 		).resolves.toMatchObject({ ok: false, error: { code: "not_supported" } });
 	});
 
+	test("routes opaque server services and publishes attachment changes out of band", async () => {
+		const backing = new TestServerHost();
+		await backing.seed("session-1");
+		let releaseCount = 0;
+		const host: PiServerHost = {
+			sessions: backing.sessions,
+			openSession: (metadata, context) => backing.openSession(metadata, context),
+			serverServices: {
+				attachClient(presentation) {
+					return {
+						async invokeService(call, _publish, context) {
+							if (call.serviceId !== "session-management") throw new Error("Unexpected service");
+							if (call.member === "attach" && typeof call.args[0] === "string") {
+								await presentation.attachSession(call.args[0], context);
+								return undefined;
+							}
+							if (call.member === "detach") {
+								await presentation.detachSession(context);
+								return undefined;
+							}
+							throw new Error("Unexpected service member");
+						},
+						release() {
+							releaseCount += 1;
+						},
+					};
+				},
+			},
+		};
+		const server = createServer(host);
+		const client = connect(server);
+		await client.hello();
+		const attached = client.next((message) => message.type === "attachment" && message.attachment !== null);
+		const attachResponse = client.next((message) => message.type === "response" && message.id === "service-attach");
+		await client.sendMessage({
+			type: "request",
+			id: "service-attach",
+			target: { serverId: "00000000-0000-4000-8000-000000000001" },
+			call: { serviceId: "session-management", member: "attach", args: ["session-1"] },
+		});
+		await expect(attached).resolves.toMatchObject({
+			type: "attachment",
+			attachment: { sessionId: "session-1", attachmentId: expect.any(String) },
+		});
+		await expect(attachResponse).resolves.toEqual({
+			type: "response",
+			id: "service-attach",
+			ok: true,
+		});
+		expect(backing.latestHarness("session-1").attachedClients).toBe(1);
+
+		const detached = client.next((message) => message.type === "attachment" && message.attachment === null);
+		const detachResponse = client.next((message) => message.type === "response" && message.id === "service-detach");
+		await client.sendMessage({
+			type: "request",
+			id: "service-detach",
+			target: { serverId: "00000000-0000-4000-8000-000000000001" },
+			call: { serviceId: "session-management", member: "detach", args: [] },
+		});
+		await expect(detached).resolves.toMatchObject({ type: "attachment", attachment: null });
+		await expect(detachResponse).resolves.toEqual({
+			type: "response",
+			id: "service-detach",
+			ok: true,
+		});
+		expect(backing.latestHarness("session-1").attachedClients).toBe(0);
+		await client.close();
+		await expect.poll(() => releaseCount).toBe(1);
+	});
+
 	test("delivers a lane snapshot before buffered and live watch events", async () => {
 		const host = new TestServerHost();
 		await host.seed("session-1");