From 48a07ddbfd3abff279187286cd6da97f722a75a3 Mon Sep 17 00:00:00 2001 From: Armin Ronacher Date: Thu, 20 Aug 2026 10:13:28 +0200 Subject: [PATCH] feat(agent): add server-scoped session services --- packages/agent/docs/plugins.md | 2 +- packages/agent/docs/rpc.md | 2 +- packages/client/README.md | 4 +- packages/client/src/client.ts | 43 ++++- packages/client/test/client.test.ts | 26 +++ .../coding-agent/src/experimental/client.ts | 39 ++++- .../coding-agent/src/experimental/server.ts | 93 ++++++++--- .../src/experimental/services/connection.ts | 60 +++++-- .../src/experimental/services/server.ts | 152 ++++++++++++++++++ .../src/experimental/services/sessions.ts | 22 +++ .../experimental/session-worker-manager.ts | 5 + .../test/experimental-remote-runtime.test.ts | 61 ++++++- packages/protocol/README.md | 8 +- packages/protocol/src/protocol.ts | 7 + packages/protocol/test/protocol.test.ts | 21 ++- packages/server/README.md | 12 +- packages/server/src/connection.ts | 3 +- packages/server/src/server.ts | 54 ++++++- packages/server/src/session-router.ts | 58 ++++++- packages/server/src/types.ts | 23 +++ packages/server/test/conformance.test.ts | 72 ++++++++- 21 files changed, 690 insertions(+), 77 deletions(-) create mode 100644 packages/coding-agent/src/experimental/services/server.ts create mode 100644 packages/coding-agent/src/experimental/services/sessions.ts diff --git a/packages/agent/docs/plugins.md b/packages/agent/docs/plugins.md index 58bac7636..93e95e7bb 100644 --- a/packages/agent/docs/plugins.md +++ b/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. diff --git a/packages/agent/docs/rpc.md b/packages/agent/docs/rpc.md index 5695eea91..6f55e3998 100644 --- a/packages/agent/docs/rpc.md +++ b/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. diff --git a/packages/client/README.md b/packages/client/README.md index 80a579e0d..009dd6158 100644 --- a/packages/client/README.md +++ b/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. diff --git a/packages/client/src/client.ts b/packages/client/src/client.ts index 85126acc4..98060c0f7 100644 --- a/packages/client/src/client.ts +++ b/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>; @@ -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; diff --git a/packages/client/test/client.test.ts b/packages/client/test/client.test.ts index 890d43c7c..925181b12 100644 --- a/packages/client/test/client.test.ts +++ b/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 = []; + 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); diff --git a/packages/coding-agent/src/experimental/client.ts b/packages/coding-agent/src/experimental/client.ts index 4925b261c..d8043d23d 100644 --- a/packages/coding-agent/src/experimental/client.ts +++ b/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(); + const serviceNamespaces = new Set(); 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(state: RemoteState): Promise { + 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) : ""; diff --git a/packages/coding-agent/src/experimental/server.ts b/packages/coding-agent/src/experimental/server.ts index 55dba09de..eb1edd6b9 100644 --- a/packages/coding-agent/src/experimental/server.ts +++ b/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; +} + async function startServerBackend( options: StartServerBackendOptions, workers: SessionWorkerManager, onConnectionCountChanged?: (count: number) => void, -): Promise { +): Promise { 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 = { - 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["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["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 => { + 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 = { + serverServices: serverServices.host, + sessions: { list: listSessions, create: createSession }, openSession: (metadata, context) => workers.openSession(metadata, context), }; const socketPath = options.path; const closeCatalog = async (): Promise => { - 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 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 { + 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 { diff --git a/packages/coding-agent/src/experimental/services/server.ts b/packages/coding-agent/src/experimental/services/server.ts new file mode 100644 index 000000000..c3904adf3 --- /dev/null +++ b/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; + dispose(): Promise; +} + +export async function createExperimentalServerServices(options: { + list(context: Context): Promise; + create(createOptions: SessionCreateOptions, context: Context): Promise; + remove(sessionId: string, context: Context): Promise; +}): Promise { + let revision = 1; + const directory = remoteState({ + revision, + sessions: await options.list(BACKGROUND_CONTEXT), + }); + const attachments = new Set(); + let mutationTail = Promise.resolve(); + + const refreshNow = async (context: Context): Promise => { + revision += 1; + directory.set({ revision, sessions: await options.list(context) }, context); + }; + const serialize = (operation: () => Promise): Promise => { + 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(); + 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; +} diff --git a/packages/coding-agent/src/experimental/services/sessions.ts b/packages/coding-agent/src/experimental/services/sessions.ts new file mode 100644 index 000000000..13191b188 --- /dev/null +++ b/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; +} + +export const SessionDirectory = defineRemoteService("session-directory"); + +export interface SessionManagementService { + create(options: SessionCreateOptions, context: Context): Promise; + remove(sessionId: string, context: Context): Promise; + attach(sessionId: string, context: Context): Promise; + detach(context: Context): Promise; +} + +export const SessionManagement = defineRemoteService("session-management"); diff --git a/packages/coding-agent/src/experimental/session-worker-manager.ts b/packages/coding-agent/src/experimental/session-worker-manager.ts index edd409c31..b4a72d32c 100644 --- a/packages/coding-agent/src/experimental/session-worker-manager.ts +++ b/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 { + 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, diff --git a/packages/coding-agent/test/experimental-remote-runtime.test.ts b/packages/coding-agent/test/experimental-remote-runtime.test.ts index 986f3d941..c5201a726 100644 --- a/packages/coding-agent/test/experimental-remote-runtime.test.ts +++ b/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"); diff --git a/packages/protocol/README.md b/packages/protocol/README.md index 0983a3e55..1ae09df04 100644 --- a/packages/protocol/README.md +++ b/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. diff --git a/packages/protocol/src/protocol.ts b/packages/protocol/src/protocol.ts index a05a30fa2..97bc52bc0 100644 --- a/packages/protocol/src/protocol.ts +++ b/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; export type ServerHelloError = Static; export type ResponseEnvelope = Static; export type EventEnvelope = Static; export type ServiceEventEnvelope = Static; +export type AttachmentEnvelope = Static; export type ServerEventEnvelope = EventEnvelope | ServiceEventEnvelope; export type ServerMessage = Static; diff --git a/packages/protocol/test/protocol.test.ts b/packages/protocol/test/protocol.test.ts index 44950f642..e5917bb93 100644 --- a/packages/protocol/test/protocol.test.ts +++ b/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([ diff --git a/packages/server/README.md b/packages/server/README.md index bf2acd721..17ff332af 100644 --- a/packages/server/README.md +++ b/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"; diff --git a/packages/server/src/connection.ts b/packages/server/src/connection.ts index 3eadb8ede..e9ae22aad 100644 --- a/packages/server/src/connection.ts +++ b/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; handshakeTimeout: NodeJS.Timeout; + serverServices?: RoutedServerServiceAttachment; activeRequests: Map; } diff --git a/packages/server/src/server.ts b/packages/server/src/server.ts index d68a50a58..3c7e029b3 100644 --- a/packages/server/src/server.ts +++ b/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 { /** Resolves after shutdown, or rejects when listener or routed-Session cleanup fails. */ readonly closed: Promise; + private readonly host: PiServerHost; private readonly listeners: readonly PiServerListener[]; private readonly maxFrameLength: number; private readonly handshakeTimeoutMs: number; @@ -57,6 +64,7 @@ export class PiServer { constructor(host: PiServerHost, 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 { 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 { } 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 { 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 { } 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 { } 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) { diff --git a/packages/server/src/session-router.ts b/packages/server/src/session-router.ts index e834a617c..0ddacef95 100644 --- a/packages/server/src/session-router.ts +++ b/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 { host: PiServerHost; + serverId: string; isClosing: () => boolean; + publishAttachment(client: object, attachment: SessionTarget | undefined, context: Context): Promise; reportError: (error: unknown) => void; } @@ -102,7 +105,7 @@ export class SessionRouter 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 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 { + 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 { + 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 { 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 return result; } - private async attachClient( + private async attachClientNow( client: object, sessionId: string, context: Context, @@ -223,7 +261,7 @@ export class SessionRouter 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 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 if ("sessionId" in target) throw new ProtocolValidationError("Server operation cannot target a Session"); } - private releaseAttachment(attachment: ClientAttachment, context: Context): Promise { + private releaseAttachment(attachment: ClientAttachment, context: Context, publish = true): Promise { attachment.releasing ??= (async () => { const errors: unknown[] = []; try { @@ -389,16 +432,17 @@ export class SessionRouter 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 { 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); } } diff --git a/packages/server/src/types.ts b/packages/server/src/types.ts index b358c0a9b..f3066258a 100644 --- a/packages/server/src/types.ts +++ b/packages/server/src/types.ts @@ -38,6 +38,28 @@ export interface RoutedSessionAttachment { release(context: Context): MaybePromise; } +/** Presentation-scoped routing capabilities available to server service implementations. */ +export interface RoutedServerPresentation { + attachSession(sessionId: string, context: Context): Promise; + detachSession(context: Context): Promise; + /** Release routed attachments and handles before the application deletes durable metadata. */ + prepareSessionRemoval(sessionId: string, context: Context): Promise; +} + +/** One connection's server-scoped service endpoint. */ +export interface RoutedServerServiceAttachment { + invokeService( + call: ProtocolRpcCall, + publish: (subscriptionId: string, update: ServiceProviderUpdate, context: Context) => MaybePromise, + context: Context, + ): Promise; + release(context: Context): MaybePromise; +} + +export interface RoutedServerServiceHost { + attachClient(presentation: RoutedServerPresentation, context: Context): MaybePromise; +} + /** 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 { + readonly serverServices?: RoutedServerServiceHost; readonly sessions: { list(context: Context): Promise; create(options: SessionCreateOptions, context: Context): Promise; diff --git a/packages/server/test/conformance.test.ts b/packages/server/test/conformance.test.ts index 5017fe36d..d5a4234d9 100644 --- a/packages/server/test/conformance.test.ts +++ b/packages/server/test/conformance.test.ts @@ -7,7 +7,7 @@ import type { PiServerHost } from "../src/types.ts"; const servers = new Set(); -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");