mirror of
https://github.com/badlogic/pi-mono.git
synced 2026-11-05 23:37:25 +00:00
feat(coding-agent): tighten experimental service boundaries
This commit is contained in:
Armin Ronacher
parent
5cf1b95c87
commit
8e374e4dc2
27 changed files with 338 additions and 329 deletions
|
|
@ -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 },
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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<PromptMessage, { role: "assistant" }>): string {
|
||||
function messageText(message: Extract<WireAgentMessage, { role: "assistant" }>): string {
|
||||
return message.content
|
||||
.filter((content) => content.type === "text")
|
||||
.map((content) => content.text)
|
||||
|
|
|
|||
|
|
@ -25,18 +25,18 @@ import type {
|
|||
LaneEvent,
|
||||
LaneSnapshot,
|
||||
OperationResultRecord,
|
||||
PromptImage,
|
||||
PromptMessage,
|
||||
ToolOutput,
|
||||
WireAgentMessage,
|
||||
WireImageContent,
|
||||
} from "@earendil-works/pi-protocol";
|
||||
|
||||
type WireAssistantMessage = Extract<PromptMessage, { role: "assistant" }>;
|
||||
type WireUserMessage = Extract<PromptMessage, { role: "user" }>;
|
||||
type WireToolResultMessage = Extract<PromptMessage, { role: "toolResult" }>;
|
||||
type WireBashExecutionMessage = Extract<PromptMessage, { role: "bashExecution" }>;
|
||||
type WireCustomMessage = Extract<PromptMessage, { role: "custom" }>;
|
||||
type WireBranchSummaryMessage = Extract<PromptMessage, { role: "branchSummary" }>;
|
||||
type WireCompactionSummaryMessage = Extract<PromptMessage, { role: "compactionSummary" }>;
|
||||
type WireAssistantMessage = Extract<WireAgentMessage, { role: "assistant" }>;
|
||||
type WireUserMessage = Extract<WireAgentMessage, { role: "user" }>;
|
||||
type WireToolResultMessage = Extract<WireAgentMessage, { role: "toolResult" }>;
|
||||
type WireBashExecutionMessage = Extract<WireAgentMessage, { role: "bashExecution" }>;
|
||||
type WireCustomMessage = Extract<WireAgentMessage, { role: "custom" }>;
|
||||
type WireBranchSummaryMessage = Extract<WireAgentMessage, { role: "branchSummary" }>;
|
||||
type WireCompactionSummaryMessage = Extract<WireAgentMessage, { role: "compactionSummary" }>;
|
||||
type SameKeys<TLeft, TRight> = Exclude<keyof TLeft, keyof TRight> extends never
|
||||
? Exclude<keyof TRight, keyof TLeft> extends never
|
||||
? true
|
||||
|
|
@ -46,7 +46,7 @@ type Assert<T extends true> = T;
|
|||
|
||||
/** Compile-time alarms for field additions on either side of each closed message adapter. */
|
||||
export type HarnessWireAdapterCompatibility = [
|
||||
Assert<SameKeys<PromptImage, ImageContent>>,
|
||||
Assert<SameKeys<WireImageContent, ImageContent>>,
|
||||
Assert<SameKeys<WireUserMessage, UserMessage>>,
|
||||
Assert<SameKeys<WireAssistantMessage, AssistantMessage>>,
|
||||
Assert<SameKeys<WireToolResultMessage, ToolResultMessage>>,
|
||||
|
|
@ -55,7 +55,7 @@ export type HarnessWireAdapterCompatibility = [
|
|||
Assert<SameKeys<WireBranchSummaryMessage, BranchSummaryMessage>>,
|
||||
Assert<SameKeys<WireCompactionSummaryMessage, CompactionSummaryMessage>>,
|
||||
];
|
||||
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<unknown>): 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) {
|
||||
|
|
|
|||
|
|
@ -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<void>;
|
||||
readonly #abortController = new AbortController();
|
||||
readonly #removeConnectionListener: () => void;
|
||||
|
|
@ -340,7 +351,7 @@ export class RadiusClientReconnect {
|
|||
#reconnecting: Promise<void> | undefined;
|
||||
#disposed = false;
|
||||
|
||||
constructor(client: Client, reattach: (sessionId: string) => Promise<void>) {
|
||||
constructor(client: RadiusReconnectClient, reattach: (sessionId: string) => Promise<void>) {
|
||||
this.#client = client;
|
||||
this.#reattach = reattach;
|
||||
this.#desiredSessionId = client.attachment?.sessionId;
|
||||
|
|
|
|||
|
|
@ -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<JsonlSessionMetadata>["sessions"]["list"] = async (context) => {
|
||||
const listSessions = async (context: Context): Promise<JsonlSessionMetadata[]> => {
|
||||
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<JsonlSessionMetadata> => {
|
||||
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<JsonlSessionMetadata> = {
|
||||
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);
|
||||
|
|
|
|||
|
|
@ -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.
|
||||
|
||||
|
|
|
|||
|
|
@ -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 {
|
||||
|
|
|
|||
|
|
@ -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<void> => {
|
||||
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<void> => {
|
||||
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);
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
|
|
|
|||
|
|
@ -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";
|
||||
|
||||
|
|
|
|||
|
|
@ -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");
|
||||
},
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
90
packages/coding-agent/test/experimental-service-binding.ts
Normal file
90
packages/coding-agent/test/experimental-service-binding.ts
Normal file
|
|
@ -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<void> => {
|
||||
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<void> => {
|
||||
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<unknown>[], 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);
|
||||
}
|
||||
|
|
@ -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.
|
||||
|
||||
|
|
|
|||
|
|
@ -6,25 +6,25 @@ const TimestampSchema = Type.Integer({ minimum: 0 });
|
|||
const StrictObject = <const T extends Parameters<typeof Type.Object>[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<typeof PromptImageSchema>;
|
||||
export type WireImageContent = Static<typeof WireImageContentSchema>;
|
||||
|
||||
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<typeof PromptMessageSchema>;
|
||||
export type WireAgentMessage = Static<typeof WireAgentMessageSchema>;
|
||||
|
||||
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<typeof OperationResultRecordSchema>;
|
||||
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<typeof ToolOutputSchema>;
|
||||
|
||||
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<typeof LaneEntrySchema>;
|
||||
|
||||
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<typeof LaneQueuedItemSchema>;
|
||||
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,
|
||||
}),
|
||||
|
|
|
|||
|
|
@ -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";
|
||||
|
|
|
|||
|
|
@ -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 = <const T extends Parameters<typeof Type.Object>[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<typeof ServerIdSchema>;
|
||||
export type SessionId = Static<typeof SessionIdSchema>;
|
||||
|
||||
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<typeof SessionSummarySchema>;
|
||||
|
||||
export const SessionCreateOptionsSchema = StrictObject({
|
||||
id: Type.Optional(SessionIdSchema),
|
||||
});
|
||||
export type SessionCreateOptions = Static<typeof SessionCreateOptionsSchema>;
|
||||
|
||||
/** 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<LaneWatchRpcManifest>;
|
||||
export type LaneWatchRpcArgs<TMethod extends LaneWatchRpcMethod> = RpcArgs<LaneWatchRpcManifest, TMethod>;
|
||||
export type LaneWatchRpcResult<TMethod extends LaneWatchRpcMethod> = RpcResult<LaneWatchRpcManifest, TMethod>;
|
||||
export type LaneWatchRpcCall = RpcCall<LaneWatchRpcManifest>;
|
||||
export type LaneWatchRpcResultUnion = RpcResultUnion<LaneWatchRpcManifest>;
|
||||
export const LaneWatchRpcCallSchema = Type.Unsafe<LaneWatchRpcCall>(createRpcCallSchema(LaneWatchRpc));
|
||||
export const LaneWatchRpcResultSchema = Type.Unsafe<LaneWatchRpcResultUnion>(createRpcResultSchema(LaneWatchRpc));
|
||||
type LaneWatchRpcMethod = RpcMethodName<typeof LaneWatchRpc>;
|
||||
export type LaneWatchRpcCall = RpcCall<typeof LaneWatchRpc>;
|
||||
export type LaneWatchRpcResultUnion = RpcResultUnion<typeof LaneWatchRpc>;
|
||||
const LaneWatchRpcCallSchema = Type.Unsafe<LaneWatchRpcCall>(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<typeof ServiceModeSchema>;
|
||||
|
||||
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<typeof ServiceCatalogueEntrySchema>;
|
||||
|
||||
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<typeof ServiceInstanceAddressSchema>;
|
||||
|
||||
/** 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<typeof ProtocolRpcCallSchema>;
|
||||
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<typeof ServiceMemberSnapshotSchema>;
|
||||
|
||||
export const ServiceInstanceSnapshotSchema = StrictObject({
|
||||
const ServiceInstanceSnapshotSchema = StrictObject({
|
||||
instance: Type.Optional(ServiceInstanceAddressSchema),
|
||||
members: Type.Array(ServiceMemberSnapshotSchema),
|
||||
});
|
||||
export type ServiceInstanceSnapshot = Static<typeof ServiceInstanceSnapshotSchema>;
|
||||
|
||||
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<typeof ServiceProviderUpdateSchema>;
|
||||
|
||||
// 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<typeof ServiceErrorCodeSchema>;
|
||||
|
||||
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<typeof ProtocolErrorCodeSchema>;
|
|||
export type ProtocolError = Static<typeof ProtocolErrorSchema>;
|
||||
|
||||
/** 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<typeof ClientHelloSchema>;
|
||||
|
||||
/** A server-wide call, fenced to one logical server. */
|
||||
export const ServerTargetSchema = StrictObject({
|
||||
const ServerTargetSchema = StrictObject({
|
||||
serverId: ServerIdSchema,
|
||||
});
|
||||
export type ServerTarget = Static<typeof ServerTargetSchema>;
|
||||
|
||||
/** Durable Session identity, unique within the addressed logical server. */
|
||||
export const SessionAddressSchema = StrictObject({
|
||||
serverId: ServerIdSchema,
|
||||
sessionId: SessionIdSchema,
|
||||
});
|
||||
export type SessionAddress = Static<typeof SessionAddressSchema>;
|
||||
|
||||
/** 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<typeof SessionTargetSchema>;
|
||||
export const RpcTargetSchema = Type.Union([ServerTargetSchema, SessionTargetSchema]);
|
||||
const RpcTargetSchema = Type.Union([ServerTargetSchema, SessionTargetSchema]);
|
||||
export type RpcTarget = Static<typeof RpcTargetSchema>;
|
||||
|
||||
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<typeof CancelEnvelopeSchema>;
|
|||
export const ClientMessageSchema = Type.Union([ClientHelloSchema, RequestEnvelopeSchema, CancelEnvelopeSchema]);
|
||||
export type ClientMessage = Static<typeof ClientMessageSchema>;
|
||||
|
||||
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()]),
|
||||
});
|
||||
|
|
|
|||
|
|
@ -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",
|
||||
|
|
|
|||
|
|
@ -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.
|
||||
|
||||
|
|
|
|||
|
|
@ -261,24 +261,21 @@ export class Server<TMetadata extends SessionMetadata = SessionMetadata> {
|
|||
}
|
||||
|
||||
if (this.closing || state.disconnected || state.stage !== "handshaking" || state.connection.closed) return;
|
||||
const serviceHost = this.host.serverServices;
|
||||
if (serviceHost !== undefined) {
|
||||
const services = await serviceHost.attachClient(
|
||||
{
|
||||
attachSession: async (sessionId, context) => {
|
||||
await this.sessions.attachClient(state, sessionId, context);
|
||||
},
|
||||
detachSession: (context) => this.sessions.detachClient(state, context),
|
||||
prepareSessionRemoval: (sessionId, context) => this.sessions.removeSession(sessionId, context),
|
||||
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,
|
||||
|
|
|
|||
|
|
@ -16,9 +16,7 @@ import {
|
|||
import {
|
||||
NotSupportedError,
|
||||
ServerDrainingError,
|
||||
SessionAmbiguousError,
|
||||
SessionNotAttachedError,
|
||||
SessionNotFoundError,
|
||||
WatchInUseError,
|
||||
WatchNotFoundError,
|
||||
} from "./errors.ts";
|
||||
|
|
@ -417,12 +415,7 @@ export class SessionRouter<TMetadata extends SessionMetadata = SessionMetadata>
|
|||
}
|
||||
|
||||
private async open(sessionId: string, context: Context): Promise<HostedSession> {
|
||||
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 {
|
||||
|
|
|
|||
|
|
@ -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<T> {
|
||||
|
|
@ -208,11 +209,6 @@ export class TestHarness {
|
|||
}
|
||||
}
|
||||
|
||||
interface ListDelay {
|
||||
entered: Deferred<void>;
|
||||
release: Deferred<void>;
|
||||
}
|
||||
|
||||
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<SessionMetadata> {
|
||||
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<RoutedSessionHandle> {
|
||||
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<void>(), release: new Deferred<void>() };
|
||||
this.nextListDelay = delay;
|
||||
return delay;
|
||||
}
|
||||
|
||||
gateNextOpenSession(): OpenGate {
|
||||
const gate = { entered: new Deferred<void>(), release: new Deferred<void>() };
|
||||
this.nextOpenSessionGate = gate;
|
||||
|
|
|
|||
|
|
@ -73,9 +73,8 @@ export interface RoutedSessionHandle {
|
|||
|
||||
/** Application capabilities used by server-wide management and Session routing. */
|
||||
export interface ServerHost<TMetadata extends SessionMetadata = SessionMetadata> {
|
||||
readonly serverServices?: RoutedServerServiceHost;
|
||||
readonly sessions: {
|
||||
list(context: Context): Promise<TMetadata[]>;
|
||||
};
|
||||
readonly serverServices: RoutedServerServiceHost;
|
||||
/** Resolve one durable Session ID or throw a bounded routing error. */
|
||||
resolveSession(sessionId: string, context: Context): Promise<TMetadata>;
|
||||
openSession(metadata: TMetadata, context: Context): Promise<RoutedSessionHandle>;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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<BackendMetadata> = {
|
||||
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<BackendMetadata> = {
|
||||
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 () => {
|
||||
|
|
|
|||
|
|
@ -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" },
|
||||
});
|
||||
});
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue