diff --git a/packages/agent/README.md b/packages/agent/README.md index c7be41669..81422bd90 100644 --- a/packages/agent/README.md +++ b/packages/agent/README.md @@ -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 diff --git a/packages/agent/docs/plugins.md b/packages/agent/docs/plugins.md index 82712f4af..aaf7bbe39 100644 --- a/packages/agent/docs/plugins.md +++ b/packages/agent/docs/plugins.md @@ -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. diff --git a/packages/agent/src/plugins/services/events.ts b/packages/agent/src/plugins/services/events.ts index 7ec583f65..4cc04c0df 100644 --- a/packages/agent/src/plugins/services/events.ts +++ b/packages/agent/src/plugins/services/events.ts @@ -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 { + subscribe(listener: RemoteEventListener): () => void; } export interface RemoteEventsSource extends MutableRemoteEvents { - readonly [REMOTE_EVENTS_INTERNALS]: RemoteEventsInternals; + readonly [REMOTE_EVENTS_INTERNALS]: RemoteEventsInternals; } -class UnimplementedRemoteEvents implements RemoteEventsSource { - readonly [REMOTE_EVENTS_INTERNALS] = { deliveryImplemented: false } as const; +class MutableRemoteEventsImpl implements RemoteEventsSource { + readonly #listeners = new Set>(); + readonly #sourceListeners = new Set>(); - subscribe(_listener: RemoteEventListener): () => void { - throw new ServiceSliceNotImplemented("RemoteEvents.subscribe"); + subscribe(listener: RemoteEventListener): () => void { + this.#listeners.add(listener); + return () => this.#listeners.delete(listener); } on>( - _type: TType, - _listener: RemoteEventListener>, + type: TType, + listener: RemoteEventListener>, ): () => 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 { + 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(): MutableRemoteEvents { - return new UnimplementedRemoteEvents(); + return new MutableRemoteEventsImpl(); } 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(event: T, type: TType): event is Extract { + return typeof event === "object" && event !== null && "type" in event && event.type === type; } export type { RemoteEvents }; diff --git a/packages/agent/src/plugins/services/namespace.ts b/packages/agent/src/plugins/services/namespace.ts index a8fe5d93e..5b3350258 100644 --- a/packages/agent/src/plugins/services/namespace.ts +++ b/packages/agent/src/plugins/services/namespace.ts @@ -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 { } } +class RemoteEventsReplica implements RemoteEvents { + readonly #listeners = new Set>(); + readonly #reportError: ErrorReporter; + + constructor(reportError: ErrorReporter) { + this.#reportError = reportError; + } + + subscribe(listener: RemoteEventListener): () => void { + this.#listeners.add(listener); + return () => this.#listeners.delete(listener); + } + + on>( + type: TType, + listener: RemoteEventListener>, + ): () => 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; + remove: (() => void) | undefined; + closed: boolean; +} + class MemberSlot { readonly #serviceId: string; readonly #member: string; readonly #invoke: (args: readonly JsonValue[], context: Context) => Promise; readonly #state: RemoteStateReplica; + readonly #events: RemoteEventsReplica; readonly #isActive: () => boolean; + readonly #reportError: ErrorReporter; + readonly #pendingSubscriptions = new Set(); 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): () => 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 { 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(event: T, type: TType): event is Extract { + 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)); } diff --git a/packages/agent/src/plugins/services/provider.ts b/packages/agent/src/plugins/services/provider.ts index 1cd555124..45e9efadf 100644 --- a/packages/agent/src/plugins/services/provider.ts +++ b/packages/agent/src/plugins/services/provider.ts @@ -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; - 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(); - 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`); diff --git a/packages/agent/src/plugins/services/types.ts b/packages/agent/src/plugins/services/types.ts index b66e3066e..7e5753127 100644 --- a/packages/agent/src/plugins/services/types.ts +++ b/packages/agent/src/plugins/services/types.ts @@ -29,9 +29,9 @@ export interface MutableRemoteState extends RemoteState { } export type RemoteEventType = T extends { readonly type: infer TType extends string } ? TType : never; -export type RemoteEventListener = (event: T, context: Context) => void | Promise; +export type RemoteEventListener = (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 { subscribe(listener: RemoteEventListener): () => void; on>( @@ -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 }; diff --git a/packages/agent/test/plugins/services.test.ts b/packages/agent/test/plugins/services.test.ts index 471c509fa..8851bd27b 100644 --- a/packages/agent/test/plugins/services.test.ts +++ b/packages/agent/test/plugins/services.test.ts @@ -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("question-dialog"); +type ActivityEvent = { type: "changed"; revision: number } | { type: "ignored"; revision: number }; interface ActivityService { - readonly events: RemoteEvents<{ type: "changed"; revision: number }>; + readonly events: RemoteEvents; } const Activity = defineRemoteService("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({ 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(); 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[] = []; + 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((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(); + 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(); }); diff --git a/packages/coding-agent/src/experimental/client-runtime.ts b/packages/coding-agent/src/experimental/client-runtime.ts new file mode 100644 index 000000000..c7358fb0b --- /dev/null +++ b/packages/coding-agent/src/experimental/client-runtime.ts @@ -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; +} + +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 { + 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 => { + 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(state: { + readonly value: T | undefined; + subscribe(listener: (value: T) => void): () => void; +}): Promise { + 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 .sock"); + return { serverId, path }; +} diff --git a/packages/coding-agent/src/experimental/client-tui.ts b/packages/coding-agent/src/experimental/client-tui.ts new file mode 100644 index 000000000..7d18d7952 --- /dev/null +++ b/packages/coding-agent/src/experimental/client-tui.ts @@ -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; + readonly attachment: RemoteState; + 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(); + #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 { + 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 { + const runtime = await openClientRuntime(command, options); + const terminal = new ProcessTerminal(); + const tui = new TuiMainScreen(terminal); + let component: ExperimentalClientTui | undefined; + try { + await new Promise((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, sessionId: string): Promise { + 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(); + } + }); + }); +} diff --git a/packages/coding-agent/src/experimental/client.ts b/packages/coding-agent/src/experimental/client.ts index cdf2b2449..97a19413e 100644 --- a/packages/coding-agent/src/experimental/client.ts +++ b/packages/coding-agent/src/experimental/client.ts @@ -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 { - 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(); - const serviceNamespaces = new Set(); - 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(); // 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): st .map((content) => content.text) .join(""); } - -function waitForState(state: RemoteState): Promise { - 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 .sock"); - } - return { serverId, path }; -} diff --git a/packages/coding-agent/src/experimental/services/README.md b/packages/coding-agent/src/experimental/services/README.md index b46e10138..ea56343d6 100644 --- a/packages/coding-agent/src/experimental/services/README.md +++ b/packages/coding-agent/src/experimental/services/README.md @@ -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. diff --git a/packages/coding-agent/src/experimental/services/connection.ts b/packages/coding-agent/src/experimental/services/connection.ts index 7df9c362c..56b3724df 100644 --- a/packages/coding-agent/src/experimental/services/connection.ts +++ b/packages/coding-agent/src/experimental/services/connection.ts @@ -89,13 +89,25 @@ class SessionServiceNamespace extends RemoteServiceNamespace implements SessionS const attachmentState = remoteState(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)); + }); }); } diff --git a/packages/coding-agent/src/experimental/services/models-provider.ts b/packages/coding-agent/src/experimental/services/models-provider.ts index fd5156579..3abe41419 100644 --- a/packages/coding-agent/src/experimental/services/models-provider.ts +++ b/packages/coding-agent/src/experimental/services/models-provider.ts @@ -27,11 +27,16 @@ export async function provideModelsService( }; const readCatalog = async (context: Context = BACKGROUND_CONTEXT): Promise => { 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, diff --git a/packages/coding-agent/src/experimental/services/server.ts b/packages/coding-agent/src/experimental/services/server.ts index 0ac4e78ff..1e833d8fa 100644 --- a/packages/coding-agent/src/experimental/services/server.ts +++ b/packages/coding-agent/src/experimental/services/server.ts @@ -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 => { + 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 = (operation: () => Promise): Promise => { 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, + 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, diff --git a/packages/coding-agent/src/main.ts b/packages/coding-agent/src/main.ts index 6d4e1e92c..de486baf7 100644 --- a/packages/coding-agent/src/main.ts +++ b/packages/coding-agent/src/main.ts @@ -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 { + 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) { diff --git a/packages/coding-agent/test/experimental-client-tui.test.ts b/packages/coding-agent/test/experimental-client-tui.test.ts new file mode 100644 index 000000000..249723f82 --- /dev/null +++ b/packages/coding-agent/test/experimental-client-tui.test.ts @@ -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({ revision: 1, sessions: [session("one", 1)] }); + const directoryEvents = remoteEvents(); + const attachment = remoteState({ status: "detached" }); + const modelsState = remoteState({ + 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(); + } + }); +}); diff --git a/packages/coding-agent/test/experimental-remote-runtime.test.ts b/packages/coding-agent/test/experimental-remote-runtime.test.ts index 803427448..70791ddd8 100644 --- a/packages/coding-agent/test/experimental-remote-runtime.test.ts +++ b/packages/coding-agent/test/experimental-remote-runtime.test.ts @@ -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; diff --git a/packages/protocol/README.md b/packages/protocol/README.md index cb8d5de70..29eea1c16 100644 --- a/packages/protocol/README.md +++ b/packages/protocol/README.md @@ -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. diff --git a/packages/protocol/src/protocol.ts b/packages/protocol/src/protocol.ts index 8499f5b77..e9e06e4a9 100644 --- a/packages/protocol/src/protocol.ts +++ b/packages/protocol/src/protocol.ts @@ -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, diff --git a/packages/protocol/test/protocol.test.ts b/packages/protocol/test/protocol.test.ts index 16fe36dd7..0fdef2a75 100644 --- a/packages/protocol/test/protocol.test.ts +++ b/packages/protocol/test/protocol.test.ts @@ -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",