diff --git a/packages/client/test/unix.test.ts b/packages/client/test/unix.test.ts index 1702e1c93..e9b6d5ca1 100644 --- a/packages/client/test/unix.test.ts +++ b/packages/client/test/unix.test.ts @@ -5,6 +5,7 @@ import { createServer, type Server, type Socket } from "node:net"; import { join } from "node:path"; import { afterEach, describe, expect, test } from "vitest"; import { Server as RuntimeServer } from "../../server/src/server.ts"; +import { createTestServerServices } from "../../server/src/testing/host.ts"; import { createUnixListener } from "../../server/src/transports/unix/listener.ts"; import { discoverUnixServers } from "../src/unix.ts"; @@ -32,9 +33,8 @@ async function startServer( const path = join(directory, `${fileServerId}.sock`); const server = new RuntimeServer( { - sessions: { - list: async () => [], - }, + serverServices: createTestServerServices(), + resolveSession: async () => Promise.reject(new Error("unused")), openSession: async () => Promise.reject(new Error("unused")), }, { listeners: [createUnixListener({ path })], serverId: reportedServerId }, diff --git a/packages/coding-agent/src/experimental/client-tui.ts b/packages/coding-agent/src/experimental/client-tui.ts index 5bacc1ee1..97999b4b7 100644 --- a/packages/coding-agent/src/experimental/client-tui.ts +++ b/packages/coding-agent/src/experimental/client-tui.ts @@ -8,7 +8,7 @@ import { type LoadedFacets, } from "@earendil-works/chord"; import { BACKGROUND_CONTEXT } from "@earendil-works/chord/context"; -import type { JsonValue, SessionSummary } from "@earendil-works/pi-protocol"; +import type { JsonValue } from "@earendil-works/pi-protocol"; import { CombinedAutocompleteProvider, type Component, @@ -42,7 +42,7 @@ import type { } from "./services/connection.ts"; import { PresentationPlugins } from "./services/plugins.ts"; import { PresentationUI } from "./services/presentation-ui.ts"; -import { SessionDirectory, SessionManagement } from "./services/sessions.ts"; +import { SessionDirectory, SessionManagement, type SessionSummary } from "./services/sessions.ts"; import { SlashCommands } from "./services/slash-commands.ts"; import { createBuiltInSlashCommandsFacet, diff --git a/packages/coding-agent/src/experimental/client.ts b/packages/coding-agent/src/experimental/client.ts index 2d7eef906..d65fff29b 100644 --- a/packages/coding-agent/src/experimental/client.ts +++ b/packages/coding-agent/src/experimental/client.ts @@ -1,9 +1,10 @@ import { resolve } from "node:path"; import { BACKGROUND_CONTEXT } from "@earendil-works/pi-agent-core"; -import type { LaneEvent, PromptMessage, SessionAddress } from "@earendil-works/pi-protocol"; +import type { LaneEvent, WireAgentMessage } from "@earendil-works/pi-protocol"; import type { ClientCommand } from "../cli/experimental/commands/client.ts"; import { activateBuiltinClientServices, openClientRuntime } from "./client-runtime.ts"; import type { AgentOperationResponse } from "./services/agent-controller.ts"; +import type { SessionAddress } from "./services/sessions.ts"; export type ClientResult = | { @@ -107,7 +108,7 @@ export async function runClient(command: ClientCommand, options: RunClientOption } } -function messageText(message: Extract): string { +function messageText(message: Extract): string { return message.content .filter((content) => content.type === "text") .map((content) => content.text) diff --git a/packages/coding-agent/src/experimental/harness-wire-adapter.ts b/packages/coding-agent/src/experimental/harness-wire-adapter.ts index bd30c06d1..680d1189e 100644 --- a/packages/coding-agent/src/experimental/harness-wire-adapter.ts +++ b/packages/coding-agent/src/experimental/harness-wire-adapter.ts @@ -25,18 +25,18 @@ import type { LaneEvent, LaneSnapshot, OperationResultRecord, - PromptImage, - PromptMessage, ToolOutput, + WireAgentMessage, + WireImageContent, } from "@earendil-works/pi-protocol"; -type WireAssistantMessage = Extract; -type WireUserMessage = Extract; -type WireToolResultMessage = Extract; -type WireBashExecutionMessage = Extract; -type WireCustomMessage = Extract; -type WireBranchSummaryMessage = Extract; -type WireCompactionSummaryMessage = Extract; +type WireAssistantMessage = Extract; +type WireUserMessage = Extract; +type WireToolResultMessage = Extract; +type WireBashExecutionMessage = Extract; +type WireCustomMessage = Extract; +type WireBranchSummaryMessage = Extract; +type WireCompactionSummaryMessage = Extract; type SameKeys = Exclude extends never ? Exclude extends never ? true @@ -46,7 +46,7 @@ type Assert = T; /** Compile-time alarms for field additions on either side of each closed message adapter. */ export type HarnessWireAdapterCompatibility = [ - Assert>, + Assert>, Assert>, Assert>, Assert>, @@ -55,7 +55,7 @@ export type HarnessWireAdapterCompatibility = [ Assert>, Assert>, ]; -function toHarnessUsage(usage: Usage): Usage { +function toWireUsage(usage: Usage): Usage { return { input: usage.input, output: usage.output, @@ -159,7 +159,7 @@ function toWireAssistantMessage(message: AssistantMessage): WireAssistantMessage }), })), }), - usage: toHarnessUsage(message.usage), + usage: toWireUsage(message.usage), stopReason: message.stopReason, ...(message.deferred === undefined ? {} : { deferred: toWireDeferred(message.deferred) }), ...(message.errorMessage === undefined ? {} : { errorMessage: message.errorMessage }), @@ -185,7 +185,7 @@ function toWireJsonValue(value: unknown): JsonValue { return value as JsonValue; } -function toWireMessage(message: AgentMessage): PromptMessage { +function toWireMessage(message: AgentMessage): WireAgentMessage { switch (message.role) { case "user": return { @@ -223,7 +223,7 @@ function toWireMessage(message: AgentMessage): PromptMessage { : { type: "image" as const, data: content.data, mimeType: content.mimeType }, ), ...(message.details === undefined ? {} : { details: toWireJsonValue(message.details) }), - ...(message.usage === undefined ? {} : { usage: toHarnessUsage(message.usage) }), + ...(message.usage === undefined ? {} : { usage: toWireUsage(message.usage) }), ...(message.addedToolNames === undefined ? {} : { addedToolNames: [...message.addedToolNames] }), isError: message.isError, timestamp: message.timestamp, @@ -299,7 +299,7 @@ function toWireEntry(entry: Entry): LaneEntry { retainedTail: entry.retainedTail.map(toWireMessage), tokensBefore: entry.tokensBefore, ...(entry.details === undefined ? {} : { details: entry.details }), - ...(entry.usage === undefined ? {} : { usage: toHarnessUsage(entry.usage) }), + ...(entry.usage === undefined ? {} : { usage: toWireUsage(entry.usage) }), fromHook: entry.fromHook, }; case "branch_summary": @@ -309,7 +309,7 @@ function toWireEntry(entry: Entry): LaneEntry { fromId: entry.fromId, summary: entry.summary, ...(entry.details === undefined ? {} : { details: entry.details }), - ...(entry.usage === undefined ? {} : { usage: toHarnessUsage(entry.usage) }), + ...(entry.usage === undefined ? {} : { usage: toWireUsage(entry.usage) }), fromHook: entry.fromHook, }; case "custom": @@ -336,7 +336,7 @@ function toWireToolOutput(result: AgentToolResult): ToolOutput { : { type: "image" as const, data: content.data, mimeType: content.mimeType }, ), ...(result.details === undefined ? {} : { details: toWireJsonValue(result.details) }), - ...(result.usage === undefined ? {} : { usage: toHarnessUsage(result.usage) }), + ...(result.usage === undefined ? {} : { usage: toWireUsage(result.usage) }), }; } @@ -403,7 +403,7 @@ export function toWireLaneSnapshot(snapshot: HarnessLaneSnapshot): LaneSnapshot thinkingLevel: snapshot.configuration.thinkingLevel, activeToolNames: [...snapshot.configuration.activeToolNames], }, - stats: { messageCount: snapshot.stats.messageCount, usage: toHarnessUsage(snapshot.stats.usage) }, + stats: { messageCount: snapshot.stats.messageCount, usage: toWireUsage(snapshot.stats.usage) }, operation: snapshot.operation === null ? null @@ -526,10 +526,10 @@ export function toWireLaneEvent(event: HarnessEvent): LaneEvent | undefined { lane: event.lane, row: { ...event.row, - usage: toHarnessUsage(event.row.usage), + usage: toWireUsage(event.row.usage), ...(event.row.details === undefined ? {} : { details: toWireJsonValue(event.row.details) }), }, - totals: toHarnessUsage(event.totals), + totals: toWireUsage(event.totals), }; case "config_update": switch (event.property) { diff --git a/packages/coding-agent/src/experimental/radius-relay.ts b/packages/coding-agent/src/experimental/radius-relay.ts index d5ab0a5bb..0eaffe805 100644 --- a/packages/coding-agent/src/experimental/radius-relay.ts +++ b/packages/coding-agent/src/experimental/radius-relay.ts @@ -329,9 +329,20 @@ export function createRadiusClientTransportFactory(options: { }; } +type RadiusReconnectClient = Pick< + Client, + | "attachment" + | "connected" + | "connectionState" + | "disconnect" + | "onAttachmentChange" + | "onConnectionStateChange" + | "reconnect" +>; + /** Reconnect one established Radius client and restore its last selected Session. */ export class RadiusClientReconnect { - readonly #client: Client; + readonly #client: RadiusReconnectClient; readonly #reattach: (sessionId: string) => Promise; readonly #abortController = new AbortController(); readonly #removeConnectionListener: () => void; @@ -340,7 +351,7 @@ export class RadiusClientReconnect { #reconnecting: Promise | undefined; #disposed = false; - constructor(client: Client, reattach: (sessionId: string) => Promise) { + constructor(client: RadiusReconnectClient, reattach: (sessionId: string) => Promise) { this.#client = client; this.#reattach = reattach; this.#desiredSessionId = client.attachment?.sessionId; diff --git a/packages/coding-agent/src/experimental/server.ts b/packages/coding-agent/src/experimental/server.ts index d30b7e820..5ec02aaa2 100644 --- a/packages/coding-agent/src/experimental/server.ts +++ b/packages/coding-agent/src/experimental/server.ts @@ -13,7 +13,7 @@ import { import { NodeExecutionEnv } from "@earendil-works/pi-agent-core/node"; import { Client, ServerError as ClientServerError, DisconnectedError } from "@earendil-works/pi-client"; import { createUnixTransportFactory, type UnixServerRoute } from "@earendil-works/pi-client/unix"; -import { isServerId, type ServerId, type SessionCreateOptions, type SessionSummary } from "@earendil-works/pi-protocol"; +import { isServerId, type ServerId } from "@earendil-works/pi-protocol"; import { ServerError as RoutedServerError, type Server, @@ -45,6 +45,7 @@ import { import { RadiusRelayAuthResolver } from "./radius-auth.ts"; import { RadiusRelayHost, type RadiusRelayHostStatus } from "./radius-relay.ts"; import { createExperimentalServerServices } from "./services/server.ts"; +import type { SessionCreateOptions, SessionSummary } from "./services/sessions.ts"; import { SessionPluginSelectionConflictError, SessionWorkerManager } from "./session-worker-manager.ts"; export const ENV_SERVER_DIR = "PI_SERVER_DIR"; @@ -381,11 +382,17 @@ async function startServerBackend( const sessionDir = resolveSessionDirectory(options.sessionDir); const executionEnv = new NodeExecutionEnv({ cwd: process.cwd() }); const repo = new JsonlSessionRepo({ fileSystem: executionEnv, sessionsRoot: sessionDir }); - const listSessions: ServerHost["sessions"]["list"] = async (context) => { + const listSessions = async (context: Context): Promise => { 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 resolveSession = async (sessionId: string, context: Context): Promise => { + 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(); + return matches[0]!; + }; const createSession = async ( createOptions: SessionCreateOptions, context: Context, @@ -415,21 +422,17 @@ async function startServerBackend( .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); - await options.removeSessionPlugins(matches[0]!); + const metadata = await resolveSession(sessionId, context); + await workers.closeSession(metadata, context); + await repo.delete(metadata, context); + await options.removeSessionPlugins(metadata); }, async prepareSessionPlugins(sessionId, packagePaths, 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(); + const metadata = await resolveSession(sessionId, context); let selected: ResolvedSessionPlugins; try { - selected = await options.resolveSessionPlugins(matches[0]!, packagePaths, context); - workers.assertSessionPluginManifestPaths(matches[0]!, selected.manifestPaths); + selected = await options.resolveSessionPlugins(metadata, packagePaths, context); + workers.assertSessionPluginManifestPaths(metadata, selected.manifestPaths); } catch (error) { if (error instanceof SessionPluginSelectionConflictError) { throw new RoutedServerError("service_invalid_value", error.message); @@ -454,7 +457,7 @@ async function startServerBackend( }); const host: ServerHost = { serverServices: serverServices.host, - sessions: { list: listSessions }, + resolveSession, openSession: async (metadata, context) => { const selected = await options.resolveSessionPlugins(metadata, undefined, context); return workers.openSession(metadata, context, selected.manifestPaths); diff --git a/packages/coding-agent/src/experimental/services/README.md b/packages/coding-agent/src/experimental/services/README.md index 5ba6d6f83..ede4db2dc 100644 --- a/packages/coding-agent/src/experimental/services/README.md +++ b/packages/coding-agent/src/experimental/services/README.md @@ -13,7 +13,7 @@ Facet setup generates each host's RPC service catalogue from its provided non-lo | presentation | `SlashCommands` | process-local contribution registry with model, thinking, compact, reload, and example hello commands | add more presentation hookpoints only as concrete plugin slices require them | | presentation | `PresentationUI` | process-local selection and status capabilities | add narrowly scoped UI capabilities only when commands require them | -`ServerServiceSource.connection` and `SessionServiceSource.attachment` are implemented local control states. +`ServerServiceSource.connection` and `SessionServiceSource.attachment` are implemented local control states. Session directory, creation, and address DTOs are owned by these coding-agent service contracts rather than `pi-protocol`; the transport treats their payloads as opaque service data. With `PI_EXPERIMENTAL=1`, an interactive `pi client` creates and attaches a Session before opening the service-only chat TUI. `pi client -c` and `pi client -r` attach the newest existing Session instead, preserving its durable model and thinking configuration. Model selection is available on demand through `/model`; it is never a startup screen. The presentation always uses the stable coding agent's alternate-screen renderer and shared transcript/dock viewport. It loads configured theme resources and uses the stable terminal light/dark detection and appearance-change notifications. Its replicated state feeds the stable editor, message, tool, status, theme, and tool-renderer components. The presentation drives the worker-owned main lane through `AgentController` and renders a coherent lane-watch replica with the Harness reducer while controller calls are pending. A future transcript service can replace that compatibility watch. diff --git a/packages/coding-agent/src/experimental/services/agent-controller.ts b/packages/coding-agent/src/experimental/services/agent-controller.ts index 6fbd80cb5..fc74962ad 100644 --- a/packages/coding-agent/src/experimental/services/agent-controller.ts +++ b/packages/coding-agent/src/experimental/services/agent-controller.ts @@ -1,9 +1,14 @@ import { type Context, defineService } from "@earendil-works/chord"; -import type { PromptImage } from "@earendil-works/pi-protocol"; + +export interface AgentPromptImage { + type: "image"; + data: string; + mimeType: string; +} export interface AgentPromptRequest { message: string; - images: PromptImage[] | null; + images: AgentPromptImage[] | null; } export interface AgentOperationError { diff --git a/packages/coding-agent/src/experimental/services/connection.ts b/packages/coding-agent/src/experimental/services/connection.ts index 9b4a58b94..92b676762 100644 --- a/packages/coding-agent/src/experimental/services/connection.ts +++ b/packages/coding-agent/src/experimental/services/connection.ts @@ -48,10 +48,6 @@ export interface ServiceSourceOptions { readonly onError?: (error: Error) => void; } -export interface ServiceBindingOptions extends ServiceSourceOptions { - readonly services: readonly { readonly id: string }[]; -} - class RoutedServiceBinding implements RemoteServices { readonly #services: RemoteServiceBinding; readonly #getBound: () => boolean; @@ -350,76 +346,6 @@ class SessionServiceSourceImpl implements SessionServiceSource { } } -/** Create an explicitly bound server-service binding outside a facet host. */ -export function createServerServiceBinding( - client: Client, - options: ServiceBindingOptions, -): ServerServiceSource & RemoteServices { - const source = createServerServiceSource(client, options); - const services = source.open({ - services: options.services, - assertAccess() {}, - onError: options.onError ?? (() => {}), - }); - const activation = services.ready(BACKGROUND_CONTEXT); - void activation.catch(options.onError ?? (() => {})); - const ready = async (context: Context): Promise => { - await activation; - await services.ready(context); - }; - return { - acceptsUnavailableServices: source.acceptsUnavailableServices, - connection: source.connection, - catalogue: (context) => source.catalogue(context), - open: (openOptions) => source.open(openOptions), - use: (service) => services.use(service), - observe: (service, handler) => services.observe(service, handler), - ready, - async dispose(context) { - const results = await Promise.allSettled([services.dispose(context), source.dispose(context)]); - throwFailures(results, "Failed to dispose server service binding"); - }, - }; -} - -/** Create an explicitly bound selected-Session service binding outside a facet host. */ -export function createSessionServiceBinding( - client: Client, - options: ServiceBindingOptions, -): SessionServiceSource & RemoteServices { - const source = createSessionServiceSource(client, options); - const services = source.open({ - services: options.services, - assertAccess() {}, - onError: options.onError ?? (() => {}), - }); - const activation = services.ready(BACKGROUND_CONTEXT); - void activation.catch(options.onError ?? (() => {})); - const ready = async (context: Context): Promise => { - await activation; - const attachment = source.attachment.value; - if (attachment?.status === "detached") await source.whenDetached(context); - else if (attachment !== undefined) await source.whenAttached(attachment.sessionId, context); - }; - return { - get acceptsUnavailableServices() { - return source.acceptsUnavailableServices; - }, - attachment: source.attachment, - catalogue: (context) => source.catalogue(context), - open: (openOptions) => source.open(openOptions), - use: (service) => services.use(service), - observe: (service, handler) => services.observe(service, handler), - ready, - whenAttached: (sessionId, context) => source.whenAttached(sessionId, context), - whenDetached: (context) => source.whenDetached(context), - async dispose(context) { - const results = await Promise.allSettled([services.dispose(context), source.dispose(context)]); - throwFailures(results, "Failed to dispose Session service binding"); - }, - }; -} - /** Create the server-scoped remote service source for one presentation client. */ export function createServerServiceSource(client: Client, options: ServiceSourceOptions = {}): ServerServiceSource { return new ServerServiceSourceImpl(client, options); diff --git a/packages/coding-agent/src/experimental/services/server.ts b/packages/coding-agent/src/experimental/services/server.ts index e857fc2eb..adc8d74ef 100644 --- a/packages/coding-agent/src/experimental/services/server.ts +++ b/packages/coding-agent/src/experimental/services/server.ts @@ -11,13 +11,17 @@ import { type JsonValue, 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 { PresentationPlugins } from "./plugins.ts"; -import { SessionDirectory, type SessionDirectoryState, SessionManagement } from "./sessions.ts"; +import { + type SessionCreateOptions, + SessionDirectory, + type SessionDirectoryState, + SessionManagement, + type SessionSummary, +} from "./sessions.ts"; export interface ExperimentalServerServices { readonly host: RoutedServerServiceHost; diff --git a/packages/coding-agent/src/experimental/services/sessions.ts b/packages/coding-agent/src/experimental/services/sessions.ts index b2840c78a..79dbec2f5 100644 --- a/packages/coding-agent/src/experimental/services/sessions.ts +++ b/packages/coding-agent/src/experimental/services/sessions.ts @@ -1,5 +1,18 @@ import { type Context, defineService, type ReplicatedState } from "@earendil-works/chord"; -import type { SessionCreateOptions, SessionSummary } from "@earendil-works/pi-protocol"; +import type { ServerId } from "@earendil-works/pi-protocol"; + +export interface SessionAddress { + serverId: ServerId; + sessionId: string; +} + +export interface SessionSummary extends SessionAddress { + createdAt: number; +} + +export interface SessionCreateOptions { + id?: string; +} export interface SessionDirectoryState { revision: number; diff --git a/packages/coding-agent/test/experimental-client-tui.test.ts b/packages/coding-agent/test/experimental-client-tui.test.ts index 5a714191e..40819c7e1 100644 --- a/packages/coding-agent/test/experimental-client-tui.test.ts +++ b/packages/coding-agent/test/experimental-client-tui.test.ts @@ -13,7 +13,7 @@ import { type FacetBundleArtifact, } from "@earendil-works/chord/node"; import type { AgentLane } from "@earendil-works/pi-agent-core"; -import type { LaneEvent, LaneSnapshot, SessionSummary } from "@earendil-works/pi-protocol"; +import type { LaneEvent, LaneSnapshot } from "@earendil-works/pi-protocol"; import { ProcessTerminal, TuiMainScreen } from "@earendil-works/pi-tui"; import { beforeAll, describe, expect, test, vi } from "vitest"; import { type ClientTuiServer, ExperimentalClientTui } from "../src/experimental/client-tui.ts"; @@ -32,6 +32,7 @@ import { SessionDirectory, type SessionDirectoryState, SessionManagement, + type SessionSummary, } from "../src/experimental/services/sessions.ts"; import { initTheme } from "../src/modes/interactive/theme/theme.ts"; diff --git a/packages/coding-agent/test/experimental-radius-relay.test.ts b/packages/coding-agent/test/experimental-radius-relay.test.ts index 3391be90a..7ef6795ce 100644 --- a/packages/coding-agent/test/experimental-radius-relay.test.ts +++ b/packages/coding-agent/test/experimental-radius-relay.test.ts @@ -165,11 +165,19 @@ describe("experimental Radius relay", () => { const hostWebSockets = socketFactory(); const clientWebSockets = socketFactory(); const serverHost: ServerHost = { - sessions: { - async list() { - return []; + serverServices: { + attachClient() { + return { + async invokeService() { + throw new Error("not used"); + }, + release() {}, + }; }, }, + async resolveSession() { + throw new Error("not used"); + }, async openSession() { throw new Error("not used"); }, diff --git a/packages/coding-agent/test/experimental-remote-runtime.test.ts b/packages/coding-agent/test/experimental-remote-runtime.test.ts index 3d84ccd0c..bb44b56e5 100644 --- a/packages/coding-agent/test/experimental-remote-runtime.test.ts +++ b/packages/coding-agent/test/experimental-remote-runtime.test.ts @@ -12,15 +12,11 @@ import { activateBuiltinClientServices, openClientRuntime } from "../src/experim import { createPresentationFacetLoaders } from "../src/experimental/plugins/bundled.ts"; import * as processRuntime from "../src/experimental/process.ts"; import { activateServer, type RunningServer, startServer } from "../src/experimental/server.ts"; -import { - createServerServiceBinding, - createSessionServiceBinding, - createSessionServiceSource, - type SessionAttachmentState, -} from "../src/experimental/services/connection.ts"; +import { createSessionServiceSource, type SessionAttachmentState } from "../src/experimental/services/connection.ts"; import { Models } from "../src/experimental/services/models.ts"; import { PresentationPlugins } from "../src/experimental/services/plugins.ts"; import { SessionDirectory, SessionManagement } from "../src/experimental/services/sessions.ts"; +import { createServerServiceBinding, createSessionServiceBinding } from "./experimental-service-binding.ts"; import { configureExperimentalWorkerModel, createExperimentalSessions, diff --git a/packages/coding-agent/test/experimental-service-binding.ts b/packages/coding-agent/test/experimental-service-binding.ts new file mode 100644 index 000000000..ecd725479 --- /dev/null +++ b/packages/coding-agent/test/experimental-service-binding.ts @@ -0,0 +1,90 @@ +import type { Context, RemoteServices } from "@earendil-works/chord"; +import { BACKGROUND_CONTEXT } from "@earendil-works/chord/context"; +import type { Client } from "@earendil-works/pi-client"; +import { + createServerServiceSource, + createSessionServiceSource, + type ServerServiceSource, + type ServiceSourceOptions, + type SessionServiceSource, +} from "../src/experimental/services/connection.ts"; + +interface ServiceBindingOptions extends ServiceSourceOptions { + readonly services: readonly { readonly id: string }[]; +} + +/** Test helper for binding server services without a facet host. */ +export function createServerServiceBinding( + client: Client, + options: ServiceBindingOptions, +): ServerServiceSource & RemoteServices { + const source = createServerServiceSource(client, options); + const services = source.open({ + services: options.services, + assertAccess() {}, + onError: options.onError ?? (() => {}), + }); + const activation = services.ready(BACKGROUND_CONTEXT); + void activation.catch(options.onError ?? (() => {})); + const ready = async (context: Context): Promise => { + await activation; + await services.ready(context); + }; + return { + acceptsUnavailableServices: source.acceptsUnavailableServices, + connection: source.connection, + catalogue: (context) => source.catalogue(context), + open: (openOptions) => source.open(openOptions), + use: (service) => services.use(service), + observe: (service, handler) => services.observe(service, handler), + ready, + async dispose(context) { + const results = await Promise.allSettled([services.dispose(context), source.dispose(context)]); + throwFailures(results, "Failed to dispose server service binding"); + }, + }; +} + +/** Test helper for binding selected-Session services without a facet host. */ +export function createSessionServiceBinding( + client: Client, + options: ServiceBindingOptions, +): SessionServiceSource & RemoteServices { + const source = createSessionServiceSource(client, options); + const services = source.open({ + services: options.services, + assertAccess() {}, + onError: options.onError ?? (() => {}), + }); + const activation = services.ready(BACKGROUND_CONTEXT); + void activation.catch(options.onError ?? (() => {})); + const ready = async (context: Context): Promise => { + await activation; + const attachment = source.attachment.value; + if (attachment?.status === "detached") await source.whenDetached(context); + else if (attachment !== undefined) await source.whenAttached(attachment.sessionId, context); + }; + return { + get acceptsUnavailableServices() { + return source.acceptsUnavailableServices; + }, + attachment: source.attachment, + catalogue: (context) => source.catalogue(context), + open: (openOptions) => source.open(openOptions), + use: (service) => services.use(service), + observe: (service, handler) => services.observe(service, handler), + ready, + whenAttached: (sessionId, context) => source.whenAttached(sessionId, context), + whenDetached: (context) => source.whenDetached(context), + async dispose(context) { + const results = await Promise.allSettled([services.dispose(context), source.dispose(context)]); + throwFailures(results, "Failed to dispose Session service binding"); + }, + }; +} + +function throwFailures(results: readonly PromiseSettledResult[], message: string): void { + const errors = results.flatMap((result) => (result.status === "rejected" ? [result.reason] : [])); + if (errors.length === 1) throw errors[0]; + if (errors.length > 1) throw new AggregateError(errors, message); +} diff --git a/packages/protocol/README.md b/packages/protocol/README.md index 064da80d9..ac28f65d1 100644 --- a/packages/protocol/README.md +++ b/packages/protocol/README.md @@ -10,7 +10,7 @@ Protocol version `6` carries the current server- and Session-scoped facet-servic - singleton and keyed-service subscription snapshots plus ordered state, spawn, and close updates; - 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 an application 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. Service updates carry state, keyed spawn/close, temporary unavailability, and complete singleton replacement; there is no remote service-event member kind. Session-directory state and management results contain no working directory, owner, storage, or parent metadata. Management `attach()` and `detach()` return no routing identifiers; the server publishes the selected live route in an out-of-band `attachment` message. The real `Session` and `AgentHarness` remain process-local. Disconnecting releases only that presentation's attachment after admitted service calls settle. +The transport carries `{ serviceId, instance?, member, args }` calls. A server target contains `{ serverId }`; a Session target contains `{ serverId, sessionId, attachmentId }`. Keyed addresses contain an application 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. Service updates carry state, keyed spawn/close, temporary unavailability, and complete singleton replacement; there is no remote service-event member kind. Session-directory state and management results are opaque application service data, not protocol DTOs. Management `attach()` and `detach()` return no routing identifiers; the server publishes the selected live route in an out-of-band `attachment` message. The real `Session` and `AgentHarness` remain process-local. Disconnecting releases only that presentation's attachment after admitted service calls settle. Experimental presentations consume server-owned `SessionDirectory` and `SessionManagement` plus provider-generated Session service catalogues; there is no handwritten worker service inventory. `Models` has an implemented provider. `AgentController` is the presentation-safe command facade over the worker-owned main `AgentLane`. Session service calls route opaquely to the attached worker, where the provider validates and invokes them. The temporary `LaneWatchRpc` manifest remains only for snapshot-first main-lane observation. diff --git a/packages/protocol/src/harness.ts b/packages/protocol/src/harness.ts index 42e871a79..c0816f497 100644 --- a/packages/protocol/src/harness.ts +++ b/packages/protocol/src/harness.ts @@ -6,25 +6,25 @@ const TimestampSchema = Type.Integer({ minimum: 0 }); const StrictObject = [0]>(properties: T) => Type.Object(properties, { additionalProperties: false }); -export const PromptImageSchema = StrictObject({ +const WireImageContentSchema = StrictObject({ type: Type.Literal("image"), data: Type.String(), mimeType: Type.String({ minLength: 1 }), }); -export type PromptImage = Static; +export type WireImageContent = Static; -export const TextContentSchema = StrictObject({ +const TextContentSchema = StrictObject({ type: Type.Literal("text"), text: Type.String(), textSignature: Type.Optional(Type.String()), }); -export const ThinkingContentSchema = StrictObject({ +const ThinkingContentSchema = StrictObject({ type: Type.Literal("thinking"), thinking: Type.String(), thinkingSignature: Type.Optional(Type.String()), redacted: Type.Optional(Type.Boolean()), }); -export const ToolCallSchema = StrictObject({ +const ToolCallSchema = StrictObject({ type: Type.Literal("toolCall"), id: Type.String(), name: Type.String(), @@ -32,7 +32,7 @@ export const ToolCallSchema = StrictObject({ thoughtSignature: Type.Optional(Type.String()), namespace: Type.Optional(Type.String()), }); -export const UsageSchema = StrictObject({ +const UsageSchema = StrictObject({ input: Type.Number(), output: Type.Number(), cacheRead: Type.Number(), @@ -69,7 +69,7 @@ const AssistantMessageDiagnosticSchema = StrictObject({ error: Type.Optional(DiagnosticErrorSchema), details: Type.Optional(Type.Record(Type.String(), JsonValueSchema)), }); -export const AssistantMessageSchema = StrictObject({ +const AssistantMessageSchema = StrictObject({ role: Type.Literal("assistant"), content: Type.Array(Type.Union([TextContentSchema, ThinkingContentSchema, ToolCallSchema])), api: Type.String(), @@ -96,14 +96,14 @@ export const AssistantMessageSchema = StrictObject({ }); const UserMessageSchema = StrictObject({ role: Type.Literal("user"), - content: Type.Union([Type.String(), Type.Array(Type.Union([TextContentSchema, PromptImageSchema]))]), + content: Type.Union([Type.String(), Type.Array(Type.Union([TextContentSchema, WireImageContentSchema]))]), timestamp: TimestampSchema, }); const ToolResultMessageSchema = StrictObject({ role: Type.Literal("toolResult"), toolCallId: Type.String(), toolName: Type.String(), - content: Type.Array(Type.Union([TextContentSchema, PromptImageSchema])), + content: Type.Array(Type.Union([TextContentSchema, WireImageContentSchema])), details: Type.Optional(JsonValueSchema), usage: Type.Optional(UsageSchema), addedToolNames: Type.Optional(Type.Array(Type.String())), @@ -124,7 +124,7 @@ const BashExecutionMessageSchema = StrictObject({ const CustomMessageSchema = StrictObject({ role: Type.Literal("custom"), customType: Type.String(), - content: Type.Union([Type.String(), Type.Array(Type.Union([TextContentSchema, PromptImageSchema]))]), + content: Type.Union([Type.String(), Type.Array(Type.Union([TextContentSchema, WireImageContentSchema]))]), display: Type.Boolean(), details: Type.Optional(JsonValueSchema), timestamp: TimestampSchema, @@ -143,7 +143,7 @@ const CompactionSummaryMessageSchema = StrictObject({ }); /** Closed set of built-in AgentMessage shapes supported by the wire protocol. */ -export const PromptMessageSchema = Type.Union([ +const WireAgentMessageSchema = Type.Union([ UserMessageSchema, AssistantMessageSchema, ToolResultMessageSchema, @@ -152,7 +152,7 @@ export const PromptMessageSchema = Type.Union([ BranchSummaryMessageSchema, CompactionSummaryMessageSchema, ]); -export type PromptMessage = Static; +export type WireAgentMessage = Static; const OperationErrorSchema = StrictObject({ code: Type.String(), @@ -167,7 +167,7 @@ const OperationResultBase = { startedAt: TimestampSchema, endedAt: TimestampSchema, }; -export const OperationResultRecordSchema = Type.Union([ +const OperationResultRecordSchema = Type.Union([ StrictObject({ ...OperationResultBase, status: Type.Union([Type.Literal("completed"), Type.Literal("declined"), Type.Literal("aborted")]), @@ -175,14 +175,14 @@ export const OperationResultRecordSchema = Type.Union([ StrictObject({ ...OperationResultBase, status: Type.Literal("failed"), error: OperationErrorSchema }), ]); export type OperationResultRecord = Static; -export const ToolOutputSchema = StrictObject({ - content: Type.Array(Type.Union([TextContentSchema, PromptImageSchema])), +const ToolOutputSchema = StrictObject({ + content: Type.Array(Type.Union([TextContentSchema, WireImageContentSchema])), details: Type.Optional(JsonValueSchema), usage: Type.Optional(UsageSchema), }); export type ToolOutput = Static; -export const AssistantMessageFrameSchema = Type.Union([ +const AssistantMessageFrameSchema = Type.Union([ StrictObject({ type: Type.Literal("start"), partial: AssistantMessageSchema }), StrictObject({ type: Type.Literal("text_start"), @@ -246,7 +246,7 @@ const MessageEntrySchema = StrictObject({ seq: Type.Integer({ minimum: 1 }), timestamp: TimestampSchema, type: Type.Literal("message"), - message: PromptMessageSchema, + message: WireAgentMessageSchema, terminate: Type.Optional(Type.Literal(true)), }); const CompactionEntrySchema = StrictObject({ @@ -256,7 +256,7 @@ const CompactionEntrySchema = StrictObject({ timestamp: TimestampSchema, type: Type.Literal("compaction"), summary: Type.String(), - retainedTail: Type.Array(PromptMessageSchema), + retainedTail: Type.Array(WireAgentMessageSchema), tokensBefore: Type.Number(), details: Type.Optional(JsonValueSchema), usage: Type.Optional(UsageSchema), @@ -283,7 +283,7 @@ const CustomEntrySchema = StrictObject({ customType: Type.String({ minLength: 1 }), data: Type.Optional(JsonValueSchema), }); -export const LaneEntrySchema = Type.Union([ +const LaneEntrySchema = Type.Union([ MessageEntrySchema, CompactionEntrySchema, BranchSummaryEntrySchema, @@ -291,7 +291,7 @@ export const LaneEntrySchema = Type.Union([ ]); export type LaneEntry = Static; -export const LaneQueuedItemSchema = Type.Union([ +const LaneQueuedItemSchema = Type.Union([ StrictObject({ entryId: IdSchema, kind: Type.Union([ @@ -301,7 +301,7 @@ export const LaneQueuedItemSchema = Type.Union([ Type.Literal("write"), ]), type: Type.Literal("message"), - message: PromptMessageSchema, + message: WireAgentMessageSchema, }), StrictObject({ entryId: IdSchema, @@ -311,7 +311,6 @@ export const LaneQueuedItemSchema = Type.Union([ data: Type.Optional(JsonValueSchema), }), ]); -export type LaneQueuedItem = Static; const RunningToolSchema = StrictObject({ toolCallId: Type.String(), toolName: Type.String(), @@ -390,8 +389,8 @@ export const LaneEventSchema = Type.Union([ StrictObject({ type: Type.Literal("operation_abort"), operationId: IdSchema, - steer: Type.Array(PromptMessageSchema), - followUp: Type.Array(PromptMessageSchema), + steer: Type.Array(WireAgentMessageSchema), + followUp: Type.Array(WireAgentMessageSchema), ...LaneEventBase, }), StrictObject({ @@ -402,7 +401,7 @@ export const LaneEventSchema = Type.Union([ StrictObject({ type: Type.Literal("message_start"), runId: Type.Optional(IdSchema), - message: PromptMessageSchema, + message: WireAgentMessageSchema, ...LaneEventBase, }), StrictObject({ @@ -415,7 +414,7 @@ export const LaneEventSchema = Type.Union([ StrictObject({ type: Type.Literal("message_end"), runId: Type.Optional(IdSchema), - message: PromptMessageSchema, + message: WireAgentMessageSchema, entryId: Type.Optional(IdSchema), ...LaneEventBase, }), diff --git a/packages/protocol/src/index.ts b/packages/protocol/src/index.ts index 76901e2f9..43b9f4500 100644 --- a/packages/protocol/src/index.ts +++ b/packages/protocol/src/index.ts @@ -3,5 +3,46 @@ export * from "./codec.ts"; export * from "./framing.ts"; export * from "./harness.ts"; export * from "./json-value.ts"; -export * from "./protocol.ts"; +export { + type AttachmentEnvelope, + type CancelEnvelope, + type ClientHello, + type ClientMessage, + createServiceCatalogueCall, + createServiceSubscribeCall, + createServiceUnsubscribeCall, + decodeLaneWatchRpcCall, + decodeServiceControlCall, + type EventEnvelope, + encodeLaneWatchRpcCall, + isServerId, + LaneWatchRpc, + type LaneWatchRpcCall, + type LaneWatchRpcResultUnion, + PROTOCOL_VERSION, + type ProtocolError, + type ProtocolErrorCode, + type ProtocolRpcCall, + ProtocolRpcCallSchema, + type ProtocolRpcResult, + parseServiceCatalogue, + parseServiceSubscriptionSnapshot, + type RequestEnvelope, + type ResponseEnvelope, + type RpcTarget, + type ServerEventEnvelope, + type ServerHello, + type ServerHelloError, + type ServerId, + type ServerMessage, + type ServiceCatalogueEntry, + type ServiceErrorCode, + ServiceErrorCodeSchema, + type ServiceEventEnvelope, + type ServiceMode, + type ServiceProviderUpdate, + ServiceProviderUpdateSchema, + type ServiceSubscriptionSnapshot, + type SessionTarget, +} from "./protocol.ts"; export * from "./rpc.ts"; diff --git a/packages/protocol/src/protocol.ts b/packages/protocol/src/protocol.ts index a0649c6e3..1e0d0177d 100644 --- a/packages/protocol/src/protocol.ts +++ b/packages/protocol/src/protocol.ts @@ -2,48 +2,23 @@ import Type, { type Static } from "typebox"; import { Check } from "typebox/value"; import { LaneEventSchema, LaneSnapshotSchema } from "./harness.ts"; import { type JsonValue, JsonValueSchema } from "./json-value.ts"; -import { - createRpcCallSchema, - createRpcResultSchema, - defineRpc, - type RpcArgs, - type RpcCall, - type RpcMethodName, - type RpcResult, - type RpcResultUnion, -} from "./rpc.ts"; +import { createRpcCallSchema, defineRpc, type RpcCall, type RpcMethodName, type RpcResultUnion } from "./rpc.ts"; export const PROTOCOL_VERSION = 6 as const; const IdSchema = Type.String({ minLength: 1 }); -const TimestampSchema = Type.Integer({ minimum: 0 }); const StrictObject = [0]>(properties: T) => Type.Object(properties, { additionalProperties: false }); -export const ServerIdSchema = Type.String({ +const ServerIdSchema = Type.String({ pattern: "^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$", }); -export const SessionIdSchema = IdSchema; export type ServerId = Static; -export type SessionId = Static; export function isServerId(value: unknown): value is ServerId { return Check(ServerIdSchema, value); } -/** Presentation-safe Session directory record. */ -export const SessionSummarySchema = StrictObject({ - serverId: ServerIdSchema, - sessionId: SessionIdSchema, - createdAt: TimestampSchema, -}); -export type SessionSummary = Static; - -export const SessionCreateOptionsSchema = StrictObject({ - id: Type.Optional(SessionIdSchema), -}); -export type SessionCreateOptions = Static; - /** Temporary typed facade for snapshot-first main-lane observation. */ export const LaneWatchRpc = defineRpc({ watch: { @@ -63,23 +38,19 @@ export const LaneWatchRpc = defineRpc({ result: StrictObject({ watchId: IdSchema }), }, }); -export type LaneWatchRpcManifest = typeof LaneWatchRpc; -export type LaneWatchRpcMethod = RpcMethodName; -export type LaneWatchRpcArgs = RpcArgs; -export type LaneWatchRpcResult = RpcResult; -export type LaneWatchRpcCall = RpcCall; -export type LaneWatchRpcResultUnion = RpcResultUnion; -export const LaneWatchRpcCallSchema = Type.Unsafe(createRpcCallSchema(LaneWatchRpc)); -export const LaneWatchRpcResultSchema = Type.Unsafe(createRpcResultSchema(LaneWatchRpc)); +type LaneWatchRpcMethod = RpcMethodName; +export type LaneWatchRpcCall = RpcCall; +export type LaneWatchRpcResultUnion = RpcResultUnion; +const LaneWatchRpcCallSchema = Type.Unsafe(createRpcCallSchema(LaneWatchRpc)); -export const ServiceModeSchema = Type.Union([Type.Literal("singleton"), Type.Literal("keyed")]); +const ServiceModeSchema = Type.Union([Type.Literal("singleton"), Type.Literal("keyed")]); export type ServiceMode = Static; -export const ServiceCatalogueEntrySchema = StrictObject({ +const ServiceCatalogueEntrySchema = StrictObject({ serviceId: IdSchema, mode: ServiceModeSchema, }); -export const ServiceCatalogueSchema = Type.Array(ServiceCatalogueEntrySchema); +const ServiceCatalogueSchema = Type.Array(ServiceCatalogueEntrySchema); export type ServiceCatalogueEntry = Static; export function parseServiceCatalogue(value: unknown): readonly ServiceCatalogueEntry[] { @@ -89,12 +60,10 @@ export function parseServiceCatalogue(value: unknown): readonly ServiceCatalogue return value; } -export const ServiceInstanceAddressSchema = StrictObject({ +const ServiceInstanceAddressSchema = StrictObject({ key: IdSchema, generation: Type.Integer({ minimum: 1 }), }); -export type ServiceInstanceAddress = Static; - /** Contract-agnostic service/member invocation carried by the transport. */ export const ProtocolRpcCallSchema = StrictObject({ serviceId: Type.String({ minLength: 1 }), @@ -105,7 +74,7 @@ export const ProtocolRpcCallSchema = StrictObject({ export type ProtocolRpcCall = Static; export type ProtocolRpcResult = JsonValue | undefined; -export const ServiceMemberSnapshotSchema = Type.Union([ +const ServiceMemberSnapshotSchema = Type.Union([ StrictObject({ name: IdSchema, kind: Type.Literal("method"), @@ -117,15 +86,11 @@ export const ServiceMemberSnapshotSchema = Type.Union([ value: JsonValueSchema, }), ]); -export type ServiceMemberSnapshot = Static; - -export const ServiceInstanceSnapshotSchema = StrictObject({ +const ServiceInstanceSnapshotSchema = StrictObject({ instance: Type.Optional(ServiceInstanceAddressSchema), members: Type.Array(ServiceMemberSnapshotSchema), }); -export type ServiceInstanceSnapshot = Static; - -export const ServiceSubscriptionSnapshotSchema = StrictObject({ +const ServiceSubscriptionSnapshotSchema = StrictObject({ serviceId: Type.String({ minLength: 1 }), mode: ServiceModeSchema, instances: Type.Array(ServiceInstanceSnapshotSchema), @@ -166,12 +131,12 @@ export const ServiceProviderUpdateSchema = Type.Union([ export type ServiceProviderUpdate = Static; // TODO: check if this should be part of Chord. -export const SERVICE_CONTROL_ID = "$chord.service"; -export const SERVICE_CATALOGUE_MEMBER = "catalogue"; -export const SERVICE_SUBSCRIBE_MEMBER = "subscribe"; -export const SERVICE_UNSUBSCRIBE_MEMBER = "unsubscribe"; +const SERVICE_CONTROL_ID = "$chord.service"; +const SERVICE_CATALOGUE_MEMBER = "catalogue"; +const SERVICE_SUBSCRIBE_MEMBER = "subscribe"; +const SERVICE_UNSUBSCRIBE_MEMBER = "unsubscribe"; -export type ServiceControlCall = +type ServiceControlCall = | { readonly type: "catalogue" } | { readonly type: "subscribe"; @@ -255,7 +220,7 @@ export const ServiceErrorCodeSchema = Type.Union([ ]); export type ServiceErrorCode = Static; -export const ProtocolErrorCodeSchema = Type.Union([ +const ProtocolErrorCodeSchema = Type.Union([ Type.Literal("version"), Type.Literal("wrong_server"), Type.Literal("session_not_found"), @@ -270,7 +235,7 @@ export const ProtocolErrorCodeSchema = Type.Union([ Type.Literal("cancelled"), Type.Literal("internal_error"), ]); -export const ProtocolErrorSchema = StrictObject({ +const ProtocolErrorSchema = StrictObject({ code: ProtocolErrorCodeSchema, message: Type.String(), }); @@ -278,46 +243,33 @@ export type ProtocolErrorCode = Static; export type ProtocolError = Static; /** Must be the first frame sent by a client. */ -export const ClientHelloSchema = StrictObject({ +const ClientHelloSchema = StrictObject({ type: Type.Literal("hello"), version: Type.Integer({ minimum: 0 }), }); export type ClientHello = Static; /** A server-wide call, fenced to one logical server. */ -export const ServerTargetSchema = StrictObject({ +const ServerTargetSchema = StrictObject({ serverId: ServerIdSchema, }); -export type ServerTarget = Static; - -/** Durable Session identity, unique within the addressed logical server. */ -export const SessionAddressSchema = StrictObject({ - serverId: ServerIdSchema, - sessionId: SessionIdSchema, -}); -export type SessionAddress = Static; - /** A session call, fenced to one logical server, durable session, and live attachment. */ -export const SessionTargetSchema = StrictObject({ +const SessionTargetSchema = StrictObject({ serverId: ServerIdSchema, - sessionId: SessionIdSchema, + sessionId: IdSchema, attachmentId: IdSchema, }); export type SessionTarget = Static; -export const RpcTargetSchema = Type.Union([ServerTargetSchema, SessionTargetSchema]); +const RpcTargetSchema = Type.Union([ServerTargetSchema, SessionTargetSchema]); export type RpcTarget = Static; -export function isSessionTarget(target: RpcTarget): target is SessionTarget { - return "sessionId" in target; -} - -export const RequestEnvelopeSchema = StrictObject({ +const RequestEnvelopeSchema = StrictObject({ type: Type.Literal("request"), id: IdSchema, target: RpcTargetSchema, call: ProtocolRpcCallSchema, }); -export const CancelEnvelopeSchema = StrictObject({ +const CancelEnvelopeSchema = StrictObject({ type: Type.Literal("cancel"), id: IdSchema, target: RpcTargetSchema, @@ -327,16 +279,16 @@ export type CancelEnvelope = Static; export const ClientMessageSchema = Type.Union([ClientHelloSchema, RequestEnvelopeSchema, CancelEnvelopeSchema]); export type ClientMessage = Static; -export const ServerHelloSchema = StrictObject({ +const ServerHelloSchema = StrictObject({ type: Type.Literal("hello"), version: Type.Literal(PROTOCOL_VERSION), serverId: ServerIdSchema, }); -export const ServerHelloErrorSchema = StrictObject({ +const ServerHelloErrorSchema = StrictObject({ type: Type.Literal("hello_error"), error: ProtocolErrorSchema, }); -export const ResponseEnvelopeSchema = Type.Union([ +const ResponseEnvelopeSchema = Type.Union([ StrictObject({ type: Type.Literal("response"), id: IdSchema, @@ -350,18 +302,18 @@ export const ResponseEnvelopeSchema = Type.Union([ error: ProtocolErrorSchema, }), ]); -export const EventEnvelopeSchema = StrictObject({ +const EventEnvelopeSchema = StrictObject({ type: Type.Literal("event"), watchId: IdSchema, event: LaneEventSchema, }); -export const ServiceEventEnvelopeSchema = StrictObject({ +const ServiceEventEnvelopeSchema = StrictObject({ type: Type.Literal("service_update"), subscriptionId: IdSchema, update: ServiceProviderUpdateSchema, }); /** Out-of-band update to this presentation's selected Session route. */ -export const AttachmentEnvelopeSchema = StrictObject({ +const AttachmentEnvelopeSchema = StrictObject({ type: Type.Literal("attachment"), attachment: Type.Union([SessionTargetSchema, Type.Null()]), }); diff --git a/packages/protocol/test/protocol.test.ts b/packages/protocol/test/protocol.test.ts index 1ad680fa9..fe4bda4cf 100644 --- a/packages/protocol/test/protocol.test.ts +++ b/packages/protocol/test/protocol.test.ts @@ -1,4 +1,3 @@ -import { Check } from "typebox/value"; import { describe, expect, test } from "vitest"; import { type ClientHello, @@ -31,8 +30,6 @@ import { type ServerHello, type ServerMessage, ServerMessageDecoder, - type SessionSummary, - SessionSummarySchema, } from "../src/index.ts"; const clientHello: ClientHello = { type: "hello", version: PROTOCOL_VERSION }; @@ -42,12 +39,6 @@ const serverHello: ServerHello = { serverId: "00000000-0000-4000-8000-000000000001", }; -const summary = { - serverId: "00000000-0000-4000-8000-000000000001", - sessionId: "session-1", - createdAt: 1, -} as const satisfies SessionSummary; - const laneSnapshot = { lane: "main", transcript: [], @@ -329,18 +320,6 @@ describe("protocol validation", () => { expect(() => parseClientMessage({ ...cancel, extra: true })).toThrow(ProtocolValidationError); }); - test("keeps presentation-safe Session summary validation at the typed service boundary", () => { - expect(Check(SessionSummarySchema, summary)).toBe(true); - expect(Check(SessionSummarySchema, { ...summary, cwd: "/private" })).toBe(false); - const message: ServerMessage = { - type: "response", - id: "request-1", - ok: true, - result: [summary], - }; - expect(parseServerMessage(message)).toEqual(message); - }); - test("validates lane watch snapshots and events", () => { const snapshotMessage: ServerMessage = { type: "response", diff --git a/packages/server/README.md b/packages/server/README.md index 74b73f40a..bb51f2fb4 100644 --- a/packages/server/README.md +++ b/packages/server/README.md @@ -21,6 +21,8 @@ import { type RoutedServerServiceHost, type RoutedSessionHandle, type ServerHost, + SessionAmbiguousError, + SessionNotFoundError, } from "@earendil-works/pi-server"; import { createUnixServer, getUnixSocketPath } from "@earendil-works/pi-server/unix"; @@ -31,8 +33,14 @@ async function startServer( const sessions = new MemorySessionRepo(); const host: ServerHost = { serverServices, - sessions: { - list: (context) => sessions.list(undefined, context), + async resolveSession(sessionId, context) { + const matches = (await sessions.list(undefined, context)) + .filter((metadata) => metadata.id === sessionId); + if (matches.length === 0) { + throw new SessionNotFoundError(`Unknown session: ${sessionId}`); + } + if (matches.length > 1) throw new SessionAmbiguousError(); + return matches[0]; }, async openSession(metadata, context) { const session = await sessions.open(metadata, context); @@ -62,7 +70,7 @@ async function startServer( } ``` -Applications supply a private Session catalog, server service host, and routed Session factory. Session discovery and management are application-owned services; the router uses the private catalog only to resolve attachment requests to concrete metadata. The host owns acquiring the worker-local Session and Harness. Failures are cleaned up in that worker. Neither an open JavaScript Session nor a Harness crosses the process boundary. +Applications supply a required server service host, a bounded Session resolver, and a routed Session factory. Session discovery and management are application-owned services; the protocol server only asks the resolver for metadata when routing an attachment. The host owns acquiring the worker-local Session and Harness. Failures are cleaned up in that worker. Neither an open JavaScript Session nor a Harness crosses the process boundary. `serverId` is a logical identity supplied by the launcher, not a socket address. The Unix preset requires an explicit physical `path`; `getUnixSocketPath()` derives one from a caller-selected directory. Choose a short, private runtime directory rather than deriving the route from an unbounded home-directory path. A long-lived launcher can reuse the same ID and path when replacing a server process. diff --git a/packages/server/src/server.ts b/packages/server/src/server.ts index 8b9b5961a..c0a3a9ea7 100644 --- a/packages/server/src/server.ts +++ b/packages/server/src/server.ts @@ -261,24 +261,21 @@ export class Server { } 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), + const services = await this.host.serverServices.attachClient( + { + attachSession: async (sessionId, context) => { + await this.sessions.attachClient(state, sessionId, context); }, - TODO_CONTEXT, - ); - if (this.closing || state.disconnected || state.stage !== "handshaking" || state.connection.closed) { - await services.release(TODO_CONTEXT); - return; - } - state.serverServices = services; + 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, diff --git a/packages/server/src/session-router.ts b/packages/server/src/session-router.ts index 362e86031..7507737e5 100644 --- a/packages/server/src/session-router.ts +++ b/packages/server/src/session-router.ts @@ -16,9 +16,7 @@ import { import { NotSupportedError, ServerDrainingError, - SessionAmbiguousError, SessionNotAttachedError, - SessionNotFoundError, WatchInUseError, WatchNotFoundError, } from "./errors.ts"; @@ -417,12 +415,7 @@ export class SessionRouter } private async open(sessionId: string, context: Context): Promise { - const matches = (await this.options.host.sessions.list(context)).filter( - (candidate) => candidate.id === sessionId, - ); - if (matches.length === 0) throw new SessionNotFoundError(`Unknown session: ${sessionId}`); - if (matches.length > 1) throw new SessionAmbiguousError(); - const metadata = matches[0]!; + const metadata = await this.options.host.resolveSession(sessionId, context); const handle = await this.options.host.openSession(metadata, context); if (this.options.isClosing()) { try { diff --git a/packages/server/src/testing/host.ts b/packages/server/src/testing/host.ts index 9f2a79f91..8d6d40379 100644 --- a/packages/server/src/testing/host.ts +++ b/packages/server/src/testing/host.ts @@ -1,6 +1,7 @@ import type { Context, Session, SessionMetadata } from "@earendil-works/pi-agent-core"; import { BACKGROUND_CONTEXT, MemorySessionRepo } from "@earendil-works/pi-agent-core"; import type { LaneEvent, LaneSnapshot, ProtocolRpcCall, ProtocolRpcResult } from "@earendil-works/pi-protocol"; +import { SessionAmbiguousError, SessionNotFoundError } from "../errors.ts"; import type { RoutedServerServiceHost, RoutedSessionHandle, RoutedSessionWatch, ServerHost } from "../types.ts"; export class Deferred { @@ -208,11 +209,6 @@ export class TestHarness { } } -interface ListDelay { - entered: Deferred; - release: Deferred; -} - export function createTestServerServices(): RoutedServerServiceHost { return { attachClient(presentation) { @@ -252,20 +248,15 @@ export class TestServerHost implements ServerHost { openSessionCount = 0; nextOpenSessionError?: Error; nextHarnessCloseError?: Error; - readonly sessions: ServerHost["sessions"] = { - list: async (context) => { - const delay = this.nextListDelay; - if (delay) { - this.nextListDelay = undefined; - delay.entered.resolve(undefined); - await delay.release.promise; - } - return this.repo.list(undefined, context); - }, - }; - private nextListDelay?: ListDelay; private nextOpenSessionGate?: OpenGate; + async resolveSession(sessionId: string, context: Context): Promise { + const matches = (await this.repo.list(undefined, context)).filter(({ id }) => id === sessionId); + if (matches.length === 0) throw new SessionNotFoundError(`Unknown session: ${sessionId}`); + if (matches.length > 1) throw new SessionAmbiguousError(); + return matches[0]!; + } + async openSession(metadata: SessionMetadata, context: Context): Promise { this.openSessionCount += 1; const gate = this.nextOpenSessionGate; @@ -303,12 +294,6 @@ export class TestServerHost implements ServerHost { return metadata; } - delayNextList(): ListDelay { - const delay = { entered: new Deferred(), release: new Deferred() }; - this.nextListDelay = delay; - return delay; - } - gateNextOpenSession(): OpenGate { const gate = { entered: new Deferred(), release: new Deferred() }; this.nextOpenSessionGate = gate; diff --git a/packages/server/src/types.ts b/packages/server/src/types.ts index be13c674a..3879272e8 100644 --- a/packages/server/src/types.ts +++ b/packages/server/src/types.ts @@ -73,9 +73,8 @@ export interface RoutedSessionHandle { /** Application capabilities used by server-wide management and Session routing. */ export interface ServerHost { - readonly serverServices?: RoutedServerServiceHost; - readonly sessions: { - list(context: Context): Promise; - }; + readonly serverServices: RoutedServerServiceHost; + /** Resolve one durable Session ID or throw a bounded routing error. */ + resolveSession(sessionId: string, context: Context): Promise; openSession(metadata: TMetadata, context: Context): Promise; } diff --git a/packages/server/test/conformance.test.ts b/packages/server/test/conformance.test.ts index 0d7667892..254a8e1c1 100644 --- a/packages/server/test/conformance.test.ts +++ b/packages/server/test/conformance.test.ts @@ -2,6 +2,7 @@ import { BACKGROUND_CONTEXT, type SessionMetadata } from "@earendil-works/pi-age import type { ProtocolRpcCall } from "@earendil-works/pi-protocol"; import { afterEach, describe, expect, test } from "vitest"; import type { ByteConnection, ByteConnectionHandler } from "../src/connection.ts"; +import { SessionAmbiguousError } from "../src/errors.ts"; import { Server } from "../src/server.ts"; import { createTestServerServices, @@ -99,7 +100,7 @@ describe("Session protocol", () => { let received: BackendMetadata | undefined; const host: ServerHost = { serverServices: createTestServerServices(), - sessions: { list: async () => [metadata] }, + resolveSession: async () => metadata, openSession: async (candidate) => { received = candidate; return { @@ -124,7 +125,7 @@ describe("Session protocol", () => { await backing.seed("session-1"); let releaseCount = 0; const host: ServerHost = { - sessions: backing.sessions, + resolveSession: (sessionId, context) => backing.resolveSession(sessionId, context), openSession: (metadata, context) => backing.openSession(metadata, context), serverServices: { attachClient(presentation) { @@ -398,11 +399,8 @@ describe("Session protocol", () => { type BackendMetadata = SessionMetadata & { path: string }; const host: ServerHost = { serverServices: createTestServerServices(), - sessions: { - list: async () => [ - { id: "duplicate", createdAt: 1, storageVersion: 1, cwd: "/one", path: "/one/session.jsonl" }, - { id: "duplicate", createdAt: 2, storageVersion: 1, cwd: "/two", path: "/two/session.jsonl" }, - ], + resolveSession: async () => { + throw new SessionAmbiguousError(); }, openSession: async () => { throw new Error("must not create a Harness for an ambiguous session"); @@ -460,7 +458,7 @@ describe("routed Session acquisition failures", () => { let releaseCount = 0; const host: ServerHost = { serverServices: createTestServerServices(), - sessions: { list: async () => [metadata] }, + resolveSession: async () => metadata, openSession: async () => ({ terminated: terminated.promise, attachClient: async () => { diff --git a/packages/server/test/protocol.test.ts b/packages/server/test/protocol.test.ts index 7dfba1f9a..44e272b2d 100644 --- a/packages/server/test/protocol.test.ts +++ b/packages/server/test/protocol.test.ts @@ -91,7 +91,7 @@ test("accepts fragmented hello and request frames", async () => { }; const frame = encodeClientMessage(request); await client.sendFragmentedMessage(request, Math.floor(frame.byteLength / 2)); - await expect(response).resolves.toMatchObject({ ok: true, result: [] }); + await expect(response).resolves.toMatchObject({ ok: false, error: { code: "internal_error" } }); }); test.each([ @@ -137,8 +137,8 @@ test("processes a hello and request coalesced in one byte chunk", async () => { await expect(client.next((message) => message.type === "response")).resolves.toMatchObject({ type: "response", id: "request-1", - ok: true, - result: [], + ok: false, + error: { code: "internal_error" }, }); });