sscoble/pi-mono
1
0
Fork 0
mirror of https://github.com/badlogic/pi-mono.git synced 2026-11-05 23:37:25 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 4

feat: move service wire semantics into Chord

Browse source 
Keep protocol payloads transport-opaque and validate service calls, state snapshots, updates, and errors at Chord boundaries. Ensure successful OpenAI Responses messages remain strict JSON so terminal Transcript updates reach clients.
This commit is contained in:
Armin Ronacher 2026-09-01 00:53:40 +02:00
parent fbf2db6353
commit 1a7bc80e7c
38 changed files with 841 additions and 645 deletions
Download patch file Download diff file Expand all files Collapse all files

1
package-lock.json generated
View file

@ -5383,6 +5383,7 @@
"version": "0.84.4",
"license": "MIT",
"dependencies": {
"@earendil-works/chord": "^0.84.4",
"@earendil-works/pi-protocol": "^0.84.4"
},
"devDependencies": {

3
packages/ai/src/api/openai-responses-shared.ts
View file

@ -590,7 +590,8 @@ export async function processResponsesStream<TApi extends Api>(
output.rawStopReason = incompleteReason ? `${status}.${incompleteReason}` : status;
const mappedStop = mapStopReason(status, incompleteReason);
output.stopReason = mappedStop.stopReason;
output.errorMessage = mappedStop.errorMessage;
if (mappedStop.errorMessage === undefined) delete output.errorMessage;
else output.errorMessage = mappedStop.errorMessage;
if (output.content.some((b) => b.type === "toolCall") && output.stopReason === "stop") {
output.stopReason = "toolUse";
}

9
packages/ai/test/openai-responses-namespace.test.ts
View file

@ -109,6 +109,15 @@ function getToolCall(output: AssistantMessage): ToolCall {
return block;
}
describe("OpenAI Responses terminal messages", () => {
it("omits an absent error message", async () => {
const output = createOutput();
await processResponsesStream(createFunctionCallEvents(), output, new AssistantMessageEventStream(), model);
expect(output).not.toHaveProperty("errorMessage");
});
});
describe("OpenAI Responses tool-call namespaces", () => {
it("round-trips a function namespace received only on output_item.done", async () => {
const output = createOutput();

19
packages/chord/README.md
View file

@ -66,6 +66,25 @@ the root API.
Chord-owned identifiers use the `chord.*` namespace and its reserved service
prefix is `$chord.*`.
## Remote service adapters
Chord owns its transport-independent service wire grammar. Adapters use
`createServiceCatalogueCall()`, `createServiceSubscribeCall()`, and
`createServiceUnsubscribeCall()` for `$chord.service` control calls, then
`decodeServiceControlCall()` on the provider side. `parseServiceCall()`,
`parseServiceCatalogue()`, and the decoded/wire snapshot and update parsers
validate Chord semantics after an adapter has established a strict-JSON
boundary. `RemoteServiceErrorCode` and `REMOTE_SERVICE_ERROR_CODES` define the
service errors that may cross that boundary.
Replicated state operations use one `createServiceStateEncoder()` at the
provider side and one `createServiceStateDecoder()` at the consumer side for
each subscription. Those registries create an independent Delta path dictionary
for every instance/member state and reset it on replacement, unavailability,
close, or fresh hydration. Applications may place these values inside any
routing, request, response, or event envelope; Chord does not prescribe that
outer protocol.
## Tracking JSON deltas
Import the standalone delta primitive from `@earendil-works/chord/delta`:

31
packages/chord/src/index.ts
View file

@ -11,8 +11,36 @@ export {
replicatedState,
} from "./api.ts";
export { isJsonValue } from "./json.ts";
export { RemoteServiceError } from "./services/errors.ts";
export {
isRemoteServiceErrorCode,
REMOTE_SERVICE_ERROR_CODES,
RemoteServiceError,
type RemoteServiceErrorCode,
} from "./services/errors.ts";
export { RemoteServiceProvider } from "./services/provider.ts";
export {
createServiceStateDecoder,
createServiceStateEncoder,
type ServiceStateDecoder,
type ServiceStateEncoder,
} from "./services/state-codec.ts";
export {
createServiceCatalogueCall,
createServiceSubscribeCall,
createServiceUnsubscribeCall,
decodeServiceControlCall,
parseServiceCall,
parseServiceCatalogue,
parseServiceProviderUpdate,
parseServiceSubscriptionSnapshot,
parseWireServiceProviderUpdate,
parseWireServiceSubscriptionSnapshot,
type ServiceControlCall,
type WireServiceInstanceSnapshot,
type WireServiceMemberSnapshot,
type WireServiceProviderUpdate,
type WireServiceSubscriptionSnapshot,
} from "./services/wire.ts";
export type {
Context,
ContextKey,
@ -27,7 +55,6 @@ export type {
MutableReplicatedState,
RemoteServiceBinding,
RemoteServiceBindingOptions,
RemoteServiceErrorCode,
RemoteServiceSource,
RemoteServices,
RemoteServiceTransport,

17
packages/chord/src/services/errors.ts
View file

@ -1,4 +1,19 @@
import type { RemoteServiceErrorCode } from "../types.ts";
export const REMOTE_SERVICE_ERROR_CODES = [
"service_not_allowed",
"service_not_found",
"service_mode_mismatch",
"service_member_not_found",
"service_member_mismatch",
"service_instance_not_found",
"service_stale_instance",
"service_invalid_value",
] as const;
export type RemoteServiceErrorCode = (typeof REMOTE_SERVICE_ERROR_CODES)[number];
export function isRemoteServiceErrorCode(value: unknown): value is RemoteServiceErrorCode {
return typeof value === "string" && REMOTE_SERVICE_ERROR_CODES.includes(value as RemoteServiceErrorCode);
}
export class RemoteServiceError extends Error {
readonly code: RemoteServiceErrorCode;

33
packages/protocol/src/service-state.ts → packages/chord/src/services/state-codec.ts
View file

@ -1,27 +1,22 @@
import { type Decoder, decoder, type Encoder, encoder } from "../delta/index.ts";
import type {
ServiceInstanceSnapshot as DecodedServiceInstanceSnapshot,
ServiceProviderUpdate as DecodedServiceProviderUpdate,
ServiceSubscriptionSnapshot as DecodedServiceSubscriptionSnapshot,
ServiceInstanceAddress,
} from "@earendil-works/chord";
import { type Decoder, decoder, type Encoder, encoder } from "@earendil-works/chord/delta";
import type {
ServiceProviderUpdate as WireServiceProviderUpdate,
ServiceSubscriptionSnapshot as WireServiceSubscriptionSnapshot,
} from "./protocol.ts";
ServiceInstanceSnapshot,
ServiceProviderUpdate,
ServiceSubscriptionSnapshot,
} from "../types.ts";
import type { WireServiceProviderUpdate, WireServiceSubscriptionSnapshot } from "./wire.ts";
export type { DecodedServiceProviderUpdate, DecodedServiceSubscriptionSnapshot };
/** Stateful operation encoders for every state in one service subscription. */
/** Stateful operation encoders for every replicated state in one service subscription. */
export interface ServiceStateEncoder {
encodeSnapshot(snapshot: DecodedServiceSubscriptionSnapshot): WireServiceSubscriptionSnapshot;
encodeUpdate(update: DecodedServiceProviderUpdate): WireServiceProviderUpdate;
encodeSnapshot(snapshot: ServiceSubscriptionSnapshot): WireServiceSubscriptionSnapshot;
encodeUpdate(update: ServiceProviderUpdate): WireServiceProviderUpdate;
}
/** Stateful operation decoders for every state in one service subscription. */
/** Stateful operation decoders for every replicated state in one service subscription. */
export interface ServiceStateDecoder {
decodeSnapshot(snapshot: WireServiceSubscriptionSnapshot): DecodedServiceSubscriptionSnapshot;
decodeUpdate(update: WireServiceProviderUpdate): DecodedServiceProviderUpdate;
decodeSnapshot(snapshot: WireServiceSubscriptionSnapshot): ServiceSubscriptionSnapshot;
decodeUpdate(update: WireServiceProviderUpdate): ServiceProviderUpdate;
}
interface CodecEntry<C> {
@ -123,7 +118,7 @@ export function createServiceStateDecoder(): ServiceStateDecoder {
}
function encodeInstance(
instance: DecodedServiceInstanceSnapshot,
instance: ServiceInstanceSnapshot,
codecs: StateCodecRegistry<Encoder>,
): WireServiceSubscriptionSnapshot["instances"][number] {
return {
@ -139,7 +134,7 @@ function encodeInstance(
function decodeInstance(
instance: WireServiceSubscriptionSnapshot["instances"][number],
codecs: StateCodecRegistry<Decoder>,
): DecodedServiceInstanceSnapshot {
): ServiceInstanceSnapshot {
return {
...instance,
members: instance.members.map((member) =>

234
packages/chord/src/services/wire.ts Normal file
View file

@ -0,0 +1,234 @@
import { assertValidOp, assertValidWireOp, type WireOp } from "../delta/index.ts";
import type {
ServiceCall,
ServiceCatalogueEntry,
ServiceInstanceAddress,
ServiceMode,
ServiceProviderUpdate,
ServiceSubscriptionSnapshot,
} from "../types.ts";
export type WireServiceMemberSnapshot =
| { readonly name: string; readonly kind: "method" }
| { readonly name: string; readonly kind: "state"; readonly sequence: number; readonly ops: readonly WireOp[] };
export type WireServiceInstanceSnapshot = {
readonly instance?: ServiceInstanceAddress;
readonly members: readonly WireServiceMemberSnapshot[];
};
export type WireServiceSubscriptionSnapshot = {
readonly serviceId: string;
readonly mode: ServiceMode;
readonly instances: readonly WireServiceInstanceSnapshot[];
};
export type WireServiceProviderUpdate =
| {
readonly type: "state";
readonly instance?: ServiceInstanceAddress;
readonly member: string;
readonly sequence: number;
readonly ops: readonly WireOp[];
}
| { readonly type: "unavailable" }
| { readonly type: "replaced"; readonly snapshot: WireServiceInstanceSnapshot }
| { readonly type: "spawned"; readonly instance: WireServiceInstanceSnapshot }
| { readonly type: "closed"; readonly instance: ServiceInstanceAddress };
const SERVICE_CONTROL_ID = "$chord.service";
const SERVICE_CATALOGUE_MEMBER = "catalogue";
const SERVICE_SUBSCRIBE_MEMBER = "subscribe";
const SERVICE_UNSUBSCRIBE_MEMBER = "unsubscribe";
export type ServiceControlCall =
| { readonly type: "catalogue" }
| {
readonly type: "subscribe";
readonly subscriptionId: string;
readonly serviceId: string;
readonly mode: ServiceMode;
}
| { readonly type: "unsubscribe"; readonly subscriptionId: string };
export function createServiceCatalogueCall(): ServiceCall {
return { serviceId: SERVICE_CONTROL_ID, member: SERVICE_CATALOGUE_MEMBER, args: [] };
}
export function createServiceSubscribeCall(subscriptionId: string, serviceId: string, mode: ServiceMode): ServiceCall {
return { serviceId: SERVICE_CONTROL_ID, member: SERVICE_SUBSCRIBE_MEMBER, args: [subscriptionId, serviceId, mode] };
}
export function createServiceUnsubscribeCall(subscriptionId: string): ServiceCall {
return { serviceId: SERVICE_CONTROL_ID, member: SERVICE_UNSUBSCRIBE_MEMBER, args: [subscriptionId] };
}
export function decodeServiceControlCall(call: ServiceCall): ServiceControlCall | undefined {
if (call.serviceId !== SERVICE_CONTROL_ID || call.instance !== undefined) return undefined;
if (call.member === SERVICE_CATALOGUE_MEMBER && call.args.length === 0) return { type: "catalogue" };
if (
call.member === SERVICE_SUBSCRIBE_MEMBER &&
call.args.length === 3 &&
isId(call.args[0]) &&
isId(call.args[1]) &&
(call.args[2] === "singleton" || call.args[2] === "keyed")
) {
return {
type: "subscribe",
subscriptionId: call.args[0],
serviceId: call.args[1],
mode: call.args[2],
};
}
if (call.member === SERVICE_UNSUBSCRIBE_MEMBER && call.args.length === 1 && isId(call.args[0])) {
return { type: "unsubscribe", subscriptionId: call.args[0] };
}
return undefined;
}
export function parseServiceCall(value: unknown): ServiceCall {
const call = record(value, "service call");
assertKeys(call, ["serviceId", "member", "args"], ["instance"], "service call");
if (!isId(call.serviceId) || !isId(call.member) || !Array.isArray(call.args)) {
throw new TypeError("Invalid service call");
}
if (call.instance !== undefined) assertAddress(call.instance);
return value as ServiceCall;
}
export function parseServiceCatalogue(value: unknown): readonly ServiceCatalogueEntry[] {
if (!Array.isArray(value)) throw new TypeError("Invalid service catalogue");
const ids = new Set<string>();
for (const candidate of value) {
const entry = record(candidate, "service catalogue entry");
assertKeys(entry, ["serviceId", "mode"], [], "service catalogue entry");
if (!isId(entry.serviceId) || !isMode(entry.mode) || ids.has(entry.serviceId)) {
throw new TypeError("Invalid service catalogue");
}
ids.add(entry.serviceId);
}
return value as unknown as readonly ServiceCatalogueEntry[];
}
export function parseServiceSubscriptionSnapshot(value: unknown): ServiceSubscriptionSnapshot {
assertSubscriptionSnapshot(value, assertValidOp);
return value as ServiceSubscriptionSnapshot;
}
export function parseWireServiceSubscriptionSnapshot(value: unknown): WireServiceSubscriptionSnapshot {
assertSubscriptionSnapshot(value, assertValidWireOp);
return value as WireServiceSubscriptionSnapshot;
}
export function parseServiceProviderUpdate(value: unknown): ServiceProviderUpdate {
assertProviderUpdate(value, assertValidOp);
return value as ServiceProviderUpdate;
}
export function parseWireServiceProviderUpdate(value: unknown): WireServiceProviderUpdate {
assertProviderUpdate(value, assertValidWireOp);
return value as WireServiceProviderUpdate;
}
function assertSubscriptionSnapshot(value: unknown, assertOp: (value: unknown) => void): void {
const snapshot = record(value, "service subscription snapshot");
assertKeys(snapshot, ["serviceId", "mode", "instances"], [], "service subscription snapshot");
if (!isId(snapshot.serviceId) || !isMode(snapshot.mode) || !Array.isArray(snapshot.instances)) {
throw new TypeError("Invalid service subscription snapshot");
}
for (const instance of snapshot.instances) assertInstance(instance, assertOp);
}
function assertProviderUpdate(value: unknown, assertOp: (value: unknown) => void): void {
const update = record(value, "service provider update");
switch (update.type) {
case "state":
assertKeys(update, ["type", "member", "sequence", "ops"], ["instance"], "state update");
if (!isId(update.member) || !isInteger(update.sequence, 1) || !Array.isArray(update.ops)) {
throw new TypeError("Invalid service state update");
}
if (update.instance !== undefined) assertAddress(update.instance);
for (const op of update.ops) assertOp(op);
return;
case "unavailable":
assertKeys(update, ["type"], [], "unavailable update");
return;
case "replaced":
assertKeys(update, ["type", "snapshot"], [], "replacement update");
assertInstance(update.snapshot, assertOp);
return;
case "spawned":
assertKeys(update, ["type", "instance"], [], "spawn update");
assertInstance(update.instance, assertOp);
return;
case "closed":
assertKeys(update, ["type", "instance"], [], "close update");
assertAddress(update.instance);
return;
default:
throw new TypeError("Invalid service provider update");
}
}
function assertInstance(value: unknown, assertOp: (value: unknown) => void): void {
const instance = record(value, "service instance snapshot");
assertKeys(instance, ["members"], ["instance"], "service instance snapshot");
if (instance.instance !== undefined) assertAddress(instance.instance);
if (!Array.isArray(instance.members)) throw new TypeError("Invalid service instance snapshot");
for (const candidate of instance.members) {
const member = record(candidate, "service member snapshot");
if (member.kind === "method") {
assertKeys(member, ["name", "kind"], [], "service method snapshot");
if (!isId(member.name)) throw new TypeError("Invalid service method snapshot");
continue;
}
if (member.kind === "state") {
assertKeys(member, ["name", "kind", "sequence", "ops"], [], "service state snapshot");
if (!isId(member.name) || !isInteger(member.sequence, 0) || !Array.isArray(member.ops)) {
throw new TypeError("Invalid service state snapshot");
}
for (const op of member.ops) assertOp(op);
continue;
}
throw new TypeError("Invalid service member snapshot");
}
}
function assertAddress(value: unknown): asserts value is ServiceInstanceAddress {
const address = record(value, "service instance address");
assertKeys(address, ["key", "generation"], [], "service instance address");
if (!isId(address.key) || !isInteger(address.generation, 1)) {
throw new TypeError("Invalid service instance address");
}
}
function record(value: unknown, description: string): Record<string, unknown> {
if (typeof value !== "object" || value === null || Array.isArray(value)) {
throw new TypeError(`Invalid ${description}`);
}
return value as Record<string, unknown>;
}
function assertKeys(
value: Record<string, unknown>,
required: readonly string[],
optional: readonly string[],
description: string,
): void {
const allowed = new Set([...required, ...optional]);
if (required.some((key) => !Object.hasOwn(value, key)) || Object.keys(value).some((key) => !allowed.has(key))) {
throw new TypeError(`Invalid ${description}`);
}
}
function isId(value: unknown): value is string {
return typeof value === "string" && value.length > 0;
}
function isMode(value: unknown): value is ServiceMode {
return value === "singleton" || value === "keyed";
}
function isInteger(value: unknown, minimum: number): value is number {
return typeof value === "number" && Number.isInteger(value) && value >= minimum;
}

10
packages/chord/src/types.ts
View file

@ -203,16 +203,6 @@ export interface RemoteServiceBinding extends RemoteServices {
rebind(bound: boolean, context: Context): Promise<void>;
}
export type RemoteServiceErrorCode =
| "service_not_allowed"
| "service_not_found"
| "service_mode_mismatch"
| "service_member_not_found"
| "service_member_mismatch"
| "service_instance_not_found"
| "service_stale_instance"
| "service_invalid_value";
export interface FacetEnvironment {
/** Declare a hard dependency on one singleton service and return its stable handle. */
use<T>(service: Service<T>): T;

226
packages/chord/test/service-wire.test.ts Normal file
View file

@ -0,0 +1,226 @@
import { describe, expect, test } from "vitest";
import {
createServiceCatalogueCall,
createServiceStateDecoder,
createServiceStateEncoder,
createServiceSubscribeCall,
createServiceUnsubscribeCall,
decodeServiceControlCall,
parseServiceCall,
parseServiceCatalogue,
parseServiceProviderUpdate,
parseServiceSubscriptionSnapshot,
parseWireServiceProviderUpdate,
parseWireServiceSubscriptionSnapshot,
type ServiceProviderUpdate,
type ServiceSubscriptionSnapshot,
} from "../src/index.ts";
describe("service wire protocol", () => {
test("encodes control calls and validates service values", () => {
expect(decodeServiceControlCall(createServiceCatalogueCall())).toEqual({ type: "catalogue" });
expect(
parseServiceCatalogue([
{ serviceId: "pi.models", mode: "singleton" },
{ serviceId: "pi.dialogs", mode: "keyed" },
]),
).toEqual([
{ serviceId: "pi.models", mode: "singleton" },
{ serviceId: "pi.dialogs", mode: "keyed" },
]);
const subscribe = createServiceSubscribeCall("subscription-1", "pi.models", "singleton");
expect(decodeServiceControlCall(subscribe)).toEqual({
type: "subscribe",
subscriptionId: "subscription-1",
serviceId: "pi.models",
mode: "singleton",
});
expect(decodeServiceControlCall(createServiceUnsubscribeCall("subscription-1"))).toEqual({
type: "unsubscribe",
subscriptionId: "subscription-1",
});
expect(
parseServiceCall({
serviceId: "pi.question-dialog",
instance: { key: "invocation-1", generation: 2 },
member: "submit",
args: [{ outcome: "selected", index: 0 }],
}),
).toMatchObject({ member: "submit" });
});
test("rejects malformed service values", () => {
expect(() => parseServiceCall({ serviceId: "pi.models", member: "list", args: [], extra: true })).toThrow(
"Invalid service call",
);
expect(() => parseServiceCatalogue([{ serviceId: "pi.models", mode: "unknown" }])).toThrow(
"Invalid service catalogue",
);
expect(() => parseServiceProviderUpdate({ type: "state", member: "state", sequence: 0, ops: [] })).toThrow(
"Invalid service state update",
);
expect(() =>
parseWireServiceProviderUpdate({ type: "state", member: "state", sequence: 1, ops: [["?", 0]] }),
).toThrow();
});
test("validates decoded and wire snapshots and updates", () => {
const snapshot: ServiceSubscriptionSnapshot = {
serviceId: "pi.models",
mode: "singleton",
instances: [
{
members: [{ name: "state", kind: "state", sequence: 0, ops: [["r", { revision: 1 }]] }],
},
],
};
expect(parseServiceSubscriptionSnapshot(snapshot)).toBe(snapshot);
const encoder = createServiceStateEncoder();
const wireSnapshot = encoder.encodeSnapshot(snapshot);
expect(parseWireServiceSubscriptionSnapshot(wireSnapshot)).toBe(wireSnapshot);
const update: ServiceProviderUpdate = {
type: "state",
member: "state",
sequence: 1,
ops: [["s", ["revision"], 2]],
};
expect(parseServiceProviderUpdate(update)).toBe(update);
const wireUpdate = encoder.encodeUpdate(update);
expect(parseWireServiceProviderUpdate(wireUpdate)).toBe(wireUpdate);
});
test("keeps one operation codec pair for one subscription state", () => {
const enc = createServiceStateEncoder();
const dec = createServiceStateDecoder();
const snapshot: ServiceSubscriptionSnapshot = {
serviceId: "pi.models",
mode: "singleton",
instances: [
{
members: [{ name: "state", kind: "state", sequence: 0, ops: [["r", { revision: 0 }]] }],
},
],
};
expect(dec.decodeSnapshot(enc.encodeSnapshot(snapshot))).toEqual(snapshot);
const first: ServiceProviderUpdate = {
type: "state",
member: "state",
sequence: 1,
ops: [["s", ["revision"], 1]],
};
const second: ServiceProviderUpdate = {
type: "state",
member: "state",
sequence: 2,
ops: [["s", ["revision"], 2]],
};
const firstWire = enc.encodeUpdate(first);
const secondWire = enc.encodeUpdate(second);
expect(firstWire).toMatchObject({ ops: [["s", ["revision"], 1]] });
expect(secondWire).toMatchObject({
ops: [
["#", 0, ["revision"]],
["s", 0, 2],
],
});
expect(dec.decodeUpdate(firstWire)).toEqual(first);
expect(dec.decodeUpdate(secondWire)).toEqual(second);
});
test("isolates operation dictionaries between states and subscriptions", () => {
const snapshot: ServiceSubscriptionSnapshot = {
serviceId: "pi.states",
mode: "singleton",
instances: [
{
members: [
{ name: "left", kind: "state", sequence: 0, ops: [["r", { revision: 0 }]] },
{ name: "right", kind: "state", sequence: 0, ops: [["r", { revision: 0 }]] },
],
},
],
};
const firstEncoder = createServiceStateEncoder();
const firstDecoder = createServiceStateDecoder();
const secondEncoder = createServiceStateEncoder();
const secondDecoder = createServiceStateDecoder();
firstDecoder.decodeSnapshot(firstEncoder.encodeSnapshot(snapshot));
secondDecoder.decodeSnapshot(secondEncoder.encodeSnapshot(snapshot));
const update = (member: string, sequence: number, revision: number): ServiceProviderUpdate => ({
type: "state",
member,
sequence,
ops: [["s", ["revision"], revision]],
});
const firstLeft = firstEncoder.encodeUpdate(update("left", 1, 1));
const firstRight = firstEncoder.encodeUpdate(update("right", 1, 1));
const secondLeft = firstEncoder.encodeUpdate(update("left", 2, 2));
const secondRight = firstEncoder.encodeUpdate(update("right", 2, 2));
expect(firstLeft).toMatchObject({ ops: [["s", ["revision"], 1]] });
expect(firstRight).toMatchObject({ ops: [["s", ["revision"], 1]] });
expect(secondLeft).toMatchObject({
ops: [
["#", 0, ["revision"]],
["s", 0, 2],
],
});
expect(secondRight).toMatchObject({
ops: [
["#", 0, ["revision"]],
["s", 0, 2],
],
});
expect(firstDecoder.decodeUpdate(firstLeft)).toEqual(update("left", 1, 1));
expect(firstDecoder.decodeUpdate(firstRight)).toEqual(update("right", 1, 1));
expect(firstDecoder.decodeUpdate(secondLeft)).toEqual(update("left", 2, 2));
expect(firstDecoder.decodeUpdate(secondRight)).toEqual(update("right", 2, 2));
const independentLeft = secondEncoder.encodeUpdate(update("left", 1, 1));
expect(independentLeft).toMatchObject({ ops: [["s", ["revision"], 1]] });
expect(secondDecoder.decodeUpdate(independentLeft)).toEqual(update("left", 1, 1));
const leftBase: ServiceProviderUpdate = {
type: "state",
member: "left",
sequence: 3,
ops: [["r", { revision: 3 }]],
};
expect(firstDecoder.decodeUpdate(firstEncoder.encodeUpdate(leftBase))).toEqual(leftBase);
const thirdRight = firstEncoder.encodeUpdate(update("right", 3, 3));
expect(thirdRight).toMatchObject({ ops: [["s", 0, 3]] });
expect(firstDecoder.decodeUpdate(thirdRight)).toEqual(update("right", 3, 3));
});
test("creates and removes keyed instance codecs with their lifecycle", () => {
const enc = createServiceStateEncoder();
const dec = createServiceStateDecoder();
const snapshot: ServiceSubscriptionSnapshot = {
serviceId: "pi.dialogs",
mode: "keyed",
instances: [],
};
expect(dec.decodeSnapshot(enc.encodeSnapshot(snapshot))).toEqual(snapshot);
const address = { key: "dialog-1", generation: 1 };
const spawned: ServiceProviderUpdate = {
type: "spawned",
instance: {
instance: address,
members: [{ name: "request", kind: "state", sequence: 0, ops: [["r", { value: 0 }]] }],
},
};
expect(dec.decodeUpdate(enc.encodeUpdate(spawned))).toEqual(spawned);
const update: ServiceProviderUpdate = {
type: "state",
instance: address,
member: "request",
sequence: 1,
ops: [["s", ["value"], 1]],
};
expect(dec.decodeUpdate(enc.encodeUpdate(update))).toEqual(update);
const closed: ServiceProviderUpdate = { type: "closed", instance: address };
expect(dec.decodeUpdate(enc.encodeUpdate(closed))).toEqual(closed);
expect(() => enc.encodeUpdate({ ...update, sequence: 2 })).toThrow("Unknown service state");
});
});

2
packages/client/README.md
View file

@ -27,7 +27,7 @@ const result = await client.request(
The client verifies that the physical endpoint reports the expected logical `serverId`. Server-wide requests carry that ID, and every Session request carries the full live target `{ serverId, sessionId, attachmentId }`. The combined durable address prevents cross-server or cross-session misrouting; the server-generated attachment ID rejects delayed frames after switching or reattaching.
Typed server and Session APIs are provided by Chord service bindings owned by the application. `request()` and `subscribeService()` are low-level transport adapters. A service subscription returns a complete provider snapshot; the binding installs it and then calls `start()` to release updates buffered during hydration. `Client` applies ordered out-of-band attachment changes but deliberately does not construct typed service proxies or interpret service contracts.
Typed server and Session APIs are provided by Chord service bindings owned by the application. `request()` and `subscribeService()` are low-level transport adapters. The client uses Chord's service-control parsers and per-subscription state decoder; `pi-protocol` only validates the routed envelope and strict-JSON boundary. A service subscription returns a complete provider snapshot; the binding installs it and then calls `start()` to release updates buffered during hydration. `Client` applies ordered out-of-band attachment changes but deliberately does not construct typed service proxies or interpret application contracts.
Application observation APIs such as the coding agent's `Transcript` are ordinary Chord services. The client does not interpret their snapshots or updates.

1
packages/client/package.json
View file

@ -47,6 +47,7 @@
"node": ">=22.19.0"
},
"dependencies": {
"@earendil-works/chord": "^0.84.4",
"@earendil-works/pi-protocol": "^0.84.4"
},
"devDependencies": {

59
packages/client/src/client.ts
View file

@ -1,27 +1,30 @@
import {
type AttachmentEnvelope,
createServiceCatalogueCall,
createServiceStateDecoder,
createServiceSubscribeCall,
createServiceUnsubscribeCall,
type DecodedServiceProviderUpdate,
type DecodedServiceSubscriptionSnapshot,
type JsonValue,
parseServiceCall,
parseServiceCatalogue,
parseWireServiceProviderUpdate,
parseWireServiceSubscriptionSnapshot,
type ServiceCall,
type ServiceCatalogueEntry,
type ServiceMode,
type ServiceProviderUpdate,
type ServiceStateDecoder,
type ServiceSubscriptionSnapshot,
} from "@earendil-works/chord";
import {
type AttachmentEnvelope,
encodeClientMessage,
isServerId,
type ProtocolRpcCall,
type ProtocolRpcResult,
ProtocolValidationError,
parseServiceCatalogue,
parseServiceSubscriptionSnapshot,
type ResponseEnvelope,
type RpcTarget,
type ServerHello,
type ServiceCatalogueEntry,
type ServiceEventEnvelope,
type ServiceMode,
type ServiceStateDecoder,
type SessionTarget,
type ServiceProviderUpdate as WireServiceProviderUpdate,
} from "@earendil-works/pi-protocol";
import { Connection } from "./connection.ts";
import { ClientDisposedError, DisconnectedError, ServerError, toError } from "./errors.ts";
@ -35,18 +38,20 @@ import type {
Unsubscribe,
} from "./types.ts";
type ServiceResult = JsonValue | undefined;
interface PendingRequest {
resolve(result: ProtocolRpcResult): void;
resolve(result: ServiceResult): void;
reject(error: Error): void;
cleanup(): void;
}
interface ActiveServiceListener {
readonly target: RpcTarget;
readonly listener: (update: DecodedServiceProviderUpdate) => void | Promise<void>;
readonly listener: (update: ServiceProviderUpdate) => void | Promise<void>;
readonly decoder: ServiceStateDecoder;
readonly queuedWireUpdates: WireServiceProviderUpdate[];
readonly queued: DecodedServiceProviderUpdate[];
readonly queuedWireUpdates: JsonValue[];
readonly queued: ServiceProviderUpdate[];
deliveryTail: Promise<void>;
hydrated: boolean;
ready: boolean;
@ -145,7 +150,7 @@ export class Client {
}
/** Invoke one low-level protocol call against an explicit routed target. */
request(target: RpcTarget, call: ProtocolRpcCall, signal?: AbortSignal): Promise<ProtocolRpcResult> {
request(target: RpcTarget, call: ServiceCall, signal?: AbortSignal): Promise<ServiceResult> {
return this.#request(target, call, signal);
}
@ -166,7 +171,7 @@ export class Client {
target: RpcTarget,
serviceId: string,
mode: ServiceMode,
listener: (update: DecodedServiceProviderUpdate) => void | Promise<void>,
listener: (update: ServiceProviderUpdate) => void | Promise<void>,
signal?: AbortSignal,
): Promise<ServiceSubscription> {
const subscriptionId = `service-${++this.#serviceSubscriptionSequence}`;
@ -181,17 +186,17 @@ export class Client {
ready: false,
};
this.#serviceListeners.set(subscriptionId, active);
let snapshot: DecodedServiceSubscriptionSnapshot;
let snapshot: ServiceSubscriptionSnapshot;
try {
snapshot = await this.#request(
target,
createServiceSubscribeCall(subscriptionId, serviceId, mode),
signal,
(result) => {
const decoded = active.decoder.decodeSnapshot(parseServiceSubscriptionSnapshot(result));
const decoded = active.decoder.decodeSnapshot(parseWireServiceSubscriptionSnapshot(result));
active.hydrated = true;
for (const update of active.queuedWireUpdates.splice(0)) {
active.queued.push(active.decoder.decodeUpdate(update));
active.queued.push(active.decoder.decodeUpdate(parseWireServiceProviderUpdate(update)));
}
return decoded;
},
@ -228,11 +233,11 @@ export class Client {
};
}
#request<T = ProtocolRpcResult>(
#request<T = ServiceResult>(
target: RpcTarget,
call: ProtocolRpcCall,
call: ServiceCall,
signal?: AbortSignal,
transform?: (result: ProtocolRpcResult) => T,
transform?: (result: ServiceResult) => T,
): Promise<T> {
if (this.#disposed) return Promise.reject(new ClientDisposedError());
if (!this.connected) return Promise.reject(new DisconnectedError());
@ -281,7 +286,7 @@ export class Client {
let frame: Uint8Array;
try {
frame = encodeClientMessage(
{ type: "request", id, target, call },
{ type: "request", id, target, call: parseServiceCall(call) as unknown as JsonValue },
{ maxFrameLength: this.#connection.maxFrameLength },
);
} catch (error) {
@ -310,9 +315,9 @@ export class Client {
active.queuedWireUpdates.push(message.update);
return;
}
let update: DecodedServiceProviderUpdate;
let update: ServiceProviderUpdate;
try {
update = active.decoder.decodeUpdate(message.update);
update = active.decoder.decodeUpdate(parseWireServiceProviderUpdate(message.update));
} catch (error) {
this.#connection.fail(
new ProtocolValidationError(error instanceof Error ? error.message : "Invalid service operation stream"),
@ -407,7 +412,7 @@ export class Client {
}
}
#deliverServiceUpdate(active: ActiveServiceListener, update: DecodedServiceProviderUpdate): void {
#deliverServiceUpdate(active: ActiveServiceListener, update: ServiceProviderUpdate): void {
active.deliveryTail = active.deliveryTail
.then(() => active.listener(update))
.catch((error: unknown) => this.#reportListenerError(error));

5
packages/client/src/types.ts
View file

@ -1,4 +1,5 @@
import type { DecodedServiceSubscriptionSnapshot, RpcTarget, SessionTarget } from "@earendil-works/pi-protocol";
import type { ServiceSubscriptionSnapshot } from "@earendil-works/chord";
import type { RpcTarget, SessionTarget } from "@earendil-works/pi-protocol";
import type { ByteTransportFactory } from "./transport.ts";
export type ConnectionState = "disconnected" | "connecting" | "connected";
@ -15,7 +16,7 @@ export type AttachmentChangeListener = (attachment: SessionTarget | undefined) =
export interface ServiceSubscription {
readonly id: string;
readonly target: RpcTarget;
readonly snapshot: DecodedServiceSubscriptionSnapshot;
readonly snapshot: ServiceSubscriptionSnapshot;
/** Begin ordered update delivery after the caller has installed the snapshot. */
start(): void;
dispose(): Promise<void>;

4
packages/client/test/unix-transport.test.ts
View file

@ -1,6 +1,7 @@
import { mkdtemp, rm } from "node:fs/promises";
import { createServer, type Server, type Socket } from "node:net";
import { join } from "node:path";
import { parseServiceCall } from "@earendil-works/chord";
import { ClientMessageDecoder, encodeServerMessage, PROTOCOL_VERSION } from "@earendil-works/pi-protocol";
import { afterEach, describe, expect, test } from "vitest";
import { Client } from "../src/index.ts";
@ -68,7 +69,8 @@ describe.runIf(process.platform !== "win32")("createUnixTransportFactory", () =>
continue;
}
if (message.type === "cancel") continue;
receivedMembers.push(`${message.call.serviceId}.${message.call.member}`);
const call = parseServiceCall(message.call);
receivedMembers.push(`${call.serviceId}.${call.member}`);
const frame = encodeServerMessage({
type: "response",
id: message.id,

1
packages/coding-agent/install-lock/package-lock.json generated
View file

@ -506,6 +506,7 @@
"resolved": "https://registry.npmjs.org/@earendil-works/pi-client/-/pi-client-0.84.4.tgz",
"license": "MIT",
"dependencies": {
"@earendil-works/chord": "^0.84.4",
"@earendil-works/pi-protocol": "^0.84.4"
},
"engines": {

1
packages/coding-agent/npm-shrinkwrap.json generated
View file

@ -533,6 +533,7 @@
"resolved": "https://registry.npmjs.org/@earendil-works/pi-client/-/pi-client-0.84.4.tgz",
"license": "MIT",
"dependencies": {
"@earendil-works/chord": "^0.84.4",
"@earendil-works/pi-protocol": "^0.84.4"
},
"engines": {

6
packages/coding-agent/src/experimental/services/connection.ts
View file

@ -9,10 +9,12 @@ import {
type ReplicatedState,
replicatedState,
type Service,
type ServiceCall,
type ServiceCatalogueEntry,
} from "@earendil-works/chord";
import { BACKGROUND_CONTEXT } from "@earendil-works/chord/context";
import type { Client } from "@earendil-works/pi-client";
import type { ProtocolRpcCall, RpcTarget, ServiceCatalogueEntry, SessionTarget } from "@earendil-works/pi-protocol";
import type { RpcTarget, SessionTarget } from "@earendil-works/pi-protocol";
export type ServerConnectionState =
| { status: "connecting"; attempt: number }
@ -374,7 +376,7 @@ function createRemoteServiceTransport(client: Client, getTarget: () => RpcTarget
invoke: async (call, context) => {
const target = getTarget();
if (target === undefined) throw new Error("Service connection is not routed");
const wireCall: ProtocolRpcCall = {
const wireCall: ServiceCall = {
serviceId: call.serviceId,
...(call.instance === undefined ? {} : { instance: call.instance }),
member: call.member,

2
packages/coding-agent/src/experimental/services/server.ts
View file

@ -1,12 +1,12 @@
import {
type Context,
decodeServiceControlCall,
type JsonValue,
RemoteServiceProvider,
replicatedState,
type ServiceSubscription,
} from "@earendil-works/chord";
import { BACKGROUND_CONTEXT } from "@earendil-works/chord/context";
import { decodeServiceControlCall } from "@earendil-works/pi-protocol";
import type { RoutedServerServiceAttachment, RoutedServerServiceHost } from "@earendil-works/pi-server";
import { PresentationPlugins } from "./plugins.ts";
import {

5
packages/coding-agent/src/experimental/services/worker.ts
View file

@ -2,16 +2,17 @@ import {
type Context,
createFacetHost,
createStaticFacetLoader,
decodeServiceControlCall,
defineFacet,
type FacetHost,
type FacetLoader,
isJsonValue,
type JsonValue,
type ServiceCall,
type ServiceProviderUpdate,
type ServiceSubscription,
} from "@earendil-works/chord";
import type { AgentHarness, AgentLane } from "@earendil-works/pi-agent-core";
import { decodeServiceControlCall, type ProtocolRpcCall } from "@earendil-works/pi-protocol";
import Type, { type Static } from "typebox";
import type { ModelRuntime } from "../../core/model-runtime.ts";
import type { SettingsManager } from "../../core/settings-manager.ts";
@ -48,7 +49,7 @@ interface WorkerServiceSubscription {
}
export interface SessionWorkerServices {
invoke(call: ProtocolRpcCall, scope: WorkerServiceScope, context: Context): Promise<JsonValue | undefined>;
invoke(call: ServiceCall, scope: WorkerServiceScope, context: Context): Promise<JsonValue | undefined>;
removeSubscriptions(matches: (scope: WorkerServiceScope) => boolean): void;
dispose(): Promise<void>;
}

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

@ -1,19 +1,20 @@
import type { ChildProcess } from "node:child_process";
import { randomUUID } from "node:crypto";
import { isAbsolute } from "node:path";
import type { ServiceProviderUpdate } from "@earendil-works/chord";
import {
createServiceUnsubscribeCall,
decodeServiceControlCall,
type JsonValue,
parseServiceProviderUpdate,
type ServiceCall,
type ServiceProviderUpdate,
} from "@earendil-works/chord";
import {
BACKGROUND_CONTEXT,
type Context,
type JsonlSessionMetadata,
TODO_CONTEXT,
} from "@earendil-works/pi-agent-core";
import {
createServiceUnsubscribeCall,
decodeServiceControlCall,
type ProtocolRpcCall,
type ProtocolRpcResult,
} from "@earendil-works/pi-protocol";
import { type RoutedSessionAttachment, type RoutedSessionHandle, ServerError } from "@earendil-works/pi-server";
import { Check } from "typebox/value";
import type { CoordinatorConnection, CoordinatorConnectionEvent } from "./coordinator.ts";
@ -241,10 +242,10 @@ export class SessionWorkerManager {
async #invokeService(
worker: WorkerRecord,
scope: WorkerOperationScope,
call: ProtocolRpcCall,
call: ServiceCall,
publish: (subscriptionId: string, update: ServiceProviderUpdate, context: Context) => void | Promise<void>,
context: Context,
): Promise<ProtocolRpcResult> {
): Promise<JsonValue | undefined> {
const control = decodeServiceControlCall(call);
let addedSubscriptionKey: string | undefined;
if (control?.type === "subscribe") {
@ -321,7 +322,7 @@ export class SessionWorkerManager {
#invoke(
worker: WorkerRecord,
scope: WorkerOperationScope,
call: ProtocolRpcCall,
call: ServiceCall,
context: Context,
): Promise<ServiceOperationResult> {
try {
@ -573,7 +574,7 @@ export class SessionWorkerManager {
message.sessionKey,
message.scope,
message.subscriptionId,
message.update as unknown as ServiceProviderUpdate,
message.update,
);
return;
}
@ -620,7 +621,7 @@ export class SessionWorkerManager {
sessionKey: string,
scope: WorkerOperationScope,
subscriptionId: string,
update: ServiceProviderUpdate,
update: unknown,
): void {
const entry = this.#serviceSubscriptions.get(scopedServiceSubscriptionKey(scope, subscriptionId));
if (
@ -632,7 +633,13 @@ export class SessionWorkerManager {
) {
return;
}
entry.deliveryTail = entry.deliveryTail.then(() => entry.listener(update, TODO_CONTEXT)).catch(() => {});
let parsed: ServiceProviderUpdate;
try {
parsed = parseServiceProviderUpdate(update);
} catch {
return;
}
entry.deliveryTail = entry.deliveryTail.then(() => entry.listener(parsed, TODO_CONTEXT)).catch(() => {});
}
#recordReadyWorker(peerId: string, message: Extract<SessionWorkerEvent, { type: "worker_ready" }>): void {

46
packages/coding-agent/src/experimental/session-worker.ts
View file

@ -1,6 +1,14 @@
import { createConnection, type Socket } from "node:net";
import { isAbsolute } from "node:path";
import { RemoteServiceError, type ServiceProviderUpdate } from "@earendil-works/chord";
import {
isJsonValue,
parseServiceProviderUpdate,
REMOTE_SERVICE_ERROR_CODES,
RemoteServiceError,
type RemoteServiceErrorCode,
type ServiceCall,
type ServiceProviderUpdate,
} from "@earendil-works/chord";
import {
AgentHarness,
type AgentHarness as AgentHarnessInstance,
@ -16,12 +24,6 @@ import {
withCancel,
} from "@earendil-works/pi-agent-core";
import { NodeExecutionEnv } from "@earendil-works/pi-agent-core/node";
import {
ProtocolRpcCallSchema,
type ServiceErrorCode,
ServiceErrorCodeSchema,
ServiceProviderUpdateSchema,
} from "@earendil-works/pi-protocol";
import lockfile from "proper-lockfile";
import Type, { type Static } from "typebox";
import { Check } from "typebox/value";
@ -48,6 +50,19 @@ export type { SessionWorkerRuntime } from "./services/worker.ts";
const StrictObject = <const T extends Parameters<typeof Type.Object>[0]>(properties: T) =>
Type.Object(properties, { additionalProperties: false });
const ServiceCallSchema = Type.Unsafe<ServiceCall>(
StrictObject({
serviceId: Type.String({ minLength: 1 }),
instance: Type.Optional(
StrictObject({ key: Type.String({ minLength: 1 }), generation: Type.Integer({ minimum: 1 }) }),
),
member: Type.String({ minLength: 1 }),
args: Type.Array(Type.Unknown()),
}),
);
const RemoteServiceErrorCodeSchema = Type.Unsafe<RemoteServiceErrorCode>(
Type.String({ pattern: `^(?:${REMOTE_SERVICE_ERROR_CODES.join("|")})$` }),
);
export const SESSION_WORKER_CONTROL_ADDRESS_ENV = "PI_SESSION_WORKER_CONTROL_ADDRESS";
export const SESSION_WORKER_CONTROL_TOKEN_ENV = "PI_SESSION_WORKER_CONTROL_TOKEN";
@ -84,7 +99,7 @@ export const WorkerOperationRequestSchema = StrictObject({
type: Type.Literal("operation"),
requestId: Type.String({ minLength: 1 }),
scope: WorkerOperationScopeSchema,
call: ProtocolRpcCallSchema,
call: ServiceCallSchema,
});
export type WorkerOperationRequest = Static<typeof WorkerOperationRequestSchema>;
@ -99,7 +114,7 @@ export const WorkerOperationResponseSchema = Type.Union([
type: Type.Literal("operation_error"),
requestId: Type.String({ minLength: 1 }),
scope: WorkerOperationScopeSchema,
code: Type.Optional(ServiceErrorCodeSchema),
code: Type.Optional(RemoteServiceErrorCodeSchema),
message: Type.String(),
}),
]);
@ -167,7 +182,7 @@ export const SessionWorkerEventSchema = Type.Union([
sessionKey: Type.String(),
scope: WorkerOperationScopeSchema,
subscriptionId: Type.String({ minLength: 1 }),
update: ServiceProviderUpdateSchema,
update: Type.Unknown(),
}),
]);
export type SessionWorkerEvent = Static<typeof SessionWorkerEventSchema>;
@ -441,10 +456,9 @@ function writeJsonLine(socket: Socket, message: unknown): Promise<void> {
});
}
function toWorkerServiceUpdate(update: ServiceProviderUpdate): Static<typeof ServiceProviderUpdateSchema> {
const candidate: unknown = update;
if (!Check(ServiceProviderUpdateSchema, candidate)) throw new Error("Service produced an invalid update");
return candidate;
function toWorkerServiceUpdate(update: ServiceProviderUpdate): ServiceProviderUpdate {
if (!isJsonValue(update)) throw new Error("Service produced a non-JSON update");
return parseServiceProviderUpdate(update);
}
function demandKey(serverConnectionId: string, attachmentId: string): string {
@ -627,12 +641,12 @@ async function run(options: SessionWorkerOptions, createHarness: CreateSessionWo
response: { type: "operation_result", requestId: request.requestId, scope: request.scope, result },
});
} catch (error) {
let code: ServiceErrorCode | undefined;
let code: RemoteServiceErrorCode | undefined;
if (error instanceof RemoteServiceError) {
code = error.code;
} else if (error instanceof Error && "code" in error) {
const candidate = error.code;
if (Check(ServiceErrorCodeSchema, candidate)) code = candidate;
if (Check(RemoteServiceErrorCodeSchema, candidate)) code = candidate;
}
await control.send({
type: "operation_response",

23
packages/coding-agent/test/experimental-client-tui.test.ts
View file

@ -151,7 +151,18 @@ describe("experimental client TUI", () => {
transcriptState.state.event = event;
transcriptState.publish(BACKGROUND_CONTEXT);
};
let finishPrompt!: () => void;
const promptFinished = new Promise<void>((resolve) => {
finishPrompt = resolve;
});
const prompt = vi.fn(async () => {
emitTranscriptEvent({
type: "run_start",
lane: "main",
runId: "run-1",
startedAt: 1,
});
await promptFinished;
emitTranscriptEvent({
type: "entry_added",
lane: "main",
@ -185,6 +196,15 @@ describe("experimental client TUI", () => {
},
},
});
emitTranscriptEvent({
type: "run_end",
lane: "main",
runId: "run-1",
status: "completed",
fromTipId: null,
tipId: "entry-assistant",
endedAt: 2,
});
return {
ok: true as const,
value: {
@ -349,8 +369,11 @@ describe("experimental client TUI", () => {
component.handleInput("hello");
component.handleInput("\r");
await vi.waitFor(() => expect(prompt).toHaveBeenCalledWith("hello", undefined, BACKGROUND_CONTEXT));
await vi.waitFor(() => expect(component.render(80).join("\n")).toContain("Working..."));
finishPrompt();
await vi.waitFor(() => expect(component.render(80).join("\n")).toContain("remote answer"));
expect(component.render(80).join("\n")).toContain("hello");
expect(component.render(80).join("\n")).not.toContain("Working...");
expect(component.render(80).join("\n")).not.toContain("Operation run-1 completed");
component.handleInput("/reload");

41
packages/coding-agent/test/experimental-remote-runtime.test.ts
View file

@ -9,13 +9,16 @@ import { afterEach, beforeEach, describe, expect, test, vi } from "vitest";
import { ExampleFacetService } from "../examples/plugins/pi-example-plugin/src/contract.ts";
import { runClient } from "../src/experimental/client.ts";
import { activateBuiltinClientServices, openClientRuntime } from "../src/experimental/client-runtime.ts";
import { openLaneReplica } from "../src/experimental/lane-replica.ts";
import { createPresentationFacetLoaders } from "../src/experimental/plugins/bundled.ts";
import * as processRuntime from "../src/experimental/process.ts";
import { type RunningServer, startServer } from "../src/experimental/server.ts";
import { AgentController } from "../src/experimental/services/agent-controller.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 { PresentationPlugins, SessionPlugins } from "../src/experimental/services/plugins.ts";
import { SessionDirectory, SessionManagement } from "../src/experimental/services/sessions.ts";
import { Transcript } from "../src/experimental/services/transcript.ts";
import { createServerServiceBinding, createSessionServiceBinding } from "./experimental-service-binding.ts";
import {
configureExperimentalWorkerModel,
@ -624,6 +627,42 @@ describe("experimental durable server composition", () => {
);
});
test("replicates terminal operation state after consecutive prompts", async ({ onTestFinished }) => {
const spawn = vi
.spyOn(processRuntime, "spawnInternalProcess")
.mockImplementation((role, args, options) =>
realSpawnInternalProcess(
role,
args,
role === "session-worker" ? { ...options, entryUrl: fauxWorkerEntryUrl } : options,
),
);
onTestFinished(() => spawn.mockRestore());
const { runtime } = await makeServer();
const client = await attachClient(runtime, "demo-1");
const services = createSessionServiceBinding(client, { services: [AgentController, SessionPlugins, Transcript] });
const controller = services.use(AgentController);
const transcript = services.use(Transcript);
await services.ready(BACKGROUND_CONTEXT);
await services.use(SessionPlugins).reload(BACKGROUND_CONTEXT);
const replica = await openLaneReplica(transcript);
try {
for (const message of ["first question", "second question"]) {
const response = await controller.prompt({ message, images: null }, BACKGROUND_CONTEXT);
expect(response).toMatchObject({ accepted: true, operationId: expect.any(String) });
await vi.waitFor(() => {
expect(replica.state()).toMatchObject({
operation: null,
lastResult: { operationId: response.operationId },
});
});
}
} finally {
await replica.close();
await services.dispose(BACKGROUND_CONTEXT);
}
});
test("stops an idle Session worker after its client disconnects", async () => {
const { runtime } = await makeServer();
const client = await attachClient(runtime, "demo-1");

4
packages/coding-agent/test/experimental-session-worker-manager.test.ts
View file

@ -1,5 +1,5 @@
import type { ServiceCall } from "@earendil-works/chord";
import { BACKGROUND_CONTEXT, type JsonlSessionMetadata } from "@earendil-works/pi-agent-core";
import type { ProtocolRpcCall } from "@earendil-works/pi-protocol";
import { afterEach, describe, expect, test, vi } from "vitest";
import type { CoordinatorConnectionEvent } from "../src/experimental/coordinator.ts";
import { SessionWorkerManager } from "../src/experimental/session-worker-manager.ts";
@ -194,7 +194,7 @@ describe("Session worker lifecycle failures", () => {
});
describe("Session worker operations", () => {
const serviceCall = { serviceId: "test.session", member: "run", args: ["Hello"] } satisfies ProtocolRpcCall;
const serviceCall = { serviceId: "test.session", member: "run", args: ["Hello"] } satisfies ServiceCall;
test("correlates service results to the worker generation and attachment", async () => {
const { coordinator, workers, attachment, release } = await createAttachedWorker();

18
packages/protocol/README.md
View file

@ -1,18 +1,20 @@
# @earendil-works/pi-protocol
Runtime-neutral schemas, types, CBOR encoding, and byte-stream framing for the experimental Pi protocol.
Runtime-neutral routed envelopes, CBOR encoding, and byte-stream framing for the experimental Pi protocol.
Protocol version `8` carries the current server- and Session-scoped facet-service slice, including operation-based replicated state and Session-branch plugin preparation:
Protocol version `8` defines:
- a version handshake that identifies the logical `serverId`;
- explicit server and Session request targets;
- contract-agnostic service calls with optional keyed-instance identity;
- singleton and keyed-service subscription snapshots plus ordered operation-based state, spawn, and close updates;
- correlated responses, request cancellation, out-of-band attachment updates, attachment-scoped events, and bounded protocol errors.
- correlated requests and responses with opaque strict-JSON payloads;
- request cancellation, opaque subscription updates, and out-of-band attachment changes;
- non-empty opaque error codes and bounded transport messages.
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.
A server target contains `{ serverId }`; a Session target contains `{ serverId, sessionId, attachmentId }`. The combined route fences calls to one logical server, durable Session, and live presentation attachment. Management `attach()` and `detach()` return no routing identifiers; the server publishes the selected live route in an out-of-band `attachment` message. Disconnecting releases only that presentation's attachment after admitted 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`. `Transcript` publishes through ordinary tracked Chord state. Chord flushes decoded operations behind the state API; each client subscription owns one stateful encoder and decoder per replicated member so hydration bases and path dictionaries remain independent. Session service calls and updates route opaquely to the attached worker, where the provider validates and invokes them. The protocol does not define Agent, lane, transcript-entry, branch, model, or plugin business schemas.
Chord owns the payload semantics carried inside these envelopes: `{ serviceId, instance?, member, args }` calls, the `$chord.service` control vocabulary, service catalogues, subscription snapshots and updates, service error codes, and the independent Delta path codecs for replicated states. `pi-protocol` validates that each opaque payload is strict JSON but does not validate or export its Chord grammar. Clients and servers parse those values through `@earendil-works/chord` at the service adapter boundary.
Session-directory state, management results, transcripts, models, plugins, and all other application values remain opaque service data. The real `Session` and `AgentHarness` remain process-local. Server and Session calls route opaquely to their owning providers, where Chord and the application validate and invoke them.
Server and worker lifecycle is intentionally outside this public protocol. The experimental local coordinator is only an opaque message router; each replaceable server process owns the private lifecycle protocol.
@ -34,6 +36,6 @@ for (const message of decoder.push(incomingChunk)) handleServerMessage(message);
decoder.end();
```
All control schemas reject unknown object properties, and codecs recursively reject non-JSON opaque service payloads, including non-finite numbers, byte arrays, `undefined`, prototypes, and cycles. Schema violations, malformed CBOR, and invalid framing throw `ProtocolValidationError`. Transports must preserve byte order. Peer authentication and authenticated service contexts are not implemented by the experimental transport.
All envelope schemas reject unknown object properties, and codecs recursively reject non-JSON opaque payloads, including non-finite numbers, byte arrays, `undefined`, prototypes, and cycles. Envelope violations, malformed CBOR, and invalid framing throw `ProtocolValidationError`. Payload-specific adapters must perform their own semantic validation after decoding. Transports must preserve byte order. Peer authentication and authenticated service contexts are not implemented by the experimental transport.
Default limits are 16 MiB per CBOR payload/frame, 1,000,000 array elements or map entries, and 64 nested item levels. The protocol is experimental and has no compatibility guarantees.

17
packages/protocol/src/index.ts
View file

@ -6,19 +6,10 @@ export {
type CancelEnvelope,
type ClientHello,
type ClientMessage,
createServiceCatalogueCall,
createServiceSubscribeCall,
createServiceUnsubscribeCall,
decodeServiceControlCall,
isServerId,
PROTOCOL_VERSION,
type ProtocolError,
type ProtocolErrorCode,
type ProtocolRpcCall,
ProtocolRpcCallSchema,
type ProtocolRpcResult,
parseServiceCatalogue,
parseServiceSubscriptionSnapshot,
type RequestEnvelope,
type ResponseEnvelope,
type RpcTarget,
@ -26,14 +17,6 @@ export {
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 "./service-state.ts";

216
packages/protocol/src/protocol.ts
View file

@ -1,13 +1,11 @@
import { isJsonValue, type JsonValue } from "@earendil-works/chord";
import type { WireOp } from "@earendil-works/chord/delta";
import type { JsonValue } from "@earendil-works/chord";
import Type, { type Static } from "typebox";
import { Check } from "typebox/value";
export const PROTOCOL_VERSION = 8 as const;
const OpaqueJsonValueSchema = Type.Unsafe<JsonValue>(Type.Unknown());
const IdSchema = Type.String({ minLength: 1 });
const OpaqueJsonValueSchema = Type.Unsafe<JsonValue>(Type.Unknown());
const StrictObject = <const T extends Parameters<typeof Type.Object>[0]>(properties: T) =>
Type.Object(properties, { additionalProperties: false });
@ -20,213 +18,11 @@ export function isServerId(value: unknown): value is ServerId {
return Check(ServerIdSchema, value);
}
const ServiceModeSchema = Type.Union([Type.Literal("singleton"), Type.Literal("keyed")]);
export type ServiceMode = Static<typeof ServiceModeSchema>;
const ServiceCatalogueEntrySchema = StrictObject({
serviceId: IdSchema,
mode: ServiceModeSchema,
});
const ServiceCatalogueSchema = Type.Array(ServiceCatalogueEntrySchema);
export type ServiceCatalogueEntry = Static<typeof ServiceCatalogueEntrySchema>;
export function parseServiceCatalogue(value: unknown): readonly ServiceCatalogueEntry[] {
if (!Check(ServiceCatalogueSchema, value)) throw new TypeError("Invalid service catalogue");
const ids = value.map(({ serviceId }) => serviceId);
if (new Set(ids).size !== ids.length) throw new TypeError("Service catalogue contains duplicate IDs");
return value;
}
const ServiceInstanceAddressSchema = StrictObject({
key: IdSchema,
generation: Type.Integer({ minimum: 1 }),
});
/** Contract-agnostic service/member invocation carried by the transport. */
export const ProtocolRpcCallSchema = StrictObject({
serviceId: Type.String({ minLength: 1 }),
instance: Type.Optional(ServiceInstanceAddressSchema),
member: Type.String({ minLength: 1 }),
args: Type.Array(OpaqueJsonValueSchema),
});
export type ProtocolRpcCall = Static<typeof ProtocolRpcCallSchema>;
export type ProtocolRpcResult = JsonValue | undefined;
const DeltaPathSegmentSchema = Type.Union([Type.String(), Type.Integer({ minimum: 0 })]);
const DeltaPathSchema = Type.Array(DeltaPathSegmentSchema);
const DeltaNonEmptyPathSchema = Type.Array(DeltaPathSegmentSchema, { minItems: 1 });
const DeltaPathRefSchema = Type.Union([DeltaPathSchema, Type.Integer({ minimum: 0 })]);
const DeltaNonEmptyPathRefSchema = Type.Union([DeltaNonEmptyPathSchema, Type.Integer({ minimum: 0 })]);
const DeltaOpsSchema = Type.Array(
Type.Unsafe<WireOp>(
Type.Union([
Type.Tuple([Type.Literal("r"), OpaqueJsonValueSchema]),
Type.Tuple([Type.Literal("#"), Type.Integer({ minimum: 0 }), DeltaPathSchema]),
Type.Tuple([Type.Literal("s"), DeltaNonEmptyPathRefSchema, OpaqueJsonValueSchema]),
Type.Tuple([Type.Literal("s"), OpaqueJsonValueSchema]),
Type.Tuple([Type.Literal("d"), DeltaNonEmptyPathRefSchema]),
Type.Tuple([Type.Literal("d")]),
Type.Tuple([Type.Literal("a"), DeltaNonEmptyPathRefSchema, Type.String()]),
Type.Tuple([Type.Literal("a"), Type.String()]),
Type.Tuple([Type.Literal("t"), DeltaNonEmptyPathRefSchema, Type.Integer({ minimum: 0 })]),
Type.Tuple([Type.Literal("t"), Type.Integer({ minimum: 0 })]),
Type.Tuple([
Type.Literal("p"),
DeltaPathRefSchema,
Type.Integer({ minimum: 0 }),
Type.Integer({ minimum: 0 }),
Type.Array(OpaqueJsonValueSchema),
]),
Type.Tuple([
Type.Literal("p"),
Type.Integer({ minimum: 0 }),
Type.Integer({ minimum: 0 }),
Type.Array(OpaqueJsonValueSchema),
]),
]),
),
);
const ServiceMemberSnapshotSchema = Type.Union([
StrictObject({
name: IdSchema,
kind: Type.Literal("method"),
}),
StrictObject({
name: IdSchema,
kind: Type.Literal("state"),
sequence: Type.Integer({ minimum: 0 }),
ops: DeltaOpsSchema,
}),
]);
const ServiceInstanceSnapshotSchema = StrictObject({
instance: Type.Optional(ServiceInstanceAddressSchema),
members: Type.Array(ServiceMemberSnapshotSchema),
});
const ServiceSubscriptionSnapshotSchema = StrictObject({
serviceId: Type.String({ minLength: 1 }),
mode: ServiceModeSchema,
instances: Type.Array(ServiceInstanceSnapshotSchema),
});
export type ServiceSubscriptionSnapshot = Static<typeof ServiceSubscriptionSnapshotSchema>;
export function parseServiceSubscriptionSnapshot(value: unknown): ServiceSubscriptionSnapshot {
if (!Check(ServiceSubscriptionSnapshotSchema, value) || !isJsonValue(value)) {
throw new TypeError("Invalid service subscription snapshot");
}
return value;
}
export const ServiceProviderUpdateSchema = Type.Union([
StrictObject({
type: Type.Literal("state"),
instance: Type.Optional(ServiceInstanceAddressSchema),
member: IdSchema,
sequence: Type.Integer({ minimum: 1 }),
ops: DeltaOpsSchema,
}),
StrictObject({
type: Type.Literal("unavailable"),
}),
StrictObject({
type: Type.Literal("replaced"),
snapshot: ServiceInstanceSnapshotSchema,
}),
StrictObject({
type: Type.Literal("spawned"),
instance: ServiceInstanceSnapshotSchema,
}),
StrictObject({
type: Type.Literal("closed"),
instance: ServiceInstanceAddressSchema,
}),
]);
export type ServiceProviderUpdate = Static<typeof ServiceProviderUpdateSchema>;
// TODO: check if this should be part of Chord.
const SERVICE_CONTROL_ID = "$chord.service";
const SERVICE_CATALOGUE_MEMBER = "catalogue";
const SERVICE_SUBSCRIBE_MEMBER = "subscribe";
const SERVICE_UNSUBSCRIBE_MEMBER = "unsubscribe";
type ServiceControlCall =
| { readonly type: "catalogue" }
| {
readonly type: "subscribe";
readonly subscriptionId: string;
readonly serviceId: string;
readonly mode: ServiceMode;
}
| { readonly type: "unsubscribe"; readonly subscriptionId: string };
export function createServiceCatalogueCall(): ProtocolRpcCall {
return { serviceId: SERVICE_CONTROL_ID, member: SERVICE_CATALOGUE_MEMBER, args: [] };
}
export function createServiceSubscribeCall(
subscriptionId: string,
serviceId: string,
mode: ServiceMode,
): ProtocolRpcCall {
return { serviceId: SERVICE_CONTROL_ID, member: SERVICE_SUBSCRIBE_MEMBER, args: [subscriptionId, serviceId, mode] };
}
export function createServiceUnsubscribeCall(subscriptionId: string): ProtocolRpcCall {
return { serviceId: SERVICE_CONTROL_ID, member: SERVICE_UNSUBSCRIBE_MEMBER, args: [subscriptionId] };
}
export function decodeServiceControlCall(call: ProtocolRpcCall): ServiceControlCall | undefined {
if (call.serviceId !== SERVICE_CONTROL_ID || call.instance !== undefined) return undefined;
if (call.member === SERVICE_CATALOGUE_MEMBER && call.args.length === 0) return { type: "catalogue" };
if (
call.member === SERVICE_SUBSCRIBE_MEMBER &&
call.args.length === 3 &&
typeof call.args[0] === "string" &&
typeof call.args[1] === "string" &&
(call.args[2] === "singleton" || call.args[2] === "keyed")
) {
return {
type: "subscribe",
subscriptionId: call.args[0],
serviceId: call.args[1],
mode: call.args[2],
};
}
if (call.member === SERVICE_UNSUBSCRIBE_MEMBER && call.args.length === 1 && typeof call.args[0] === "string") {
return { type: "unsubscribe", subscriptionId: call.args[0] };
}
return undefined;
}
export const ServiceErrorCodeSchema = Type.Union([
Type.Literal("service_not_allowed"),
Type.Literal("service_not_found"),
Type.Literal("service_mode_mismatch"),
Type.Literal("service_member_not_found"),
Type.Literal("service_member_mismatch"),
Type.Literal("service_instance_not_found"),
Type.Literal("service_stale_instance"),
Type.Literal("service_invalid_value"),
]);
export type ServiceErrorCode = Static<typeof ServiceErrorCodeSchema>;
const ProtocolErrorCodeSchema = Type.Union([
Type.Literal("version"),
Type.Literal("wrong_server"),
Type.Literal("session_not_found"),
Type.Literal("session_ambiguous"),
Type.Literal("session_not_attached"),
Type.Literal("not_supported"),
Type.Literal("server_draining"),
ServiceErrorCodeSchema,
Type.Literal("invalid_request"),
Type.Literal("cancelled"),
Type.Literal("internal_error"),
]);
const ProtocolErrorSchema = StrictObject({
code: ProtocolErrorCodeSchema,
code: IdSchema,
message: Type.String(),
});
export type ProtocolErrorCode = Static<typeof ProtocolErrorCodeSchema>;
export type ProtocolErrorCode = string;
export type ProtocolError = Static<typeof ProtocolErrorSchema>;
/** Must be the first frame sent by a client. */
@ -254,7 +50,7 @@ const RequestEnvelopeSchema = StrictObject({
type: Type.Literal("request"),
id: IdSchema,
target: RpcTargetSchema,
call: ProtocolRpcCallSchema,
call: OpaqueJsonValueSchema,
});
const CancelEnvelopeSchema = StrictObject({
type: Type.Literal("cancel"),
@ -292,7 +88,7 @@ const ResponseEnvelopeSchema = Type.Union([
const ServiceEventEnvelopeSchema = StrictObject({
type: Type.Literal("service_update"),
subscriptionId: IdSchema,
update: ServiceProviderUpdateSchema,
update: OpaqueJsonValueSchema,
});
/** Out-of-band update to this presentation's selected Session route. */
const AttachmentEnvelopeSchema = StrictObject({

300
packages/protocol/test/protocol.test.ts
View file

@ -3,15 +3,7 @@ import {
type ClientHello,
type ClientMessage,
ClientMessageDecoder,
createServiceCatalogueCall,
createServiceStateDecoder,
createServiceStateEncoder,
createServiceSubscribeCall,
createServiceUnsubscribeCall,
type DecodedServiceProviderUpdate,
type DecodedServiceSubscriptionSnapshot,
decodeCbor,
decodeServiceControlCall,
encodeCbor,
encodeClientMessage,
encodeFrame,
@ -22,8 +14,6 @@ import {
ProtocolValidationError,
parseClientMessage,
parseServerMessage,
parseServiceCatalogue,
parseServiceSubscriptionSnapshot,
type ServerHello,
type ServerMessage,
ServerMessageDecoder,
@ -68,237 +58,14 @@ describe("protocol validation", () => {
parseClientMessage({
type: "request",
id: "request-1",
serverId,
call: { method: "list", args: [] },
target: { serverId },
call: { serviceId: "pi.models", member: "list", args: [] },
}),
).toThrow(ProtocolValidationError);
});
test("encodes service control and keyed instance addresses", () => {
expect(decodeServiceControlCall(createServiceCatalogueCall())).toEqual({ type: "catalogue" });
expect(
parseServiceCatalogue([
{ serviceId: "pi.models", mode: "singleton" },
{ serviceId: "pi.dialogs", mode: "keyed" },
]),
).toEqual([
{ serviceId: "pi.models", mode: "singleton" },
{ serviceId: "pi.dialogs", mode: "keyed" },
]);
const subscribe = createServiceSubscribeCall("subscription-1", "pi.models", "singleton");
expect(decodeServiceControlCall(subscribe)).toEqual({
type: "subscribe",
subscriptionId: "subscription-1",
serviceId: "pi.models",
mode: "singleton",
});
expect(decodeServiceControlCall(createServiceUnsubscribeCall("subscription-1"))).toEqual({
type: "unsubscribe",
subscriptionId: "subscription-1",
});
expect(
parseServiceSubscriptionSnapshot({
serviceId: "pi.models",
mode: "singleton",
instances: [
{
members: [{ name: "state", kind: "state", sequence: 0, ops: [["r", { revision: 1 }]] }],
},
],
}),
).toEqual({
serviceId: "pi.models",
mode: "singleton",
instances: [
{
members: [{ name: "state", kind: "state", sequence: 0, ops: [["r", { revision: 1 }]] }],
},
],
});
expect(
parseServerMessage({
type: "service_update",
subscriptionId: "subscription-1",
update: {
type: "state",
member: "state",
sequence: 1,
ops: [["s", ["revision"], 2]],
},
}),
).toMatchObject({ update: { type: "state", member: "state" } });
expect(
parseServerMessage({
type: "service_update",
subscriptionId: "subscription-1",
update: { type: "unavailable" },
}),
).toMatchObject({ update: { type: "unavailable" } });
expect(
parseServerMessage({
type: "service_update",
subscriptionId: "subscription-1",
update: {
type: "replaced",
snapshot: { members: [{ name: "select", kind: "method" }] },
},
}),
).toMatchObject({ update: { type: "replaced", snapshot: { members: [{ name: "select" }] } } });
const keyed: ClientMessage = {
type: "request",
id: "request-1",
target: {
serverId: "00000000-0000-4000-8000-000000000001",
sessionId: "session-1",
attachmentId: "attachment-1",
},
call: {
serviceId: "pi.question-dialog",
instance: { key: "invocation-1", generation: 2 },
member: "submit",
args: [{ outcome: "selected", index: 0 }],
},
};
expect(parseClientMessage(keyed)).toEqual(keyed);
});
test("keeps one operation codec pair for one subscription state", () => {
const enc = createServiceStateEncoder();
const dec = createServiceStateDecoder();
const snapshot: DecodedServiceSubscriptionSnapshot = {
serviceId: "pi.models",
mode: "singleton",
instances: [
{
members: [{ name: "state", kind: "state", sequence: 0, ops: [["r", { revision: 0 }]] }],
},
],
};
expect(dec.decodeSnapshot(enc.encodeSnapshot(snapshot))).toEqual(snapshot);
const first: DecodedServiceProviderUpdate = {
type: "state",
member: "state",
sequence: 1,
ops: [["s", ["revision"], 1]],
};
const second: DecodedServiceProviderUpdate = {
type: "state",
member: "state",
sequence: 2,
ops: [["s", ["revision"], 2]],
};
const firstWire = enc.encodeUpdate(first);
const secondWire = enc.encodeUpdate(second);
expect(firstWire).toMatchObject({ ops: [["s", ["revision"], 1]] });
expect(secondWire).toMatchObject({
ops: [
["#", 0, ["revision"]],
["s", 0, 2],
],
});
expect(dec.decodeUpdate(firstWire)).toEqual(first);
expect(dec.decodeUpdate(secondWire)).toEqual(second);
});
test("isolates operation dictionaries between states and subscriptions", () => {
const snapshot: DecodedServiceSubscriptionSnapshot = {
serviceId: "pi.states",
mode: "singleton",
instances: [
{
members: [
{ name: "left", kind: "state", sequence: 0, ops: [["r", { revision: 0 }]] },
{ name: "right", kind: "state", sequence: 0, ops: [["r", { revision: 0 }]] },
],
},
],
};
const firstEncoder = createServiceStateEncoder();
const firstDecoder = createServiceStateDecoder();
const secondEncoder = createServiceStateEncoder();
const secondDecoder = createServiceStateDecoder();
firstDecoder.decodeSnapshot(firstEncoder.encodeSnapshot(snapshot));
secondDecoder.decodeSnapshot(secondEncoder.encodeSnapshot(snapshot));
const update = (member: string, sequence: number, revision: number): DecodedServiceProviderUpdate => ({
type: "state",
member,
sequence,
ops: [["s", ["revision"], revision]],
});
const firstLeft = firstEncoder.encodeUpdate(update("left", 1, 1));
const firstRight = firstEncoder.encodeUpdate(update("right", 1, 1));
const secondLeft = firstEncoder.encodeUpdate(update("left", 2, 2));
const secondRight = firstEncoder.encodeUpdate(update("right", 2, 2));
expect(firstLeft).toMatchObject({ ops: [["s", ["revision"], 1]] });
expect(firstRight).toMatchObject({ ops: [["s", ["revision"], 1]] });
expect(secondLeft).toMatchObject({
ops: [
["#", 0, ["revision"]],
["s", 0, 2],
],
});
expect(secondRight).toMatchObject({
ops: [
["#", 0, ["revision"]],
["s", 0, 2],
],
});
expect(firstDecoder.decodeUpdate(firstLeft)).toEqual(update("left", 1, 1));
expect(firstDecoder.decodeUpdate(firstRight)).toEqual(update("right", 1, 1));
expect(firstDecoder.decodeUpdate(secondLeft)).toEqual(update("left", 2, 2));
expect(firstDecoder.decodeUpdate(secondRight)).toEqual(update("right", 2, 2));
const independentLeft = secondEncoder.encodeUpdate(update("left", 1, 1));
expect(independentLeft).toMatchObject({ ops: [["s", ["revision"], 1]] });
expect(secondDecoder.decodeUpdate(independentLeft)).toEqual(update("left", 1, 1));
const leftBase: DecodedServiceProviderUpdate = {
type: "state",
member: "left",
sequence: 3,
ops: [["r", { revision: 3 }]],
};
expect(firstDecoder.decodeUpdate(firstEncoder.encodeUpdate(leftBase))).toEqual(leftBase);
const thirdRight = firstEncoder.encodeUpdate(update("right", 3, 3));
expect(thirdRight).toMatchObject({ ops: [["s", 0, 3]] });
expect(firstDecoder.decodeUpdate(thirdRight)).toEqual(update("right", 3, 3));
});
test("creates and removes keyed instance codecs with their lifecycle", () => {
const enc = createServiceStateEncoder();
const dec = createServiceStateDecoder();
const snapshot: DecodedServiceSubscriptionSnapshot = {
serviceId: "pi.dialogs",
mode: "keyed",
instances: [],
};
expect(dec.decodeSnapshot(enc.encodeSnapshot(snapshot))).toEqual(snapshot);
const address = { key: "dialog-1", generation: 1 };
const spawned: DecodedServiceProviderUpdate = {
type: "spawned",
instance: {
instance: address,
members: [{ name: "request", kind: "state", sequence: 0, ops: [["r", { value: 0 }]] }],
},
};
expect(dec.decodeUpdate(enc.encodeUpdate(spawned))).toEqual(spawned);
const update: DecodedServiceProviderUpdate = {
type: "state",
instance: address,
member: "request",
sequence: 1,
ops: [["s", ["value"], 1]],
};
expect(dec.decodeUpdate(enc.encodeUpdate(update))).toEqual(update);
const closed: DecodedServiceProviderUpdate = { type: "closed", instance: address };
expect(dec.decodeUpdate(enc.encodeUpdate(closed))).toEqual(closed);
expect(() => enc.encodeUpdate({ ...update, sequence: 2 })).toThrow("Unknown service state");
});
test("keeps transport RPC payloads opaque while validating their envelope", () => {
const message: ClientMessage = {
test("keeps routed request and event payloads opaque", () => {
const request: ClientMessage = {
type: "request",
id: "request-1",
target: {
@ -308,14 +75,28 @@ describe("protocol validation", () => {
},
call: {
serviceId: "application.custom",
instance: { key: "instance-1", generation: 2 },
member: "invoke",
args: [{ arbitrary: true }, ["opaque"]],
},
};
expect(parseClientMessage(message)).toEqual(message);
expect(parseClientMessage(request)).toEqual(request);
expect(
parseClientMessage({
...request,
call: { arbitrary: "strict JSON whose service meaning belongs to Chord" },
}),
).toMatchObject({ call: { arbitrary: expect.any(String) } });
expect(
parseServerMessage({
type: "service_update",
subscriptionId: "subscription-1",
update: { applicationDefined: true },
}),
).toMatchObject({ update: { applicationDefined: true } });
});
test("rejects non-JSON opaque service payloads", () => {
test("rejects non-JSON opaque payloads", () => {
const request = {
type: "request",
id: "request-1",
@ -374,17 +155,18 @@ describe("protocol validation", () => {
{
type: "request",
id: "",
serverId: "00000000-0000-4000-8000-000000000001",
call: { method: "list", args: [] },
target: { serverId: "00000000-0000-4000-8000-000000000001" },
call: { serviceId: "pi.models", member: "list", args: [] },
},
],
[
"extra call field",
"extra envelope field",
{
type: "request",
id: "request-1",
serverId: "00000000-0000-4000-8000-000000000001",
call: { method: "list", args: [], extra: true },
target: { serverId: "00000000-0000-4000-8000-000000000001" },
call: { serviceId: "pi.models", member: "list", args: [] },
extra: true,
},
],
] as const)("rejects malformed request boundaries: %s", (_label, message) => {
@ -402,27 +184,23 @@ describe("protocol validation", () => {
test.each([
["invalid server id", { ...serverHello, serverId: "server-1" }],
["extra response field", { type: "response", id: "request-1", ok: true, result: [], extra: true }],
["empty error code", { type: "response", id: "request-1", ok: false, error: { code: "", message: "bad" } }],
] as const)("rejects malformed server boundaries: %s", (_label, message) => {
expect(() => parseServerMessage(message)).toThrow(ProtocolValidationError);
});
test.each([
"wrong_server",
"session_not_found",
"session_not_attached",
"not_supported",
"server_draining",
"cancelled",
"internal_error",
] as const)("accepts the %s error code", (code) => {
const message: ServerMessage = {
type: "response",
id: "request-1",
ok: false,
error: { code, message: "safe" },
};
expect(parseServerMessage(message)).toEqual(message);
});
test.each(["wrong_server", "cancelled", "service_not_found", "application_error"] as const)(
"accepts the opaque %s error code",
(code) => {
const message: ServerMessage = {
type: "response",
id: "request-1",
ok: false,
error: { code, message: "safe" },
};
expect(parseServerMessage(message)).toEqual(message);
},
);
test("rejects unknown messages and fields", () => {
expect(() => parseServerMessage({ ...serverHello, snapshot: {} })).toThrow(ProtocolValidationError);

2
packages/server/README.md
View file

@ -74,6 +74,6 @@ Applications supply a required server service host, a bounded Session resolver,
`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.
`Server` composes transports through `ServerListener`; peer authentication remains application policy and is not implemented by the experimental Unix transport. The Unix submodule provides `createUnixListener()` and `createUnixServer()`. Low-level CBOR framing and validation come from `@earendil-works/pi-protocol`.
`Server` composes transports through `ServerListener`; peer authentication remains application policy and is not implemented by the experimental Unix transport. The Unix submodule provides `createUnixListener()` and `createUnixServer()`. Low-level routed-envelope validation, CBOR, and framing come from `@earendil-works/pi-protocol`; Chord owns service-control parsing, error codes, snapshots and updates, and each subscription's replicated-state encoder.
Server and worker lifecycle is managed outside the public Pi protocol. The replaceable application server converts connection attachments into private demand updates; the worker combines generation-tagged demand with authoritative Harness activity. The experimental coordinator only supplies stable routing and reports generic server-generation connection changes.

3
packages/server/src/connection.ts
View file

@ -1,4 +1,5 @@
import type { ClientMessageDecoder, RpcTarget, ServiceStateEncoder } from "@earendil-works/pi-protocol";
import type { ServiceStateEncoder } from "@earendil-works/chord";
import type { ClientMessageDecoder, RpcTarget } from "@earendil-works/pi-protocol";
import type { MaybePromise, RoutedServerServiceAttachment } from "./types.ts";

19
packages/server/src/errors.ts
View file

@ -1,16 +1,13 @@
import type { ProtocolErrorCode, ServiceErrorCode } from "@earendil-works/pi-protocol";
import type { RemoteServiceErrorCode } from "@earendil-works/chord";
type ServerOperationErrorCode =
| ServiceErrorCode
| Extract<
ProtocolErrorCode,
| "wrong_server"
| "session_not_found"
| "session_ambiguous"
| "session_not_attached"
| "not_supported"
| "server_draining"
>;
| RemoteServiceErrorCode
| "wrong_server"
| "session_not_found"
| "session_ambiguous"
| "session_not_attached"
| "not_supported"
| "server_draining";
export const INTERNAL_SERVER_ERROR_MESSAGE = "Internal server error";

38
packages/server/src/server.ts
View file

@ -1,7 +1,12 @@
import {
createServiceStateEncoder,
decodeServiceControlCall,
type JsonValue,
parseServiceCall,
parseServiceSubscriptionSnapshot,
RemoteServiceError,
type ServiceCall,
type ServiceProviderUpdate,
type ServiceSubscriptionSnapshot,
} from "@earendil-works/chord";
import { BACKGROUND_CONTEXT, type SessionMetadata, TODO_CONTEXT, withAbortSignal } from "@earendil-works/pi-agent-core";
import {
@ -9,15 +14,12 @@ import {
type ClientHello,
type ClientMessage,
ClientMessageDecoder,
createServiceStateEncoder,
DEFAULT_MAX_FRAME_LENGTH,
decodeServiceControlCall,
encodeServerMessage,
isServerId,
isSupportedProtocolVersion,
PROTOCOL_VERSION,
type ProtocolError,
type ProtocolRpcResult,
ProtocolValidationError,
type RequestEnvelope,
type ResponseEnvelope,
@ -311,11 +313,23 @@ export class Server<TMetadata extends SessionMetadata = SessionMetadata> {
} satisfies ResponseEnvelope);
return;
}
let call: ServiceCall;
try {
call = parseServiceCall(envelope.call);
} catch {
await this.sendMessage(state, {
type: "response",
id: envelope.id,
ok: false,
error: { code: "invalid_request", message: "Invalid service call" },
} satisfies ResponseEnvelope);
return;
}
const controller = new AbortController();
const active = { controller, target: envelope.target };
state.activeRequests.set(envelope.id, active);
const context = withAbortSignal(controller.signal, TODO_CONTEXT);
const control = decodeServiceControlCall(envelope.call);
const control = decodeServiceControlCall(call);
const subscribing = control?.type === "subscribe" ? control : undefined;
const pendingUpdates: { readonly update: ServiceProviderUpdate }[] = [];
let subscriptionReady = subscribing === undefined;
@ -333,21 +347,19 @@ export class Server<TMetadata extends SessionMetadata = SessionMetadata> {
if (subscribing !== undefined && state.serviceStateEncoders.has(subscribing.subscriptionId)) {
throw new ProtocolValidationError(`Duplicate service subscription ${subscribing.subscriptionId}`);
}
let result: ProtocolRpcResult;
let result: JsonValue | undefined;
if ("sessionId" in envelope.target) {
result = await this.sessions.executeServiceCall(envelope.call, envelope.target, state, publish, context);
result = await this.sessions.executeServiceCall(call, envelope.target, state, publish, context);
} else if (state.serverServices !== undefined) {
result = await state.serverServices.invokeService(envelope.call, publish, context);
result = await state.serverServices.invokeService(call, publish, context);
} else {
throw new ProtocolValidationError(
`Unknown service member ${envelope.call.serviceId}.${envelope.call.member}`,
);
throw new ProtocolValidationError(`Unknown service member ${call.serviceId}.${call.member}`);
}
if (subscribing !== undefined) {
if (result === undefined)
throw new ProtocolValidationError("Service subscription did not return a snapshot");
const stateEncoder = createServiceStateEncoder();
result = stateEncoder.encodeSnapshot(result as unknown as ServiceSubscriptionSnapshot) as ProtocolRpcResult;
result = stateEncoder.encodeSnapshot(parseServiceSubscriptionSnapshot(result)) as unknown as JsonValue;
state.serviceStateEncoders.set(subscribing.subscriptionId, stateEncoder);
installedSubscriptionEncoder = true;
} else if (control?.type === "unsubscribe") {
@ -433,7 +445,7 @@ export class Server<TMetadata extends SessionMetadata = SessionMetadata> {
await this.sendMessage(connection, {
type: "service_update",
subscriptionId,
update: stateEncoder.encodeUpdate(update),
update: stateEncoder.encodeUpdate(update) as unknown as JsonValue,
});
}

12
packages/server/src/session-router.ts
View file

@ -1,7 +1,7 @@
import { randomUUID } from "node:crypto";
import type { ServiceProviderUpdate } from "@earendil-works/chord";
import type { JsonValue, ServiceCall, ServiceProviderUpdate } from "@earendil-works/chord";
import { BACKGROUND_CONTEXT, type Context, type SessionMetadata } from "@earendil-works/pi-agent-core";
import type { ProtocolRpcCall, ProtocolRpcResult, RpcTarget, SessionTarget } from "@earendil-works/pi-protocol";
import type { RpcTarget, SessionTarget } from "@earendil-works/pi-protocol";
import { NotSupportedError, ServerDrainingError, SessionNotAttachedError } from "./errors.ts";
import type { RoutedSessionAttachment, RoutedSessionHandle, ServerHost } from "./types.ts";
@ -45,12 +45,12 @@ export class SessionRouter<TMetadata extends SessionMetadata = SessionMetadata>
}
async executeServiceCall(
call: ProtocolRpcCall,
call: ServiceCall,
target: RpcTarget,
client: object,
publish: (subscriptionId: string, update: ServiceProviderUpdate, context: Context) => Promise<void>,
context: Context,
): Promise<ProtocolRpcResult> {
): Promise<JsonValue | undefined> {
const admitted = await this.runForClient(client, () =>
this.startServiceCall(client, target, call, publish, context),
);
@ -199,10 +199,10 @@ export class SessionRouter<TMetadata extends SessionMetadata = SessionMetadata>
private async startServiceCall(
client: object,
target: RpcTarget,
call: ProtocolRpcCall,
call: ServiceCall,
publish: (subscriptionId: string, update: ServiceProviderUpdate, context: Context) => Promise<void>,
context: Context,
): Promise<{ result: Promise<ProtocolRpcResult> }> {
): Promise<{ result: Promise<JsonValue | undefined> }> {
const attachment = this.requireAttachment(client, target);
const invoke = attachment.lease?.invokeService;
if (invoke === undefined) throw new NotSupportedError("Routed Session does not support plugin services");

8
packages/server/src/testing/client.ts
View file

@ -1,10 +1,10 @@
import { once } from "node:events";
import { createConnection, type Socket } from "node:net";
import type { JsonValue, ServiceCall } from "@earendil-works/chord";
import {
type ClientMessage,
encodeClientMessage,
PROTOCOL_VERSION,
type ProtocolRpcCall,
type ResponseEnvelope,
type RpcTarget,
type ServerMessage,
@ -50,13 +50,13 @@ export class ProtocolTestClient {
async requestService(
target: RpcTarget,
call: ProtocolRpcCall,
call: ServiceCall,
id = `request-${++this.requestSequence}`,
): Promise<ResponseEnvelope> {
const response = this.next(
(message): message is ResponseEnvelope => message.type === "response" && message.id === id,
);
await this.sendMessage({ type: "request", id, target, call });
await this.sendMessage({ type: "request", id, target, call: call as unknown as JsonValue });
return (await response) as ResponseEnvelope;
}
@ -70,7 +70,7 @@ export class ProtocolTestClient {
requestSessionService(
serverId: string,
sessionId: string,
call: ProtocolRpcCall,
call: ServiceCall,
id?: string,
): Promise<ResponseEnvelope> {
const attachment = this.attachment;

8
packages/server/src/testing/host.ts
View file

@ -1,6 +1,6 @@
import type { JsonValue, ServiceCall } from "@earendil-works/chord";
import type { Context, Session, SessionMetadata } from "@earendil-works/pi-agent-core";
import { BACKGROUND_CONTEXT, MemorySessionRepo } from "@earendil-works/pi-agent-core";
import type { ProtocolRpcCall, ProtocolRpcResult } from "@earendil-works/pi-protocol";
import { SessionAmbiguousError, SessionNotFoundError } from "../errors.ts";
import type { RoutedServerServiceHost, RoutedSessionHandle, ServerHost } from "../types.ts";
@ -32,11 +32,11 @@ export class TestHarness {
attachedClients = 0;
attachmentReleaseCount = 0;
closeCount = 0;
readonly serviceCalls: ProtocolRpcCall[] = [];
readonly serviceCalls: ServiceCall[] = [];
failAttachmentRelease?: Error;
failClose?: Error;
nextServiceError?: Error;
nextServiceResult: ProtocolRpcResult = { ok: true };
nextServiceResult: JsonValue | undefined = { ok: true };
private nextCloseGate?: OpenGate;
private nextServiceGate?: OpenGate;
@ -62,7 +62,7 @@ export class TestHarness {
};
}
async invokeService(call: ProtocolRpcCall): Promise<ProtocolRpcResult> {
async invokeService(call: ServiceCall): Promise<JsonValue | undefined> {
this.serviceCalls.push(call);
if (this.nextServiceError) {
const error = this.nextServiceError;

11
packages/server/src/types.ts
View file

@ -1,6 +1,5 @@
import type { ServiceProviderUpdate } from "@earendil-works/chord";
import type { JsonValue, ServiceCall, ServiceProviderUpdate } from "@earendil-works/chord";
import type { Context, SessionMetadata } from "@earendil-works/pi-agent-core";
import type { ProtocolRpcCall, ProtocolRpcResult } from "@earendil-works/pi-protocol";
import type { ServerListener } from "./listener.ts";
export interface ServerOptions {
@ -19,10 +18,10 @@ export type MaybePromise<T> = T | Promise<T>;
export interface RoutedSessionAttachment {
/** Route one contract-agnostic service operation to the attached Session endpoint. */
invokeService?(
call: ProtocolRpcCall,
call: ServiceCall,
publish: (subscriptionId: string, update: ServiceProviderUpdate, context: Context) => MaybePromise<void>,
context: Context,
): Promise<ProtocolRpcResult>;
): Promise<JsonValue | undefined>;
release(context: Context): MaybePromise<void>;
}
@ -37,10 +36,10 @@ export interface RoutedServerPresentation {
/** One connection's server-scoped service endpoint. */
export interface RoutedServerServiceAttachment {
invokeService(
call: ProtocolRpcCall,
call: ServiceCall,
publish: (subscriptionId: string, update: ServiceProviderUpdate, context: Context) => MaybePromise<void>,
context: Context,
): Promise<ProtocolRpcResult>;
): Promise<JsonValue | undefined>;
release(context: Context): MaybePromise<void>;
}

18
packages/server/test/conformance.test.ts
View file

@ -1,5 +1,5 @@
import type { ServiceCall } from "@earendil-works/chord";
import { BACKGROUND_CONTEXT, type SessionMetadata } from "@earendil-works/pi-agent-core";
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";
@ -59,7 +59,7 @@ function connect(server: Server): ProtocolTestClient {
return client;
}
function sessionCall(member: string, args: ProtocolRpcCall["args"] = []): ProtocolRpcCall {
function sessionCall(member: string, args: ServiceCall["args"] = []): ServiceCall {
return { serviceId: "test.session", member, args };
}
@ -87,6 +87,20 @@ describe("Session protocol", () => {
expect(host.harnesses.size).toBe(0);
});
test("rejects a semantically invalid service call after envelope decoding", async () => {
const host = new TestServerHost();
const client = connect(createServer(host));
await client.hello();
const response = client.next((message) => message.type === "response" && message.id === "invalid-call");
await client.sendMessage({
type: "request",
id: "invalid-call",
target: { serverId },
call: { arbitrary: true },
});
expect(await response).toMatchObject({ ok: false, error: { code: "invalid_request" } });
});
test("attach passes concrete repository metadata to the Harness host", async () => {
type BackendMetadata = SessionMetadata & { path: string; modifiedAt: number };
const metadata: BackendMetadata = {