feat(agent): implement remote events and client TUI

This commit is contained in:
Armin Ronacher 2026-08-20 12:37:46 +02:00
parent bafbdecd1b
commit 8df02a436c
20 changed files with 950 additions and 178 deletions

View file

@ -44,7 +44,7 @@ await agent.prompt("Hello!");
## Experimental plugin services
The package exports the transport-neutral plugin-service primitives used by the experimental application hosts: `defineRemoteService()`, `remoteState()`, `RemoteServiceProvider`, and `RemoteServiceNamespace`. Providers classify implementation methods, state members, and scaffolded event members at runtime; consumers receive stable singleton facades through `use()` or lifecycle-scoped keyed instances through `observe()`. `RemoteState` values are borrowed immutable JSON: the runtime does not defensively clone them, so callers must not mutate or retain them. `RemoteEvents` is currently a contract surface only: listener registration and emission throw `ServiceSliceNotImplemented` until its transport slice lands. Wire framing and host routing remain outside this package.
The package exports the transport-neutral plugin-service primitives used by the experimental application hosts: `defineRemoteService()`, `remoteState()`, `RemoteServiceProvider`, and `RemoteServiceNamespace`. Providers classify implementation methods, state members, and event members at runtime; consumers receive stable singleton facades through `use()` or lifecycle-scoped keyed instances through `observe()`. `RemoteState` values and `RemoteEvents` deliveries are borrowed immutable JSON: the runtime does not defensively clone them, so callers must not mutate or retain them. Remote events are ordered, non-durable, and not replayed to late consumers. Wire framing and host routing remain outside this package.
## Core Concepts

View file

@ -1,6 +1,6 @@
# Coding-Agent Application Hosts and Plugin Facets
> **Status:** Tentative design input, not a normative contract or implementation handoff. The transport-neutral service token, provider, singleton `use()`, keyed `observe()`, and `RemoteState` substrate now have an experimental implementation, including `Models`, `Chat`, `SessionDirectory`, and `SessionManagement` vertical slices. `RemoteEvents` contract/member metadata and the remaining documented built-in service contracts are scaffolded with explicit `ServiceSliceNotImplemented` behavior; event delivery is not implemented. The application host contexts, plugin kernel, references, telemetry propagation, and most other example facets remain illustrative. Reconcile this design with `rpc.md`, `telemetry.md`, and the final harness contract before adding it to `harness.md` or creating a work package.
> **Status:** Tentative design input, not a normative contract or implementation handoff. The transport-neutral service token, provider, singleton `use()`, keyed `observe()`, and `RemoteState` substrate now have an experimental implementation, including `Models`, `Chat`, `SessionDirectory`, and `SessionManagement` vertical slices. `RemoteEvents` now has an experimental transport-neutral implementation with ordered non-replayed delivery, while the remaining documented built-in service contracts retain explicit `ServiceSliceNotImplemented` members. The application host contexts, plugin kernel, references, telemetry propagation, and most other example facets remain illustrative. Reconcile this design with `rpc.md`, `telemetry.md`, and the final harness contract before adding it to `harness.md` or creating a work package.
This document assumes you already understand `AgentHarness`, `AgentLane`, `Session`, `SessionTree`, `SessionRepo`, invocation `Context`, and telemetry. Read `rpc.md` for wire frames, remote references, subscriptions, and trace carriers, and `telemetry.md` for context propagation and cancellation semantics.

View file

