feat(coding-agent): replace lane RPC with Chord service

This commit is contained in:
Armin Ronacher 2026-08-31 08:03:17 +02:00
parent ba74f038c6
commit ae2cc5116f
49 changed files with 760 additions and 2567 deletions

View file

@ -1,3 +1,4 @@
import type { JsonRepresentation } from "@earendil-works/chord";
import type {
Api,
AssistantMessage,
@ -383,6 +384,20 @@ export type HarnessEvent =
| (GlobalConfigEventPayload & { lane?: never; recovery?: never })
| (HandlerErrorPayload & ({ lane: string; recovery?: true } | { lane?: never; recovery?: never }));
type LaneWatchSourceEvent =
| Exclude<
HarnessEvent,
| { type: "handler_error" | "turn_start" | "turn_end" | "value_update" | "lane_created" | "message_update" }
| ({ type: "config_update" } & { property: string })
>
| Extract<HarnessEvent, { type: "config_update"; property: "model" | "thinkingLevel" | "activeTools" }>
| Omit<Extract<HarnessEvent, { type: "message_update" }>, "event">;
/** Strict-JSON snapshot representation published to remote transcript consumers. */
export type LaneTranscriptSnapshot = JsonRepresentation<LaneSnapshot>;
/** Reducer-relevant strict-JSON Harness events published to remote transcript consumers. */
export type LaneWatchEvent = JsonRepresentation<LaneWatchSourceEvent>;
export type HarnessEventType = HarnessEvent["type"];
export type EventListener<TEvent extends HarnessEvent = HarnessEvent> = (
event: TEvent,

View file

@ -1,4 +1,4 @@
import type { HarnessEvent, LaneSnapshot } from "../agent-harness.ts";
import type { HarnessEvent, LaneSnapshot, LaneWatchEvent } from "../agent-harness.ts";
import type { OperationResultRecord } from "../session/types.ts";
export type LaneSnapshotReduction = LaneSnapshot | { rebase: true };
@ -11,7 +11,10 @@ function matchingOperation(
}
/** Apply one harness event to a replicated lane snapshot. Navigation completion requires a fresh snapshot. */
export function reduceLaneSnapshot(snapshot: LaneSnapshot, event: HarnessEvent): LaneSnapshotReduction {
export function reduceLaneSnapshot(
snapshot: LaneSnapshot,
event: HarnessEvent | LaneWatchEvent,
): LaneSnapshotReduction {
if ("lane" in event && event.lane !== undefined && event.lane !== snapshot.lane && event.type !== "usage") {
return snapshot;
}

View file

@ -90,7 +90,7 @@ Applications may define their own context keys. A Pi adapter can carry telemetry
A context object never crosses RPC as a business value. The calling peer sends cancellation control and, if configured, an opaque strict-JSON metadata carrier. The receiving adapter constructs a fresh local context. The adapter, not remote business arguments, installs authenticated local identity.
Chord owns the static `JsonValue` contract. The current implementation deliberately performs no recursive runtime validation; concrete serializers are responsible for rejecting unsupported values. Remote arguments, results, errors, snapshots, updates, catalogues, and RPC envelopes are expected to be finite strict JSON:
Chord owns the static `JsonValue` contract and provides `JsonRepresentation<T>`, `isJsonValue()`, and `cloneJsonValue()` for adapter boundaries. The service runtime deliberately performs no automatic recursive validation; concrete serializers remain responsible for rejecting or normalizing unsupported values. Remote arguments, results, errors, snapshots, updates, catalogues, and RPC envelopes are expected to be finite strict JSON:
- finite numbers only;
- no `undefined`, sparse arrays, symbols, prototypes, cycles, classes, functions, `Map`, or `Set`; and

View file

@ -46,8 +46,11 @@ The design has a few connected pieces:
calls and subscriptions through an application-supplied adapter. Chord
requires strict-JSON arguments, results, snapshots, updates, and catalogues,
but does not prescribe framing, routing, transport, or an application wire
envelope. Symmetric RPC peers are planned as one optional implementation of
this boundary.
envelope. `JsonRepresentation<T>` derives a wire-safe type for application data
with unknown payloads; `isJsonValue()` validates received values, while
`cloneJsonValue()` validates, normalizes optional fields, and detaches outbound
values at an adapter boundary. Symmetric RPC peers are planned as one optional
implementation of this boundary.
- **Context** Chord provides a Go-like context system for cancellation and
invocation-scoped application values. Applications can carry permissions or

View file

@ -10,6 +10,7 @@ export {
defineService,
replicatedState,
} from "./api.ts";
export { cloneJsonValue, isJsonValue } from "./json.ts";
export { RemoteServiceError } from "./services/errors.ts";
export { RemoteServiceProvider } from "./services/provider.ts";
export type {
@ -20,6 +21,7 @@ export type {
FacetHost,
FacetLoader,
FacetOptions,
JsonRepresentation,
JsonValue,
LoadedFacets,
MutableReplicatedState,

110
packages/chord/src/json.ts Normal file
View file

@ -0,0 +1,110 @@
import type { JsonRepresentation, JsonValue } from "./types.ts";
/** Return whether a value is finite strict JSON with plain objects and no cycles. */
export function isJsonValue(value: unknown): value is JsonValue {
return check(value, new Set<object>(), 0);
}
function check(value: unknown, ancestors: Set<object>, depth: number): boolean {
if (depth > 512) return false;
if (value === null || typeof value === "string" || typeof value === "boolean") return true;
if (typeof value === "number") return Number.isFinite(value);
if (Array.isArray(value)) {
const keys = Reflect.ownKeys(value);
if (keys.length !== value.length + 1 || keys.some((key) => typeof key !== "string")) return false;
if (ancestors.has(value)) return false;
ancestors.add(value);
try {
for (let index = 0; index < value.length; index++) {
const descriptor = Object.getOwnPropertyDescriptor(value, String(index));
if (
descriptor === undefined ||
!descriptor.enumerable ||
!("value" in descriptor) ||
!check(descriptor.value, ancestors, depth + 1)
) {
return false;
}
}
return true;
} finally {
ancestors.delete(value);
}
}
if (typeof value !== "object" || value === null) return false;
const prototype = Object.getPrototypeOf(value);
if (prototype !== Object.prototype && prototype !== null) return false;
if (Reflect.ownKeys(value).some((key) => typeof key !== "string")) return false;
if (ancestors.has(value)) return false;
ancestors.add(value);
try {
for (const descriptor of Object.values(Object.getOwnPropertyDescriptors(value))) {
if (!descriptor.enumerable || !("value" in descriptor) || !check(descriptor.value, ancestors, depth + 1)) {
return false;
}
}
return true;
} finally {
ancestors.delete(value);
}
}
/** Clone one finite plain JavaScript value as strict JSON, omitting undefined object properties. */
export function cloneJsonValue<T>(value: T): JsonRepresentation<T> {
return clone(value, new Set<object>()) as JsonRepresentation<T>;
}
function clone(value: unknown, ancestors: Set<object>): JsonValue {
if (value === null || typeof value === "string" || typeof value === "boolean") return value;
if (typeof value === "number") {
if (!Number.isFinite(value)) throw new TypeError("JSON values must contain only finite numbers");
return value;
}
if (Array.isArray(value)) {
const keys = Reflect.ownKeys(value);
if (keys.length !== value.length + 1 || keys.some((key) => typeof key !== "string")) {
throw new TypeError("JSON arrays must not contain extra properties");
}
if (ancestors.has(value)) throw new TypeError("JSON values must not contain cycles");
ancestors.add(value);
try {
const result: JsonValue[] = [];
for (let index = 0; index < value.length; index++) {
const descriptor = Object.getOwnPropertyDescriptor(value, String(index));
if (descriptor === undefined) throw new TypeError("JSON arrays must not contain empty slots");
if (!descriptor.enumerable || !("value" in descriptor)) {
throw new TypeError("JSON arrays must contain only data properties");
}
if (descriptor.value === undefined) throw new TypeError("JSON arrays must not contain undefined values");
result.push(clone(descriptor.value, ancestors));
}
return result;
} finally {
ancestors.delete(value);
}
}
if (typeof value === "object" && value !== null) {
const prototype = Object.getPrototypeOf(value);
if (prototype !== Object.prototype && prototype !== null) {
throw new TypeError("JSON values must contain only plain objects");
}
if (Object.getOwnPropertySymbols(value).length > 0) {
throw new TypeError("JSON objects must not contain symbol properties");
}
if (ancestors.has(value)) throw new TypeError("JSON values must not contain cycles");
ancestors.add(value);
try {
const result: { [key: string]: JsonValue } = {};
for (const [key, descriptor] of Object.entries(Object.getOwnPropertyDescriptors(value))) {
if (!descriptor.enumerable || !("value" in descriptor)) {
throw new TypeError("JSON objects must contain only enumerable data properties");
}
if (descriptor.value !== undefined) result[key] = clone(descriptor.value, ancestors);
}
return result;
} finally {
ancestors.delete(value);
}
}
throw new TypeError(`Unsupported JSON value: ${typeof value}`);
}

View file

@ -19,6 +19,21 @@ export interface Context {
export type JsonValue = null | boolean | number | string | JsonValue[] | { [key: string]: JsonValue };
type IsAny<T> = 0 extends 1 & T ? true : false;
/** Strict-JSON representation of an application data type. Unknown payloads become JsonValue. */
export type JsonRepresentation<T> = IsAny<T> extends true
? JsonValue
: unknown extends T
? JsonValue
: T extends null | boolean | number | string
? T
: T extends readonly (infer TItem)[]
? JsonRepresentation<TItem>[]
: T extends object
? { [TKey in keyof T]: JsonRepresentation<T[TKey]> }
: never;
export interface ReplicatedState<T> {
/** Borrowed immutable value, or undefined until hydration. Do not mutate or retain it. */
readonly value: T | undefined;
@ -44,11 +59,17 @@ export interface Service<T> {
readonly [SERVICE_TYPE]?: (value: T) => T;
}
type InvalidJsonPart<T> = [T] extends [JsonValue]
? [JsonValue] extends [T]
type InvalidJsonPart<T> = IsAny<T> extends true
? T
: unknown extends T
? never
: InvalidJsonStructure<T>
: InvalidJsonStructure<T>;
: [T] extends [JsonValue]
? [JsonValue] extends [T]
? never
: InvalidJsonStructure<T>
: InvalidJsonStructure<T>;
type InvalidJsonProperty<T> = [Exclude<T, undefined>] extends [never] ? T : InvalidJsonPart<Exclude<T, undefined>>;
type InvalidJsonStructure<T> = T extends null | boolean | number | string
? never
@ -57,7 +78,7 @@ type InvalidJsonStructure<T> = T extends null | boolean | number | string
: T extends (...args: never[]) => unknown
? T
: T extends object
? { [TKey in keyof T]-?: InvalidJsonPart<T[TKey]> }[keyof T]
? { [TKey in keyof T]-?: InvalidJsonProperty<T[TKey]> }[keyof T]
: T;
type InvalidRemoteMember<T> = T extends ReplicatedState<infer TValue>

View file

@ -0,0 +1,35 @@
import { describe, expect, test } from "vitest";
import { cloneJsonValue, isJsonValue } from "../src/index.ts";
describe("cloneJsonValue", () => {
test("clones strict JSON and omits undefined object properties", () => {
const source = { nested: { value: 1 }, omitted: undefined, values: [true, null, "value"] };
const cloned = cloneJsonValue(source);
expect(cloned).toEqual({ nested: { value: 1 }, values: [true, null, "value"] });
expect(cloned).not.toBe(source);
expect(cloned.nested).not.toBe(source.nested);
});
test.each([[new Date()], [{ value: Number.NaN }], [[undefined]], [Array(1)], [1n]])(
"rejects non-JSON values",
(value) => {
expect(() => cloneJsonValue(value)).toThrow(TypeError);
},
);
test("rejects cycles", () => {
const value: { self?: unknown } = {};
value.self = value;
expect(() => cloneJsonValue(value)).toThrow(/cycles/);
});
test("checks strict JSON without normalizing it", () => {
expect(isJsonValue({ nested: [1, true, null] })).toBe(true);
expect(isJsonValue({ omitted: undefined })).toBe(false);
expect(isJsonValue(new Uint8Array([1]))).toBe(false);
expect(isJsonValue(Number.POSITIVE_INFINITY)).toBe(false);
const cyclic: { self?: unknown } = {};
cyclic.self = cyclic;
expect(isJsonValue(cyclic)).toBe(false);
});
});

View file

@ -29,7 +29,7 @@ The client verifies that the physical endpoint reports the expected logical `ser
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.
`watchSession()` is the remaining low-level compatibility API. It creates a main-lane watch and returns its authoritative snapshot without starting event delivery. Install the listener with `await watch.start(listener)`; this then flushes events buffered after the snapshot and continues with live events. `watch.dispose()` stops server delivery and waits for already-received listener work. A disconnected watch is stale and cannot be reused after reconnection.
Application observation APIs such as the coding agent's `Transcript` are ordinary Chord services. The client does not interpret their snapshots or updates.
On disconnect or disposal, pending requests reject locally, but accepted work may still complete remotely before the attachment is released. The client clears its live attachment route. It never reconnects or replays requests automatically. After disconnection, call `reconnect()`, attach through the application's management service again, and explicitly repeat only operations known to be safe.

View file

@ -1,23 +1,20 @@
import {
type AttachmentEnvelope,
createRpcClient,
createServiceCatalogueCall,
createServiceSubscribeCall,
createServiceUnsubscribeCall,
encodeClientMessage,
encodeLaneWatchRpcCall,
isServerId,
type LaneEvent,
LaneWatchRpc,
type ProtocolRpcCall,
type ProtocolRpcResult,
ProtocolValidationError,
parseServiceCatalogue,
parseServiceSubscriptionSnapshot,
type ResponseEnvelope,
type RpcTarget,
type ServerEventEnvelope,
type ServerHello,
type ServiceCatalogueEntry,
type ServiceEventEnvelope,
type ServiceMode,
type ServiceProviderUpdate,
type ServiceSubscriptionSnapshot,
@ -31,22 +28,16 @@ import type {
ClientOptions,
ConnectionState,
ConnectionStateChange,
LaneWatch,
ServiceSubscription,
Unsubscribe,
} from "./types.ts";
interface PendingRequest {
resolve(result: unknown): void;
resolve(result: ProtocolRpcResult): void;
reject(error: Error): void;
cleanup(): void;
}
interface ActiveWatchListener {
readonly listener: (event: LaneEvent) => void | Promise<void>;
deliveryTail: Promise<void>;
}
interface ActiveServiceListener {
readonly target: RpcTarget;
readonly listener: (update: ServiceProviderUpdate) => void | Promise<void>;
@ -61,9 +52,7 @@ export class Client {
readonly #pendingRequests = new Map<string, PendingRequest>();
readonly #connectionStateListeners = new Set<(change: ConnectionStateChange) => void>();
readonly #attachmentListeners = new Set<AttachmentChangeListener>();
readonly #watchListeners = new Map<string, ActiveWatchListener>();
readonly #serviceListeners = new Map<string, ActiveServiceListener>();
readonly #laneWatchRpc: ReturnType<typeof createRpcClient<typeof LaneWatchRpc>>;
#requestSequence = 0;
#serviceSubscriptionSequence = 0;
#hello: ServerHello | undefined;
@ -86,11 +75,6 @@ export class Client {
onMessage: (message) => this.#handleMessage(message),
onStateChange: (change) => this.#handleConnectionStateChange(change),
});
this.#laneWatchRpc = createRpcClient(
LaneWatchRpc,
(call) => this.#request(this.#requireSessionTarget(), encodeLaneWatchRpcCall(call)),
(message) => new ProtocolValidationError(message),
);
}
get disposed(): boolean {
@ -155,7 +139,7 @@ export class Client {
}
/** Invoke one low-level protocol call against an explicit routed target. */
request(target: RpcTarget, call: ProtocolRpcCall, signal?: AbortSignal): Promise<unknown> {
request(target: RpcTarget, call: ProtocolRpcCall, signal?: AbortSignal): Promise<ProtocolRpcResult> {
return this.#request(target, call, signal);
}
@ -235,65 +219,12 @@ export class Client {
};
}
async watchSession(sessionId: string): Promise<LaneWatch> {
this.#requireSessionTarget(sessionId);
const { watchId, snapshot } = await this.#laneWatchRpc.watch();
const connection = this.#hello;
let currentSnapshot = snapshot;
let state: "ready" | "starting" | "started" | "disposed" = "ready";
return {
id: watchId,
sessionId,
get snapshot() {
return currentSnapshot;
},
start: async (listener) => {
if (state !== "ready") throw new Error("Lane watch may be started only once");
if (this.#watchListeners.has(watchId)) {
this.#connection.fail(new ProtocolValidationError("Server reused an active lane watch ID"));
throw new ProtocolValidationError("Server reused an active lane watch ID");
}
state = "starting";
this.#watchListeners.set(watchId, { listener, deliveryTail: Promise.resolve() });
try {
this.#requireSessionTarget(sessionId);
await this.#laneWatchRpc.startWatch(watchId);
if (state === "starting") state = "started";
} catch (error) {
this.#watchListeners.delete(watchId);
state = "disposed";
throw error;
}
},
resnapshot: async () => {
if (state === "disposed") throw new Error("Lane watch is disposed");
this.#requireSessionTarget(sessionId);
const refreshed = await this.#laneWatchRpc.resnapshotWatch(watchId);
currentSnapshot = refreshed.snapshot;
return currentSnapshot;
},
dispose: async () => {
if (state === "disposed") return;
state = "disposed";
const active = this.#watchListeners.get(watchId);
try {
if (this.connected && this.#hello === connection && this.#attachment?.sessionId === sessionId) {
await this.#laneWatchRpc.stopWatch(watchId);
}
await active?.deliveryTail;
} finally {
this.#watchListeners.delete(watchId);
}
},
};
}
#request(target: RpcTarget, call: ProtocolRpcCall, signal?: AbortSignal): Promise<unknown> {
#request(target: RpcTarget, call: ProtocolRpcCall, signal?: AbortSignal): Promise<ProtocolRpcResult> {
if (this.#disposed) return Promise.reject(new ClientDisposedError());
if (!this.connected) return Promise.reject(new DisconnectedError());
if (signal?.aborted) return Promise.reject(abortError(signal));
const id = `request-${++this.#requestSequence}`;
const { promise, resolve, reject } = createPromiseResolvers<unknown>();
const { promise, resolve, reject } = createPromiseResolvers<ProtocolRpcResult>();
let sent = false;
let aborted = false;
let onAbort: (() => void) | undefined;
@ -339,7 +270,7 @@ export class Client {
return promise;
}
#handleMessage(message: ResponseEnvelope | ServerEventEnvelope | AttachmentEnvelope): void {
#handleMessage(message: ResponseEnvelope | ServiceEventEnvelope | AttachmentEnvelope): void {
if (message.type === "attachment") {
if (message.attachment !== null && message.attachment.serverId !== this.#options.serverId) {
this.#connection.fail(new ProtocolValidationError("Attachment update belongs to another server"));
@ -348,14 +279,6 @@ export class Client {
this.#setAttachment(message.attachment ?? undefined);
return;
}
if (message.type === "event") {
const active = this.#watchListeners.get(message.watchId);
if (active === undefined) return;
active.deliveryTail = active.deliveryTail
.then(() => active.listener(message.event))
.catch((error: unknown) => this.#reportListenerError(error));
return;
}
if (message.type === "service_update") {
const active = this.#serviceListeners.get(message.subscriptionId);
if (active === undefined) return;
@ -380,7 +303,6 @@ export class Client {
this.#hello = undefined;
this.#setAttachment(undefined);
this.#rejectPendingRequests(change.error ?? new DisconnectedError());
this.#watchListeners.clear();
this.#serviceListeners.clear();
}
for (const listener of this.#connectionStateListeners) {
@ -421,7 +343,6 @@ export class Client {
this.#setAttachment(undefined);
this.#connectionStateListeners.clear();
this.#attachmentListeners.clear();
this.#watchListeners.clear();
this.#serviceListeners.clear();
return this.#disposePromise;
}
@ -430,14 +351,6 @@ export class Client {
return this.dispose();
}
#requireSessionTarget(sessionId?: string): SessionTarget {
const attachment = this.#attachment;
if (attachment === undefined || (sessionId !== undefined && attachment.sessionId !== sessionId)) {
throw new ServerError({ code: "session_not_attached", message: "Session is not attached" });
}
return attachment;
}
#setAttachment(attachment: SessionTarget | undefined): void {
const previous = this.#attachment;
if (

View file

@ -6,7 +6,6 @@ export type {
ClientOptions,
ConnectionState,
ConnectionStateChange,
LaneWatch,
ListenerErrorHandler,
ServiceSubscription,
Unsubscribe,

View file

@ -1,10 +1,4 @@
import type {
LaneEvent,
LaneSnapshot,
RpcTarget,
ServiceSubscriptionSnapshot,
SessionTarget,
} from "@earendil-works/pi-protocol";
import type { RpcTarget, ServiceSubscriptionSnapshot, SessionTarget } from "@earendil-works/pi-protocol";
import type { ByteTransportFactory } from "./transport.ts";
export type ConnectionState = "disconnected" | "connecting" | "connected";
@ -18,15 +12,6 @@ export type Unsubscribe = () => void;
export type ListenerErrorHandler = (error: Error) => void;
export type AttachmentChangeListener = (attachment: SessionTarget | undefined) => void;
export interface LaneWatch {
readonly id: string;
readonly sessionId: string;
readonly snapshot: LaneSnapshot;
start(listener: (event: LaneEvent) => void | Promise<void>): Promise<void>;
resnapshot(): Promise<LaneSnapshot>;
dispose(): Promise<void>;
}
export interface ServiceSubscription {
readonly id: string;
readonly target: RpcTarget;

View file

@ -71,119 +71,6 @@ describe("Client service operations", () => {
await client.dispose();
});
test("starts lane event delivery only after receiving the authoritative snapshot", async () => {
const server = new MemoryByteServer();
const client = await connectClient(server);
await attachClient(client, server, "session-1");
const opening = client.watchSession("session-1");
await server.waitForMessages(3);
expect(server.messages[2]).toMatchObject({ call: { serviceId: "pi.transcript", member: "watch", args: [] } });
server.send({
type: "response",
id: "request-2",
ok: true,
result: {
watchId: "watch-1",
snapshot: {
lane: "main",
transcript: [],
tipId: null,
configuration: {
model: { provider: "faux", modelId: "faux-1" },
thinkingLevel: "off",
activeToolNames: [],
},
stats: {
messageCount: 0,
usage: {
input: 0,
output: 0,
cacheRead: 0,
cacheWrite: 0,
totalTokens: 0,
cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 },
},
},
operation: null,
queues: [],
faulted: false,
},
},
});
const watch = await opening;
expect(watch.snapshot).toMatchObject({ lane: "main", transcript: [] });
const events: string[] = [];
let releaseEvent!: () => void;
const eventGate = new Promise<void>((resolve) => {
releaseEvent = resolve;
});
const starting = watch.start(async (event) => {
await eventGate;
events.push(event.type);
});
await server.waitForMessages(4);
expect(server.messages[3]).toMatchObject({
call: { serviceId: "pi.transcript", member: "startWatch", args: ["watch-1"] },
});
server.send({
type: "event",
watchId: "watch-1",
event: { type: "run_start", lane: "main", runId: "run-1", startedAt: 1 },
});
await Promise.resolve();
expect(events).toEqual([]);
server.send({ type: "response", id: "request-3", ok: true, result: { watchId: "watch-1" } });
await starting;
const refreshing = watch.resnapshot();
await server.waitForMessages(5);
expect(server.messages[4]).toMatchObject({
call: { serviceId: "pi.transcript", member: "resnapshotWatch", args: ["watch-1"] },
});
const refreshed = {
...watch.snapshot,
operation: {
id: "run-1",
kind: "run" as const,
startedAt: 1,
fromTipId: null,
status: "open" as const,
runningTools: [],
},
};
server.send({
type: "response",
id: "request-4",
ok: true,
result: { watchId: "watch-1", snapshot: refreshed },
});
expect(await refreshing).toEqual(refreshed);
expect(watch.snapshot).toEqual(refreshed);
let disposed = false;
const disposing = watch.dispose().then(() => {
disposed = true;
});
await server.waitForMessages(6);
expect(server.messages[5]).toMatchObject({
call: { serviceId: "pi.transcript", member: "stopWatch", args: ["watch-1"] },
});
server.send({ type: "response", id: "request-5", ok: true, result: { watchId: "watch-1" } });
await Promise.resolve();
expect(disposed).toBe(false);
releaseEvent();
await disposing;
expect(events).toEqual(["run_start"]);
server.send({
type: "event",
watchId: "watch-1",
event: { type: "run_resume", lane: "main", runId: "run-1" },
});
expect(events).toEqual(["run_start"]);
await client.dispose();
});
test("buffers service updates until the subscription snapshot arrives", async () => {
const server = new MemoryByteServer();
const client = await connectClient(server);

View file

@ -8,8 +8,10 @@ export default defineConfig({
reporters: process.env.GITHUB_ACTIONS ? ["dot", "github-actions"] : ["dot"],
},
resolve: {
conditions: ["source"],
alias: {
"@earendil-works/pi-protocol": fileURLToPath(new URL("../protocol/src/index.ts", import.meta.url)),
},
},
ssr: { resolve: { conditions: ["source"] } },
});

View file

@ -17,6 +17,7 @@ import {
import { Models } from "./services/models.ts";
import { PresentationPlugins } from "./services/plugins.ts";
import { SessionDirectory, SessionManagement } from "./services/sessions.ts";
import { Transcript } from "./services/transcript.ts";
export type ClientRuntimeRoute =
| ({ readonly transport: "unix" } & UnixServerRoute)
@ -35,6 +36,7 @@ export interface ActivatedClientRuntimeServer extends ClientRuntimeServer {
readonly plugins: PresentationPlugins;
readonly models: Models;
readonly agent: AgentController;
readonly transcript: Transcript;
}
export interface ClientRuntime {
@ -190,7 +192,7 @@ export async function activateBuiltinClientServices(
onError() {},
});
const sessionServices = server.session.open({
services: [Models, AgentController],
services: [Models, AgentController, Transcript],
assertAccess() {},
onError() {},
});
@ -215,8 +217,9 @@ export async function activateBuiltinClientServices(
const plugins = serverServices.use(PresentationPlugins);
const models = sessionServices.use(Models);
const agent = sessionServices.use(AgentController);
const transcript = sessionServices.use(Transcript);
await Promise.all([serverServices.ready(BACKGROUND_CONTEXT), sessionServices.ready(BACKGROUND_CONTEXT)]);
return { ...server, directory, management, plugins, models, agent };
return { ...server, directory, management, plugins, models, agent, transcript };
}
function routeFromExplicitPath(path: string): UnixServerRoute {

View file

@ -31,7 +31,7 @@ import { InteractiveThemeController } from "../modes/interactive/theme/theme-con
import { createInteractiveTui } from "../modes/interactive/tui-renderer.ts";
import { type OpenClientRuntimeOptions, openClientRuntime } from "./client-runtime.ts";
import { ExperimentalChatView } from "./client-tui-chat.ts";
import { type LaneReplica, type LaneWatchSource, openLaneReplica } from "./lane-replica.ts";
import { type LaneReplica, openLaneReplica } from "./lane-replica.ts";
import { createPresentationFacetLoaders } from "./plugins/bundled.ts";
import { AgentController, type AgentOperationResponse, type AgentQueueResponse } from "./services/agent-controller.ts";
import type {
@ -48,6 +48,7 @@ import {
createBuiltInSlashCommandsFacet,
createSlashCommandsRuntimeFacet,
} from "./services/slash-commands-provider.ts";
import { Transcript, type Transcript as TranscriptService } from "./services/transcript.ts";
export interface RunClientTuiOptions extends OpenClientRuntimeOptions {
readonly facetLoader?: FacetLoader;
@ -56,7 +57,6 @@ export interface RunClientTuiOptions extends OpenClientRuntimeOptions {
export interface ClientTuiServer {
readonly serverId: string;
readonly radius: boolean;
readonly laneWatches: LaneWatchSource;
readonly server: ServerServiceSource;
readonly session: SessionServiceSource;
}
@ -64,7 +64,7 @@ export interface ClientTuiServer {
interface SessionFeature {
readonly serverId: string;
readonly session: SessionServiceSource;
readonly laneWatches: LaneWatchSource;
readonly transcript: TranscriptService;
}
interface PreparedClientSession {
@ -281,10 +281,11 @@ export class ExperimentalClientTui implements Component {
});
const commands = env.use(SlashCommands);
const controller = env.use(AgentController);
const transcript = env.use(Transcript);
const sessionFeature: SessionFeature = {
serverId: server.serverId,
session: server.session,
laneWatches: server.laneWatches,
transcript,
};
env.onActivate(() => {
if (this.#session !== undefined || this.#slashCommands !== undefined || this.#controller !== undefined) {
@ -494,7 +495,7 @@ export class ExperimentalClientTui implements Component {
async #openLane(feature: SessionFeature, sessionId: string): Promise<void> {
await this.#closeLane();
const replica = await openLaneReplica(feature.laneWatches, sessionId);
const replica = await openLaneReplica(feature.transcript, sessionId);
const view = new ExperimentalChatView(this.#ui, process.cwd());
view.apply(replica.state());
this.#laneReplica = replica;
@ -768,7 +769,6 @@ export async function runClientTui(command: ClientCommand, options: RunClientTui
servers: runtime.servers.map((server) => ({
serverId: server.route.serverId,
radius: server.route.transport === "radius",
laneWatches: server.client,
server: server.server,
session: server.session,
})),

View file

@ -1,8 +1,9 @@
import { resolve } from "node:path";
import { BACKGROUND_CONTEXT } from "@earendil-works/pi-agent-core";
import type { LaneEvent, WireAgentMessage } from "@earendil-works/pi-protocol";
import { BACKGROUND_CONTEXT, type LaneWatchEvent } from "@earendil-works/pi-agent-core";
import type { AssistantMessage } from "@earendil-works/pi-ai";
import type { ClientCommand } from "../cli/experimental/commands/client.ts";
import { activateBuiltinClientServices, openClientRuntime } from "./client-runtime.ts";
import { openLaneReplica } from "./lane-replica.ts";
import type { AgentOperationResponse } from "./services/agent-controller.ts";
import type { SessionAddress } from "./services/sessions.ts";
@ -18,7 +19,7 @@ export interface RunClientOptions {
/** Directory searched when --connect is omitted. Defaults to PI_SERVER_DIR or ~/.pi/server. */
readonly directory?: string;
/** Receives snapshot-ordered main-lane events while a prompt is active. */
readonly onEvent?: (event: LaneEvent) => void | Promise<void>;
readonly onEvent?: (event: LaneWatchEvent) => void | Promise<void>;
}
/** Discover servers, then list Sessions, attach to one, or create one for a prompt. */
@ -81,9 +82,7 @@ export async function runClient(command: ClientCommand, options: RunClientOption
const agent = match.agent;
const completedText = new Map<string, string>();
// AgentController returns no transcript content. Keep the lane watch until a transcript service owns live deltas.
const watch = await match.client.watchSession(sessionId);
await watch.start(async (event) => {
const replica = await openLaneReplica(match.transcript, sessionId, async (event) => {
if (event.type === "message_end" && event.runId !== undefined && event.message.role === "assistant") {
completedText.set(event.runId, messageText(event.message));
}
@ -93,7 +92,7 @@ export async function runClient(command: ClientCommand, options: RunClientOption
try {
response = await agent.prompt({ message: command.prompt, images: null }, BACKGROUND_CONTEXT);
} finally {
await watch.dispose();
await replica.close();
}
if (!response.accepted) throw new Error(response.error.message);
if (response.error !== null) throw new Error(response.error.message);
@ -108,7 +107,7 @@ export async function runClient(command: ClientCommand, options: RunClientOption
}
}
function messageText(message: Extract<WireAgentMessage, { role: "assistant" }>): string {
function messageText(message: AssistantMessage): string {
return message.content
.filter((content) => content.type === "text")
.map((content) => content.text)

View file

@ -1,579 +0,0 @@
import type {
AgentMessage,
AgentToolResult,
BashExecutionMessage,
BranchSummaryMessage,
CompactionSummaryMessage,
CustomMessage,
Entry,
HarnessEvent,
LaneSnapshot as HarnessLaneSnapshot,
} from "@earendil-works/pi-agent-core";
import type {
AssistantMessageFrame as AiAssistantMessageFrame,
AssistantMessage,
DeferredHandle,
ImageContent,
ToolResultMessage,
Usage,
UserMessage,
} from "@earendil-works/pi-ai";
import type {
AssistantMessageFrame,
JsonValue,
LaneEntry,
LaneEvent,
LaneSnapshot,
OperationResultRecord,
ToolOutput,
WireAgentMessage,
WireImageContent,
} from "@earendil-works/pi-protocol";
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
: false
: false;
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<WireImageContent, ImageContent>>,
Assert<SameKeys<WireUserMessage, UserMessage>>,
Assert<SameKeys<WireAssistantMessage, AssistantMessage>>,
Assert<SameKeys<WireToolResultMessage, ToolResultMessage>>,
Assert<SameKeys<WireBashExecutionMessage, BashExecutionMessage>>,
Assert<SameKeys<WireCustomMessage, CustomMessage>>,
Assert<SameKeys<WireBranchSummaryMessage, BranchSummaryMessage>>,
Assert<SameKeys<WireCompactionSummaryMessage, CompactionSummaryMessage>>,
];
function toWireUsage(usage: Usage): Usage {
return {
input: usage.input,
output: usage.output,
cacheRead: usage.cacheRead,
cacheWrite: usage.cacheWrite,
...(usage.cacheWrite1h === undefined ? {} : { cacheWrite1h: usage.cacheWrite1h }),
...(usage.reasoning === undefined ? {} : { reasoning: usage.reasoning }),
totalTokens: usage.totalTokens,
cost: {
input: usage.cost.input,
output: usage.cost.output,
cacheRead: usage.cost.cacheRead,
cacheWrite: usage.cost.cacheWrite,
total: usage.cost.total,
},
};
}
function toWireOperationResult(value: NonNullable<HarnessLaneSnapshot["lastResult"]>): OperationResultRecord {
const base = {
operationId: value.operationId,
kind: value.kind,
fromTipId: value.fromTipId,
tipId: value.tipId,
startedAt: value.startedAt,
endedAt: value.endedAt,
};
if (value.status !== "failed") return { ...base, status: value.status };
if (value.error === undefined) throw new TypeError("Failed harness result is missing its error");
return {
...base,
status: "failed",
error: {
code: value.error.code,
message: value.error.message,
...(value.error.details === undefined ? {} : { details: toWireJsonValue(value.error.details) }),
},
};
}
function toWireAssistantMessage(message: AssistantMessage): WireAssistantMessage {
return {
role: "assistant",
content: message.content.map((content) => {
switch (content.type) {
case "text":
return {
type: "text" as const,
text: content.text,
...(content.textSignature === undefined ? {} : { textSignature: content.textSignature }),
};
case "thinking":
return {
type: "thinking" as const,
thinking: content.thinking,
...(content.thinkingSignature === undefined ? {} : { thinkingSignature: content.thinkingSignature }),
...(content.redacted === undefined ? {} : { redacted: content.redacted }),
};
case "toolCall":
return {
type: "toolCall" as const,
id: content.id,
name: content.name,
arguments: Object.fromEntries(
Object.entries(content.arguments).map(([key, value]) => [key, toWireJsonValue(value)]),
),
...(content.thoughtSignature === undefined ? {} : { thoughtSignature: content.thoughtSignature }),
...(content.namespace === undefined ? {} : { namespace: content.namespace }),
};
default:
return assertNever(content);
}
}),
api: message.api,
provider: message.provider,
model: message.model,
...(message.responseModel === undefined ? {} : { responseModel: message.responseModel }),
...(message.responseId === undefined ? {} : { responseId: message.responseId }),
...(message.diagnostics === undefined
? {}
: {
diagnostics: message.diagnostics.map((diagnostic) => ({
type: diagnostic.type,
timestamp: diagnostic.timestamp,
...(diagnostic.error === undefined
? {}
: {
error: {
...(diagnostic.error.name === undefined ? {} : { name: diagnostic.error.name }),
message: diagnostic.error.message,
...(diagnostic.error.stack === undefined ? {} : { stack: diagnostic.error.stack }),
...(diagnostic.error.code === undefined ? {} : { code: diagnostic.error.code }),
},
}),
...(diagnostic.details === undefined
? {}
: {
details: Object.fromEntries(
Object.entries(diagnostic.details).map(([key, value]) => [key, toWireJsonValue(value)]),
),
}),
})),
}),
usage: toWireUsage(message.usage),
stopReason: message.stopReason,
...(message.deferred === undefined ? {} : { deferred: toWireDeferred(message.deferred) }),
...(message.errorMessage === undefined ? {} : { errorMessage: message.errorMessage }),
...(message.rawStopReason === undefined ? {} : { rawStopReason: message.rawStopReason }),
...(message.endTurn === undefined ? {} : { endTurn: message.endTurn }),
timestamp: message.timestamp,
};
}
function toWireDeferred(deferred: DeferredHandle): WireAssistantMessage["deferred"] & object {
return {
provider: deferred.provider,
modelId: deferred.modelId,
api: deferred.api,
id: deferred.id,
...(deferred.expiresAt === undefined ? {} : { expiresAt: deferred.expiresAt }),
...(deferred.pollAfterMs === undefined ? {} : { pollAfterMs: deferred.pollAfterMs }),
...(deferred.data === undefined ? {} : { data: toWireJsonValue(deferred.data) }),
};
}
function toWireJsonValue(value: unknown): JsonValue {
return value as JsonValue;
}
function toWireMessage(message: AgentMessage): WireAgentMessage {
switch (message.role) {
case "user":
return {
role: "user",
content:
typeof message.content === "string"
? message.content
: message.content.map((content) =>
content.type === "text"
? {
type: "text" as const,
text: content.text,
...(content.textSignature === undefined
? {}
: { textSignature: content.textSignature }),
}
: { type: "image" as const, data: content.data, mimeType: content.mimeType },
),
timestamp: message.timestamp,
};
case "assistant":
return toWireAssistantMessage(message);
case "toolResult":
return {
role: "toolResult",
toolCallId: message.toolCallId,
toolName: message.toolName,
content: message.content.map((content) =>
content.type === "text"
? {
type: "text" as const,
text: content.text,
...(content.textSignature === undefined ? {} : { textSignature: content.textSignature }),
}
: { type: "image" as const, data: content.data, mimeType: content.mimeType },
),
...(message.details === undefined ? {} : { details: toWireJsonValue(message.details) }),
...(message.usage === undefined ? {} : { usage: toWireUsage(message.usage) }),
...(message.addedToolNames === undefined ? {} : { addedToolNames: [...message.addedToolNames] }),
isError: message.isError,
timestamp: message.timestamp,
};
case "bashExecution":
return {
role: "bashExecution",
command: message.command,
output: message.output,
...(message.exitCode === undefined ? {} : { exitCode: message.exitCode }),
cancelled: message.cancelled,
truncated: message.truncated,
...(message.fullOutputPath === undefined ? {} : { fullOutputPath: message.fullOutputPath }),
timestamp: message.timestamp,
...(message.excludeFromContext === undefined ? {} : { excludeFromContext: message.excludeFromContext }),
};
case "custom":
return {
role: "custom",
customType: message.customType,
content:
typeof message.content === "string"
? message.content
: message.content.map((content) =>
content.type === "text"
? {
type: "text" as const,
text: content.text,
...(content.textSignature === undefined
? {}
: { textSignature: content.textSignature }),
}
: { type: "image" as const, data: content.data, mimeType: content.mimeType },
),
display: message.display,
...(message.details === undefined ? {} : { details: toWireJsonValue(message.details) }),
timestamp: message.timestamp,
};
case "branchSummary":
return {
role: "branchSummary",
summary: message.summary,
fromId: message.fromId,
timestamp: message.timestamp,
};
case "compactionSummary":
return {
role: "compactionSummary",
summary: message.summary,
tokensBefore: message.tokensBefore,
timestamp: message.timestamp,
};
default:
return assertNever(message);
}
}
function toWireEntry(entry: Entry): LaneEntry {
const base = { id: entry.id, parentId: entry.parentId, seq: entry.seq, timestamp: entry.timestamp };
switch (entry.type) {
case "message":
return {
...base,
type: "message",
message: toWireMessage(entry.message),
...(entry.terminate === undefined ? {} : { terminate: entry.terminate }),
};
case "compaction":
return {
...base,
type: "compaction",
summary: entry.summary,
retainedTail: entry.retainedTail.map(toWireMessage),
tokensBefore: entry.tokensBefore,
...(entry.details === undefined ? {} : { details: entry.details }),
...(entry.usage === undefined ? {} : { usage: toWireUsage(entry.usage) }),
fromHook: entry.fromHook,
};
case "branch_summary":
return {
...base,
type: "branch_summary",
fromId: entry.fromId,
summary: entry.summary,
...(entry.details === undefined ? {} : { details: entry.details }),
...(entry.usage === undefined ? {} : { usage: toWireUsage(entry.usage) }),
fromHook: entry.fromHook,
};
case "custom":
return {
...base,
type: "custom",
customType: entry.customType,
...(entry.data === undefined ? {} : { data: entry.data }),
};
default:
return assertNever(entry);
}
}
function toWireToolOutput(result: AgentToolResult<unknown>): ToolOutput {
return {
content: result.content.map((content) =>
content.type === "text"
? {
type: "text" as const,
text: content.text,
...(content.textSignature === undefined ? {} : { textSignature: content.textSignature }),
}
: { type: "image" as const, data: content.data, mimeType: content.mimeType },
),
...(result.details === undefined ? {} : { details: toWireJsonValue(result.details) }),
...(result.usage === undefined ? {} : { usage: toWireUsage(result.usage) }),
};
}
function toWireFrame(frame: AiAssistantMessageFrame): AssistantMessageFrame {
switch (frame.type) {
case "start":
return { type: "start", partial: toWireAssistantMessage(frame.partial) };
case "text_start":
return { type: "text_start", contentIndex: frame.contentIndex, content: { ...frame.content } };
case "text_delta":
case "thinking_delta":
case "toolcall_checkpoint":
case "toolcall_delta":
return { ...frame };
case "text_end":
return { ...frame };
case "thinking_start":
return { type: "thinking_start", contentIndex: frame.contentIndex, content: { ...frame.content } };
case "thinking_end":
return { ...frame };
case "toolcall_start":
return {
type: "toolcall_start",
contentIndex: frame.contentIndex,
toolCall: {
...frame.toolCall,
arguments: Object.fromEntries(
Object.entries(frame.toolCall.arguments).map(([key, value]) => [key, toWireJsonValue(value)]),
),
},
};
case "toolcall_end":
return {
...frame,
arguments: Object.fromEntries(
Object.entries(frame.arguments).map(([key, value]) => [key, toWireJsonValue(value)]),
),
};
default:
return assertNever(frame);
}
}
function toWireQueuedItem(item: HarnessLaneSnapshot["queues"][number]): LaneSnapshot["queues"][number] {
return item.type === "message"
? { entryId: item.entryId, kind: item.kind, type: "message", message: toWireMessage(item.message) }
: {
entryId: item.entryId,
kind: "write",
type: "custom",
customType: item.customType,
...(item.data === undefined ? {} : { data: item.data }),
};
}
export function toWireLaneSnapshot(snapshot: HarnessLaneSnapshot): LaneSnapshot {
return {
lane: snapshot.lane,
transcript: snapshot.transcript.map(toWireEntry),
tipId: snapshot.tipId,
...(snapshot.lastResult === undefined ? {} : { lastResult: toWireOperationResult(snapshot.lastResult) }),
configuration: {
model: { ...snapshot.configuration.model },
thinkingLevel: snapshot.configuration.thinkingLevel,
activeToolNames: [...snapshot.configuration.activeToolNames],
},
stats: { messageCount: snapshot.stats.messageCount, usage: toWireUsage(snapshot.stats.usage) },
operation:
snapshot.operation === null
? null
: {
id: snapshot.operation.id,
kind: snapshot.operation.kind,
status: snapshot.operation.status,
startedAt: snapshot.operation.startedAt,
fromTipId: snapshot.operation.fromTipId,
...(snapshot.operation.deferred === undefined
? {}
: {
deferred: {
handle: toWireDeferred(snapshot.operation.deferred.handle),
poll: snapshot.operation.deferred.poll,
},
}),
...(snapshot.operation.streamingMessage === undefined
? {}
: { streamingMessage: toWireAssistantMessage(snapshot.operation.streamingMessage) }),
runningTools: snapshot.operation.runningTools.map((tool) => ({
toolCallId: tool.toolCallId,
toolName: tool.toolName,
args: toWireJsonValue(tool.args),
...(tool.partialResult === undefined
? {}
: { partialResult: toWireToolOutput(tool.partialResult) }),
})),
...(snapshot.operation.retry === undefined ? {} : { retry: { ...snapshot.operation.retry } }),
},
queues: snapshot.queues.map(toWireQueuedItem),
faulted: snapshot.faulted,
};
}
function toWireOperationError(error: Extract<HarnessEvent, { type: "run_end" }>["error"]) {
if (error === undefined) return undefined;
return {
code: error.code,
message: error.message,
...(error.details === undefined ? {} : { details: toWireJsonValue(error.details) }),
};
}
export function toWireLaneEvent(event: HarnessEvent): LaneEvent | undefined {
const lane = "lane" in event && typeof event.lane === "string" ? event.lane : undefined;
const base = lane === undefined ? {} : { lane, ...(event.recovery === true ? { recovery: true as const } : {}) };
switch (event.type) {
case "run_start":
return { type: "run_start", runId: event.runId, startedAt: event.startedAt, ...base, lane: event.lane };
case "run_resume":
return { type: "run_resume", runId: event.runId, ...base, lane: event.lane };
case "run_suspend":
return {
type: "run_suspend",
runId: event.runId,
reason: "deferred",
deferred: toWireDeferred(event.deferred),
poll: event.poll,
...base,
lane: event.lane,
};
case "operation_abort":
return {
type: "operation_abort",
operationId: event.operationId,
steer: event.steer.map(toWireMessage),
followUp: event.followUp.map(toWireMessage),
...base,
lane: event.lane,
};
case "run_end":
return event.status === "failed"
? { ...event, error: toWireOperationError(event.error)!, ...base, lane: event.lane }
: { ...event, ...base, lane: event.lane };
case "message_start":
case "message_end":
return { ...event, message: toWireMessage(event.message), ...base, lane: event.lane };
case "message_update":
if (event.message.role !== "assistant") {
throw new TypeError("Harness message_update did not carry an assistant message");
}
return {
type: "message_update",
runId: event.runId,
message: toWireAssistantMessage(event.message),
...(event.frame === undefined ? {} : { frame: toWireFrame(event.frame) }),
...base,
lane: event.lane,
};
case "tool_start":
return {
...event,
args: toWireJsonValue(event.args),
...base,
lane: event.lane,
};
case "tool_update":
return { ...event, partialResult: toWireToolOutput(event.partialResult), ...base, lane: event.lane };
case "tool_end":
return { ...event, result: toWireToolOutput(event.result), ...base, lane: event.lane };
case "entry_added":
return { type: "entry_added", entry: toWireEntry(event.entry), ...base, lane: event.lane };
case "queue_update":
return { type: "queue_update", queues: event.queues.map(toWireQueuedItem), ...base, lane: event.lane };
case "retry_scheduled":
case "retry_start":
case "retry_end":
case "compaction_start":
case "navigation_start":
return { ...event, ...base, lane: event.lane };
case "compaction_end":
case "navigation_end":
return event.status === "failed"
? { ...event, error: toWireOperationError(event.error)!, ...base, lane: event.lane }
: { ...event, ...base, lane: event.lane };
case "usage":
return {
type: "usage",
lane: event.lane,
row: {
...event.row,
usage: toWireUsage(event.row.usage),
...(event.row.details === undefined ? {} : { details: toWireJsonValue(event.row.details) }),
},
totals: toWireUsage(event.totals),
};
case "config_update":
switch (event.property) {
case "tools":
case "resources":
return { type: "config_update", property: event.property };
case "model":
case "thinkingLevel":
case "activeTools":
return {
type: "config_update",
property: event.property,
value: toWireJsonValue(event.value),
previous: toWireJsonValue(event.previous),
...base,
lane: event.lane,
};
case "streamOptions":
case "retryPolicy":
case "compactionSettings":
case "steeringMode":
case "followUpMode":
return {
type: "config_update",
property: event.property,
value: toWireJsonValue(event.value),
previous: toWireJsonValue(event.previous),
};
default:
return assertNever(event);
}
case "fault":
return { type: "fault", code: event.code, message: event.message };
case "handler_error":
case "turn_start":
case "turn_end":
case "value_update":
case "lane_created":
return undefined;
default:
return assertNever(event);
}
}
function assertNever(value: never): never {
throw new TypeError(`Unsupported Harness value: ${String(value)}`);
}

View file

@ -1,10 +1,6 @@
import { type HarnessEvent, type LaneSnapshot, reduceLaneSnapshot } from "@earendil-works/pi-agent-core";
import type { LaneWatch } from "@earendil-works/pi-client";
import type { LaneEvent as WireLaneEvent, LaneSnapshot as WireLaneSnapshot } from "@earendil-works/pi-protocol";
export interface LaneWatchSource {
watchSession(sessionId: string): Promise<LaneWatch>;
}
import { BACKGROUND_CONTEXT } from "@earendil-works/chord/context";
import { type LaneSnapshot, type LaneWatchEvent, reduceLaneSnapshot } from "@earendil-works/pi-agent-core";
import type { Transcript, TranscriptSnapshot, TranscriptUpdate } from "./services/transcript.ts";
export interface LaneReplica {
readonly sessionId: string;
@ -13,33 +9,80 @@ export interface LaneReplica {
close(): Promise<void>;
}
/**
* Open the protocol's coherent snapshot/event watch as a presentation-local Harness replica.
*
* The protocol DTOs are the strict-JSON projection of these Harness types. Keeping the cast here
* lets every presentation use the Harness's normative reducer instead of maintaining a second fold.
*/
export async function openLaneReplica(source: LaneWatchSource, sessionId: string): Promise<LaneReplica> {
const watch = await source.watchSession(sessionId);
/** Open a presentation-local Harness replica from the Session's Transcript service. */
export async function openLaneReplica(
transcript: Transcript,
sessionId: string,
onEvent?: (event: LaneWatchEvent) => void | Promise<void>,
): Promise<LaneReplica> {
const listeners = new Set<() => void>();
let snapshot = fromWireSnapshot(watch.snapshot);
const pending: TranscriptUpdate[] = [];
let snapshot: LaneSnapshot | undefined;
let revision = 0;
let initialized = false;
let closed = false;
let deliveryTail = Promise.resolve();
const publish = (): void => {
if (closed) return;
for (const listener of listeners) listener();
};
await watch.start(async (wireEvent) => {
if (closed) return;
const folded = reduceLaneSnapshot(snapshot, fromWireEvent(wireEvent));
if ("rebase" in folded) snapshot = fromWireSnapshot(await watch.resnapshot());
else snapshot = folded;
const installSnapshot = (captured: TranscriptSnapshot): void => {
revision = captured.revision;
snapshot = captured.snapshot;
publish();
};
const apply = async (update: TranscriptUpdate): Promise<void> => {
if (closed || update.revision <= revision) return;
if (update.revision !== revision + 1) {
installSnapshot(await transcript.snapshot(BACKGROUND_CONTEXT));
if (update.revision <= revision) return;
if (update.revision !== revision + 1) throw new Error("Transcript snapshot did not repair its revision gap");
}
if (update.type === "snapshot") {
revision = update.revision;
snapshot = update.snapshot;
publish();
return;
}
const event = update.event;
const current = snapshot;
if (current === undefined) throw new Error("Transcript replica has no snapshot");
const reduced = reduceLaneSnapshot(current, event);
revision = update.revision;
if (!("rebase" in reduced)) {
snapshot = reduced;
publish();
}
await onEvent?.(event);
};
const enqueue = (update: TranscriptUpdate): void => {
const detached = structuredClone(update);
deliveryTail = deliveryTail.then(() => apply(detached));
};
const unsubscribe = transcript.updates.subscribe((update) => {
if (closed || update === null) return;
if (initialized) enqueue(update);
else pending.push(structuredClone(update));
});
try {
installSnapshot(await transcript.snapshot(BACKGROUND_CONTEXT));
initialized = true;
for (const update of pending.splice(0)) enqueue(update);
await deliveryTail;
} catch (error) {
closed = true;
unsubscribe();
throw error;
}
return {
sessionId,
state: () => snapshot,
state() {
if (snapshot === undefined) throw new Error("Transcript replica has no snapshot");
return snapshot;
},
subscribe(listener) {
listeners.add(listener);
return () => listeners.delete(listener);
@ -47,16 +90,9 @@ export async function openLaneReplica(source: LaneWatchSource, sessionId: string
async close() {
if (closed) return;
closed = true;
unsubscribe();
listeners.clear();
await watch.dispose();
await deliveryTail;
},
};
}
function fromWireSnapshot(snapshot: WireLaneSnapshot): LaneSnapshot {
return snapshot as unknown as LaneSnapshot;
}
function fromWireEvent(event: WireLaneEvent): HarnessEvent {
return event as unknown as HarnessEvent;
}

View file

@ -10,12 +10,13 @@ Facet setup generates each host's RPC service catalogue from its provided non-lo
| session | `SessionPlugins` | reloads the configured Session facet generation | add coordinated multi-worker reload reporting |
| session | `Models` | state, default-persisting selection, thinking, refresh implemented | move provider/auth composition behind plugin facets |
| session | `AgentController` | presentation-safe `AgentLane` facade for prompting, queueing, abort, resume, compaction, and navigation | add new lane operations only when a presentation requires them |
| session | `Transcript` | coherent main-lane snapshots and ordered reducer events | move to a dedicated Chord feed primitive if other high-frequency observations need the same semantics |
| 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. 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.
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 the `Transcript` service's coherent replica with the Harness reducer while controller calls are pending.
A foreground server uses repeatable `-e` options to establish its default Session and TUI facets. A local client may instead select packages for the Session it creates or resumes; that branch selection is persisted with the Session and does not alter other workers or the server default. Before attaching, the server asks Chord to build conventional `src/session.ts` and `src/tui.ts` entries into separate `plugin-builds/` directories, passes the generated manifest paths to that Session worker, and returns the matching TUI artifacts. The Session worker loads built-in facets and separately owned plugin generations, then creates one active `FacetHost`. `/reload` atomically rebuilds the packages, loads fresh plugin candidates, cuts them over through `FacetHost.reload()`, and disposes the retired generations. Host-created implementation dependencies such as `AgentLane`, `ModelRuntime`, and `SettingsManager` are passed directly to built-in facet factories; they are not exposed as services. The TUI loads configured presentation facets and adds one private bridge facet that consumes the connected server and selected-Session services for `ExperimentalClientTui`. Its local presentation services are `SlashCommands`, which owns command contributions, and `PresentationUI`, which exposes selection and status rendering without exposing the raw TUI. Command callbacks receive Chord `Context` directly. Facets consume `AgentController` explicitly for prompting, steering, or queueing and return structured controller results to the command dispatcher.

View file

@ -1,7 +1,6 @@
import {
type Context,
createRemoteServiceBinding,
type JsonValue,
type MutableReplicatedState,
type RemoteServiceBinding,
type RemoteServiceSource,
@ -367,7 +366,7 @@ function createRemoteServiceTransport(client: Client, getTarget: () => RpcTarget
member: call.member,
args: [...call.args],
};
return client.request(target, wireCall, context.abortSignal) as Promise<JsonValue | undefined>;
return client.request(target, wireCall, context.abortSignal);
},
subscribe: async (serviceId, mode, listener, context) => {
const target = getTarget();

View file

@ -0,0 +1,143 @@
import { cloneJsonValue, defineFacet, type Facet, type MutableReplicatedState } from "@earendil-works/chord";
import { BACKGROUND_CONTEXT } from "@earendil-works/chord/context";
import {
type AgentLane,
type HarnessEvent,
type LaneSnapshot,
type LaneWatchEvent,
reduceLaneSnapshot,
type WatchHandle,
} from "@earendil-works/pi-agent-core";
import { Transcript, type Transcript as TranscriptService, type TranscriptUpdate } from "./transcript.ts";
type PendingTranscriptUpdate =
| Omit<Extract<TranscriptUpdate, { type: "event" }>, "revision">
| Omit<Extract<TranscriptUpdate, { type: "snapshot" }>, "revision">;
interface TranscriptRuntime {
readonly service: TranscriptService;
activate(): Promise<void>;
dispose(): Promise<void>;
}
export function createTranscriptService(
lane: AgentLane,
createState: (initial: TranscriptUpdate | null) => MutableReplicatedState<TranscriptUpdate | null>,
): TranscriptRuntime {
const updates = createState(null);
let watch: WatchHandle<LaneSnapshot> | undefined;
let snapshot: LaneSnapshot | undefined;
let revision = 0;
let rebase: Promise<void> | undefined;
let rebaseError: Error | undefined;
const publish = (update: PendingTranscriptUpdate, context: Parameters<typeof updates.set>[1]): void => {
revision += 1;
updates.set(
update.type === "event"
? { type: "event", revision, event: update.event }
: { type: "snapshot", revision, snapshot: update.snapshot },
context,
);
};
const scheduleRebase = (context: Parameters<typeof updates.set>[1]): void => {
if (rebase !== undefined) return;
const activeWatch = watch;
if (activeWatch === undefined) return;
const pending = (async () => {
const refreshed = await activeWatch.resnapshot(context);
snapshot = refreshed;
publish({ type: "snapshot", snapshot: cloneJsonValue(refreshed) }, context);
})();
rebase = pending;
void pending.then(
() => {
if (rebase === pending) rebase = undefined;
},
(error: unknown) => {
rebaseError = error instanceof Error ? error : new Error(String(error));
if (rebase === pending) rebase = undefined;
},
);
};
const onEvent = (event: HarnessEvent, context: Parameters<typeof updates.set>[1]): void => {
const forwarded = toLaneWatchEvent(event);
if (forwarded === undefined) return;
const current = snapshot;
if (current === undefined) throw new Error("Transcript service is not active");
const reduced = reduceLaneSnapshot(current, event);
if ("rebase" in reduced) scheduleRebase(context);
else snapshot = reduced;
publish({ type: "event", event: forwarded }, context);
};
return {
service: {
updates,
async snapshot(_context) {
await rebase;
if (rebaseError !== undefined) throw rebaseError;
if (snapshot === undefined) throw new Error("Transcript service is not active");
return { revision, snapshot: cloneJsonValue(snapshot) };
},
},
async activate() {
if (watch !== undefined) throw new Error("Transcript service is already active");
const opened = await lane.watch(BACKGROUND_CONTEXT);
watch = opened;
snapshot = opened.snapshot;
opened.start(onEvent);
},
async dispose() {
let failure: unknown;
try {
await rebase;
} catch (error) {
failure = error;
}
watch?.unsubscribe();
watch = undefined;
snapshot = undefined;
if (failure !== undefined) throw failure;
},
};
}
export function createTranscriptServiceFacet(lane: AgentLane): Facet {
return defineFacet({
id: "@pi/transcript",
setup(env) {
const runtime = createTranscriptService(lane, env.replicatedState);
env.provide(Transcript, runtime.service);
env.onActivate(() => runtime.activate());
env.own(() => runtime.dispose());
},
});
}
function toLaneWatchEvent(event: HarnessEvent): LaneWatchEvent | undefined {
switch (event.type) {
case "handler_error":
case "turn_start":
case "turn_end":
case "value_update":
case "lane_created":
return undefined;
case "config_update":
if (event.property !== "model" && event.property !== "thinkingLevel" && event.property !== "activeTools") {
return undefined;
}
return cloneJsonValue(event);
case "message_update": {
if (event.message.role !== "assistant") {
throw new TypeError("Harness message_update did not carry an assistant message");
}
const { event: _providerEvent, ...update } = event;
return cloneJsonValue(update);
}
default:
return cloneJsonValue(event);
}
}

View file

@ -0,0 +1,19 @@
import { type Context, defineService, type ReplicatedState } from "@earendil-works/chord";
import type { LaneTranscriptSnapshot, LaneWatchEvent } from "@earendil-works/pi-agent-core";
export type TranscriptUpdate =
| { readonly type: "event"; readonly revision: number; readonly event: LaneWatchEvent }
| { readonly type: "snapshot"; readonly revision: number; readonly snapshot: LaneTranscriptSnapshot };
export interface TranscriptSnapshot {
readonly revision: number;
readonly snapshot: LaneTranscriptSnapshot;
}
/** Coherent main-lane observation published as an ordinary Chord service. */
export interface Transcript {
readonly updates: ReplicatedState<TranscriptUpdate | null>;
snapshot(context: Context): Promise<TranscriptSnapshot>;
}
export const Transcript = defineService<Transcript>("pi.transcript");

View file

@ -1,6 +1,7 @@
import {
type Context,
type ServiceProviderUpdate as CoreServiceProviderUpdate,
cloneJsonValue,
createFacetHost,
createStaticFacetLoader,
defineFacet,
@ -26,6 +27,7 @@ import { AgentController } from "./agent-controller.ts";
import { createAgentController } from "./agent-controller-provider.ts";
import { createModelsServiceFacet } from "./models-provider.ts";
import { SessionPlugins } from "./plugins.ts";
import { createTranscriptServiceFacet } from "./transcript-provider.ts";
export const ServiceOperationResultSchema = Type.Object(
{ result: Type.Optional(JsonValueSchema) },
@ -83,6 +85,7 @@ export async function createSessionWorkerServices(options: {
agentControllerRuntimeFacet,
pluginRuntimeFacet,
createModelsServiceFacet(options),
createTranscriptServiceFacet(options.lane),
]).load();
const pluginLoader = options.facetLoader ?? createStaticFacetLoader([]);
let loadedPlugins = await pluginLoader.load();
@ -133,7 +136,7 @@ export async function createSessionWorkerServices(options: {
catalogue: provider.catalogue,
async invoke(call, scope, context) {
const controlCall = decodeServiceControlCall(call);
if (controlCall?.type === "catalogue") return toProtocolJson(provider.catalogue);
if (controlCall?.type === "catalogue") return cloneJsonValue(provider.catalogue);
if (controlCall?.type === "subscribe") {
const key = scopedSubscriptionKey(scope, controlCall.subscriptionId);
if (subscriptions.has(key)) throw new Error("Service subscription ID is already active");
@ -142,7 +145,7 @@ export async function createSessionWorkerServices(options: {
});
subscriptions.set(key, { scope, subscription });
subscription.activate();
return toProtocolJson(subscription.snapshot);
return cloneJsonValue(subscription.snapshot);
}
if (controlCall?.type === "unsubscribe") {
const key = scopedSubscriptionKey(scope, controlCall.subscriptionId);
@ -177,10 +180,6 @@ function scopedSubscriptionKey(scope: WorkerServiceScope, subscriptionId: string
return `${scope.serverConnectionId}\0${scope.attachmentId}\0${subscriptionId}`;
}
function toProtocolJson(value: unknown): JsonValue {
return value as JsonValue;
}
function toProtocolServiceUpdate(update: CoreServiceProviderUpdate): ProtocolServiceProviderUpdate {
const candidate: unknown = update;
if (!Check(ServiceProviderUpdateSchema, candidate)) throw new Error("Service produced an invalid update");

View file

@ -8,20 +8,13 @@ import {
TODO_CONTEXT,
} from "@earendil-works/pi-agent-core";
import {
createRpcClient,
createServiceUnsubscribeCall,
decodeServiceControlCall,
type LaneEvent,
type ProtocolRpcCall,
type ProtocolRpcResult,
type ServiceProviderUpdate,
} from "@earendil-works/pi-protocol";
import {
type RoutedSessionAttachment,
type RoutedSessionHandle,
type RoutedSessionWatch,
ServerError,
} from "@earendil-works/pi-server";
import { type RoutedSessionAttachment, type RoutedSessionHandle, ServerError } from "@earendil-works/pi-server";
import { Check } from "typebox/value";
import type { CoordinatorConnection, CoordinatorConnectionEvent } from "./coordinator.ts";
import { spawnInternalProcess } from "./process.ts";
@ -33,8 +26,6 @@ import {
SESSION_WORKER_SESSION_KEY_ENV,
type SessionWorkerEvent,
SessionWorkerEventSchema,
type SessionWorkerOperationCall,
SessionWorkerOperations,
type SessionWorkerOptions,
type WorkerOperationResponse,
type WorkerOperationScope,
@ -76,18 +67,11 @@ interface PendingDemand {
reject(error: Error): void;
}
interface WorkerLaneWatch {
readonly worker: WorkerRecord;
readonly scope: WorkerOperationScope;
listener?: (event: LaneEvent, context: Context) => void | Promise<void>;
deliveryTail: Promise<void>;
}
interface PendingWorkerOperation {
readonly worker: WorkerRecord;
readonly scope: WorkerOperationScope;
cleanup(): void;
resolve(result: unknown): void;
resolve(result: ServiceOperationResult): void;
reject(error: Error): void;
}
@ -124,7 +108,6 @@ export class SessionWorkerManager {
readonly #pending = new Map<string, PendingLaunch>();
readonly #pendingDemand = new Map<string, PendingDemand>();
readonly #pendingOperations = new Map<string, PendingWorkerOperation>();
readonly #laneWatches = new Map<string, WorkerLaneWatch>();
readonly #serviceSubscriptions = new Map<string, WorkerServiceSubscription>();
readonly #removeListener: () => void;
readonly #onWorkerCountChanged: ((count: number) => void) | undefined;
@ -219,14 +202,6 @@ export class SessionWorkerManager {
};
}
#operations(
worker: WorkerRecord,
scope: WorkerOperationScope,
context: Context,
): ReturnType<typeof createRpcClient<typeof SessionWorkerOperations>> {
return createRpcClient(SessionWorkerOperations, (call) => this.#invoke(worker, scope, call, context));
}
async #attachClient(worker: WorkerRecord, context: Context): Promise<RoutedSessionAttachment> {
if (this.#detached || this.#shuttingDown || worker.stopping) {
throw new Error("Experimental Session worker is stopping");
@ -247,7 +222,6 @@ export class SessionWorkerManager {
return {
invokeService: (call, publish, serviceContext) =>
this.#invokeService(worker, scope, call, publish, serviceContext),
watch: (watchContext) => this.#createLaneWatch(worker, scope, watchContext),
release: async (releaseContext) => {
if (released) return;
released = true;
@ -286,13 +260,16 @@ export class SessionWorkerManager {
}
let response: ServiceOperationResult;
try {
response = await this.#operations(worker, scope, context).service(call);
response = await this.#invoke(worker, scope, call, context);
} catch (error) {
if (addedSubscriptionKey !== undefined && control?.type === "subscribe") {
this.#serviceSubscriptions.delete(addedSubscriptionKey);
void this.#operations(worker, scope, BACKGROUND_CONTEXT)
.service(createServiceUnsubscribeCall(control.subscriptionId))
.catch(() => {});
void this.#invoke(
worker,
scope,
createServiceUnsubscribeCall(control.subscriptionId),
BACKGROUND_CONTEXT,
).catch(() => {});
}
throw error;
}
@ -302,57 +279,6 @@ export class SessionWorkerManager {
return response.result;
}
async #createLaneWatch(
worker: WorkerRecord,
scope: WorkerOperationScope,
context: Context,
): Promise<RoutedSessionWatch> {
const { watchId, snapshot } = await this.#operations(worker, scope, context).watch();
if (this.#laneWatches.has(watchId)) {
await this.#operations(worker, scope, context)
.stopWatch(watchId)
.catch(() => {});
throw new Error("Session worker reused an active lane watch ID");
}
const watch: WorkerLaneWatch = { worker, scope, deliveryTail: Promise.resolve() };
this.#laneWatches.set(watchId, watch);
let state: "ready" | "started" | "unsubscribed" = "ready";
return {
snapshot,
start: async (listener, startContext) => {
if (state !== "ready" || this.#laneWatches.get(watchId) !== watch) {
throw new Error("Session worker lane watch may be started only once");
}
state = "started";
watch.listener = listener;
try {
await this.#operations(worker, scope, startContext).startWatch(watchId);
} catch (error) {
this.#laneWatches.delete(watchId);
delete watch.listener;
state = "unsubscribed";
await this.#operations(worker, scope, startContext)
.stopWatch(watchId)
.catch(() => {});
throw error;
}
},
resnapshot: async (resnapshotContext) => {
if (state === "unsubscribed" || this.#laneWatches.get(watchId) !== watch) {
throw new Error("Session worker lane watch is unsubscribed");
}
return (await this.#operations(worker, scope, resnapshotContext).resnapshotWatch(watchId)).snapshot;
},
unsubscribe: async (unsubscribeContext) => {
if (state === "unsubscribed") return;
state = "unsubscribed";
if (this.#laneWatches.get(watchId) === watch) this.#laneWatches.delete(watchId);
delete watch.listener;
await this.#operations(worker, scope, unsubscribeContext).stopWatch(watchId);
},
};
}
async #applyDemand(
worker: WorkerRecord,
attachmentId: string,
@ -395,9 +321,9 @@ export class SessionWorkerManager {
#invoke(
worker: WorkerRecord,
scope: WorkerOperationScope,
call: SessionWorkerOperationCall,
call: ProtocolRpcCall,
context: Context,
): Promise<unknown> {
): Promise<ServiceOperationResult> {
try {
this.#operationScope(worker, scope.attachmentId);
} catch (error) {
@ -405,9 +331,9 @@ export class SessionWorkerManager {
}
if (context.abortSignal?.aborted) return Promise.reject(abortError(context.abortSignal));
const requestId = randomUUID();
let resolve!: (result: unknown) => void;
let resolve!: (result: ServiceOperationResult) => void;
let reject!: (error: Error) => void;
const result = new Promise<unknown>((resolvePromise, rejectPromise) => {
const result = new Promise<ServiceOperationResult>((resolvePromise, rejectPromise) => {
resolve = resolvePromise;
reject = rejectPromise;
});
@ -640,17 +566,6 @@ export class SessionWorkerManager {
this.#handleOperationResponse(event.from, message.token, message.sessionKey, message.response);
return;
}
if (message.type === "lane_event") {
this.#handleLaneEvent(
event.from,
message.token,
message.sessionKey,
message.scope,
message.watchId,
message.event,
);
return;
}
if (message.type === "service_update") {
this.#handleServiceEvent(
event.from,
@ -699,30 +614,6 @@ export class SessionWorkerManager {
}
}
#handleLaneEvent(
peerId: string,
token: string,
sessionKey: string,
scope: WorkerOperationScope,
watchId: string,
event: LaneEvent,
): void {
const watch = this.#laneWatches.get(watchId);
const listener = watch?.listener;
if (
watch === undefined ||
listener === undefined ||
watch.worker.peerId !== peerId ||
watch.worker.token !== token ||
watch.worker.metadata.path !== sessionKey ||
watch.scope.serverConnectionId !== scope.serverConnectionId ||
watch.scope.attachmentId !== scope.attachmentId
) {
return;
}
watch.deliveryTail = watch.deliveryTail.then(() => listener(event, TODO_CONTEXT)).catch(() => {});
}
#handleServiceEvent(
peerId: string,
token: string,
@ -883,9 +774,6 @@ export class SessionWorkerManager {
#removeWorker(worker: WorkerRecord, error: Error | undefined): void {
if (this.#workersByPeer.get(worker.peerId) !== worker) return;
this.#rejectWorkerOperations(worker, new Error("Session worker disconnected during an operation"));
for (const [watchId, watch] of this.#laneWatches) {
if (watch.worker === worker) this.#laneWatches.delete(watchId);
}
this.#removeServiceSubscriptions((entry) => entry.worker === worker);
for (const pending of [...this.#pendingDemand.values()]) {
if (pending.worker === worker) {
@ -929,7 +817,6 @@ export class SessionWorkerManager {
pending.resolve();
}
this.#pending.clear();
this.#laneWatches.clear();
this.#serviceSubscriptions.clear();
this.#workersByPeer.clear();
this.#workersBySession.clear();

View file

@ -1,4 +1,3 @@
import { randomUUID } from "node:crypto";
import { createConnection, type Socket } from "node:net";
import { isAbsolute } from "node:path";
import { RemoteServiceError } from "@earendil-works/chord";
@ -7,7 +6,6 @@ import {
type AgentHarness as AgentHarnessInstance,
type AgentLane,
BACKGROUND_CONTEXT,
type Context,
createBashTool,
createReadTool,
createWriteTool,
@ -19,15 +17,7 @@ import {
} from "@earendil-works/pi-agent-core";
import { NodeExecutionEnv } from "@earendil-works/pi-agent-core/node";
import {
createRpcCallSchema,
createRpcDispatcher,
createRpcResultSchema,
defineRpc,
LaneEventSchema,
LaneSnapshotSchema,
ProtocolRpcCallSchema,
type RpcCall,
type RpcResultUnion,
type ServiceErrorCode,
ServiceErrorCodeSchema,
ServiceProviderUpdateSchema,
@ -39,7 +29,6 @@ import { findInitialModel, resolveCliModel } from "../core/model-resolver.ts";
import { ModelRuntime } from "../core/model-runtime.ts";
import { SettingsManager } from "../core/settings-manager.ts";
import { COORDINATOR_PROTOCOL_VERSION } from "./coordinator.ts";
import { toWireLaneEvent, toWireLaneSnapshot } from "./harness-wire-adapter.ts";
import { createSessionPluginFacetLoader } from "./plugins/bundled.ts";
import {
consumeInternalProcessRole,
@ -49,6 +38,7 @@ import {
} from "./process.ts";
import {
createSessionWorkerServices,
type ServiceOperationResult,
ServiceOperationResultSchema,
type SessionWorkerRuntime,
type SessionWorkerServices,
@ -84,38 +74,6 @@ export const SessionWorkerOptionsSchema = StrictObject({
});
export type SessionWorkerOptions = Static<typeof SessionWorkerOptionsSchema>;
export const SessionWorkerOperations = defineRpc({
watch: {
args: Type.Tuple([]),
result: StrictObject({ watchId: Type.String({ minLength: 1 }), snapshot: LaneSnapshotSchema }),
},
startWatch: {
args: Type.Tuple([Type.String({ minLength: 1 })]),
result: StrictObject({ watchId: Type.String({ minLength: 1 }) }),
},
resnapshotWatch: {
args: Type.Tuple([Type.String({ minLength: 1 })]),
result: StrictObject({ watchId: Type.String({ minLength: 1 }), snapshot: LaneSnapshotSchema }),
},
stopWatch: {
args: Type.Tuple([Type.String({ minLength: 1 })]),
result: StrictObject({ watchId: Type.String({ minLength: 1 }) }),
},
service: {
args: Type.Tuple([ProtocolRpcCallSchema]),
result: ServiceOperationResultSchema,
},
});
export type SessionWorkerOperationCall = RpcCall<typeof SessionWorkerOperations>;
type SessionWorkerOperationResult = RpcResultUnion<typeof SessionWorkerOperations>;
const SessionWorkerOperationCallSchema = Type.Unsafe<SessionWorkerOperationCall>(
createRpcCallSchema(SessionWorkerOperations),
);
const SessionWorkerOperationResultSchema = Type.Unsafe<SessionWorkerOperationResult>(
createRpcResultSchema(SessionWorkerOperations),
);
export const WorkerOperationScopeSchema = StrictObject({
serverConnectionId: Type.String(),
attachmentId: Type.String(),
@ -126,7 +84,7 @@ export const WorkerOperationRequestSchema = StrictObject({
type: Type.Literal("operation"),
requestId: Type.String({ minLength: 1 }),
scope: WorkerOperationScopeSchema,
call: SessionWorkerOperationCallSchema,
call: ProtocolRpcCallSchema,
});
export type WorkerOperationRequest = Static<typeof WorkerOperationRequestSchema>;
@ -135,7 +93,7 @@ export const WorkerOperationResponseSchema = Type.Union([
type: Type.Literal("operation_result"),
requestId: Type.String({ minLength: 1 }),
scope: WorkerOperationScopeSchema,
result: SessionWorkerOperationResultSchema,
result: ServiceOperationResultSchema,
}),
StrictObject({
type: Type.Literal("operation_error"),
@ -203,14 +161,6 @@ export const SessionWorkerEventSchema = Type.Union([
sessionKey: Type.String(),
response: WorkerOperationResponseSchema,
}),
Type.Object({
type: Type.Literal("lane_event"),
token: Type.String(),
sessionKey: Type.String(),
scope: WorkerOperationScopeSchema,
watchId: Type.String({ minLength: 1 }),
event: LaneEventSchema,
}),
Type.Object({
type: Type.Literal("service_update"),
token: Type.String(),
@ -602,28 +552,16 @@ async function run(options: SessionWorkerOptions, createHarness: CreateSessionWo
throw error;
}
const laneWatches = new Map<
string,
{ readonly scope: WorkerOperationScope; readonly handle: Awaited<ReturnType<AgentLane["watch"]>> }
>();
const activeRequests = new Map<
string,
{ readonly scope: WorkerOperationScope; readonly cancel: (reason?: unknown) => void }
>();
const removeLaneWatches = (matches: (scope: WorkerOperationScope) => boolean): void => {
for (const [watchId, watch] of laneWatches) {
if (!matches(watch.scope)) continue;
watch.handle.unsubscribe();
laneWatches.delete(watchId);
}
};
let lifecycle: WorkerLifecycle | undefined;
let removeLifecycleListeners: (() => void)[] = [];
let closing: Promise<void> | undefined;
const close = (): Promise<void> => {
if (closing) return closing;
lifecycle?.close();
removeLaneWatches(() => true);
services.removeSubscriptions(() => true);
for (const request of activeRequests.values()) request.cancel(new Error("Session worker is closing"));
activeRequests.clear();
@ -668,54 +606,14 @@ async function run(options: SessionWorkerOptions, createHarness: CreateSessionWo
harness.events.on("fault", closeAndExit),
];
interface WorkerOperationContext {
readonly scope: WorkerOperationScope;
readonly context: Context;
}
const dispatchWorkerOperation = createRpcDispatcher(SessionWorkerOperations, {
watch: async ({ scope, context }: WorkerOperationContext, ..._args: never[]) => {
const handle = await lane.watch(context);
const watchId = randomUUID();
laneWatches.set(watchId, { scope, handle });
return { watchId, snapshot: toWireLaneSnapshot(handle.snapshot) };
},
startWatch: async ({ scope }: WorkerOperationContext, watchId) => {
const watch = laneWatches.get(watchId);
if (!watch || !sameScope(watch.scope, scope)) throw new Error("Session worker lane watch was not found");
watch.handle.start(async (event) => {
const wireEvent = toWireLaneEvent(event);
if (wireEvent === undefined) return;
await control.send({ type: "lane_event", token, sessionKey, scope, watchId, event: wireEvent });
});
return { watchId };
},
resnapshotWatch: async ({ scope, context }: WorkerOperationContext, watchId) => {
const watch = laneWatches.get(watchId);
if (!watch || !sameScope(watch.scope, scope)) throw new Error("Session worker lane watch was not found");
return { watchId, snapshot: toWireLaneSnapshot(await watch.handle.resnapshot(context)) };
},
stopWatch: async ({ scope }: WorkerOperationContext, watchId) => {
const watch = laneWatches.get(watchId);
if (!watch || !sameScope(watch.scope, scope)) throw new Error("Session worker lane watch was not found");
watch.handle.unsubscribe();
laneWatches.delete(watchId);
return { watchId };
},
service: async ({ scope, context }: WorkerOperationContext, call) => {
const result = await services.invoke(call, scope, context);
return result === undefined ? {} : { result };
},
});
const handleOperation = async (request: WorkerOperationRequest): Promise<void> => {
let releaseRequest = (): void => {};
const cancellable = withCancel(BACKGROUND_CONTEXT);
try {
releaseRequest = lifecycle!.beginRequest(request.scope.serverConnectionId, request.scope.attachmentId);
activeRequests.set(request.requestId, { scope: request.scope, cancel: cancellable.cancel });
const result = await dispatchWorkerOperation(request.call, {
scope: request.scope,
context: cancellable.context,
});
const serviceResult = await services.invoke(request.call, request.scope, cancellable.context);
const result: ServiceOperationResult = serviceResult === undefined ? {} : { result: serviceResult };
await control.send({
type: "operation_response",
token,
@ -776,7 +674,6 @@ async function run(options: SessionWorkerOptions, createHarness: CreateSessionWo
const matches = (scope: WorkerOperationScope): boolean =>
scope.serverConnectionId === command.serverConnectionId &&
scope.attachmentId === command.attachmentId;
removeLaneWatches(matches);
services.removeSubscriptions(matches);
}
lifecycle?.setDemand(command.serverConnectionId, command.attachmentId, command.attached);
@ -814,7 +711,6 @@ async function run(options: SessionWorkerOptions, createHarness: CreateSessionWo
onServerConnected: (serverConnectionId) => lifecycle?.serverConnected(serverConnectionId),
onServerDisconnected: (serverConnectionId) => {
const matches = (scope: WorkerOperationScope): boolean => scope.serverConnectionId === serverConnectionId;
removeLaneWatches(matches);
services.removeSubscriptions(matches);
for (const request of activeRequests.values()) {
if (matches(request.scope)) request.cancel(new Error("Server disconnected"));

View file

@ -1,6 +1,7 @@
import { createHash } from "node:crypto";
import { resolve } from "node:path";
import {
cloneJsonValue,
createRemoteServiceBinding,
RemoteServiceProvider,
type RemoteServiceTransport,
@ -12,8 +13,12 @@ import {
FACET_BUNDLE_ARTIFACT_FORMAT_VERSION,
type FacetBundleArtifact,
} from "@earendil-works/chord/node";
import type { AgentLane } from "@earendil-works/pi-agent-core";
import type { LaneEvent, LaneSnapshot } from "@earendil-works/pi-protocol";
import {
type AgentLane,
type LaneSnapshot,
type LaneWatchEvent,
reduceLaneSnapshot,
} from "@earendil-works/pi-agent-core";
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";
@ -34,6 +39,7 @@ import {
SessionManagement,
type SessionSummary,
} from "../src/experimental/services/sessions.ts";
import { Transcript, type TranscriptUpdate } from "../src/experimental/services/transcript.ts";
import { initTheme } from "../src/modes/interactive/theme/theme.ts";
const serverId = "00000000-0000-4000-8000-000000000001";
@ -127,22 +133,21 @@ describe("experimental client TUI", () => {
BACKGROUND_CONTEXT,
);
});
let watchListener: ((event: LaneEvent) => void | Promise<void>) | undefined;
const snapshot = laneSnapshot();
const watchSession = vi.fn(async (sessionId: string) => ({
id: "watch-1",
sessionId,
snapshot,
async start(listener: (event: LaneEvent) => void | Promise<void>) {
watchListener = listener;
},
async resnapshot() {
return snapshot;
},
async dispose() {},
let snapshot = laneSnapshot();
let transcriptRevision = 0;
const transcriptUpdates = replicatedState<TranscriptUpdate | null>(null);
const transcriptSnapshot = vi.fn(async () => ({
revision: transcriptRevision,
snapshot: cloneJsonValue(snapshot),
}));
const emitTranscriptEvent = (event: LaneWatchEvent): void => {
const reduced = reduceLaneSnapshot(snapshot, event);
if (!("rebase" in reduced)) snapshot = reduced;
transcriptRevision += 1;
transcriptUpdates.set({ type: "event", revision: transcriptRevision, event }, BACKGROUND_CONTEXT);
};
const prompt = vi.fn(async () => {
await watchListener?.({
emitTranscriptEvent({
type: "entry_added",
lane: "main",
entry: {
@ -154,7 +159,7 @@ describe("experimental client TUI", () => {
message: { role: "user", content: [{ type: "text", text: "hello" }], timestamp: 1 },
},
});
await watchListener?.({
emitTranscriptEvent({
type: "entry_added",
lane: "main",
entry: {
@ -223,7 +228,7 @@ describe("experimental client TUI", () => {
attachment.set({ status: "detached" }, BACKGROUND_CONTEXT);
},
});
const sessionProvider = new RemoteServiceProvider([Models, AgentController, SessionPlugins]);
const sessionProvider = new RemoteServiceProvider([Models, AgentController, SessionPlugins, Transcript]);
sessionProvider.provide(SessionPlugins, { reload: reloadSessionPlugins });
sessionProvider.provide(Models, {
state: modelsState,
@ -236,6 +241,7 @@ describe("experimental client TUI", () => {
selectThinking,
});
sessionProvider.provide(AgentController, createAgentController({ prompt } as unknown as AgentLane));
sessionProvider.provide(Transcript, { updates: transcriptUpdates, snapshot: transcriptSnapshot });
const serverNamespace = createRemoteServiceBinding({
services: [SessionDirectory, SessionManagement, PresentationPlugins],
@ -266,7 +272,7 @@ describe("experimental client TUI", () => {
},
});
const sessionNamespace = createRemoteServiceBinding({
services: [Models, AgentController, SessionPlugins],
services: [Models, AgentController, SessionPlugins, Transcript],
transport: createLoopbackServiceTransport(sessionProvider),
bound: false,
});
@ -297,7 +303,6 @@ describe("experimental client TUI", () => {
const server: ClientTuiServer = {
serverId,
radius: true,
laneWatches: { watchSession },
server: serverServices,
session: sessionServices,
};
@ -326,7 +331,7 @@ describe("experimental client TUI", () => {
expect.anything(),
);
expect(attachment.value).toEqual({ status: "attached", sessionId });
expect(watchSession).toHaveBeenCalledWith(sessionId);
expect(transcriptSnapshot).toHaveBeenCalledOnce();
expect(select).not.toHaveBeenCalled();
expect(component.render(80).join("\n")).toContain(`Server: ${serverId}`);
expect(component.render(80).join("\n")).toContain(`Session: ${sessionId}`);
@ -368,7 +373,7 @@ describe("experimental client TUI", () => {
connectionState.set({ status: "connecting", attempt: 1 }, BACKGROUND_CONTEXT);
connectionState.set({ status: "connected", since: "reconnected" }, BACKGROUND_CONTEXT);
attachment.set({ status: "attached", sessionId }, BACKGROUND_CONTEXT);
await vi.waitFor(() => expect(watchSession).toHaveBeenCalledTimes(2));
await vi.waitFor(() => expect(transcriptSnapshot).toHaveBeenCalledTimes(2));
await vi.waitFor(() => expect(component.render(80).join("\n")).not.toContain("Reattaching"));
component.handleInput("/model");

View file

@ -1,100 +0,0 @@
import type { HarnessEvent, LaneSnapshot as HarnessLaneSnapshot } from "@earendil-works/pi-agent-core";
import type { AssistantMessage } from "@earendil-works/pi-ai";
import { LaneEventSchema, LaneSnapshotSchema } from "@earendil-works/pi-protocol";
import { Check } from "typebox/value";
import { describe, expect, test } from "vitest";
import { toWireLaneEvent, toWireLaneSnapshot } from "../src/experimental/harness-wire-adapter.ts";
const usage = {
input: 1,
output: 2,
cacheRead: 3,
cacheWrite: 4,
totalTokens: 10,
cost: { input: 0.1, output: 0.2, cacheRead: 0.3, cacheWrite: 0.4, total: 1 },
};
const finalMessage: AssistantMessage = {
role: "assistant",
content: [
{ type: "text", text: "answer", textSignature: "text-signature" },
{ type: "thinking", thinking: "reason", thinkingSignature: "thinking-signature", redacted: false },
{ type: "toolCall", id: "tool-1", name: "read", arguments: { path: "/tmp" }, namespace: "fs" },
],
api: "test",
provider: "test",
model: "test",
diagnostics: [{ type: "retry", timestamp: 1, details: { attempt: 2 } }],
usage,
stopReason: "stop",
timestamp: 2,
};
describe("Harness wire adapter", () => {
test("projects lane snapshots through the closed wire schema", () => {
const snapshot: HarnessLaneSnapshot = {
lane: "main",
transcript: [
{
id: "entry-1",
parentId: null,
seq: 1,
timestamp: 1,
type: "message",
message: { role: "user", content: "hello", timestamp: 1 },
},
],
tipId: "entry-1",
configuration: {
model: { provider: "faux", modelId: "faux-1" },
thinkingLevel: "off",
activeToolNames: [],
},
stats: { messageCount: 1, usage: finalMessage.usage },
operation: null,
queues: [],
faulted: false,
};
const wire = toWireLaneSnapshot(snapshot);
expect(Check(LaneSnapshotSchema, wire)).toBe(true);
expect(wire).toMatchObject({ lane: "main", transcript: [{ id: "entry-1" }] });
});
test("projects family-neutral operation abort events", () => {
const event: HarnessEvent = {
type: "operation_abort",
lane: "main",
operationId: "operation-1",
steer: [{ role: "user", content: "steer", timestamp: 1 }],
followUp: [{ role: "user", content: "follow", timestamp: 2 }],
};
const wire = toWireLaneEvent(event);
expect(Check(LaneEventSchema, wire)).toBe(true);
expect(wire).toMatchObject({ type: "operation_abort", operationId: "operation-1" });
});
test("projects assistant updates as compact frames", () => {
const partial: AssistantMessage = {
...finalMessage,
content: [{ type: "text", text: "answer" }],
stopReason: "pending",
};
const event: HarnessEvent = {
type: "message_update",
lane: "main",
runId: "run-1",
message: partial,
event: { type: "text_delta", contentIndex: 0, delta: "swer", partial },
frame: { type: "text_delta", contentIndex: 0, delta: "swer" },
};
const wire = toWireLaneEvent(event);
expect(Check(LaneEventSchema, wire)).toBe(true);
expect(wire).toEqual({
type: "message_update",
lane: "main",
runId: "run-1",
message: partial,
frame: { type: "text_delta", contentIndex: 0, delta: "swer" },
});
});
});

View file

@ -1,6 +1,9 @@
import type { LaneEvent, LaneSnapshot } from "@earendil-works/pi-protocol";
import { cloneJsonValue, replicatedState } from "@earendil-works/chord";
import { BACKGROUND_CONTEXT } from "@earendil-works/chord/context";
import type { LaneSnapshot } from "@earendil-works/pi-agent-core";
import { describe, expect, test, vi } from "vitest";
import { openLaneReplica } from "../src/experimental/lane-replica.ts";
import type { TranscriptUpdate } from "../src/experimental/services/transcript.ts";
function snapshot(tipId: string | null = null): LaneSnapshot {
return {
@ -30,48 +33,69 @@ function snapshot(tipId: string | null = null): LaneSnapshot {
}
describe("experimental lane replica", () => {
test("uses the Harness reducer and resnapshots navigation rebases", async () => {
let listener: ((event: LaneEvent) => void | Promise<void>) | undefined;
const replacement = snapshot("replacement-tip");
const resnapshot = vi.fn(async () => replacement);
const dispose = vi.fn(async () => {});
const replica = await openLaneReplica(
{
async watchSession(sessionId) {
return {
id: "watch-1",
sessionId,
snapshot: snapshot(),
async start(next) {
listener = next;
},
resnapshot,
dispose,
};
},
},
"session-1",
);
test("uses the Harness reducer and installs provider rebases", async () => {
const updates = replicatedState<TranscriptUpdate | null>(null);
const readSnapshot = vi.fn(async () => ({ revision: 0, snapshot: cloneJsonValue(snapshot()) }));
const replica = await openLaneReplica({ updates, snapshot: readSnapshot }, "session-1");
const changed = vi.fn();
replica.subscribe(changed);
await listener?.({ type: "run_start", lane: "main", runId: "run-1", startedAt: 1 });
expect(replica.state().operation).toMatchObject({ id: "run-1", kind: "run" });
updates.set(
{
type: "event",
revision: 1,
event: { type: "run_start", lane: "main", runId: "run-1", startedAt: 1 },
},
BACKGROUND_CONTEXT,
);
await vi.waitFor(() => expect(replica.state().operation).toMatchObject({ id: "run-1", kind: "run" }));
expect(changed).toHaveBeenCalledOnce();
await listener?.({
type: "navigation_end",
lane: "main",
runId: "navigation-1",
status: "completed",
fromTipId: null,
tipId: "replacement-tip",
endedAt: 2,
});
expect(resnapshot).toHaveBeenCalledOnce();
expect(replica.state().tipId).toBe("replacement-tip");
updates.set(
{
type: "event",
revision: 2,
event: {
type: "navigation_end",
lane: "main",
runId: "navigation-1",
status: "completed",
fromTipId: null,
tipId: "replacement-tip",
endedAt: 2,
},
},
BACKGROUND_CONTEXT,
);
updates.set(
{ type: "snapshot", revision: 3, snapshot: cloneJsonValue(snapshot("replacement-tip")) },
BACKGROUND_CONTEXT,
);
await vi.waitFor(() => expect(replica.state().tipId).toBe("replacement-tip"));
expect(readSnapshot).toHaveBeenCalledOnce();
await replica.close();
expect(dispose).toHaveBeenCalledOnce();
});
test("repairs an application revision gap from a fresh snapshot", async () => {
const updates = replicatedState<TranscriptUpdate | null>(null);
const readSnapshot = vi
.fn()
.mockResolvedValueOnce({ revision: 0, snapshot: cloneJsonValue(snapshot()) })
.mockResolvedValueOnce({ revision: 2, snapshot: cloneJsonValue(snapshot("caught-up")) });
const replica = await openLaneReplica({ updates, snapshot: readSnapshot }, "session-1");
updates.set(
{
type: "event",
revision: 2,
event: { type: "run_start", lane: "main", runId: "run-2", startedAt: 2 },
},
BACKGROUND_CONTEXT,
);
await vi.waitFor(() => expect(replica.state().tipId).toBe("caught-up"));
expect(readSnapshot).toHaveBeenCalledTimes(2);
await replica.close();
});
});

View file

@ -1,5 +1,5 @@
import { defineFacet, type FacetLoader } from "@earendil-works/chord";
import type { AgentLane } from "@earendil-works/pi-agent-core";
import type { AgentLane, LaneSnapshot } from "@earendil-works/pi-agent-core";
import { describe, expect, test, vi } from "vitest";
import { createSessionWorkerServices } from "../src/experimental/services/worker.ts";
@ -28,7 +28,41 @@ describe("experimental plugin reload", () => {
};
},
};
const snapshot: LaneSnapshot = {
lane: "main",
transcript: [],
tipId: null,
configuration: {
model: { provider: "test", modelId: "model" },
thinkingLevel: "off",
activeToolNames: [],
},
stats: {
messageCount: 0,
usage: {
input: 0,
output: 0,
cacheRead: 0,
cacheWrite: 0,
totalTokens: 0,
cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 },
},
},
operation: null,
queues: [],
faulted: false,
};
const lane = {
async watch() {
return {
snapshot,
start() {},
async resnapshot() {
return snapshot;
},
unsubscribe() {},
};
},
async getModel() {
return undefined;
},

View file

@ -229,7 +229,7 @@ describe("Session worker operations", () => {
.find(({ type }) => type === "operation");
expect(operation).toMatchObject({
scope: { serverConnectionId: "server-generation-1", attachmentId: expect.any(String) },
call: { method: "service", args: [serviceCall] },
call: serviceCall,
});
workers.detach();
await release();

View file

@ -0,0 +1,93 @@
import { replicatedState } from "@earendil-works/chord";
import { BACKGROUND_CONTEXT } from "@earendil-works/chord/context";
import type { AgentLane, EventListener, HarnessEvent, LaneSnapshot, WatchHandle } from "@earendil-works/pi-agent-core";
import { describe, expect, test, vi } from "vitest";
import { createTranscriptService } from "../src/experimental/services/transcript-provider.ts";
function laneSnapshot(tipId: string | null = null): LaneSnapshot {
return {
lane: "main",
transcript: [],
tipId,
configuration: {
model: { provider: "test", modelId: "model" },
thinkingLevel: "off",
activeToolNames: [],
},
stats: {
messageCount: 0,
usage: {
input: 0,
output: 0,
cacheRead: 0,
cacheWrite: 0,
totalTokens: 0,
cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 },
},
},
operation: null,
queues: [],
faulted: false,
};
}
describe("Transcript service", () => {
test("publishes ordered reducer events and a snapshot after navigation", async () => {
let listener: EventListener | undefined;
const replacement = laneSnapshot("replacement-tip");
const resnapshot = vi.fn(async () => replacement);
const unsubscribe = vi.fn();
const handle: WatchHandle<LaneSnapshot> = {
snapshot: laneSnapshot(),
start(next) {
listener = next;
},
resnapshot,
unsubscribe,
};
const lane = { watch: async () => handle } as unknown as AgentLane;
const runtime = createTranscriptService(lane, replicatedState);
const updates: Array<{ type: string; revision: number }> = [];
runtime.service.updates.subscribe((update) => {
if (update !== null) updates.push({ type: update.type, revision: update.revision });
});
await runtime.activate();
expect(await runtime.service.snapshot(BACKGROUND_CONTEXT)).toMatchObject({
revision: 0,
snapshot: { lane: "main", operation: null },
});
await listener?.({ type: "run_start", lane: "main", runId: "run-1", startedAt: 1 }, BACKGROUND_CONTEXT);
expect(updates).toEqual([{ type: "event", revision: 1 }]);
expect(await runtime.service.snapshot(BACKGROUND_CONTEXT)).toMatchObject({
revision: 1,
snapshot: { operation: { id: "run-1" } },
});
const navigation: HarnessEvent = {
type: "navigation_end",
lane: "main",
runId: "navigation-1",
status: "completed",
fromTipId: null,
tipId: "replacement-tip",
endedAt: 2,
};
await listener?.(navigation, BACKGROUND_CONTEXT);
await vi.waitFor(() =>
expect(updates).toEqual([
{ type: "event", revision: 1 },
{ type: "event", revision: 2 },
{ type: "snapshot", revision: 3 },
]),
);
expect(await runtime.service.snapshot(BACKGROUND_CONTEXT)).toMatchObject({
revision: 3,
snapshot: { tipId: "replacement-tip" },
});
expect(resnapshot).toHaveBeenCalledOnce();
await runtime.dispose();
expect(unsubscribe).toHaveBeenCalledOnce();
});
});

View file

@ -2,7 +2,7 @@
Runtime-neutral schemas, types, CBOR encoding, and byte-stream framing for the experimental Pi protocol.
Protocol version `6` carries the current server- and Session-scoped facet-service slice, including Session-branch plugin preparation:
Protocol version `7` carries the current server- and Session-scoped facet-service slice, including Session-branch plugin preparation:
- a version handshake that identifies the logical `serverId`;
- explicit server and Session request targets;
@ -12,9 +12,7 @@ Protocol version `6` carries the current server- and Session-scoped facet-servic
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.
A lane watch uses three RPCs so snapshot ordering does not depend on response/event scheduling: `watch()` creates a buffering watch and returns its authoritative snapshot, `startWatch()` flushes buffered events and begins live delivery, and `stopWatch()` releases it. Events carry their watch ID. Streaming message updates carry compact assistant-message frames rather than cumulative partial messages. Reconnection creates a new watch and snapshot rather than replaying old events.
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 coherent snapshots and ordered reducer events through ordinary Chord service state. 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.
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.
@ -36,6 +34,6 @@ for (const message of decoder.push(incomingChunk)) handleServerMessage(message);
decoder.end();
```
All schemas reject unknown object properties. 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 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.
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.

View file

@ -1,3 +1,4 @@
import { isJsonValue } from "@earendil-works/chord";
import { Check } from "typebox/value";
import { decodeCbor, encodeCbor } from "./cbor/index.ts";
import {
@ -23,14 +24,14 @@ export class ProtocolValidationError extends Error {
}
export function parseClientMessage(value: unknown): ClientMessage {
if (!Check(ClientMessageSchema, value)) {
if (!Check(ClientMessageSchema, value) || !isJsonValue(value)) {
throw new ProtocolValidationError("Invalid client protocol message");
}
return value;
}
export function parseServerMessage(value: unknown): ServerMessage {
if (!Check(ServerMessageSchema, value)) {
if (!Check(ServerMessageSchema, value) || !isJsonValue(value)) {
throw new ProtocolValidationError("Invalid server protocol message");
}
return value;

View file

@ -1,565 +0,0 @@
import Type, { type Static } from "typebox";
import { JsonValueSchema } from "./json-value.ts";
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 });
const WireImageContentSchema = StrictObject({
type: Type.Literal("image"),
data: Type.String(),
mimeType: Type.String({ minLength: 1 }),
});
export type WireImageContent = Static<typeof WireImageContentSchema>;
const TextContentSchema = StrictObject({
type: Type.Literal("text"),
text: Type.String(),
textSignature: Type.Optional(Type.String()),
});
const ThinkingContentSchema = StrictObject({
type: Type.Literal("thinking"),
thinking: Type.String(),
thinkingSignature: Type.Optional(Type.String()),
redacted: Type.Optional(Type.Boolean()),
});
const ToolCallSchema = StrictObject({
type: Type.Literal("toolCall"),
id: Type.String(),
name: Type.String(),
arguments: Type.Record(Type.String(), JsonValueSchema),
thoughtSignature: Type.Optional(Type.String()),
namespace: Type.Optional(Type.String()),
});
const UsageSchema = StrictObject({
input: Type.Number(),
output: Type.Number(),
cacheRead: Type.Number(),
cacheWrite: Type.Number(),
cacheWrite1h: Type.Optional(Type.Number()),
reasoning: Type.Optional(Type.Number()),
totalTokens: Type.Number(),
cost: StrictObject({
input: Type.Number(),
output: Type.Number(),
cacheRead: Type.Number(),
cacheWrite: Type.Number(),
total: Type.Number(),
}),
});
const DeferredHandleSchema = StrictObject({
provider: Type.String(),
modelId: Type.String(),
api: Type.String(),
id: Type.String(),
expiresAt: Type.Optional(Type.Number()),
pollAfterMs: Type.Optional(Type.Number()),
data: Type.Optional(JsonValueSchema),
});
const DiagnosticErrorSchema = StrictObject({
name: Type.Optional(Type.String()),
message: Type.String(),
stack: Type.Optional(Type.String()),
code: Type.Optional(Type.Union([Type.String(), Type.Number()])),
});
const AssistantMessageDiagnosticSchema = StrictObject({
type: Type.String(),
timestamp: TimestampSchema,
error: Type.Optional(DiagnosticErrorSchema),
details: Type.Optional(Type.Record(Type.String(), JsonValueSchema)),
});
const AssistantMessageSchema = StrictObject({
role: Type.Literal("assistant"),
content: Type.Array(Type.Union([TextContentSchema, ThinkingContentSchema, ToolCallSchema])),
api: Type.String(),
provider: Type.String(),
model: Type.String(),
responseModel: Type.Optional(Type.String()),
responseId: Type.Optional(Type.String()),
diagnostics: Type.Optional(Type.Array(AssistantMessageDiagnosticSchema)),
usage: UsageSchema,
stopReason: Type.Union([
Type.Literal("pending"),
Type.Literal("stop"),
Type.Literal("length"),
Type.Literal("toolUse"),
Type.Literal("error"),
Type.Literal("aborted"),
Type.Literal("deferred"),
]),
deferred: Type.Optional(DeferredHandleSchema),
errorMessage: Type.Optional(Type.String()),
rawStopReason: Type.Optional(Type.String()),
endTurn: Type.Optional(Type.Boolean()),
timestamp: TimestampSchema,
});
const UserMessageSchema = StrictObject({
role: Type.Literal("user"),
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, WireImageContentSchema])),
details: Type.Optional(JsonValueSchema),
usage: Type.Optional(UsageSchema),
addedToolNames: Type.Optional(Type.Array(Type.String())),
isError: Type.Boolean(),
timestamp: TimestampSchema,
});
const BashExecutionMessageSchema = StrictObject({
role: Type.Literal("bashExecution"),
command: Type.String(),
output: Type.String(),
exitCode: Type.Optional(Type.Number()),
cancelled: Type.Boolean(),
truncated: Type.Boolean(),
fullOutputPath: Type.Optional(Type.String()),
timestamp: TimestampSchema,
excludeFromContext: Type.Optional(Type.Boolean()),
});
const CustomMessageSchema = StrictObject({
role: Type.Literal("custom"),
customType: Type.String(),
content: Type.Union([Type.String(), Type.Array(Type.Union([TextContentSchema, WireImageContentSchema]))]),
display: Type.Boolean(),
details: Type.Optional(JsonValueSchema),
timestamp: TimestampSchema,
});
const BranchSummaryMessageSchema = StrictObject({
role: Type.Literal("branchSummary"),
summary: Type.String(),
fromId: Type.Union([Type.String(), Type.Null()]),
timestamp: TimestampSchema,
});
const CompactionSummaryMessageSchema = StrictObject({
role: Type.Literal("compactionSummary"),
summary: Type.String(),
tokensBefore: Type.Number(),
timestamp: TimestampSchema,
});
/** Closed set of built-in AgentMessage shapes supported by the wire protocol. */
const WireAgentMessageSchema = Type.Union([
UserMessageSchema,
AssistantMessageSchema,
ToolResultMessageSchema,
BashExecutionMessageSchema,
CustomMessageSchema,
BranchSummaryMessageSchema,
CompactionSummaryMessageSchema,
]);
export type WireAgentMessage = Static<typeof WireAgentMessageSchema>;
const OperationErrorSchema = StrictObject({
code: Type.String(),
message: Type.String(),
details: Type.Optional(JsonValueSchema),
});
const OperationResultBase = {
operationId: IdSchema,
kind: Type.Union([Type.Literal("run"), Type.Literal("compaction"), Type.Literal("navigation")]),
fromTipId: Type.Union([IdSchema, Type.Null()]),
tipId: Type.Union([IdSchema, Type.Null()]),
startedAt: TimestampSchema,
endedAt: TimestampSchema,
};
const OperationResultRecordSchema = Type.Union([
StrictObject({
...OperationResultBase,
status: Type.Union([Type.Literal("completed"), Type.Literal("declined"), Type.Literal("aborted")]),
}),
StrictObject({ ...OperationResultBase, status: Type.Literal("failed"), error: OperationErrorSchema }),
]);
export type OperationResultRecord = Static<typeof OperationResultRecordSchema>;
const ToolOutputSchema = StrictObject({
content: Type.Array(Type.Union([TextContentSchema, WireImageContentSchema])),
details: Type.Optional(JsonValueSchema),
usage: Type.Optional(UsageSchema),
});
export type ToolOutput = Static<typeof ToolOutputSchema>;
const AssistantMessageFrameSchema = Type.Union([
StrictObject({ type: Type.Literal("start"), partial: AssistantMessageSchema }),
StrictObject({
type: Type.Literal("text_start"),
contentIndex: Type.Integer({ minimum: 0 }),
content: TextContentSchema,
}),
StrictObject({ type: Type.Literal("text_delta"), contentIndex: Type.Integer({ minimum: 0 }), delta: Type.String() }),
StrictObject({
type: Type.Literal("text_end"),
contentIndex: Type.Integer({ minimum: 0 }),
content: Type.String(),
textSignature: Type.Optional(Type.String()),
}),
StrictObject({
type: Type.Literal("thinking_start"),
contentIndex: Type.Integer({ minimum: 0 }),
content: ThinkingContentSchema,
}),
StrictObject({
type: Type.Literal("thinking_delta"),
contentIndex: Type.Integer({ minimum: 0 }),
delta: Type.String(),
}),
StrictObject({
type: Type.Literal("thinking_end"),
contentIndex: Type.Integer({ minimum: 0 }),
content: Type.String(),
thinkingSignature: Type.Optional(Type.String()),
redacted: Type.Optional(Type.Boolean()),
}),
StrictObject({
type: Type.Literal("toolcall_start"),
contentIndex: Type.Integer({ minimum: 0 }),
toolCall: ToolCallSchema,
}),
StrictObject({
type: Type.Literal("toolcall_checkpoint"),
contentIndex: Type.Integer({ minimum: 0 }),
json: Type.String(),
}),
StrictObject({
type: Type.Literal("toolcall_delta"),
contentIndex: Type.Integer({ minimum: 0 }),
delta: Type.String(),
}),
StrictObject({
type: Type.Literal("toolcall_end"),
contentIndex: Type.Integer({ minimum: 0 }),
id: Type.String(),
name: Type.String(),
arguments: Type.Record(Type.String(), JsonValueSchema),
thoughtSignature: Type.Optional(Type.String()),
namespace: Type.Optional(Type.String()),
}),
]);
export type AssistantMessageFrame = Static<typeof AssistantMessageFrameSchema>;
const MessageEntrySchema = StrictObject({
id: IdSchema,
parentId: Type.Union([IdSchema, Type.Null()]),
seq: Type.Integer({ minimum: 1 }),
timestamp: TimestampSchema,
type: Type.Literal("message"),
message: WireAgentMessageSchema,
terminate: Type.Optional(Type.Literal(true)),
});
const CompactionEntrySchema = StrictObject({
id: IdSchema,
parentId: Type.Union([IdSchema, Type.Null()]),
seq: Type.Integer({ minimum: 1 }),
timestamp: TimestampSchema,
type: Type.Literal("compaction"),
summary: Type.String(),
retainedTail: Type.Array(WireAgentMessageSchema),
tokensBefore: Type.Number(),
details: Type.Optional(JsonValueSchema),
usage: Type.Optional(UsageSchema),
fromHook: Type.Boolean(),
});
const BranchSummaryEntrySchema = StrictObject({
id: IdSchema,
parentId: Type.Union([IdSchema, Type.Null()]),
seq: Type.Integer({ minimum: 1 }),
timestamp: TimestampSchema,
type: Type.Literal("branch_summary"),
fromId: Type.Union([IdSchema, Type.Null()]),
summary: Type.String(),
details: Type.Optional(JsonValueSchema),
usage: Type.Optional(UsageSchema),
fromHook: Type.Boolean(),
});
const CustomEntrySchema = StrictObject({
id: IdSchema,
parentId: Type.Union([IdSchema, Type.Null()]),
seq: Type.Integer({ minimum: 1 }),
timestamp: TimestampSchema,
type: Type.Literal("custom"),
customType: Type.String({ minLength: 1 }),
data: Type.Optional(JsonValueSchema),
});
const LaneEntrySchema = Type.Union([
MessageEntrySchema,
CompactionEntrySchema,
BranchSummaryEntrySchema,
CustomEntrySchema,
]);
export type LaneEntry = Static<typeof LaneEntrySchema>;
const LaneQueuedItemSchema = Type.Union([
StrictObject({
entryId: IdSchema,
kind: Type.Union([
Type.Literal("steer"),
Type.Literal("followUp"),
Type.Literal("nextRun"),
Type.Literal("write"),
]),
type: Type.Literal("message"),
message: WireAgentMessageSchema,
}),
StrictObject({
entryId: IdSchema,
kind: Type.Literal("write"),
type: Type.Literal("custom"),
customType: Type.String(),
data: Type.Optional(JsonValueSchema),
}),
]);
const RunningToolSchema = StrictObject({
toolCallId: Type.String(),
toolName: Type.String(),
args: JsonValueSchema,
partialResult: Type.Optional(ToolOutputSchema),
});
const LaneOperationSchema = StrictObject({
id: IdSchema,
kind: Type.Union([Type.Literal("run"), Type.Literal("compaction"), Type.Literal("navigation")]),
status: Type.Union([Type.Literal("running"), Type.Literal("open"), Type.Literal("aborting")]),
startedAt: TimestampSchema,
fromTipId: Type.Union([IdSchema, Type.Null()]),
deferred: Type.Optional(StrictObject({ handle: DeferredHandleSchema, poll: Type.Integer({ minimum: 0 }) })),
streamingMessage: Type.Optional(AssistantMessageSchema),
runningTools: Type.Array(RunningToolSchema),
retry: Type.Optional(
StrictObject({
attempt: Type.Integer({ minimum: 1 }),
maxAttempts: Type.Integer({ minimum: 1 }),
nextAttemptAt: TimestampSchema,
}),
),
});
const ModelIdentitySchema = StrictObject({ provider: Type.String(), modelId: Type.String() });
const ThinkingLevelSchema = Type.Union([
Type.Literal("off"),
Type.Literal("minimal"),
Type.Literal("low"),
Type.Literal("medium"),
Type.Literal("high"),
Type.Literal("xhigh"),
Type.Literal("max"),
]);
const LaneConfigurationSchema = StrictObject({
model: ModelIdentitySchema,
thinkingLevel: ThinkingLevelSchema,
activeToolNames: Type.Array(Type.String()),
});
const SessionStatsSchema = StrictObject({ messageCount: Type.Integer({ minimum: 0 }), usage: UsageSchema });
export const LaneSnapshotSchema = StrictObject({
lane: Type.String(),
transcript: Type.Array(LaneEntrySchema),
tipId: Type.Union([IdSchema, Type.Null()]),
lastResult: Type.Optional(OperationResultRecordSchema),
configuration: LaneConfigurationSchema,
stats: SessionStatsSchema,
operation: Type.Union([LaneOperationSchema, Type.Null()]),
queues: Type.Array(LaneQueuedItemSchema),
faulted: Type.Boolean(),
});
export type LaneSnapshot = Static<typeof LaneSnapshotSchema>;
const LaneEventBase = {
lane: Type.String(),
recovery: Type.Optional(Type.Literal(true)),
};
const RunEndBase = {
type: Type.Literal("run_end"),
runId: IdSchema,
fromTipId: Type.Union([IdSchema, Type.Null()]),
tipId: Type.Union([IdSchema, Type.Null()]),
endedAt: TimestampSchema,
...LaneEventBase,
};
export const LaneEventSchema = Type.Union([
StrictObject({ type: Type.Literal("run_start"), runId: IdSchema, startedAt: TimestampSchema, ...LaneEventBase }),
StrictObject({ type: Type.Literal("run_resume"), runId: IdSchema, ...LaneEventBase }),
StrictObject({
type: Type.Literal("run_suspend"),
runId: IdSchema,
reason: Type.Literal("deferred"),
deferred: DeferredHandleSchema,
poll: Type.Integer({ minimum: 0 }),
...LaneEventBase,
}),
StrictObject({
type: Type.Literal("operation_abort"),
operationId: IdSchema,
steer: Type.Array(WireAgentMessageSchema),
followUp: Type.Array(WireAgentMessageSchema),
...LaneEventBase,
}),
StrictObject({
...RunEndBase,
status: Type.Union([Type.Literal("completed"), Type.Literal("aborted")]),
}),
StrictObject({ ...RunEndBase, status: Type.Literal("failed"), error: OperationErrorSchema }),
StrictObject({
type: Type.Literal("message_start"),
runId: Type.Optional(IdSchema),
message: WireAgentMessageSchema,
...LaneEventBase,
}),
StrictObject({
type: Type.Literal("message_update"),
runId: IdSchema,
message: AssistantMessageSchema,
frame: Type.Optional(AssistantMessageFrameSchema),
...LaneEventBase,
}),
StrictObject({
type: Type.Literal("message_end"),
runId: Type.Optional(IdSchema),
message: WireAgentMessageSchema,
entryId: Type.Optional(IdSchema),
...LaneEventBase,
}),
StrictObject({
type: Type.Literal("tool_start"),
runId: IdSchema,
turnId: IdSchema,
toolCallId: Type.String(),
toolName: Type.String(),
args: JsonValueSchema,
...LaneEventBase,
}),
StrictObject({
type: Type.Literal("tool_update"),
runId: IdSchema,
turnId: IdSchema,
toolCallId: Type.String(),
toolName: Type.String(),
partialResult: ToolOutputSchema,
...LaneEventBase,
}),
StrictObject({
type: Type.Literal("tool_end"),
runId: IdSchema,
turnId: IdSchema,
toolCallId: Type.String(),
toolName: Type.String(),
result: ToolOutputSchema,
isError: Type.Boolean(),
terminate: Type.Boolean(),
...LaneEventBase,
}),
StrictObject({ type: Type.Literal("entry_added"), entry: LaneEntrySchema, ...LaneEventBase }),
StrictObject({
type: Type.Literal("queue_update"),
queues: Type.Array(LaneQueuedItemSchema),
...LaneEventBase,
}),
StrictObject({
type: Type.Literal("retry_scheduled"),
runId: IdSchema,
step: IdSchema,
attempt: Type.Integer({ minimum: 1 }),
maxAttempts: Type.Integer({ minimum: 1 }),
delayMs: Type.Integer({ minimum: 0 }),
notBefore: TimestampSchema,
errorMessage: Type.String(),
...LaneEventBase,
}),
StrictObject({
type: Type.Literal("retry_start"),
runId: IdSchema,
step: IdSchema,
attempt: Type.Integer({ minimum: 1 }),
...LaneEventBase,
}),
StrictObject({
type: Type.Literal("retry_end"),
runId: IdSchema,
step: IdSchema,
attempt: Type.Integer({ minimum: 1 }),
success: Type.Boolean(),
finalError: Type.Optional(Type.String()),
...LaneEventBase,
}),
StrictObject({
type: Type.Literal("compaction_start"),
runId: IdSchema,
reason: Type.Union([Type.Literal("manual"), Type.Literal("threshold"), Type.Literal("overflow")]),
startedAt: TimestampSchema,
...LaneEventBase,
}),
StrictObject({
type: Type.Literal("compaction_end"),
runId: IdSchema,
reason: Type.Union([Type.Literal("manual"), Type.Literal("threshold"), Type.Literal("overflow")]),
status: Type.Union([
Type.Literal("completed"),
Type.Literal("declined"),
Type.Literal("failed"),
Type.Literal("aborted"),
]),
entryId: Type.Optional(IdSchema),
error: Type.Optional(OperationErrorSchema),
endedAt: TimestampSchema,
...LaneEventBase,
}),
StrictObject({
type: Type.Literal("navigation_start"),
runId: IdSchema,
targetId: Type.Union([IdSchema, Type.Null()]),
startedAt: TimestampSchema,
...LaneEventBase,
}),
StrictObject({
type: Type.Literal("navigation_end"),
runId: IdSchema,
status: Type.Union([
Type.Literal("completed"),
Type.Literal("declined"),
Type.Literal("failed"),
Type.Literal("aborted"),
]),
fromTipId: Type.Union([IdSchema, Type.Null()]),
tipId: Type.Union([IdSchema, Type.Null()]),
error: Type.Optional(OperationErrorSchema),
endedAt: TimestampSchema,
...LaneEventBase,
}),
StrictObject({
type: Type.Literal("usage"),
lane: Type.String(),
row: StrictObject({
id: IdSchema,
seq: Type.Integer({ minimum: 1 }),
usage: UsageSchema,
entryId: Type.Optional(IdSchema),
adjustment: Type.Boolean(),
details: Type.Optional(JsonValueSchema),
}),
totals: UsageSchema,
}),
StrictObject({
type: Type.Literal("config_update"),
property: Type.Union([Type.Literal("model"), Type.Literal("thinkingLevel"), Type.Literal("activeTools")]),
value: JsonValueSchema,
previous: JsonValueSchema,
...LaneEventBase,
}),
StrictObject({
type: Type.Literal("config_update"),
property: Type.Union([
Type.Literal("streamOptions"),
Type.Literal("retryPolicy"),
Type.Literal("compactionSettings"),
Type.Literal("steeringMode"),
Type.Literal("followUpMode"),
]),
value: JsonValueSchema,
previous: JsonValueSchema,
}),
StrictObject({
type: Type.Literal("config_update"),
property: Type.Union([Type.Literal("tools"), Type.Literal("resources")]),
}),
StrictObject({ type: Type.Literal("fault"), code: Type.String(), message: Type.String() }),
]);
export type LaneEvent = Static<typeof LaneEventSchema>;

View file

@ -1,7 +1,6 @@
export * from "./cbor/index.ts";
export * from "./codec.ts";
export * from "./framing.ts";
export * from "./harness.ts";
export * from "./json-value.ts";
export {
type AttachmentEnvelope,
@ -11,14 +10,8 @@ export {
createServiceCatalogueCall,
createServiceSubscribeCall,
createServiceUnsubscribeCall,
decodeLaneWatchRpcCall,
decodeServiceControlCall,
type EventEnvelope,
encodeLaneWatchRpcCall,
isServerId,
LaneWatchRpc,
type LaneWatchRpcCall,
type LaneWatchRpcResultUnion,
PROTOCOL_VERSION,
type ProtocolError,
type ProtocolErrorCode,
@ -30,7 +23,6 @@ export {
type RequestEnvelope,
type ResponseEnvelope,
type RpcTarget,
type ServerEventEnvelope,
type ServerHello,
type ServerHelloError,
type ServerId,
@ -45,4 +37,3 @@ export {
type ServiceSubscriptionSnapshot,
type SessionTarget,
} from "./protocol.ts";
export * from "./rpc.ts";

View file

@ -1,7 +1,7 @@
import type { JsonValue } from "@earendil-works/chord";
import Type from "typebox";
/** Static JSON contract. Concrete serializers are responsible for rejecting unsupported runtime values. */
/** Static JSON contract. Protocol codecs perform recursive validation at the wire boundary. */
export const JsonValueSchema = Type.Unsafe<JsonValue>(Type.Unknown());
export type { JsonValue } from "@earendil-works/chord";

View file

@ -1,10 +1,9 @@
import { isJsonValue } from "@earendil-works/chord";
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, defineRpc, type RpcCall, type RpcMethodName, type RpcResultUnion } from "./rpc.ts";
export const PROTOCOL_VERSION = 6 as const;
export const PROTOCOL_VERSION = 7 as const;
const IdSchema = Type.String({ minLength: 1 });
const StrictObject = <const T extends Parameters<typeof Type.Object>[0]>(properties: T) =>
@ -19,30 +18,6 @@ export function isServerId(value: unknown): value is ServerId {
return Check(ServerIdSchema, value);
}
/** Temporary typed facade for snapshot-first main-lane observation. */
export const LaneWatchRpc = defineRpc({
watch: {
args: Type.Tuple([]),
result: StrictObject({ watchId: IdSchema, snapshot: LaneSnapshotSchema }),
},
startWatch: {
args: Type.Tuple([IdSchema]),
result: StrictObject({ watchId: IdSchema }),
},
resnapshotWatch: {
args: Type.Tuple([IdSchema]),
result: StrictObject({ watchId: IdSchema, snapshot: LaneSnapshotSchema }),
},
stopWatch: {
args: Type.Tuple([IdSchema]),
result: StrictObject({ watchId: IdSchema }),
},
});
type LaneWatchRpcMethod = RpcMethodName<typeof LaneWatchRpc>;
export type LaneWatchRpcCall = RpcCall<typeof LaneWatchRpc>;
export type LaneWatchRpcResultUnion = RpcResultUnion<typeof LaneWatchRpc>;
const LaneWatchRpcCallSchema = Type.Unsafe<LaneWatchRpcCall>(createRpcCallSchema(LaneWatchRpc));
const ServiceModeSchema = Type.Union([Type.Literal("singleton"), Type.Literal("keyed")]);
export type ServiceMode = Static<typeof ServiceModeSchema>;
@ -98,7 +73,7 @@ const ServiceSubscriptionSnapshotSchema = StrictObject({
export type ServiceSubscriptionSnapshot = Static<typeof ServiceSubscriptionSnapshotSchema>;
export function parseServiceSubscriptionSnapshot(value: unknown): ServiceSubscriptionSnapshot {
if (!Check(ServiceSubscriptionSnapshotSchema, value)) {
if (!Check(ServiceSubscriptionSnapshotSchema, value) || !isJsonValue(value)) {
throw new TypeError("Invalid service subscription snapshot");
}
return value;
@ -185,29 +160,6 @@ export function decodeServiceControlCall(call: ProtocolRpcCall): ServiceControlC
return undefined;
}
const LaneWatchRpcAddresses = {
watch: { serviceId: "pi.transcript", member: "watch" },
startWatch: { serviceId: "pi.transcript", member: "startWatch" },
resnapshotWatch: { serviceId: "pi.transcript", member: "resnapshotWatch" },
stopWatch: { serviceId: "pi.transcript", member: "stopWatch" },
} as const satisfies Record<LaneWatchRpcMethod, { serviceId: string; member: string }>;
/** Translate the lane-watch contract to its generic service envelope. */
export function encodeLaneWatchRpcCall(call: LaneWatchRpcCall): ProtocolRpcCall {
return { ...LaneWatchRpcAddresses[call.method], args: call.args };
}
/** Validate and decode one generic envelope against the lane-watch contract. */
export function decodeLaneWatchRpcCall(call: ProtocolRpcCall): LaneWatchRpcCall | undefined {
if (call.instance !== undefined) return undefined;
for (const [method, address] of Object.entries(LaneWatchRpcAddresses)) {
if (call.serviceId !== address.serviceId || call.member !== address.member) continue;
const candidate = { method, args: call.args };
return Check(LaneWatchRpcCallSchema, candidate) ? (candidate as LaneWatchRpcCall) : undefined;
}
return undefined;
}
export const ServiceErrorCodeSchema = Type.Union([
Type.Literal("service_not_allowed"),
Type.Literal("service_not_found"),
@ -226,8 +178,6 @@ const ProtocolErrorCodeSchema = Type.Union([
Type.Literal("session_not_found"),
Type.Literal("session_ambiguous"),
Type.Literal("session_not_attached"),
Type.Literal("watch_not_found"),
Type.Literal("watch_in_use"),
Type.Literal("not_supported"),
Type.Literal("server_draining"),
ServiceErrorCodeSchema,
@ -302,11 +252,6 @@ const ResponseEnvelopeSchema = Type.Union([
error: ProtocolErrorSchema,
}),
]);
const EventEnvelopeSchema = StrictObject({
type: Type.Literal("event"),
watchId: IdSchema,
event: LaneEventSchema,
});
const ServiceEventEnvelopeSchema = StrictObject({
type: Type.Literal("service_update"),
subscriptionId: IdSchema,
@ -321,15 +266,12 @@ export const ServerMessageSchema = Type.Union([
ServerHelloSchema,
ServerHelloErrorSchema,
ResponseEnvelopeSchema,
EventEnvelopeSchema,
ServiceEventEnvelopeSchema,
AttachmentEnvelopeSchema,
]);
export type ServerHello = Static<typeof ServerHelloSchema>;
export type ServerHelloError = Static<typeof ServerHelloErrorSchema>;
export type ResponseEnvelope = Static<typeof ResponseEnvelopeSchema>;
export type EventEnvelope = Static<typeof EventEnvelopeSchema>;
export type ServiceEventEnvelope = Static<typeof ServiceEventEnvelopeSchema>;
export type AttachmentEnvelope = Static<typeof AttachmentEnvelopeSchema>;
export type ServerEventEnvelope = EventEnvelope | ServiceEventEnvelope;
export type ServerMessage = Static<typeof ServerMessageSchema>;

View file

@ -1,112 +0,0 @@
import Type, { type Static, type TSchema } from "typebox";
import { Check } from "typebox/value";
export interface RpcMethodSchema<TArgs extends TSchema = TSchema, TResult extends TSchema = TSchema> {
readonly args: TArgs;
readonly result: TResult;
}
export type RpcManifest = Record<string, RpcMethodSchema>;
export function defineRpc<const TManifest extends RpcManifest>(manifest: TManifest): TManifest {
if (Object.keys(manifest).length === 0) throw new TypeError("RPC manifest must define at least one method");
return manifest;
}
export function createRpcCallSchema<TManifest extends RpcManifest>(manifest: TManifest): TSchema {
const variants = Object.entries(manifest).map(([method, definition]) =>
Type.Object(
{
method: Type.Literal(method),
args: definition.args,
},
{ additionalProperties: false },
),
);
return Type.Union(variants);
}
export function createRpcResultSchema<TManifest extends RpcManifest>(manifest: TManifest): TSchema {
return Type.Union(Object.values(manifest).map(({ result }) => result));
}
export type RpcMethodName<TManifest extends RpcManifest> = Extract<keyof TManifest, string>;
export type RpcArgs<TManifest extends RpcManifest, TMethod extends RpcMethodName<TManifest>> = Static<
TManifest[TMethod]["args"]
> &
readonly unknown[];
export type RpcResult<TManifest extends RpcManifest, TMethod extends RpcMethodName<TManifest>> = Static<
TManifest[TMethod]["result"]
>;
export type RpcCall<TManifest extends RpcManifest> = {
[TMethod in RpcMethodName<TManifest>]: {
readonly method: TMethod;
readonly args: RpcArgs<TManifest, TMethod>;
};
}[RpcMethodName<TManifest>];
export type RpcResultUnion<TManifest extends RpcManifest> = {
[TMethod in RpcMethodName<TManifest>]: RpcResult<TManifest, TMethod>;
}[RpcMethodName<TManifest>];
export type RpcClient<TManifest extends RpcManifest> = {
[TMethod in RpcMethodName<TManifest>]: (
...args: RpcArgs<TManifest, TMethod>
) => Promise<RpcResult<TManifest, TMethod>>;
};
export type RpcImplementation<TManifest extends RpcManifest, TContext> = {
[TMethod in RpcMethodName<TManifest>]: (
context: TContext,
...args: RpcArgs<TManifest, TMethod>
) => RpcResult<TManifest, TMethod> | Promise<RpcResult<TManifest, TMethod>>;
};
export type RpcValidationErrorFactory = (message: string) => Error;
function defaultValidationError(message: string): Error {
return new TypeError(message);
}
/**
* Create typed methods backed by one generic RPC call function.
*
* Only methods declared by the manifest are created. Arguments and returned
* values are validated at this boundary.
*/
export function createRpcClient<TManifest extends RpcManifest>(
manifest: TManifest,
call: (call: RpcCall<TManifest>) => Promise<unknown>,
validationError: RpcValidationErrorFactory = defaultValidationError,
): RpcClient<TManifest> {
const client: Record<string, (...args: unknown[]) => Promise<unknown>> = {};
for (const method of Object.keys(manifest)) {
const definition = manifest[method];
if (!definition) continue;
client[method] = async (...args: unknown[]) => {
if (!Check(definition.args, args)) throw validationError(`Invalid arguments for RPC method ${method}`);
const result = await call({ method, args } as RpcCall<TManifest>);
if (!Check(definition.result, result)) throw validationError(`Invalid result for RPC method ${method}`);
return result;
};
}
return client as unknown as RpcClient<TManifest>;
}
/** Bind a typed implementation to a validated RPC manifest. */
export function createRpcDispatcher<TManifest extends RpcManifest, TContext>(
manifest: TManifest,
implementation: RpcImplementation<TManifest, TContext>,
validationError: RpcValidationErrorFactory = defaultValidationError,
): (call: RpcCall<TManifest>, context: TContext) => Promise<RpcResultUnion<TManifest>> {
return async (call, context) => {
const definition = manifest[call.method];
const method = implementation[call.method];
if (!definition || typeof method !== "function") throw validationError(`Unknown RPC method ${call.method}`);
if (!Check(definition.args, call.args)) throw validationError(`Invalid arguments for RPC method ${call.method}`);
const result: unknown = await Reflect.apply(method, implementation, [context, ...call.args]);
if (!Check(definition.result, result)) throw validationError(`Invalid result for RPC method ${call.method}`);
return result as RpcResultUnion<TManifest>;
};
}

View file

@ -3,24 +3,17 @@ import {
type ClientHello,
type ClientMessage,
ClientMessageDecoder,
createRpcClient,
createRpcDispatcher,
createServiceCatalogueCall,
createServiceSubscribeCall,
createServiceUnsubscribeCall,
decodeCbor,
decodeLaneWatchRpcCall,
decodeServiceControlCall,
defineRpc,
encodeCbor,
encodeClientMessage,
encodeFrame,
encodeLaneWatchRpcCall,
encodeServerMessage,
FrameDecoder,
isSupportedProtocolVersion,
type LaneSnapshot,
LaneWatchRpc,
PROTOCOL_VERSION,
ProtocolValidationError,
parseClientMessage,
@ -39,106 +32,12 @@ const serverHello: ServerHello = {
serverId: "00000000-0000-4000-8000-000000000001",
};
const laneSnapshot = {
lane: "main",
transcript: [],
tipId: null,
configuration: {
model: { provider: "faux", modelId: "faux-1" },
thinkingLevel: "off",
activeToolNames: [],
},
stats: {
messageCount: 0,
usage: {
input: 0,
output: 0,
cacheRead: 0,
cacheWrite: 0,
totalTokens: 0,
cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 },
},
},
operation: null,
queues: [],
faulted: false,
} satisfies LaneSnapshot;
describe("RPC manifest", () => {
test("creates typed lane-watch client methods from the manifest", async () => {
const calls: unknown[] = [];
const client = createRpcClient(LaneWatchRpc, async (call) => {
calls.push(call);
switch (call.method) {
case "watch":
return { watchId: "watch-1", snapshot: laneSnapshot };
case "startWatch":
case "stopWatch":
return { watchId: call.args[0] };
case "resnapshotWatch":
return { watchId: call.args[0], snapshot: laneSnapshot };
}
});
await expect(client.watch()).resolves.toEqual({ watchId: "watch-1", snapshot: laneSnapshot });
await expect(client.startWatch("watch-1")).resolves.toEqual({ watchId: "watch-1" });
await expect(client.resnapshotWatch("watch-1")).resolves.toEqual({
watchId: "watch-1",
snapshot: laneSnapshot,
});
await expect(client.stopWatch("watch-1")).resolves.toEqual({ watchId: "watch-1" });
expect(calls).toEqual([
{ method: "watch", args: [] },
{ method: "startWatch", args: ["watch-1"] },
{ method: "resnapshotWatch", args: ["watch-1"] },
{ method: "stopWatch", args: ["watch-1"] },
]);
});
test("maps lane-watch contracts onto generic service/member envelopes", () => {
const encoded = encodeLaneWatchRpcCall({ method: "startWatch", args: ["watch-1"] });
expect(encoded).toEqual({ serviceId: "pi.transcript", member: "startWatch", args: ["watch-1"] });
expect(decodeLaneWatchRpcCall(encoded)).toEqual({ method: "startWatch", args: ["watch-1"] });
expect(decodeLaneWatchRpcCall({ serviceId: "pi.transcript", member: "startWatch", args: [] })).toBeUndefined();
expect(decodeLaneWatchRpcCall({ serviceId: "unknown", member: "method", args: [] })).toBeUndefined();
});
test("dispatches only methods and values allowed by the manifest", async () => {
const dispatch = createRpcDispatcher(LaneWatchRpc, {
watch: () => ({ watchId: "watch-1", snapshot: laneSnapshot }),
startWatch: (_context, watchId) => ({ watchId }),
resnapshotWatch: (_context, watchId) => ({ watchId, snapshot: laneSnapshot }),
stopWatch: (_context, watchId) => ({ watchId }),
});
await expect(dispatch({ method: "watch", args: [] }, undefined)).resolves.toEqual({
watchId: "watch-1",
snapshot: laneSnapshot,
});
await expect(dispatch({ method: "resnapshotWatch", args: ["watch-1"] }, undefined)).resolves.toEqual({
watchId: "watch-1",
snapshot: laneSnapshot,
});
await expect(dispatch({ method: "startWatch", args: [] } as never, undefined)).rejects.toThrow(
/Invalid arguments/,
);
});
test("rejects invalid results on client boundaries", async () => {
const client = createRpcClient(LaneWatchRpc, async () => ({ watchId: 1 }));
await expect(client.watch()).rejects.toThrow(/Invalid result.*watch/);
});
test("rejects empty manifests instead of creating unusable RPC clients", () => {
expect(() => defineRpc({})).toThrow(/at least one method/);
});
});
describe("protocol validation", () => {
test("negotiates protocol version 6", () => {
expect(PROTOCOL_VERSION).toBe(6);
expect(isSupportedProtocolVersion(6)).toBe(true);
expect(isSupportedProtocolVersion(5)).toBe(false);
expect(isSupportedProtocolVersion(6.5)).toBe(false);
test("negotiates protocol version 7", () => {
expect(PROTOCOL_VERSION).toBe(7);
expect(isSupportedProtocolVersion(7)).toBe(true);
expect(isSupportedProtocolVersion(6)).toBe(false);
expect(isSupportedProtocolVersion(7.5)).toBe(false);
});
test.each([0, PROTOCOL_VERSION, PROTOCOL_VERSION + 1])(
@ -277,6 +176,31 @@ describe("protocol validation", () => {
expect(parseClientMessage(message)).toEqual(message);
});
test("rejects non-JSON opaque service payloads", () => {
const request = {
type: "request",
id: "request-1",
target: { serverId: "00000000-0000-4000-8000-000000000001" },
call: { serviceId: "application.custom", member: "invoke", args: [] },
};
const cyclic: { self?: unknown } = {};
cyclic.self = cyclic;
for (const [label, value] of [
["byte array", new Uint8Array([1])],
["non-finite number", Number.NaN],
["undefined property", { value: undefined }],
["cycle", cyclic],
] as const) {
expect(() => parseClientMessage({ ...request, call: { ...request.call, args: [value] } }), label).toThrow(
ProtocolValidationError,
);
expect(
() => parseServerMessage({ type: "response", id: "request-1", ok: true, result: value }),
label,
).toThrow(ProtocolValidationError);
}
});
test("validates request cancellation envelopes", () => {
const cancel: ClientMessage = {
type: "cancel",
@ -288,69 +212,6 @@ describe("protocol validation", () => {
expect(() => parseClientMessage({ ...cancel, extra: true })).toThrow(ProtocolValidationError);
});
test("validates lane watch snapshots and events", () => {
const snapshotMessage: ServerMessage = {
type: "response",
id: "request-1",
ok: true,
result: { watchId: "watch-1", snapshot: laneSnapshot },
};
const eventMessage: ServerMessage = {
type: "event",
watchId: "watch-1",
event: { type: "run_start", lane: "main", runId: "run-1", startedAt: 1 },
};
expect(parseServerMessage(snapshotMessage)).toEqual(snapshotMessage);
expect(parseServerMessage(eventMessage)).toEqual(eventMessage);
expect(
parseServerMessage({
type: "event",
watchId: "watch-1",
event: {
type: "message_update",
lane: "main",
runId: "run-1",
message: {
role: "assistant",
content: [{ type: "text", text: "hello" }],
api: "test",
provider: "test",
model: "test",
usage: {
input: 0,
output: 0,
cacheRead: 0,
cacheWrite: 0,
totalTokens: 0,
cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 },
},
stopReason: "pending",
timestamp: 1,
},
frame: { type: "text_delta", contentIndex: 0, delta: "hello" },
},
}),
).toMatchObject({ event: { frame: { delta: "hello" } } });
expect(() => parseServerMessage({ ...eventMessage, event: { ...eventMessage.event, unknown: true } })).toThrow(
ProtocolValidationError,
);
expect(() =>
parseServerMessage({
type: "event",
watchId: "watch-1",
event: {
type: "run_end",
lane: "main",
runId: "run-1",
fromTipId: null,
tipId: "leaf-1",
status: "completed",
finalEntryId: "entry-1",
},
}),
).toThrow(ProtocolValidationError);
});
test("validates attachment route updates", () => {
const attached: ServerMessage = {
type: "attachment",
@ -410,8 +271,6 @@ describe("protocol validation", () => {
"wrong_server",
"session_not_found",
"session_not_attached",
"watch_not_found",
"watch_in_use",
"not_supported",
"server_draining",
"cancelled",
@ -428,7 +287,7 @@ describe("protocol validation", () => {
test("rejects unknown messages and fields", () => {
expect(() => parseServerMessage({ ...serverHello, snapshot: {} })).toThrow(ProtocolValidationError);
expect(() => parseServerMessage({ type: "event", event: {} })).toThrow(ProtocolValidationError);
expect(() => parseServerMessage({ type: "unknown", event: {} })).toThrow(ProtocolValidationError);
});
test("does not parse JSON strings as messages", () => {

View file

@ -6,4 +6,6 @@ export default defineConfig({
environment: "node",
reporters: process.env.GITHUB_ACTIONS ? ["dot", "github-actions"] : ["dot"],
},
resolve: { conditions: ["source"] },
ssr: { resolve: { conditions: ["source"] } },
});

View file

@ -2,7 +2,7 @@
Experimental local server for the new durable Session and Agent Harness interfaces.
The current slice supports server- and Session-scoped facet-service routing, multi-presentation attachment, and optional main-lane observation. `RoutedServerServiceHost.attachClient()` creates one connection-scoped server service endpoint with narrow attachment-management capabilities. `RoutedSessionHandle.attachClient()` returns a presentation-scoped Session capability. Its optional `invokeService()` forwards an opaque service/member envelope to the selected Session endpoint; the server validates the attachment route but does not load the facet contract. Its optional `watch()` supplies an authoritative snapshot plus buffered events.
The current slice supports server- and Session-scoped facet-service routing and multi-presentation attachment. `RoutedServerServiceHost.attachClient()` creates one connection-scoped server service endpoint with narrow attachment-management capabilities. `RoutedSessionHandle.attachClient()` returns a presentation-scoped Session capability. Its optional `invokeService()` forwards an opaque service/member envelope to the selected Session endpoint; the server validates the attachment route but does not load the facet contract.
- server service calls and subscriptions route opaquely through the connection's `RoutedServerServiceAttachment`;
- the application-owned `SessionDirectory` projects the private catalog into replicated presentation-safe state;
@ -10,7 +10,7 @@ The current slice supports server- and Session-scoped facet-service routing, mul
- attachment changes are published out of band after the router installs or clears the live route;
- Session service calls route through `invokeService` without server-side business-payload decoding;
- service subscription updates remain scoped to the requesting attachment;
- `watch`, `startWatch`, and `stopWatch` provide snapshot-first lane observation when supported by the host.
- application observations such as transcripts route as ordinary service state without server-owned business schemas.
A Session may have multiple presentation attachments. Repeating `attach` from one connection is idempotent; every successful attachment has a server-generated `attachmentId` delivered only as routing control data. Session requests carry `{ serverId, sessionId, attachmentId }`, and the server rejects stale or mismatched routes. Losing a connection rejects its local responses but releases its attachment only after admitted service calls settle. The host decides when zero presentation demand and worker-local Harness activity permit worker retirement. Server shutdown closes every routed Session handle, releasing its worker and Session writer ownership.

View file

@ -8,8 +8,6 @@ type ServerOperationErrorCode =
| "session_not_found"
| "session_ambiguous"
| "session_not_attached"
| "watch_not_found"
| "watch_in_use"
| "not_supported"
| "server_draining"
>;
@ -55,20 +53,6 @@ export class SessionNotAttachedError extends ServerError {
}
}
export class WatchNotFoundError extends ServerError {
constructor() {
super("watch_not_found", "Lane watch was not found");
this.name = "WatchNotFoundError";
}
}
export class WatchInUseError extends ServerError {
constructor() {
super("watch_in_use", "Session attachment already has a lane watch");
this.name = "WatchInUseError";
}
}
export class NotSupportedError extends ServerError {
constructor(message: string) {
super("not_supported", message);

View file

@ -6,7 +6,6 @@ import {
type ClientMessage,
ClientMessageDecoder,
DEFAULT_MAX_FRAME_LENGTH,
decodeLaneWatchRpcCall,
encodeServerMessage,
isServerId,
isSupportedProtocolVersion,
@ -312,18 +311,7 @@ export class Server<TMetadata extends SessionMetadata = SessionMetadata> {
try {
if (envelope.target.serverId !== this.serverId) throw new WrongServerError();
let result: ProtocolRpcResult;
const laneWatchCall = decodeLaneWatchRpcCall(envelope.call);
if (laneWatchCall !== undefined) {
result = await this.sessions.executeLaneWatchCall(
laneWatchCall,
envelope.target,
state,
async (message, _context) => {
await this.sendMessage(state, message);
},
context,
);
} else if ("sessionId" in envelope.target) {
if ("sessionId" in envelope.target) {
result = await this.sessions.executeServiceCall(
envelope.call,
envelope.target,

View file

@ -1,35 +1,17 @@
import { randomUUID } from "node:crypto";
import { BACKGROUND_CONTEXT, type Context, type SessionMetadata } from "@earendil-works/pi-agent-core";
import {
createRpcDispatcher,
type EventEnvelope,
LaneWatchRpc,
type LaneWatchRpcCall,
type LaneWatchRpcResultUnion,
type ProtocolRpcCall,
type ProtocolRpcResult,
ProtocolValidationError,
type RpcTarget,
type ServerEventEnvelope,
type SessionTarget,
import type {
ProtocolRpcCall,
ProtocolRpcResult,
RpcTarget,
ServiceEventEnvelope,
SessionTarget,
} from "@earendil-works/pi-protocol";
import {
NotSupportedError,
ServerDrainingError,
SessionNotAttachedError,
WatchInUseError,
WatchNotFoundError,
} from "./errors.ts";
import type { RoutedSessionAttachment, RoutedSessionHandle, RoutedSessionWatch, ServerHost } from "./types.ts";
import { NotSupportedError, ServerDrainingError, SessionNotAttachedError } from "./errors.ts";
import type { RoutedSessionAttachment, RoutedSessionHandle, ServerHost } from "./types.ts";
class SessionCleanupError extends AggregateError {}
interface ActiveWatch {
readonly id: string;
readonly handle: RoutedSessionWatch;
started: boolean;
}
interface ClientAttachment {
readonly id: string;
readonly client: object;
@ -38,14 +20,6 @@ interface ClientAttachment {
acquiring?: Promise<RoutedSessionAttachment>;
lease?: RoutedSessionAttachment;
releasing?: Promise<void>;
watch?: ActiveWatch;
}
interface RpcContext {
readonly client: object;
readonly context: Context;
readonly target: RpcTarget;
publish(message: EventEnvelope, context: Context): Promise<void>;
}
interface HostedSession {
@ -70,44 +44,16 @@ export class SessionRouter<TMetadata extends SessionMetadata = SessionMetadata>
private readonly disconnectedClients = new Set<object>();
private readonly clientOperations = new Map<object, Promise<void>>();
private closePromise?: Promise<void>;
private readonly dispatchLaneWatch: (
call: LaneWatchRpcCall,
context: RpcContext,
) => Promise<LaneWatchRpcResultUnion>;
constructor(options: SessionRouterOptions<TMetadata>) {
this.options = options;
this.dispatchLaneWatch = createRpcDispatcher(
LaneWatchRpc,
{
watch: ({ client, context, target }, ..._args: never[]) =>
this.runForClient(client, () => this.createWatch(client, target, context)),
startWatch: ({ client, publish, context, target }, watchId) =>
this.runForClient(client, () => this.startWatch(client, target, watchId, publish, context)),
resnapshotWatch: ({ client, context, target }, watchId) =>
this.runForClient(client, () => this.resnapshotWatch(client, target, watchId, context)),
stopWatch: ({ client, context, target }, watchId) =>
this.runForClient(client, () => this.stopWatch(client, target, watchId, context)),
},
(message) => new ProtocolValidationError(message),
);
}
executeLaneWatchCall(
call: LaneWatchRpcCall,
target: RpcTarget,
client: object,
publish: (message: EventEnvelope, context: Context) => Promise<void>,
context: Context,
): Promise<LaneWatchRpcResultUnion> {
return this.dispatchLaneWatch(call, { client, target, publish, context });
}
async executeServiceCall(
call: ProtocolRpcCall,
target: RpcTarget,
client: object,
publish: (message: ServerEventEnvelope, context: Context) => Promise<void>,
publish: (message: ServiceEventEnvelope, context: Context) => Promise<void>,
context: Context,
): Promise<ProtocolRpcResult> {
const admitted = await this.runForClient(client, () =>
@ -259,7 +205,7 @@ export class SessionRouter<TMetadata extends SessionMetadata = SessionMetadata>
client: object,
target: RpcTarget,
call: ProtocolRpcCall,
publish: (message: ServerEventEnvelope, context: Context) => Promise<void>,
publish: (message: ServiceEventEnvelope, context: Context) => Promise<void>,
context: Context,
): Promise<{ result: Promise<ProtocolRpcResult> }> {
const attachment = this.requireAttachment(client, target);
@ -284,77 +230,6 @@ export class SessionRouter<TMetadata extends SessionMetadata = SessionMetadata>
void result.then(remove, remove);
}
private async createWatch(
client: object,
target: RpcTarget,
context: Context,
): Promise<{ watchId: string; snapshot: RoutedSessionWatch["snapshot"] }> {
const attachment = this.requireAttachment(client, target);
if (attachment.watch !== undefined) throw new WatchInUseError();
const create = attachment.lease?.watch;
if (create === undefined) throw new NotSupportedError("Routed Session does not support lane watches");
const handle = await create.call(attachment.lease, context);
const watch = { id: randomUUID(), handle, started: false } satisfies ActiveWatch;
attachment.watch = watch;
return { watchId: watch.id, snapshot: handle.snapshot };
}
private async startWatch(
client: object,
target: RpcTarget,
watchId: string,
publish: (message: EventEnvelope, context: Context) => Promise<void>,
context: Context,
): Promise<{ watchId: string }> {
const attachment = this.requireAttachment(client, target);
const watch = attachment.watch;
if (watch?.id !== watchId) throw new WatchNotFoundError();
if (!watch.started) {
watch.started = true;
try {
await watch.handle.start(
(event, eventContext) => publish({ type: "event", watchId, event }, eventContext),
context,
);
} catch (error) {
delete attachment.watch;
try {
await watch.handle.unsubscribe(context);
} catch (cleanupError) {
throw new AggregateError([error, cleanupError], "Failed to start and clean up lane watch");
}
throw error;
}
}
return { watchId };
}
private async resnapshotWatch(
client: object,
target: RpcTarget,
watchId: string,
context: Context,
): Promise<{ watchId: string; snapshot: RoutedSessionWatch["snapshot"] }> {
const attachment = this.requireAttachment(client, target);
const watch = attachment.watch;
if (watch?.id !== watchId) throw new WatchNotFoundError();
return { watchId, snapshot: await watch.handle.resnapshot(context) };
}
private async stopWatch(
client: object,
target: RpcTarget,
watchId: string,
context: Context,
): Promise<{ watchId: string }> {
const attachment = this.requireAttachment(client, target);
const watch = attachment.watch;
if (watch?.id !== watchId) throw new WatchNotFoundError();
delete attachment.watch;
await watch.handle.unsubscribe(context);
return { watchId };
}
private requireAttachment(client: object, target: RpcTarget): ClientAttachment {
if (this.options.isClosing() || this.disconnectedClients.has(client)) throw new ServerDrainingError();
if (!("sessionId" in target)) throw new SessionNotAttachedError();
@ -369,13 +244,6 @@ export class SessionRouter<TMetadata extends SessionMetadata = SessionMetadata>
attachment.releasing ??= (async () => {
const errors: unknown[] = [];
try {
const watch = attachment.watch;
delete attachment.watch;
try {
await watch?.handle.unsubscribe(context);
} catch (error) {
errors.push(error);
}
await Promise.allSettled(attachment.operations);
try {
const lease = attachment.lease ?? (await attachment.acquiring);

View file

@ -3,7 +3,6 @@ import { createConnection, type Socket } from "node:net";
import {
type ClientMessage,
encodeClientMessage,
type JsonValue,
PROTOCOL_VERSION,
type ProtocolRpcCall,
type ResponseEnvelope,
@ -13,11 +12,6 @@ import {
} from "@earendil-works/pi-protocol";
import { Deferred } from "./host.ts";
interface TestRpcCall {
readonly method: string;
readonly args: JsonValue[];
}
interface MessageWaiter {
predicate: (message: ServerMessage) => boolean;
resolve: (message: ServerMessage) => void;
@ -54,11 +48,6 @@ export class ProtocolTestClient {
return response;
}
request(serverId: string, call: TestRpcCall, id?: string): Promise<ResponseEnvelope> {
const routed = this.routeLaneWatchCall(serverId, call);
return this.requestService(routed.target, routed.call, id);
}
async requestService(
target: RpcTarget,
call: ProtocolRpcCall,
@ -96,17 +85,6 @@ export class ProtocolTestClient {
return this.channel.send(encodeClientMessage(message));
}
private routeLaneWatchCall(serverId: string, call: TestRpcCall): { target: RpcTarget; call: ProtocolRpcCall } {
if (typeof call.args[0] !== "string") throw new Error(`Test call ${call.method} requires a Session ID`);
const sessionId = call.args[0];
const attachment = this.attachment;
const target: RpcTarget =
attachment === undefined || attachment.sessionId !== sessionId
? { serverId, sessionId, attachmentId: "missing-attachment" }
: { serverId, ...attachment };
return { target, call: { serviceId: "pi.transcript", member: call.method, args: call.args.slice(1) } };
}
sendBytes(chunk: Uint8Array): Promise<void> {
return this.channel.send(chunk);
}

View file

@ -1,8 +1,8 @@
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 type { ProtocolRpcCall, ProtocolRpcResult } from "@earendil-works/pi-protocol";
import { SessionAmbiguousError, SessionNotFoundError } from "../errors.ts";
import type { RoutedServerServiceHost, RoutedSessionHandle, RoutedSessionWatch, ServerHost } from "../types.ts";
import type { RoutedServerServiceHost, RoutedSessionHandle, ServerHost } from "../types.ts";
export class Deferred<T> {
readonly promise: Promise<T>;
@ -24,77 +24,6 @@ interface OpenGate {
release: Deferred<void>;
}
const emptyLaneSnapshot: LaneSnapshot = {
lane: "main",
transcript: [],
tipId: null,
configuration: {
model: { provider: "faux", modelId: "faux-1" },
thinkingLevel: "off",
activeToolNames: [],
},
stats: {
messageCount: 0,
usage: {
input: 0,
output: 0,
cacheRead: 0,
cacheWrite: 0,
totalTokens: 0,
cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 },
},
},
operation: null,
queues: [],
faulted: false,
};
class TestHarnessWatch implements RoutedSessionWatch {
readonly snapshot: LaneSnapshot;
private readonly buffered: Array<{ event: LaneEvent; context: Context }> = [];
private listener: ((event: LaneEvent, context: Context) => void | Promise<void>) | undefined;
private tail: Promise<void> = Promise.resolve();
private state: "buffering" | "started" | "unsubscribed" = "buffering";
constructor(snapshot: LaneSnapshot) {
this.snapshot = structuredClone(snapshot);
}
start(listener: (event: LaneEvent, context: Context) => void | Promise<void>, _context: Context): void {
if (this.state !== "buffering") throw new Error("Test Harness watch may be started only once");
this.state = "started";
this.listener = listener;
for (const buffered of this.buffered.splice(0)) this.enqueue(buffered.event, buffered.context);
}
resnapshot(_context: Context): Promise<LaneSnapshot> {
return Promise.resolve(structuredClone(this.snapshot));
}
unsubscribe(_context: Context): void {
this.state = "unsubscribed";
this.buffered.splice(0);
this.listener = undefined;
}
push(event: LaneEvent, context: Context): Promise<void> {
if (this.state === "unsubscribed") return Promise.resolve();
if (this.state === "buffering") {
this.buffered.push({ event: structuredClone(event), context });
return Promise.resolve();
}
return this.enqueue(event, context);
}
private enqueue(event: LaneEvent, context: Context): Promise<void> {
const listener = this.listener;
if (listener === undefined) return Promise.resolve();
const delivery = this.tail.then(() => listener(structuredClone(event), context));
this.tail = delivery.catch(() => {});
return delivery;
}
}
export class TestHarness {
readonly session: Session;
readonly closed = new Deferred<void>();
@ -104,8 +33,6 @@ export class TestHarness {
attachmentReleaseCount = 0;
closeCount = 0;
readonly serviceCalls: ProtocolRpcCall[] = [];
watchSnapshot: LaneSnapshot = structuredClone(emptyLaneSnapshot);
private readonly watches = new Set<TestHarnessWatch>();
failAttachmentRelease?: Error;
failClose?: Error;
nextServiceError?: Error;
@ -119,14 +46,12 @@ export class TestHarness {
attachClient(_context: Context): {
invokeService: TestHarness["invokeService"];
watch: TestHarness["watch"];
release(context: Context): void;
} {
this.attachedClients += 1;
let released = false;
return {
invokeService: (call) => this.invokeService(call),
watch: (context) => this.watch(context),
release: (_context) => {
if (released) return;
this.attachmentReleaseCount += 1;
@ -137,24 +62,6 @@ export class TestHarness {
};
}
async watch(_context: Context): Promise<RoutedSessionWatch> {
const watch = new TestHarnessWatch(this.watchSnapshot);
this.watches.add(watch);
return {
snapshot: watch.snapshot,
start: (listener, context) => watch.start(listener, context),
resnapshot: (context) => watch.resnapshot(context),
unsubscribe: (context) => {
watch.unsubscribe(context);
this.watches.delete(watch);
},
};
}
async emitEvent(event: LaneEvent, context: Context = BACKGROUND_CONTEXT): Promise<void> {
await Promise.all([...this.watches].map((watch) => watch.push(event, context)));
}
async invokeService(call: ProtocolRpcCall): Promise<ProtocolRpcResult> {
this.serviceCalls.push(call);
if (this.nextServiceError) {

View file

@ -1,11 +1,5 @@
import type { Context, SessionMetadata } from "@earendil-works/pi-agent-core";
import type {
LaneEvent,
LaneSnapshot,
ProtocolRpcCall,
ProtocolRpcResult,
ServiceProviderUpdate,
} from "@earendil-works/pi-protocol";
import type { ProtocolRpcCall, ProtocolRpcResult, ServiceProviderUpdate } from "@earendil-works/pi-protocol";
import type { ServerListener } from "./listener.ts";
export interface ServerOptions {
@ -28,8 +22,6 @@ export interface RoutedSessionAttachment {
publish: (subscriptionId: string, update: ServiceProviderUpdate, context: Context) => MaybePromise<void>,
context: Context,
): Promise<ProtocolRpcResult>;
/** Observe the attached main lane when supported by this host. */
watch?(context: Context): Promise<RoutedSessionWatch>;
release(context: Context): MaybePromise<void>;
}
@ -55,14 +47,6 @@ export interface RoutedServerServiceHost {
attachClient(presentation: RoutedServerPresentation, context: Context): MaybePromise<RoutedServerServiceAttachment>;
}
/** Snapshot-first observation handle supplied by a routed Session attachment. */
export interface RoutedSessionWatch {
readonly snapshot: LaneSnapshot;
start(listener: (event: LaneEvent, context: Context) => MaybePromise<void>, context: Context): MaybePromise<void>;
resnapshot(context: Context): Promise<LaneSnapshot>;
unsubscribe(context: Context): MaybePromise<void>;
}
/** A process-safe handle that acquires presentation-scoped Session capabilities. */
export interface RoutedSessionHandle {
attachClient(context: Context): MaybePromise<RoutedSessionAttachment>;

View file

@ -114,10 +114,6 @@ describe("Session protocol", () => {
await expect(client.attach(serverId, "session-1")).resolves.toMatchObject({ ok: true });
expect(received).toBe(metadata);
await expect(client.request(serverId, { method: "watch", args: ["session-1"] })).resolves.toMatchObject({
ok: false,
error: { code: "not_supported" },
});
});
test("routes opaque server services and publishes attachment changes out of band", async () => {
@ -169,68 +165,6 @@ describe("Session protocol", () => {
await expect.poll(() => releaseCount).toBe(1);
});
test("delivers a lane snapshot before buffered and live watch events", async () => {
const host = new TestServerHost();
await host.seed("session-1");
const client = connect(createServer(host));
await client.hello();
await client.attach(serverId, "session-1");
const watched = await client.request(serverId, { method: "watch", args: ["session-1"] });
expect(watched).toMatchObject({
ok: true,
result: { snapshot: { lane: "main", transcript: [], operation: null } },
});
if (
!watched.ok ||
typeof watched.result !== "object" ||
watched.result === null ||
!("watchId" in watched.result)
) {
throw new Error("Missing watch result");
}
const watchId = watched.result.watchId;
if (typeof watchId !== "string") throw new Error("Invalid watch ID");
await expect(client.request(serverId, { method: "watch", args: ["session-1"] })).resolves.toMatchObject({
ok: false,
error: { code: "watch_in_use" },
});
await host.latestHarness("session-1").emitEvent({
type: "run_start",
lane: "main",
runId: "run-1",
startedAt: 1,
});
expect(client.messages.some((message) => message.type === "event")).toBe(false);
const messageIndex = client.messages.length;
const starting = client.request(serverId, { method: "startWatch", args: ["session-1", watchId] });
await expect(client.nextFrom(messageIndex, (message) => message.type === "event")).resolves.toMatchObject({
type: "event",
watchId,
event: { type: "run_start", runId: "run-1" },
});
await expect(starting).resolves.toMatchObject({ ok: true, result: { watchId } });
await expect(
client.request(serverId, { method: "resnapshotWatch", args: ["session-1", watchId] }),
).resolves.toMatchObject({ ok: true, result: { watchId, snapshot: { lane: "main" } } });
await host.latestHarness("session-1").emitEvent({
type: "message_start",
lane: "main",
runId: "run-1",
message: { role: "user", content: "hello", timestamp: 1 },
});
await expect(client.nextFrom(messageIndex + 1, (message) => message.type === "event")).resolves.toMatchObject({
event: { type: "message_start" },
});
await client.request(serverId, { method: "stopWatch", args: ["session-1", watchId] });
const stoppedAt = client.messages.length;
await host.latestHarness("session-1").emitEvent({ type: "run_resume", lane: "main", runId: "run-1" });
expect(client.messages.slice(stoppedAt).some((message) => message.type === "event")).toBe(false);
});
test("permits multiple client attachments per Session", async () => {
const host = new TestServerHost();
await host.seed("session-1");