@ -1,52 +1,71 @@
import type { Context } from "../../harness/context.ts";
import {
isJsonValue,
type MutableRemoteEvents,
type RemoteEventListener,
type RemoteEvents,
type RemoteEventType,
ServiceSliceNotImplemented,
} from "./types.ts";
export const REMOTE_EVENTS_INTERNALS = Symbol("pi.remoteEvents.internals");
export interface RemoteEventsInternals {
readonly deliveryImplemented: false;
export interface RemoteEventsInternals<T = unknown> {
subscribe(listener: RemoteEventListener<T>): () => void;
}
export interface RemoteEventsSource<T = unknown> extends MutableRemoteEvents<T> {
readonly [REMOTE_EVENTS_INTERNALS]: RemoteEventsInternals;
readonly [REMOTE_EVENTS_INTERNALS]: RemoteEventsInternals<T>;
}
class UnimplementedRemoteEvents<T> implements RemoteEventsSource<T> {
readonly [REMOTE_EVENTS_INTERNALS] = { deliveryImplemented: false } as const;
class MutableRemoteEventsImpl<T> implements RemoteEventsSource<T> {
readonly #listeners = new Set<RemoteEventListener<T>>();
readonly #sourceListeners = new Set<RemoteEventListener<T>>();
subscribe(_listener: RemoteEventListener<T>): () => void {
throw new ServiceSliceNotImplemented("RemoteEvents.subscribe");
subscribe(listener: RemoteEventListener<T>): () => void {
this.#listeners.add(listener);
return () => this.#listeners.delete(listener);
}
on<TType extends RemoteEventType<T>>(
_type: TType,
_listener: RemoteEventListener<Extract<T, { readonly type: TType }>>,
type: TType,
listener: RemoteEventListener<Extract<T, { readonly type: TType }>>,
): () => void {
throw new ServiceSliceNotImplemented("RemoteEvents.on");
return this.subscribe((event, context) => {
if (hasEventType(event, type)) listener(event, context);
});
}
emit(_event: T, _context: Context): void {
throw new ServiceSliceNotImplemented("RemoteEvents.emit");
emit(event: T, context: Context): void {
if (!isJsonValue(event)) throw new TypeError("Remote event value must be strict JSON");
for (const listener of this.#sourceListeners) listener(event, context);
for (const listener of this.#listeners) listener(event, context);
}
get [REMOTE_EVENTS_INTERNALS](): RemoteEventsInternals<T> {
return {
subscribe: (listener) => {
this.#sourceListeners.add(listener);
return () => this.#sourceListeners.delete(listener);
},
};
}
}
/** Declare an event member now while making its unimplemented delivery explicit. */
/** Create one provider-owned non-durable semantic event source. */
export function remoteEvents<T>(): MutableRemoteEvents<T> {
return new UnimplementedRemoteEvents<T>();
return new MutableRemoteEventsImpl<T>();
}
export function getRemoteEventsInternals(value: unknown): RemoteEventsInternals | undefined {
if (typeof value !== "object" || value === null || !(REMOTE_EVENTS_INTERNALS in value)) return undefined;
const candidate = value as { readonly [REMOTE_EVENTS_INTERNALS]?: unknown };
const internals = candidate[REMOTE_EVENTS_INTERNALS];
if (typeof internals !== "object" || internals === null) return undefined;
return internals as RemoteEventsInternals;
if (typeof internals !== "object" || internals === null || !("subscribe" in internals)) return undefined;
return typeof internals.subscribe === "function" ? (internals as RemoteEventsInternals) : undefined;
}
function hasEventType<T, TType extends string>(event: T, type: TType): event is Extract<T, { readonly type: TType }> {
return typeof event === "object" && event !== null && "type" in event && event.type === type;
}
export type { RemoteEvents };

View file

@ -5,6 +5,9 @@ import { freshDeliveryContext } from "./state.ts";
import {
cloneJson,
isJsonValue,
type RemoteEventListener,
type RemoteEvents,
type RemoteEventType,
type RemoteService,
type RemoteServiceConnection,
type RemoteServiceInstance,
@ -17,7 +20,6 @@ import {
type ServiceMemberDescription,
type ServiceMemberKind,
type ServiceProviderUpdate,
ServiceSliceNotImplemented,
type ServiceStateSnapshot,
} from "./types.ts";
@ -77,12 +79,54 @@ class RemoteStateReplica implements RemoteState<JsonValue> {
}
}
class RemoteEventsReplica implements RemoteEvents<JsonValue> {
readonly #listeners = new Set<RemoteEventListener<JsonValue>>();
readonly #reportError: ErrorReporter;
constructor(reportError: ErrorReporter) {
this.#reportError = reportError;
}
subscribe(listener: RemoteEventListener<JsonValue>): () => void {
this.#listeners.add(listener);
return () => this.#listeners.delete(listener);
}
on<TType extends RemoteEventType<JsonValue>>(
type: TType,
listener: RemoteEventListener<Extract<JsonValue, { readonly type: TType }>>,
): () => void {
return this.subscribe((event, context) => {
if (hasEventType(event, type)) listener(event, context);
});
}
deliver(event: JsonValue, context: Context): void {
for (const listener of this.#listeners) {
try {
listener(event, context);
} catch (error) {
this.#reportError(toError(error));
}
}
}
}
interface PendingMemberSubscription {
readonly listener: RemoteEventListener<JsonValue>;
remove: (() => void) | undefined;
closed: boolean;
}
class MemberSlot {
readonly #serviceId: string;
readonly #member: string;
readonly #invoke: (args: readonly JsonValue[], context: Context) => Promise<JsonValue | undefined>;
readonly #state: RemoteStateReplica;
readonly #events: RemoteEventsReplica;
readonly #isActive: () => boolean;
readonly #reportError: ErrorReporter;
readonly #pendingSubscriptions = new Set<PendingMemberSubscription>();
readonly value: unknown;
#kind: ServiceMemberKind | undefined;
#expectedKind: ServiceMemberKind | undefined;
@ -98,7 +142,9 @@ class MemberSlot {
this.#member = member;
this.#invoke = invoke;
this.#isActive = isActive;
this.#reportError = reportError;
this.#state = new RemoteStateReplica(reportError);
this.#events = new RemoteEventsReplica(reportError);
const callable = (): void => {};
this.value = new Proxy(callable, {
apply: (_target, _thisArg, args) => this.#call(args),
@ -107,21 +153,10 @@ class MemberSlot {
this.#expect("state");
return this.#state.value;
}
if (property === "subscribe") {
if (this.#kind === "events") {
this.#expect("events");
return () => {
throw new ServiceSliceNotImplemented("RemoteEvents.subscribe");
};
}
this.#expect("state");
return this.#state.subscribe.bind(this.#state);
}
if (property === "subscribe") return this.#subscribe.bind(this);
if (property === "on") {
this.#expect("events");
return () => {
throw new ServiceSliceNotImplemented("RemoteEvents.on");
};
return this.#events.on.bind(this.#events);
}
if (property === Symbol.toStringTag) return "RemoteServiceMember";
if (property === "then") return undefined;
@ -141,6 +176,8 @@ class MemberSlot {
`Remote service member ${this.#serviceId}.${this.#member} is ${kind}, not ${this.#expectedKind}`,
);
}
for (const pending of this.#pendingSubscriptions) this.#activateSubscription(pending, kind);
this.#pendingSubscriptions.clear();
}
hydrate(snapshot: ServiceStateSnapshot, context: Context): void {
@ -153,10 +190,51 @@ class MemberSlot {
this.#state.update(sequence, value, context);
}
deliverEvent(event: JsonValue, context: Context): void {
this.setDescription("events");
this.#events.deliver(event, context);
}
clear(): void {
this.#state.clear();
}
#subscribe(listener: RemoteEventListener<JsonValue>): () => void {
if (typeof listener !== "function")
throw new TypeError("Remote service subscription listener must be a function");
if (this.#kind === "state" || this.#expectedKind === "state") {
this.#expect("state");
return this.#state.subscribe(listener);
}
if (this.#kind === "events" || this.#expectedKind === "events") {
this.#expect("events");
return this.#events.subscribe(listener);
}
const pending: PendingMemberSubscription = { listener, remove: undefined, closed: false };
this.#pendingSubscriptions.add(pending);
return () => {
if (pending.closed) return;
pending.closed = true;
pending.remove?.();
this.#pendingSubscriptions.delete(pending);
};
}
#activateSubscription(pending: PendingMemberSubscription, kind: ServiceMemberKind): void {
if (pending.closed) return;
if (kind === "state") pending.remove = this.#state.subscribe(pending.listener);
else if (kind === "events") pending.remove = this.#events.subscribe(pending.listener);
else {
pending.closed = true;
this.#reportError(
new RemoteServiceError(
"service_member_mismatch",
`Remote service member ${this.#serviceId}.${this.#member} is a method, not subscribable`,
),
);
}
}
#expect(kind: ServiceMemberKind): void {
if (this.#expectedKind !== undefined && this.#expectedKind !== kind) {
throw new RemoteServiceError(
@ -274,6 +352,13 @@ class ServiceFacade {
this.#slot(member).update(sequence, value, context);
}
deliverEvent(member: string, event: JsonValue, context: Context): void {
if (this.#descriptions.get(member) !== "events") {
throw new Error(`Remote service event targets non-event member ${this.#serviceId}.${member}`);
}
this.#slot(member).deliverEvent(event, context);
}
clear(): void {
for (const slot of this.#slots.values()) slot.clear();
this.#stateSnapshots.clear();
@ -458,6 +543,13 @@ class KeyedBinding<T> {
instance.facade.update(update.member, update.sequence, update.value, context);
break;
}
case "event": {
if (update.instance === undefined) throw new Error("Keyed event has no instance address");
const instance = this.#instances.get(update.instance.key);
if (instance?.address.generation !== update.instance.generation) return;
instance.facade.deliverEvent(update.member, update.event, context);
break;
}
}
} catch (error) {
this.#reportError(toError(error));
@ -634,16 +726,13 @@ export class RemoteServiceNamespace implements RemoteServiceNamespaceApi {
serviceId,
"singleton",
(update, context) => {
if (
!binding.active ||
binding.revision !== revision ||
update.type !== "state" ||
update.instance !== undefined
) {
return;
}
if (!binding.active || binding.revision !== revision || update.instance !== undefined) return;
try {
binding.facade.update(update.member, update.sequence, update.value, context);
if (update.type === "state") {
binding.facade.update(update.member, update.sequence, update.value, context);
} else if (update.type === "event") {
binding.facade.deliverEvent(update.member, update.event, context);
}
} catch (error) {
this.#reportError(toError(error));
}
@ -705,6 +794,10 @@ function isContext(value: unknown): value is Context {
return typeof candidate.value === "function" && typeof candidate.toString === "function";
}
function hasEventType<T, TType extends string>(event: T, type: TType): event is Extract<T, { readonly type: TType }> {
return typeof event === "object" && event !== null && "type" in event && event.type === type;
}
function toError(error: unknown): Error {
return error instanceof Error ? error : new Error(String(error));
}

View file

@ -1,6 +1,6 @@
import type { Context } from "../../harness/context.ts";
import type { JsonValue } from "../../harness/session/types.ts";
import { getRemoteEventsInternals } from "./events.ts";
import { getRemoteEventsInternals, type RemoteEventsInternals } from "./events.ts";
import { freshDeliveryContext, getRemoteStateInternals, type RemoteStateInternals } from "./state.ts";
import {
cloneJson,
@ -43,13 +43,13 @@ type RemoteMethod = (...args: unknown[]) => unknown;
type InstanceMember =
| { readonly kind: "method"; readonly method: RemoteMethod }
| { readonly kind: "state"; readonly state: RemoteStateInternals }
| { readonly kind: "events" };
| { readonly kind: "events"; readonly events: RemoteEventsInternals };
interface ProviderInstance {
readonly address?: ServiceInstanceAddress;
readonly implementation: object;
readonly members: ReadonlyMap<string, InstanceMember>;
readonly removeStateListeners: readonly (() => void)[];
readonly removeMemberListeners: readonly (() => void)[];
active: boolean;
}
@ -124,7 +124,7 @@ export class RemoteServiceProvider {
closed = true;
if (registration.instances.get(key) !== instance) return;
instance.active = false;
for (const remove of instance.removeStateListeners) remove();
for (const remove of instance.removeMemberListeners) remove();
registration.instances.delete(key);
this.#emit(registration, { type: "closed", instance: address });
};
@ -206,7 +206,7 @@ export class RemoteServiceProvider {
for (const instance of instances) {
if (!instance) continue;
instance.active = false;
for (const remove of instance.removeStateListeners) remove();
for (const remove of instance.removeMemberListeners) remove();
}
registration.instances.clear();
delete registration.singleton;
@ -246,12 +246,12 @@ export class RemoteServiceProvider {
throw new TypeError(`Remote service ${registration.serviceId} implementation must be an object`);
}
const members = new Map<string, InstanceMember>();
const removeStateListeners: (() => void)[] = [];
const removeMemberListeners: (() => void)[] = [];
const instance: ProviderInstance = {
...(address === undefined ? {} : { address }),
implementation,
members,
removeStateListeners,
removeMemberListeners,
active: true,
};
for (const name of Object.keys(implementation).sort()) {
@ -266,7 +266,7 @@ export class RemoteServiceProvider {
const state = getRemoteStateInternals(descriptor.value);
if (state !== undefined) {
members.set(name, { kind: "state", state });
removeStateListeners.push(
removeMemberListeners.push(
state.subscribe((value, sequence, context) => {
if (!instance.active) return;
if (!isJsonValue(value)) {
@ -287,8 +287,27 @@ export class RemoteServiceProvider {
);
continue;
}
if (getRemoteEventsInternals(descriptor.value) !== undefined) {
members.set(name, { kind: "events" });
const events = getRemoteEventsInternals(descriptor.value);
if (events !== undefined) {
members.set(name, { kind: "events", events });
removeMemberListeners.push(
events.subscribe((event, context) => {
if (!instance.active) return;
if (!isJsonValue(event)) {
throw new RemoteServiceError("service_invalid_value", "Remote event must be strict JSON");
}
this.#emit(
registration,
{
type: "event",
...(address === undefined ? {} : { instance: address }),
member: name,
event,
},
context,
);
}),
);
continue;
}
throw new TypeError(`Remote service member ${registration.serviceId}.${name} is not remotely exposable`);

View file

@ -29,9 +29,9 @@ export interface MutableRemoteState<T> extends RemoteState<T> {
}
export type RemoteEventType<T> = T extends { readonly type: infer TType extends string } ? TType : never;
export type RemoteEventListener<T> = (event: T, context: Context) => void | Promise<void>;
export type RemoteEventListener<T> = (event: T, context: Context) => void;
/** Contract surface for non-durable semantic events. Remote delivery is a later service slice. */
/** Non-durable semantic events. Event values are borrowed immutable JSON and are never replayed. */
export interface RemoteEvents<T> {
subscribe(listener: RemoteEventListener<T>): () => void;
on<TType extends RemoteEventType<T>>(
@ -134,6 +134,12 @@ export type ServiceProviderUpdate =
readonly sequence: number;
readonly value: JsonValue;
}
| {
readonly type: "event";
readonly instance?: ServiceInstanceAddress;
readonly member: string;
readonly event: JsonValue;
}
| { readonly type: "spawned"; readonly instance: ServiceInstanceSnapshot }
| { readonly type: "closed"; readonly instance: ServiceInstanceAddress };

View file

@ -11,7 +11,6 @@ import {
type RemoteState,
remoteEvents,
remoteState,
ServiceSliceNotImplemented,
} from "../../src/index.ts";
type ModelRef = { provider: string; modelId: string };
@ -34,8 +33,9 @@ interface QuestionDialogService {
const QuestionDialogs = defineRemoteService<QuestionDialogService>("question-dialog");
type ActivityEvent = { type: "changed"; revision: number } | { type: "ignored"; revision: number };
interface ActivityService {
readonly events: RemoteEvents<{ type: "changed"; revision: number }>;
readonly events: RemoteEvents<ActivityEvent>;
}
const Activity = defineRemoteService<ActivityService>("activity");
@ -139,6 +139,43 @@ describe("plugin remote services", () => {
provider.dispose();
});
test("hydrates cold RemoteState replicas and replaces them across rebinds", async () => {
const provider = new RemoteServiceProvider([Models]);
const state = remoteState<ModelsState>({ selected: null, revision: 0 });
provider.provide(Models, {
state,
async select() {},
});
const namespace = new RemoteServiceNamespace({
services: [Models],
connection: createLoopbackServiceConnection(provider),
bound: false,
});
const models = namespace.use(Models);
const revisions: number[] = [];
models.state.subscribe((value) => revisions.push(value.revision));
expect(models.state.value).toBeUndefined();
expect(revisions).toEqual([]);
state.set({ selected: null, revision: 1 }, BACKGROUND_CONTEXT);
await namespace.rebind(true, BACKGROUND_CONTEXT);
expect(models.state.value?.revision).toBe(1);
expect(revisions).toEqual([1]);
state.set({ selected: null, revision: 2 }, BACKGROUND_CONTEXT);
expect(revisions).toEqual([1, 2]);
await namespace.rebind(false, BACKGROUND_CONTEXT);
expect(models.state.value).toBeUndefined();
state.set({ selected: null, revision: 3 }, BACKGROUND_CONTEXT);
expect(revisions).toEqual([1, 2]);
await namespace.rebind(true, BACKGROUND_CONTEXT);
expect(models.state.value?.revision).toBe(3);
expect(revisions).toEqual([1, 2, 3]);
await namespace.dispose(BACKGROUND_CONTEXT);
provider.dispose();
});
test("hydrates keyed state before observe handlers and fences reused keys", async () => {
const provider = new RemoteServiceProvider([QuestionDialogs]);
const connection = createLoopbackServiceConnection(provider);
@ -203,27 +240,90 @@ describe("plugin remote services", () => {
provider.dispose();
});
test("declares RemoteEvents members while delivery remains an explicit later slice", async () => {
test("delivers filtered RemoteEvents without replay and buffers hydration races", async () => {
const provider = new RemoteServiceProvider([Activity]);
const events = remoteEvents<{ type: "changed"; revision: number }>();
const events = remoteEvents<ActivityEvent>();
provider.provide(Activity, { events });
const subscription = provider.subscribe("activity", "singleton", () => {});
expect(subscription.snapshot.instances[0]?.members).toEqual([{ name: "events", kind: "events" }]);
expect(() => events.subscribe(() => {})).toThrow(ServiceSliceNotImplemented);
expect(() => events.on("changed", () => {})).toThrow(ServiceSliceNotImplemented);
expect(() => events.emit({ type: "changed", revision: 1 }, BACKGROUND_CONTEXT)).toThrow(
ServiceSliceNotImplemented,
);
subscription.close();
const local: ActivityEvent[] = [];
const localChanged: ActivityEvent[] = [];
const removeLocal = events.subscribe((event) => local.push(event));
const removeLocalChanged = events.on("changed", (event) => localChanged.push(event));
const beforeSubscription = { type: "changed", revision: 0 } as const;
events.emit(beforeSubscription, BACKGROUND_CONTEXT);
expect(local[0]).toBe(beforeSubscription);
expect(localChanged).toEqual([beforeSubscription]);
const connection: RemoteServiceConnection = {
invoke: (call, context) => provider.invoke(call, context),
subscribe: async (serviceId, mode, listener) => {
const subscription = provider.subscribe(serviceId, mode, listener);
events.emit({ type: "changed", revision: 1 }, BACKGROUND_CONTEXT);
return {
snapshot: subscription.snapshot,
activate: () => subscription.activate(),
close: () => subscription.close(),
};
},
};
const errors: Error[] = [];
const namespace = new RemoteServiceNamespace({
services: [Activity],
connection,
onError: (error) => errors.push(error),
});
const activity = namespace.use(Activity);
const received: ActivityEvent[] = [];
const changed: Extract<ActivityEvent, { type: "changed" }>[] = [];
const unsubscribe = activity.events.subscribe((event) => received.push(event));
const unsubscribeChanged = activity.events.on("changed", (event) => changed.push(event));
await vi.waitFor(() => expect(received).toEqual([{ type: "changed", revision: 1 }]));
events.emit({ type: "ignored", revision: 2 }, BACKGROUND_CONTEXT);
events.emit({ type: "changed", revision: 3 }, BACKGROUND_CONTEXT);
expect(received).toEqual([
{ type: "changed", revision: 1 },
{ type: "ignored", revision: 2 },
{ type: "changed", revision: 3 },
]);
expect(changed).toEqual([
{ type: "changed", revision: 1 },
{ type: "changed", revision: 3 },
]);
expect(received).not.toContainEqual(beforeSubscription);
expect(errors).toEqual([]);
expect(() => events.emit({ type: "changed", revision: Number.NaN }, BACKGROUND_CONTEXT)).toThrow(
"Remote event value must be strict JSON",
);
removeLocal();
removeLocalChanged();
unsubscribe();
unsubscribeChanged();
await namespace.dispose(BACKGROUND_CONTEXT);
provider.dispose();
});
test("routes keyed RemoteEvents only for the live instance generation", async () => {
const provider = new RemoteServiceProvider([Activity]);
const namespace = new RemoteServiceNamespace({
services: [Activity],
connection: createLoopbackServiceConnection(provider),
});
const activity = namespace.use(Activity);
await new Promise<void>((resolve) => setTimeout(resolve, 0));
expect(() => activity.events.subscribe(() => {})).toThrow(ServiceSliceNotImplemented);
expect(() => activity.events.on("changed", () => {})).toThrow(ServiceSliceNotImplemented);
let observed: ActivityService | undefined;
const stop = namespace.observe(Activity, (instance) => {
observed = instance.service;
});
const events = remoteEvents<ActivityEvent>();
const close = provider.spawn(Activity, "activity-1", { events });
await vi.waitFor(() => expect(observed).toBeDefined());
const received: ActivityEvent[] = [];
observed!.events.subscribe((event) => received.push(event));
events.emit({ type: "changed", revision: 1 }, BACKGROUND_CONTEXT);
expect(received).toEqual([{ type: "changed", revision: 1 }]);
close();
events.emit({ type: "changed", revision: 2 }, BACKGROUND_CONTEXT);
expect(received).toEqual([{ type: "changed", revision: 1 }]);
stop();
await namespace.dispose(BACKGROUND_CONTEXT);
provider.dispose();
});

View file

@ -0,0 +1,147 @@
import { basename } from "node:path";
import { BACKGROUND_CONTEXT, type RemoteServiceNamespaceApi } from "@earendil-works/pi-agent-core";
import { Client } from "@earendil-works/pi-client";
import { createUnixTransportFactory, discoverUnixServers, type UnixServerRoute } from "@earendil-works/pi-client/unix";
import { isServerId } from "@earendil-works/pi-protocol";
import type { ClientCommand } from "../cli/experimental/commands/client.ts";
import { activateServer, ENV_SERVER_ID, resolveServerDirectory, resolveSessionDirectory } from "./server.ts";
import { Chat, type ChatService } from "./services/chat.ts";
import {
createBuiltinServerServiceNamespace,
createBuiltinSessionServiceNamespace,
type ServerServices,
type SessionServices,
} from "./services/connection.ts";
import { Models, type ModelsService } from "./services/models.ts";
import {
SessionDirectory,
type SessionDirectoryService,
SessionManagement,
type SessionManagementService,
} from "./services/sessions.ts";
export interface ClientRuntimeServer {
readonly route: UnixServerRoute;
readonly client: Client;
readonly server: ServerServices;
readonly session: SessionServices;
readonly directory: SessionDirectoryService;
readonly management: SessionManagementService;
readonly models: ModelsService;
readonly chat: ChatService;
}
export interface ClientRuntime {
readonly servers: readonly ClientRuntimeServer[];
dispose(): Promise<void>;
}
export interface OpenClientRuntimeOptions {
/** Directory searched when --connect is omitted. Defaults to PI_SERVER_DIR or ~/.pi/server. */
readonly directory?: string;
}
/** Open live server/session service namespaces for one experimental presentation. */
export async function openClientRuntime(
command: ClientCommand,
options: OpenClientRuntimeOptions = {},
): Promise<ClientRuntime> {
if (command.auth !== undefined) throw new Error("Authentication is not supported by the experimental local server");
if (command.provider !== undefined && command.model === undefined) {
throw new Error("Server model provider requires a model");
}
if (command.connect && command.model !== undefined) {
throw new Error("Model selection is only valid when automatically activating a new server");
}
const directory = resolveServerDirectory(options.directory);
let routes: UnixServerRoute[];
let activatedClient: Client | undefined;
if (command.connect) {
routes = [routeFromExplicitPath(command.connect.path)];
} else {
routes = await discoverUnixServers({ directory });
if (routes.length > 0 && command.model !== undefined) {
throw new Error("Model selection is only valid when automatically activating a new server");
}
if (routes.length === 0) {
const activated = await activateServer({
directory,
requestedServerId: process.env[ENV_SERVER_ID],
sessionDir: resolveSessionDirectory(),
provider: command.provider,
model: command.model,
});
routes = [activated.route];
activatedClient = activated.client;
}
}
const clients: Client[] = [];
const namespaces: RemoteServiceNamespaceApi[] = [];
const servers: ClientRuntimeServer[] = [];
let disposed = false;
const dispose = async (): Promise<void> => {
if (disposed) return;
disposed = true;
const namespaceResults = await Promise.allSettled(
namespaces.map((services) => services.dispose(BACKGROUND_CONTEXT)),
);
const clientResults = await Promise.allSettled(clients.map((client) => client.dispose()));
const errors = [...namespaceResults, ...clientResults].flatMap((result) =>
result.status === "rejected" ? [result.reason] : [],
);
if (errors.length === 1) throw errors[0];
if (errors.length > 1) throw new AggregateError(errors, "Failed to dispose experimental client runtime");
};
try {
for (const route of routes) {
const client =
activatedClient ??
(await Client.connect({
serverId: route.serverId,
transportFactory: createUnixTransportFactory({ path: route.path }),
}));
activatedClient = undefined;
clients.push(client);
const server = createBuiltinServerServiceNamespace(client);
const session = createBuiltinSessionServiceNamespace(client);
namespaces.push(server, session);
const sessionDirectory = server.use(SessionDirectory);
const management = server.use(SessionManagement);
const models = session.use(Models);
const chat = session.use(Chat);
await waitForRemoteState(sessionDirectory.state);
servers.push({ route, client, server, session, directory: sessionDirectory, management, models, chat });
}
return { servers, dispose };
} catch (error) {
try {
await dispose();
} catch (cleanupError) {
throw new AggregateError([error, cleanupError], "Experimental client startup and cleanup failed");
}
throw error;
}
}
export function waitForRemoteState<T>(state: {
readonly value: T | undefined;
subscribe(listener: (value: T) => void): () => void;
}): Promise<T> {
if (state.value !== undefined) return Promise.resolve(state.value);
return new Promise((resolve) => {
let unsubscribe: (() => void) | undefined;
unsubscribe = state.subscribe((value) => {
queueMicrotask(() => unsubscribe?.());
resolve(value);
});
});
}
function routeFromExplicitPath(path: string): UnixServerRoute {
const name = basename(path);
const serverId = name.endsWith(".sock") ? name.slice(0, -".sock".length) : "";
if (!isServerId(serverId)) throw new Error("--connect path must end with <uuidv4-server-id>.sock");
return { serverId, path };
}

View file

@ -0,0 +1,282 @@
import { BACKGROUND_CONTEXT, type RemoteState } from "@earendil-works/pi-agent-core";
import type { SessionSummary } from "@earendil-works/pi-protocol";
import {
type Component,
Container,
ProcessTerminal,
type SelectItem,
SelectList,
Text,
TuiMainScreen,
} from "@earendil-works/pi-tui";
import chalk from "chalk";
import type { ClientCommand } from "../cli/experimental/commands/client.ts";
import { type OpenClientRuntimeOptions, openClientRuntime, waitForRemoteState } from "./client-runtime.ts";
import type { SessionAttachmentState } from "./services/connection.ts";
import type { ModelsService } from "./services/models.ts";
import type { SessionDirectoryEvent, SessionDirectoryService, SessionManagementService } from "./services/sessions.ts";
export interface ClientTuiServer {
readonly serverId: string;
readonly directory: SessionDirectoryService;
readonly management: Pick<SessionManagementService, "create" | "attach" | "detach">;
readonly attachment: RemoteState<SessionAttachmentState>;
readonly models: ModelsService;
}
type ClientTuiAction =
| { readonly kind: "attach"; readonly server: ClientTuiServer; readonly sessionId: string }
| { readonly kind: "create"; readonly server: ClientTuiServer }
| { readonly kind: "model"; readonly provider: string; readonly modelId: string }
| { readonly kind: "sessions" }
| { readonly kind: "quit" };
const selectTheme = {
selectedPrefix: (text: string) => chalk.cyan(text),
selectedText: (text: string) => chalk.cyan(text),
description: (text: string) => chalk.dim(text),
scrollInfo: (text: string) => chalk.dim(text),
noMatch: (text: string) => chalk.yellow(text),
};
/** Minimal service-only presentation used before Harness execution is available. */
export class ExperimentalClientTui implements Component {
readonly #servers: readonly ClientTuiServer[];
readonly #requestRender: () => void;
readonly #finish: () => void;
readonly #container = new Container();
readonly #subscriptions: (() => void)[] = [];
readonly #actions = new Map<string, ClientTuiAction>();
#selectList: SelectList | undefined;
#screen: "sessions" | "models" = "sessions";
#selectedServer: ClientTuiServer | undefined;
#removeModelsListener: (() => void) | undefined;
#status = "Select a Session or create one.";
#busy = false;
constructor(options: {
readonly servers: readonly ClientTuiServer[];
requestRender(): void;
finish(): void;
}) {
this.#servers = options.servers;
this.#requestRender = options.requestRender;
this.#finish = options.finish;
for (const server of this.#servers) {
this.#subscriptions.push(
server.directory.state.subscribe(() => this.#rebuild()),
server.directory.events.subscribe((event) => this.#directoryChanged(event)),
server.attachment.subscribe(() => this.#rebuild()),
);
}
this.#rebuild();
}
render(width: number): string[] {
return this.#container.render(width);
}
handleInput(data: string): void {
if (!this.#busy) this.#selectList?.handleInput(data);
}
invalidate(): void {
this.#container.invalidate();
}
dispose(): void {
this.#removeModelsListener?.();
this.#removeModelsListener = undefined;
for (const unsubscribe of this.#subscriptions.splice(0)) unsubscribe();
}
#directoryChanged(event: SessionDirectoryEvent): void {
this.#status =
event.type === "deleted"
? `Session removed: ${event.sessionId}`
: `Session ${event.type}: ${event.session.sessionId}`;
this.#rebuild();
}
#rebuild(): void {
this.#container.clear();
this.#actions.clear();
const title = this.#screen === "sessions" ? "Experimental Sessions" : "Experimental Models";
this.#container.addChild(new Text(chalk.bold(title), 1, 1));
this.#container.addChild(new Text(this.#status, 1, 0));
const items = this.#screen === "sessions" ? this.#sessionItems() : this.#modelItems();
this.#selectList = new SelectList(items, Math.min(Math.max(items.length, 1), 12), selectTheme);
this.#selectList.onSelect = (item) => {
const action = this.#actions.get(item.value);
if (action !== undefined) void this.#runAction(action);
};
this.#selectList.onCancel = () => {
if (this.#screen === "models") {
this.#screen = "sessions";
this.#status = "Select a Session or create one.";
this.#rebuild();
} else {
this.#finish();
}
};
this.#container.addChild(this.#selectList);
this.#container.addChild(
new Text(
chalk.dim(this.#screen === "sessions" ? "enter select · esc quit" : "enter select · esc sessions"),
1,
1,
),
);
this.#requestRender();
}
#sessionItems(): SelectItem[] {
const items: SelectItem[] = [];
for (const server of this.#servers) {
const createKey = `create:${server.serverId}`;
this.#actions.set(createKey, { kind: "create", server });
items.push({ value: createKey, label: "+ New Session", description: server.serverId });
for (const session of server.directory.state.value?.sessions ?? []) {
const key = `session:${server.serverId}:${session.sessionId}`;
const attachment = server.attachment.value;
const attached = attachment?.status === "attached" && attachment.sessionId === session.sessionId;
this.#actions.set(key, { kind: "attach", server, sessionId: session.sessionId });
items.push({
value: key,
label: attached ? `* ${session.sessionId}` : session.sessionId,
description: attached ? `attached · ${server.serverId}` : server.serverId,
});
}
}
const quitKey = "quit";
this.#actions.set(quitKey, { kind: "quit" });
items.push({ value: quitKey, label: "Quit" });
return items;
}
#modelItems(): SelectItem[] {
const server = this.#selectedServer;
const state = server?.models.state.value;
const selected = state?.configuration.model;
const items: SelectItem[] = [];
for (const model of state?.catalog.availableModels ?? []) {
const key = `model:${model.provider}:${model.modelId}`;
this.#actions.set(key, { kind: "model", provider: model.provider, modelId: model.modelId });
items.push({
value: key,
label:
selected?.provider === model.provider && selected.modelId === model.modelId
? `${model.name} (selected)`
: model.name,
description: `${model.provider}/${model.modelId}`,
});
}
this.#actions.set("sessions", { kind: "sessions" });
items.push({ value: "sessions", label: "Back to Sessions" });
this.#actions.set("quit", { kind: "quit" });
items.push({ value: "quit", label: "Quit" });
return items;
}
async #runAction(action: ClientTuiAction): Promise<void> {
if (this.#busy) return;
if (action.kind === "quit") {
this.#finish();
return;
}
if (action.kind === "sessions") {
this.#screen = "sessions";
this.#status = "Select a Session or create one.";
this.#rebuild();
return;
}
this.#busy = true;
try {
if (action.kind === "model") {
if (this.#selectedServer === undefined) throw new Error("No Session is attached");
await this.#selectedServer.models.select(
{ provider: action.provider, modelId: action.modelId },
BACKGROUND_CONTEXT,
);
this.#status = `Selected ${action.provider}/${action.modelId}`;
return;
}
const summary =
action.kind === "create"
? await action.server.management.create({}, BACKGROUND_CONTEXT)
: findSession(action.server, action.sessionId);
this.#status = `Attaching ${summary.sessionId}…`;
this.#rebuild();
if (this.#selectedServer !== undefined && this.#selectedServer !== action.server) {
await this.#selectedServer.management.detach(BACKGROUND_CONTEXT);
}
await action.server.management.attach(summary.sessionId, BACKGROUND_CONTEXT);
await waitForAttachment(action.server.attachment, summary.sessionId);
await waitForRemoteState(action.server.models.state);
this.#selectedServer = action.server;
this.#removeModelsListener?.();
this.#removeModelsListener = action.server.models.state.subscribe(() => this.#rebuild());
this.#screen = "models";
this.#status = `Attached ${summary.sessionId}. Select a model.`;
} catch (error) {
this.#status = `Error: ${error instanceof Error ? error.message : String(error)}`;
} finally {
this.#busy = false;
this.#rebuild();
}
}
}
export async function runClientTui(command: ClientCommand, options: OpenClientRuntimeOptions = {}): Promise<void> {
const runtime = await openClientRuntime(command, options);
const terminal = new ProcessTerminal();
const tui = new TuiMainScreen(terminal);
let component: ExperimentalClientTui | undefined;
try {
await new Promise<void>((resolve) => {
const finish = (): void => {
tui.stop();
resolve();
};
component = new ExperimentalClientTui({
servers: runtime.servers.map((server) => ({
serverId: server.route.serverId,
directory: server.directory,
management: server.management,
attachment: server.session.attachment,
models: server.models,
})),
requestRender: () => tui.requestRender(),
finish,
});
tui.addChild(component);
tui.setFocus(component);
tui.start();
});
} finally {
component?.dispose();
await runtime.dispose();
}
}
function findSession(server: ClientTuiServer, sessionId: string): SessionSummary {
const session = server.directory.state.value?.sessions.find((candidate) => candidate.sessionId === sessionId);
if (session === undefined) throw new Error(`Unknown Session: ${sessionId}`);
return session;
}
function waitForAttachment(state: RemoteState<SessionAttachmentState>, sessionId: string): Promise<void> {
if (state.value?.status === "attached" && state.value.sessionId === sessionId) return Promise.resolve();
return new Promise((resolve, reject) => {
let unsubscribe: (() => void) | undefined;
unsubscribe = state.subscribe((attachment) => {
if (attachment.status === "degraded" && attachment.sessionId === sessionId) {
queueMicrotask(() => unsubscribe?.());
reject(new Error(`Session ${sessionId} is unavailable`));
} else if (attachment.status === "attached" && attachment.sessionId === sessionId) {
queueMicrotask(() => unsubscribe?.());
resolve();
}
});
});
}

View file

@ -1,13 +1,8 @@
import { basename } from "node:path";
import { BACKGROUND_CONTEXT, type RemoteServiceNamespaceApi, type RemoteState } from "@earendil-works/pi-agent-core";
import { Client } from "@earendil-works/pi-client";
import { createUnixTransportFactory, discoverUnixServers, type UnixServerRoute } from "@earendil-works/pi-client/unix";
import { isServerId, type LaneEvent, type PromptMessage, type SessionAddress } from "@earendil-works/pi-protocol";
import { BACKGROUND_CONTEXT } from "@earendil-works/pi-agent-core";
import type { LaneEvent, PromptMessage, SessionAddress } from "@earendil-works/pi-protocol";
import type { ClientCommand } from "../cli/experimental/commands/client.ts";
import { activateServer, ENV_SERVER_ID, resolveServerDirectory, resolveSessionDirectory } from "./server.ts";
import { Chat, type ChatPromptResponse } from "./services/chat.ts";
import { createBuiltinServerServiceNamespace, createBuiltinSessionServiceNamespace } from "./services/connection.ts";
import { SessionDirectory, SessionManagement, type SessionManagementService } from "./services/sessions.ts";
import { openClientRuntime } from "./client-runtime.ts";
import type { ChatPromptResponse } from "./services/chat.ts";
export type ClientResult =
| {
@ -26,71 +21,16 @@ export interface RunClientOptions {
/** Discover servers, then list Sessions, attach to one, or create one for a prompt. */
export async function runClient(command: ClientCommand, options: RunClientOptions = {}): Promise<ClientResult> {
if (command.auth !== undefined) throw new Error("Authentication is not supported by the experimental local server");
if (command.provider !== undefined && command.model === undefined) {
throw new Error("Server model provider requires a model");
}
// The current protocol cannot forward model selection to an already-running server or Session.
if (command.connect && command.model !== undefined) {
throw new Error("Model selection is only valid when automatically activating a new server");
}
const directory = resolveServerDirectory(options.directory);
let routes: UnixServerRoute[];
let activatedClient: Client | undefined;
if (command.connect) {
routes = [routeFromExplicitPath(command.connect.path)];
} else {
routes = await discoverUnixServers({ directory });
if (routes.length > 0 && command.model !== undefined) {
throw new Error("Model selection is only valid when automatically activating a new server");
}
if (routes.length === 0) {
const activated = await activateServer({
directory,
requestedServerId: process.env[ENV_SERVER_ID],
sessionDir: resolveSessionDirectory(),
provider: command.provider,
model: command.model,
});
routes = [activated.route];
activatedClient = activated.client;
}
}
const openedClients = new Set<Client>();
const serviceNamespaces = new Set<RemoteServiceNamespaceApi>();
if (activatedClient) openedClients.add(activatedClient);
const discovered: {
route: UnixServerRoute;
sessionIds: string[];
client: Client;
management: SessionManagementService;
}[] = [];
const runtime = await openClientRuntime(command, { directory: options.directory });
try {
for (const route of routes) {
const client =
activatedClient ??
(await Client.connect({
serverId: route.serverId,
transportFactory: createUnixTransportFactory({ path: route.path }),
}));
activatedClient = undefined;
openedClients.add(client);
const services = createBuiltinServerServiceNamespace(client);
serviceNamespaces.add(services);
const directory = services.use(SessionDirectory);
const management = services.use(SessionManagement);
const state = await waitForState(directory.state);
discovered.push({ route, sessionIds: state.sessions.map(({ sessionId }) => sessionId), client, management });
}
const discovered = runtime.servers;
let sessionId = command.sessionId;
if (sessionId === undefined && command.prompt === undefined) {
return {
kind: "list",
sessions: discovered
.flatMap(({ route, sessionIds }) =>
sessionIds.map((sessionId) => ({ serverId: route.serverId, sessionId })),
.flatMap(({ route, directory }) =>
directory.state.value!.sessions.map(({ sessionId }) => ({ serverId: route.serverId, sessionId })),
)
.sort(
(left, right) =>
@ -108,7 +48,9 @@ export async function runClient(command: ClientCommand, options: RunClientOption
sessionId = (await match.management.create({}, BACKGROUND_CONTEXT)).sessionId;
} else {
const selectedSessionId = sessionId;
const matches = discovered.filter((candidate) => candidate.sessionIds.includes(selectedSessionId));
const matches = discovered.filter((candidate) =>
candidate.directory.state.value!.sessions.some(({ sessionId }) => sessionId === selectedSessionId),
);
if (matches.length > 1) {
throw new Error(`Session ${selectedSessionId} is available from more than one server`);
}
@ -128,9 +70,7 @@ export async function runClient(command: ClientCommand, options: RunClientOption
return { kind: "attached", serverId: match.route.serverId, sessionId };
}
const sessionServices = createBuiltinSessionServiceNamespace(match.client);
serviceNamespaces.add(sessionServices);
const chat = sessionServices.use(Chat);
const chat = match.chat;
const completedText = new Map<string, string>();
// Chat deliberately returns no transcript content. Keep the compatibility watch until Transcript uses RemoteEvents.
const watch = await match.client.watchSession(sessionId);
@ -157,8 +97,7 @@ export async function runClient(command: ClientCommand, options: RunClientOption
text: completedText.get(response.operationId) ?? "",
};
} finally {
await Promise.all([...serviceNamespaces].map((services) => services.dispose(BACKGROUND_CONTEXT)));
await Promise.all([...openedClients].map((client) => client.dispose()));
await runtime.dispose();
}
}
@ -168,23 +107,3 @@ function messageText(message: Extract<PromptMessage, { role: "assistant" }>): st
.map((content) => content.text)
.join("");
}
function waitForState<T>(state: RemoteState<T>): Promise<T> {
if (state.value !== undefined) return Promise.resolve(state.value);
return new Promise((resolve) => {
let unsubscribe: (() => void) | undefined;
unsubscribe = state.subscribe((value) => {
queueMicrotask(() => unsubscribe?.());
resolve(value);
});
});
}
function routeFromExplicitPath(path: string): UnixServerRoute {
const name = basename(path);
const serverId = name.endsWith(".sock") ? name.slice(0, -".sock".length) : "";
if (!isServerId(serverId)) {
throw new Error("--connect path must end with <uuidv4-server-id>.sock");
}
return { serverId, path };
}

View file

@ -4,13 +4,15 @@
| Scope | Service | Current slice | Continuation point |
|---|---|---|---|
| server | `SessionDirectory` | replicated state implemented; `events` declared | implement `RemoteEvents` delivery, then publish semantic directory changes |
| server | `SessionDirectory` | replicated state and semantic directory events implemented | add authenticated per-client projection when identity lands |
| server | `SessionManagement` | create, remove, attach, detach implemented | add authenticated workspace authorization |
| session | `Models` | state, selection, thinking, refresh implemented | move provider/auth composition behind plugin facets |
| session | `Chat` | prompt and durable `requestAbort` implemented; queue, resume, compaction, and navigation contracts declared | replace each explicit throwing provider member with its Harness adapter slice |
| session | `Accounts` | state declared; mutations throw | connect the local credential service to the presentation-safe facade |
| session | `Transcript` | revisioned snapshot/events contract declared; both operations throw | implement snapshot/event gap handling and remove the compatibility lane watch |
| session | `Transcript` | revisioned events surface declared; snapshot throws and no Harness event producer is attached | implement snapshot/event gap handling and remove the compatibility lane watch |
`RemoteEvents` currently has contract types and provider member classification only. `remoteEvents()`, local listeners, remote listeners, and emission deliberately throw `ServiceSliceNotImplemented`; there are no event delivery frames yet. `ServerServices.connection` and `SessionServices.attachment` are implemented local control states.
`RemoteEvents` has local and remote ordered delivery, hydration-race buffering, keyed-instance routing, and no replay for late subscribers. `ServerServices.connection` and `SessionServices.attachment` are implemented local control states.
With `PI_EXPERIMENTAL=1`, an interactive `pi client` opens the service-only TUI. It lists the repository-backed `SessionDirectory`, creates or switches Sessions through `SessionManagement`, and selects from the attached worker's `Models` state without invoking Harness drive.
The question dialog, diff review, Git, indexing-job, and canvas examples in `packages/agent/docs/plugins.md` are extension patterns, not built-in coding-agent services. Private references, trace carriers, flow control, the plugin kernel, and host facet contexts remain protocol/host infrastructure slices rather than presentation service tokens.

View file

@ -89,13 +89,25 @@ class SessionServiceNamespace extends RemoteServiceNamespace implements SessionS
const attachmentState = remoteState<SessionAttachmentState>(toSessionAttachmentState(client));
this.attachment = attachmentState;
this.#removeAttachmentListener = client.onAttachmentChange((attachment) => {
attachmentState.set(
attachment === undefined ? { status: "detached" } : { status: "attached", sessionId: attachment.sessionId },
BACKGROUND_CONTEXT,
);
if (attachment !== undefined) {
attachmentState.set({ status: "attaching", sessionId: attachment.sessionId }, BACKGROUND_CONTEXT);
}
this.#transition = this.#transition
.then(() => this.rebind(attachment !== undefined, BACKGROUND_CONTEXT))
.catch((error: unknown) => this.#onError?.(toError(error)));
.then(async () => {
await this.rebind(attachment !== undefined, BACKGROUND_CONTEXT);
attachmentState.set(
attachment === undefined
? { status: "detached" }
: { status: "attached", sessionId: attachment.sessionId },
BACKGROUND_CONTEXT,
);
})
.catch((error: unknown) => {
if (attachment !== undefined) {
attachmentState.set({ status: "degraded", sessionId: attachment.sessionId }, BACKGROUND_CONTEXT);
}
this.#onError?.(toError(error));
});
});
}

View file

@ -27,11 +27,16 @@ export async function provideModelsService(
};
const readCatalog = async (context: Context = BACKGROUND_CONTEXT): Promise<ModelsState["catalog"]> => {
const selected = await harness.getModel(context);
const available = modelRuntime?.getAvailableSnapshot() ?? (selected === undefined ? [] : [selected]);
const available = modelRuntime?.getAvailableSnapshot() ?? [];
const catalog =
selected === undefined ||
available.some((model) => model.provider === selected.provider && model.id === selected.id)
? available
: [...available, selected];
catalogRevision += 1;
return {
revision: catalogRevision,
availableModels: available.map((model) => ({
availableModels: catalog.map((model) => ({
provider: model.provider,
modelId: model.id,
name: model.name,

View file

@ -2,6 +2,7 @@ import {
BACKGROUND_CONTEXT,
type Context,
type ServiceProviderUpdate as CoreServiceProviderUpdate,
type MutableRemoteEvents,
RemoteServiceProvider,
remoteEvents,
remoteState,
@ -47,8 +48,11 @@ export async function createExperimentalServerServices(options: {
let mutationTail = Promise.resolve();
const refreshNow = async (context: Context): Promise<void> => {
const previous = directory.value.sessions;
const sessions = await options.list(context);
revision += 1;
directory.set({ revision, sessions: await options.list(context) }, context);
directory.set({ revision, sessions }, context);
publishDirectoryChanges(previous, sessions, directoryEvents, context);
};
const serialize = <T>(operation: () => Promise<T>): Promise<T> => {
const result = mutationTail.catch(() => {}).then(operation);
@ -105,6 +109,28 @@ export async function createExperimentalServerServices(options: {
};
}
function publishDirectoryChanges(
previous: readonly SessionSummary[],
next: readonly SessionSummary[],
events: MutableRemoteEvents<SessionDirectoryEvent>,
context: Context,
): void {
const previousById = new Map(previous.map((session) => [session.sessionId, session]));
const nextById = new Map(next.map((session) => [session.sessionId, session]));
for (const session of previous) {
if (!nextById.has(session.sessionId)) events.emit({ type: "deleted", sessionId: session.sessionId }, context);
}
for (const session of next) {
const existing = previousById.get(session.sessionId);
if (existing === undefined) events.emit({ type: "created", session }, context);
else if (!sameSessionSummary(existing, session)) events.emit({ type: "changed", session }, context);
}
}
function sameSessionSummary(left: SessionSummary, right: SessionSummary): boolean {
return left.serverId === right.serverId && left.sessionId === right.sessionId && left.createdAt === right.createdAt;
}
function createProviderAttachment(
provider: RemoteServiceProvider,
onRelease: () => void,

View file

@ -65,6 +65,7 @@ import { SettingsManager } from "./core/settings-manager.ts";
import { printTimings, resetTimings, time } from "./core/timings.ts";
import { hasTrustRequiringProjectResources, ProjectTrustStore } from "./core/trust-manager.ts";
import { runClient } from "./experimental/client.ts";
import { runClientTui } from "./experimental/client-tui.ts";
import { startForegroundServer } from "./experimental/server.ts";
import { builtInExtensions } from "./extensions/index.ts";
import { runMigrations, showDeprecationWarnings } from "./migrations.ts";
@ -598,6 +599,15 @@ async function runExperimentalServerCommand(command: ServerCommand): Promise<voi
}
async function runClientCommand(command: ClientCommand): Promise<void> {
if (
command.sessionId === undefined &&
command.prompt === undefined &&
process.stdin.isTTY === true &&
process.stdout.isTTY === true
) {
await runClientTui(command);
return;
}
let streamedText = false;
const result = await runClient(command, {
onEvent(event) {

View file

@ -0,0 +1,93 @@
import { BACKGROUND_CONTEXT, remoteEvents, remoteState } from "@earendil-works/pi-agent-core";
import type { SessionSummary } from "@earendil-works/pi-protocol";
import { describe, expect, test, vi } from "vitest";
import { type ClientTuiServer, ExperimentalClientTui } from "../src/experimental/client-tui.ts";
import type { SessionAttachmentState } from "../src/experimental/services/connection.ts";
import type { ModelsState } from "../src/experimental/services/models.ts";
import type { SessionDirectoryEvent, SessionDirectoryState } from "../src/experimental/services/sessions.ts";
const serverId = "00000000-0000-4000-8000-000000000001";
function session(sessionId: string, createdAt: number): SessionSummary {
return { serverId, sessionId, createdAt };
}
describe("experimental client TUI", () => {
test("creates and switches Sessions, then selects a model using only service facades", async () => {
const directoryState = remoteState<SessionDirectoryState>({ revision: 1, sessions: [session("one", 1)] });
const directoryEvents = remoteEvents<SessionDirectoryEvent>();
const attachment = remoteState<SessionAttachmentState>({ status: "detached" });
const modelsState = remoteState<ModelsState>({
catalog: {
revision: 1,
availableModels: [
{ provider: "test", modelId: "one", name: "Model One", reasoning: false },
{ provider: "test", modelId: "two", name: "Model Two", reasoning: true },
],
},
configuration: { model: { provider: "test", modelId: "one" }, thinkingLevel: "off" },
refresh: { status: "idle" },
});
const create = vi.fn(async () => {
const created = session("two", 2);
directoryState.set({ revision: 2, sessions: [...directoryState.value.sessions, created] }, BACKGROUND_CONTEXT);
directoryEvents.emit({ type: "created", session: created }, BACKGROUND_CONTEXT);
return created;
});
const attach = vi.fn(async (sessionId: string) => {
attachment.set({ status: "attaching", sessionId }, BACKGROUND_CONTEXT);
attachment.set({ status: "attached", sessionId }, BACKGROUND_CONTEXT);
});
const detach = vi.fn(async () => {
attachment.set({ status: "detached" }, BACKGROUND_CONTEXT);
});
const select = vi.fn(async (model: { provider: string; modelId: string }) => {
modelsState.set(
{ ...modelsState.value, configuration: { ...modelsState.value.configuration, model } },
BACKGROUND_CONTEXT,
);
});
const server: ClientTuiServer = {
serverId,
directory: { state: directoryState, events: directoryEvents },
management: { create, attach, detach },
attachment,
models: {
state: modelsState,
async cycleThinking() {},
async refresh() {},
select,
},
};
let finished = false;
const component = new ExperimentalClientTui({
servers: [server],
requestRender() {},
finish() {
finished = true;
},
});
try {
expect(component.render(80).join("\n")).toContain("one");
component.handleInput("\r");
await vi.waitFor(() => expect(component.render(80).join("\n")).toContain("Experimental Models"));
expect(create).toHaveBeenCalledOnce();
expect(attach).toHaveBeenCalledWith("two", expect.anything());
expect(attachment.value).toEqual({ status: "attached", sessionId: "two" });
component.handleInput("\u001b[B");
component.handleInput("\r");
await vi.waitFor(() =>
expect(select).toHaveBeenCalledWith({ provider: "test", modelId: "two" }, expect.anything()),
);
expect(modelsState.value.configuration.model).toEqual({ provider: "test", modelId: "two" });
component.handleInput("\u001b");
expect(component.render(80).join("\n")).toContain("Experimental Sessions");
component.handleInput("\u001b");
expect(finished).toBe(true);
} finally {
component.dispose();
}
});
});

View file

@ -12,7 +12,11 @@ import {
createSessionServiceNamespace,
} from "../src/experimental/services/connection.ts";
import { Models } from "../src/experimental/services/models.ts";
import { SessionDirectory, SessionManagement } from "../src/experimental/services/sessions.ts";
import {
SessionDirectory,
type SessionDirectoryEvent,
SessionManagement,
} from "../src/experimental/services/sessions.ts";
import { Transcript } from "../src/experimental/services/transcript.ts";
import {
configureExperimentalWorkerModel,
@ -309,6 +313,10 @@ describe("experimental durable server composition", () => {
const secondDirectory = secondServices.use(SessionDirectory);
const firstManagement = firstServices.use(SessionManagement);
const secondManagement = secondServices.use(SessionManagement);
const firstEvents: SessionDirectoryEvent[] = [];
const secondEvents: SessionDirectoryEvent[] = [];
const removeFirstEvents = firstDirectory.events.subscribe((event) => firstEvents.push(event));
const removeSecondEvents = secondDirectory.events.subscribe((event) => secondEvents.push(event));
await vi.waitFor(() => {
expect(firstDirectory.state.value?.sessions.map(({ sessionId }) => sessionId)).toEqual(["demo-1", "demo-2"]);
@ -318,6 +326,11 @@ describe("experimental durable server composition", () => {
await vi.waitFor(() => {
expect(firstDirectory.state.value?.sessions.map(({ sessionId }) => sessionId)).toContain("demo-3");
expect(secondDirectory.state.value).toEqual(firstDirectory.state.value);
expect(firstEvents).toContainEqual({
type: "created",
session: expect.objectContaining({ sessionId: "demo-3" }),
});
expect(secondEvents).toEqual(firstEvents);
});
await Promise.all([
@ -334,8 +347,14 @@ describe("experimental durable server composition", () => {
expect(secondDirectory.state.value).toEqual(firstDirectory.state.value);
});
await secondManagement.detach(BACKGROUND_CONTEXT);
await vi.waitFor(() => {
expect(firstEvents).toContainEqual({ type: "deleted", sessionId: "demo-1" });
expect(secondEvents).toEqual(firstEvents);
});
expect(errors).toEqual([]);
removeFirstEvents();
removeSecondEvents();
await Promise.all([firstServices.dispose(BACKGROUND_CONTEXT), secondServices.dispose(BACKGROUND_CONTEXT)]);
});
@ -381,6 +400,9 @@ describe("experimental durable server composition", () => {
provider: "anthropic",
modelId: "claude-sonnet-4-5",
});
expect(firstModels.state.value?.catalog.availableModels).toContainEqual(
expect.objectContaining({ provider: "anthropic", modelId: "claude-sonnet-4-5" }),
);
expect(secondModels.state.value).toEqual(firstModels.state.value);
});
const previousThinking = firstModels.state.value!.configuration.thinkingLevel;

View file

@ -10,7 +10,7 @@ Protocol version `3` carries the current server- and Session-scoped plugin-servi
- singleton and keyed-service subscription snapshots plus ordered state, spawn, and close updates;
- correlated responses, request cancellation, out-of-band attachment updates, attachment-scoped events, and bounded protocol errors.
The transport carries `{ serviceId, instance?, member, args }` calls. A server target contains `{ serverId }`; a Session target contains `{ serverId, sessionId, attachmentId }`. Keyed addresses contain a plugin key and provider-owned generation so delayed calls cannot reach a replacement instance. Service subscriptions return a complete member/state snapshot before update delivery starts; member metadata may declare scaffolded `events` members, but event delivery frames are not implemented yet. Session-directory state and management results contain no working directory, owner, storage, or parent metadata. The server derives private creation fields from authenticated workspace context. Management `attach()` and `detach()` return no routing identifiers; the server publishes the selected live route in an out-of-band `attachment` message. `prompt()` targets that exact route. The real `Session` and `AgentHarness` remain process-local. Disconnecting releases only that presentation's attachment after admitted work settles.
The transport carries `{ serviceId, instance?, member, args }` calls. A server target contains `{ serverId }`; a Session target contains `{ serverId, sessionId, attachmentId }`. Keyed addresses contain a plugin key and provider-owned generation so delayed calls cannot reach a replacement instance. Service subscriptions return a complete member/state snapshot before update delivery starts; member metadata may declare `events` members whose ordered non-durable values use service update frames and are not replayed to late subscribers. Session-directory state and management results contain no working directory, owner, storage, or parent metadata. The server derives private creation fields from authenticated workspace context. Management `attach()` and `detach()` return no routing identifiers; the server publishes the selected live route in an out-of-band `attachment` message. `prompt()` targets that exact route. The real `Session` and `AgentHarness` remain process-local. Disconnecting releases only that presentation's attachment after admitted work settles.
`PromptArguments` contains one serializable `AgentLane.prompt()` overload. `PromptMessage` is the protocol's closed set of built-in message DTOs; application-defined `AgentMessage` extensions are not accepted implicitly. `RunResult` is the wire-safe structural equivalent of the Harness result and contains no JavaScript `Error` instances.

View file

@ -144,6 +144,12 @@ export const ServiceProviderUpdateSchema = Type.Union([
sequence: Type.Integer({ minimum: 1 }),
value: JsonValueSchema,
}),
StrictObject({
type: Type.Literal("event"),
instance: Type.Optional(ServiceInstanceAddressSchema),
member: IdSchema,
event: JsonValueSchema,
}),
StrictObject({
type: Type.Literal("spawned"),
instance: ServiceInstanceSnapshotSchema,

View file

@ -231,6 +231,17 @@ describe("protocol validation", () => {
mode: "singleton",
instances: [{ members: [{ name: "events", kind: "events" }], states: {} }],
});
expect(
parseServerMessage({
type: "service_event",
subscriptionId: "subscription-1",
update: {
type: "event",
member: "events",
event: { type: "changed", revision: 1 },
},
}),
).toMatchObject({ update: { type: "event", member: "events" } });
const keyed: ClientMessage = {
type: "request",
id: "request-1",