From e44ecea8fbdc7a0225918b7c35b7290b39a6c07d Mon Sep 17 00:00:00 2001 From: Armin Ronacher Date: Wed, 26 Aug 2026 00:12:57 +0200 Subject: [PATCH] feat(coding-agent): derive facet service routing --- packages/agent/docs/plugin-reloading.md | 21 +- packages/agent/docs/plugins.md | 58 +-- packages/agent/docs/rpc.md | 26 +- packages/agent/src/plugins/services/index.ts | 4 +- .../agent/src/plugins/services/namespace.ts | 12 +- .../agent/src/plugins/services/provider.ts | 28 +- packages/agent/src/plugins/services/types.ts | 30 +- packages/agent/test/plugins/services.test.ts | 14 +- packages/client/src/client.ts | 16 + .../src/experimental/client-runtime.ts | 48 +- .../src/experimental/client-tui.ts | 137 +++-- .../src/experimental/facet-loader.ts | 52 ++ .../coding-agent/src/experimental/facets.ts | 491 ++++++++++-------- .../src/experimental/services/README.md | 8 +- .../src/experimental/services/builtins.ts | 2 - .../src/experimental/services/connection.ts | 395 ++++++++++---- .../src/experimental/services/harness.ts | 4 +- .../experimental/services/models-provider.ts | 2 +- .../experimental/services/server-builtins.ts | 4 - .../src/experimental/services/server.ts | 7 +- .../experimental/services/session-builtins.ts | 10 - .../src/experimental/services/worker.ts | 60 ++- .../src/experimental/session-worker.ts | 2 +- .../test/experimental-chat-service.test.ts | 11 +- .../test/experimental-client-tui.test.ts | 35 +- .../test/experimental-facet-loader.test.ts | 67 +++ .../test/experimental-facets.test.ts | 197 ++++++- .../test/experimental-remote-runtime.test.ts | 30 +- .../test/experimental-service-stubs.test.ts | 11 - .../test/fixtures/faux-session-worker.ts | 3 +- packages/protocol/src/protocol.ts | 21 + packages/protocol/test/protocol.test.ts | 12 + 32 files changed, 1273 insertions(+), 545 deletions(-) create mode 100644 packages/coding-agent/src/experimental/facet-loader.ts delete mode 100644 packages/coding-agent/src/experimental/services/builtins.ts delete mode 100644 packages/coding-agent/src/experimental/services/server-builtins.ts delete mode 100644 packages/coding-agent/src/experimental/services/session-builtins.ts create mode 100644 packages/coding-agent/test/experimental-facet-loader.test.ts diff --git a/packages/agent/docs/plugin-reloading.md b/packages/agent/docs/plugin-reloading.md index 431e3b38b..c8ff1cec7 100644 --- a/packages/agent/docs/plugin-reloading.md +++ b/packages/agent/docs/plugin-reloading.md @@ -26,7 +26,7 @@ These operations are different: - **Rebuild** replays active contributions over a fresh draft and commits a new registry value. Settings, configuration, availability, or source data changed, but plugin module bytes did not. - **Reload** selects a new manifest generation after plugin source or the selected plugin set changed. Every facet is replaced. - **Restart** stops and recreates a concrete server, session worker, or presentation runtime while implementing a reload. -- **Rebind** moves a remote service namespace from one live provider binding to another and hydrates fresh snapshots. +- **Rebind** moves connected services from one live provider binding to another and hydrates fresh snapshots. - **Resume** continues durable Harness work from the Session repository after a worker restart. Most runtime changes should be rebuilds. Only changed source bytes, changed plugin resolution, or changed manifest membership require reload. @@ -55,7 +55,7 @@ The command completes when the system has accepted the reload and begun moving t A reload cannot be cancelled. A later implementation may report progress for each host, but cancellation would create another partially coordinated generation and is not part of the model. -`/reload` must not be implemented as a session plugin service method. Otherwise the retiring service namespace would be asked to destroy itself while its own invocation remained active. A minimal reload and failure-reporting path must remain available even when plugin facets fail to load. +`/reload` must not be implemented as a session plugin service method. Otherwise the retiring service graph would be asked to destroy itself while its own invocation remained active. A minimal reload and failure-reporting path must remain available even when plugin facets fail to load. ## Manifest generations @@ -71,9 +71,11 @@ One generation names one resolved manifest across all facet kinds. Hosts resolve Temporary cross-generation communication is expected. The system does not attempt a distributed lock-step commit. Shared service contracts, durable records, and control messages must tolerate the supported forward-compatibility window. Runtime member-kind validation remains useful but does not replace contract compatibility. +Each process receives one `FacetLoader` configured for its process kind. Calling `load()` returns the complete ordered facet set for that process plus a disposer for the loaded module or sandbox generation. The current loader may return hardcoded built-ins; a manifest-backed loader resolves the same `./server`, `./session`, `./tui`, or `./web` entry points without changing host startup. Reload creates a fresh loader result, replaces the process's complete facet host, then disposes the retired loader result. The loader collects code; facet setup collects service requirements and providers. + ## Dependency discovery and manifest validation -The service dependency graph is generated from facet setup, not reflected from erased TypeScript interfaces and not maintained as a parallel handwritten list. Every service token retains a stable runtime ID. Each host binds its concrete local services before activating the loaded facets. Setup runs synchronously as a declaration phase; service handles remain disconnected until the complete graph validates. +The service dependency graph is generated from facet setup, not reflected from erased TypeScript interfaces and not maintained as a parallel handwritten list. Every service token retains a stable runtime ID. Each host adds a runtime facet that provides its concrete local services through the ordinary facet environment. Setup runs synchronously as a declaration phase; service handles remain disconnected until the complete graph validates. During setup the host records: @@ -84,11 +86,11 @@ use(service) → this facet requires service/singleton observe(service, handler) → this facet requires service/keyed ``` -The facet kind selects its host, lifetime, capabilities, and service source. A facet always uses unqualified `env.use()` and `env.observe()`; server and Session routing are not exposed as namespaces in the facet API. The token supplies the service ID, and the operation supplies singleton or keyed mode. No runtime representation of the generic service interface is needed. +The process kind selects the facet lifetime and host capabilities. A facet always uses unqualified `env.use()` and `env.observe()`; the host routes tokens across local, server, and selected-Session services without exposing namespaces in the facet API. The token supplies the service ID, and the operation supplies singleton or keyed mode. No runtime representation of the generic service interface is needed. First acquisition and provision are setup-only operations. Commands, hooks, event handlers, and activation callbacks use handles acquired during setup. `provideMany()` returns the only `ServiceInstances` handle permitted to add that service's dynamic instances, allowing the provision to appear in the graph before any instance exists. -After setup, the host privately resolves its recorded requirements and provisions. It rejects missing providers, duplicate singleton providers, mode mismatches, and invalid dependency cycles, then records provider-to-consumer edges for lifecycle ordering. The record is kernel machinery, not a plugin-facing plan or a second declaration format. Connected service inventories and allowlists are host/manifest inputs, and their transport bindings must become ready before consuming facets activate. `use()` and `observe()` are hard requirements; optional dependencies need a separate explicit acquisition operation if introduced later. Directly importing another plugin's live implementation bypasses this ownership model and is unsupported. +After setup, the host privately resolves its recorded requirements and provisions. Connections return provider-generated catalogues containing service IDs and modes; the host matches unresolved requirements to exactly one connection, opens generation-owned bindings only for selected services, and rejects missing providers, duplicate offers, mode mismatches, and invalid dependency cycles. The record is kernel machinery, not a plugin-facing plan or a second declaration format. Transport bindings must become ready before consuming facets activate. `use()` and `observe()` are hard requirements; optional dependencies need a separate explicit acquisition operation if introduced later. Directly importing another plugin's live implementation bypasses this ownership model and is unsupported. Removing one plugin constructs a candidate manifest without that plugin and runs the same assembly. If retained facets have unresolved hard requirements, the candidate is rejected before becoming the desired generation. If it is valid, removal still uses a complete-generation reload and simply brings back fewer facets. The removed plugin's durable namespace remains stored. @@ -166,7 +168,7 @@ selected Session S + replacement attachment B A replacement worker receives fresh attachment IDs. Attachment IDs are transport-internal, so preserving one across worker generations has no plugin-level benefit. Fresh IDs reuse the existing stale-route fence: delayed calls and frames for A cannot address B. -`SessionServices.attachment` continues to use the existing presentation-safe states: +`SessionServiceConnection.attachment` continues to use the existing presentation-safe states: ```ts type AttachmentState = @@ -187,12 +189,12 @@ This matches the basic implementation: - `RemoteServiceProvider.provide()` installs one singleton for the provider's lifetime; - plugin-facing `provideMany()` registers keyed ownership during setup, while its `ServiceInstances.add()` handle delegates dynamic instances to the provider and preserves generation fencing when a key is reused; - disposing a provider removes its member listeners and subscriptions; -- a server connection or Session attachment change makes `RemoteServiceNamespace` close old subscriptions and hydrate new ones; and +- a server connection or Session attachment change makes connected services close old subscriptions and hydrate new ones; and - binding revisions and attachment IDs fence delayed frames. There is therefore no need for a singleton-provider replacement frame or public singleton generation merely to reload plugins. Replacing the server connection or Session attachment replaces the provider binding as a whole. A stable remote facade may remain a local object, but it becomes unready and receives a complete snapshot from the replacement before it is usable again. -Changing the manifest may add or remove service tokens. Provider and consumer allowlists are loaded from their respective facets in the new manifest generation. During temporary version skew, a service absent from one side fails normally; forward-compatible plugin bundles must not assume every host has switched already. +Changing the manifest may add or remove service tokens. Each provider catalogue is regenerated from the new facet generation's RPC-capable provisions, and each consumer generation resolves only requirements recorded by its own setup. During temporary version skew, a service absent from one side fails normally; forward-compatible plugin bundles must not assume every host has switched already. Local service implementations are replaced with their host generation as well. Captured local implementation objects from the retired generation are stale and must not be retained by host-global state. @@ -256,7 +258,7 @@ The following claims are explicitly out of scope: Cordis identifies two useful concerns: resources created by a component need lifecycle ownership, and consumers must respond when a provider disappears. Pi keeps both principles: - facet environments own registrations, listeners, subscriptions, and process-local effects; and -- remote namespace bindings become unavailable and rehydrate against replacement providers. +- connected services become unavailable and rehydrate against replacement providers. Pi does not adopt Cordis's in-process component reload as its primary mechanism. Durable Session authority, process-per-session isolation, server replacement, and snapshot-based service rebinding make host-generation replacement the simpler boundary. Inverse disposal remains cleanup, not rollback. @@ -285,6 +287,7 @@ This also follows Cordis's system-boundary caveat: acquiring and releasing a pro A later implementation handoff should cover at least: - one `/reload` selects one new manifest generation for server, session, and presentation facets; +- each process's facet loader returns its complete ordered facet set and disposes failed or retired module generations; - setup-time service acquisition produces the complete internal dependency graph without reflecting on TypeScript types; - missing providers, duplicate singleton providers, mode mismatches, and late dependency acquisition are rejected; - removing a plugin rejects an invalid dependent manifest or performs a complete reload with fewer facets while retaining its durable state; diff --git a/packages/agent/docs/plugins.md b/packages/agent/docs/plugins.md index bac60ae06..1c1373d20 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 `ReplicatedState` 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 facet environment, host service bindings, plugin kernel, references, telemetry propagation, and most other example facets remain illustrative. Plugin reload semantics are specified separately in [`plugin-reloading.md`](plugin-reloading.md). 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 `ReplicatedState` 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 facet environment, runtime facets, plugin kernel, references, telemetry propagation, and most other example facets remain illustrative. Plugin reload semantics are specified separately in [`plugin-reloading.md`](plugin-reloading.md). 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`, `Branch`, `SessionRepo`, invocation `Context`, and telemetry. Read `rpc.md` for wire frames, remote references, subscriptions, and trace carriers, `telemetry.md` for context propagation and cancellation semantics, and `plugin-reloading.md` for manifest-generation replacement and durable reconstruction. @@ -9,8 +9,8 @@ This document assumes you already understand `AgentHarness`, `AgentLane`, `Sessi The new coding agent is assembled from an ordered plugin manifest, not a monolith with an extension API bolted on. Three layers: 1. The **plugin kernel** owns generic lifecycle mechanics: ordered setup, activation, scoped resource ownership, setup-failure cleanup, and reverse-order disposal. It knows nothing about Harness, tools, TUI components, RPC services, connections, or coding-agent policy. -2. An **application host** owns one concrete runtime and constructs the plugin bindings for that runtime. The **session host** normally runs in a dedicated session worker and owns session authority — the real Harness. A **presentation host** (TUI today, web later) owns a user interface. A **server host** owns server-wide authority: session records (`SessionRepo`), session-worker management, authentication, attachment, and routing between presentations and session workers. -3. A **plugin package** implements one feature as one or more host-specific **facets** — setup functions, one per host kind. Each facet receives only the bindings natural to its host instance and process. +2. An **application host** owns one concrete runtime and contributes a runtime facet that provides its concrete services. The **session host** normally runs in a dedicated session worker and owns session authority — the real Harness. A **presentation host** (TUI today, web later) owns a user interface. A **server host** owns server-wide authority: session records (`SessionRepo`), session-worker management, authentication, attachment, and routing between presentations and session workers. +3. A **plugin package** implements one feature as one or more host-specific **facets** — setup functions, one per host kind. Each facet can use only the services available in its host instance and process. The initial topology has one server and no server-to-server links: @@ -24,7 +24,7 @@ server A presentation and a session worker each connect to the server. There is no direct presentation→session-worker connection; the server routes service calls to the selected worker. The server lists and manages only its own sessions. Multi-server routing and server hierarchies are out of scope. -A session worker normally owns one session, and each session facet is instantiated for that session. A server facet is instantiated once per server process and is shared across every session and presentation connected to it. Server facets should therefore be rare and limited to inherently server-wide concerns. Per-session feature state belongs in session facets; dedicated workers provide the preferred lifecycle, crash, and state isolation. Future co-location may preserve the same logical bindings without changing what objects a facet can access. +A session worker normally owns one session, and each session facet is instantiated for that session. A server facet is instantiated once per server process and is shared across every session and presentation connected to it. Server facets should therefore be rare and limited to inherently server-wide concerns. Per-session feature state belongs in session facets; dedicated workers provide the preferred lifecycle, crash, and state isolation. Future co-location may preserve the same logical service graph without changing what objects a facet can access. **Host** and **client** are roles per connection, not fixed process kinds. The server hosts presentation and session-worker connections. A session worker serves its provided session services and may consume server-provided services over the same RPC binding mechanism. "Client" below always names the connection role, never a kind of plugin. @@ -34,8 +34,8 @@ A session worker normally owns one session, and each session facet is instantiat - **One feature stays coherent.** The question plugin's tool, dialog, and renderer ship in one package around one JSON contract, yet each facet is host-native code. - **A new surface is presentation-only work.** A web facet for the question dialog or the session picker registers against existing tokens; session and server code do not change. - **Server state stays server-wide.** A server facet is shared by all sessions and clients, so features use one only when their authority is inherently global to that server. -- **No privileged built-ins.** Built-ins and third-party plugins receive identical bindings, so shipping the product continuously exercises the extension API. -- **Testable in pieces.** A facet tests against its host bindings, a contract against loopback, and a routed TUI → server → session-worker path against a real transport — independently. +- **No privileged built-ins.** Built-ins, runtime facets, and third-party plugins use the same environment, so shipping the product continuously exercises the extension API. +- **Testable in pieces.** A facet tests against service-providing fixtures, a contract against loopback, and a routed TUI → server → session-worker path against a real transport — independently. ## One feature, several host facets @@ -46,20 +46,14 @@ interface CodingAgentPlugin { readonly id: string; readonly server?: PluginFacet; readonly session?: PluginFacet; - readonly tui?: { - readonly server?: PluginFacet; - readonly session?: PluginFacet; - }; - readonly web?: { - readonly server?: PluginFacet; - readonly session?: PluginFacet; - }; + readonly tui?: PluginFacet; + readonly web?: PluginFacet; } type PluginFacet = (env: FacetEnvironment) => void; ``` -`PluginFacet` is the only plugin-facing shape the generic kernel needs. The host placement is the facet kind: it selects the lifetime and the local or remote service sources available through `env.use()`. Facets are optional: a `models.json` plugin has only a Session facet, a terminal theme only presentation facets, and the Session directory has a server provider facet plus a server TUI facet. Setup is a synchronous declaration phase; asynchronous initialization belongs in `onActivate`. Examples use an illustrative `definePlugin()` helper returning a `CodingAgentPlugin` — the loaded, in-process shape. +`PluginFacet` is the only plugin-facing shape the generic kernel needs. Each process loads its complete facet set: the server loads server facets, each Session worker loads Session facets, and each presentation loads its TUI or web facets. The host resolves every `env.use()` across facet-provided and connected services. Facets are optional: a `models.json` plugin has only a Session facet, a terminal theme only a TUI facet, and the Session directory has a server facet plus a TUI facet. Setup is a synchronous declaration phase; asynchronous initialization belongs in `onActivate`. Examples use an illustrative `definePlugin()` helper returning a `CodingAgentPlugin` — the loaded, in-process shape. A package keeps shared wire contracts separate from host dependencies: @@ -69,7 +63,7 @@ question-plugin/ session.ts dialog-service authority and tool contribution; imports agent/session code tui.ts terminal dialog and renderer; imports TUI code web.ts optional browser dialog and renderer - index.ts loopback bundle; production hosts resolve per-host entry points + index.ts loopback bundle; production hosts resolve one entry point per process kind ``` The browser build never imports `session.ts`; the session process never imports TUI or DOM code. @@ -171,7 +165,7 @@ type ConnectionState = | { status: "disconnected"; since: string; reason: string; retryAt: string | null }; ``` -After transport setup, each host gives the kernel its loaded facets and binds its concrete local services. The kernel invokes each facet against the resulting service graph: +After transport setup, each host gives the kernel its loaded facets plus a runtime facet that provides concrete host services. The kernel invokes every facet against the resulting service graph: ```ts interface FacetLifecycle { @@ -194,10 +188,10 @@ Out of scope: arbitrary undeclared object remoting, serialized functions/classes Facets communicate across processes through **services**. One token type gives a service contract its identity: ```ts -function defineService(id: string): Service; +function defineService(id: string, options?: { rpc?: boolean }): Service; ``` -The declaration lives in the shared contract module and creates nothing. `provide(service, implementation)` exposes one singleton. `provideMany(service)` registers ownership of a multi-instance service during facet setup and returns an owned collection whose later `add(key, implementation)` calls expose instances. Consumers select the same modes with `use(service)` or `observe(service, handler)`. Within one facet generation, a token must stay in one mode: mixing `provide`/`use` with `provideMany`/`observe` is an assembly or protocol error. +The declaration lives in the shared contract module and creates nothing. Services are RPC-capable by default; a process-local token declares `{ rpc: false }`. `provide(service, implementation)` adds one singleton to the host service graph. `provideMany(service)` registers ownership of a multi-instance service during facet setup and returns an owned collection whose later `add(key, implementation)` calls add instances. The host automatically publishes provided RPC-capable services across its process boundary. Consumers select the same modes with `use(service)` or `observe(service, handler)`. Within one facet generation, a token must stay in one mode: mixing `provide`/`use` with `provideMany`/`observe` is an assembly or protocol error. ```ts interface ServiceInstances { @@ -205,13 +199,13 @@ interface ServiceInstances { } ``` -TypeScript types cannot produce runtime member metadata. Plugin authors nevertheless declare no parallel member descriptor. When `provide()` or `ServiceInstances.add()` receives an implementation, the service runtime classifies functions as remote methods and recognizes branded `ReplicatedState` and `RemoteEvents` values. It rejects unsupported members and announces the resulting member table over the transport. +TypeScript types cannot produce runtime member metadata. Plugin authors nevertheless declare no parallel member descriptor. When an exposed `provide()` or `ServiceInstances.add()` implementation reaches the remote-service boundary, the runtime classifies functions as remote methods and recognizes branded `ReplicatedState` and `RemoteEvents` values. It rejects unsupported members and announces the resulting member table over the transport. Services used only inside one host may use arbitrary object contracts. `use()` on a singleton returns a stable lazy proxy synchronously, even before a remote provider is attached. Member access creates local method, state, or event slots as they are used; attachment validates those slots against the provider-announced kinds. A mismatch is an assembly or protocol error. This runtime mechanism is implemented once by the host rather than repeated in every service declaration. ### Dependency declaration and assembly -Service API calls made during facet setup are the dependency declarations. The kernel does not reflect on erased TypeScript interfaces, and plugin authors do not maintain parallel `requires` and `provides` lists. A `Service` or `LocalService` retains its stable ID at runtime; the API call supplies the mode, while the facet kind selects the host and service source. +Service API calls made during facet setup are the dependency declarations. The kernel does not reflect on erased TypeScript interfaces, and plugin authors do not maintain parallel `requires` and `provides` lists. A `Service` retains its stable ID at runtime, and the API call supplies the mode. `use()` and `observe()` initially return source-independent disconnected handles. After all setup completes, the host matches unresolved requirements against provider-generated connection catalogues and binds each token to its local provision or exactly one connection. The host records a private generation-scoped ledger: @@ -220,20 +214,20 @@ env.provide(Models, implementation) → @pi/providers-builtin:session provides pi.models/singleton env.use(Models) -→ @pi/model-selection:tui/session requires pi.models/singleton +→ @pi/model-selection:tui requires pi.models/singleton env.provideMany(QuestionDialogs) → @pi/question:session provides pi.question-dialog/keyed env.observe(QuestionDialogs, handler) -→ @pi/question:tui/session requires pi.question-dialog/keyed +→ @pi/question:tui requires pi.question-dialog/keyed ``` The first `provide()`, `provideMany()`, `use()`, or `observe()` for a token must occur during facet setup. Commands, hooks, event handlers, and activation callbacks use handles acquired during setup; they cannot introduce an undeclared service dependency later. Dynamic instances use the setup-owned `ServiceInstances` handle, so adding and closing instances do not change the graph. -After every facet has registered, the host resolves requirements to provisions, rejects missing providers, duplicate singleton owners, singleton/keyed mismatches, and invalid dependency cycles, derives service allowlists, and records consumer-to-provider edges for lifecycle ordering. `use()` and `observe()` declare hard requirements; optional dependencies require a future distinct acquisition API rather than inference from call failure. The ledger and resulting graph are private kernel machinery, not a plugin-facing plan or second declaration format. +After every facet has registered, the host generates its outgoing catalogue from RPC-capable provisions, obtains catalogues from its connections, resolves requirements to local or connected provisions, rejects missing providers, duplicate offers or singleton owners, singleton/keyed mismatches, invalid dependency cycles, and invalid RPC service implementations, then records consumer-to-provider edges for lifecycle ordering. `use()` and `observe()` declare hard requirements; optional dependencies require a future distinct acquisition API rather than inference from call failure. The ledger and resulting graph are private kernel machinery, not a plugin-facing plan or second declaration format. -Only dependencies acquired through host bindings belong to this lifecycle graph. Importing another plugin's live implementation bypasses ownership and is unsupported. The module loader separately owns the ordinary source import graph. Complete-generation reload avoids needing that module graph for partial invalidation; see [`plugin-reloading.md`](plugin-reloading.md). +Only dependencies acquired through `env.use()` or `env.observe()` belong to this lifecycle graph. Importing another plugin's live implementation bypasses ownership and is unsupported. The module loader separately owns the ordinary source import graph. Complete-generation reload avoids needing that module graph for partial invalidation; see [`plugin-reloading.md`](plugin-reloading.md). The models service — the authority behind the model picker and thinking-level control — exercises methods, replicated state, and multiple consumers. @@ -360,7 +354,7 @@ Multi-instance services use `provideMany()` and `observe()`. The service is empt An added instance member has structural identity `(service, key, generation, member)`. Its `ReplicatedState` members therefore need no independent IDs. The instance directory is control-plane metadata, not a plugin-visible `ReplicatedState` containing proxies. Switching sessions aborts all observed instance tasks before hydrating the selected session's current instances. -Every facet uses the same unqualified `env.use()` and `env.observe()` operations. A server presentation facet receives the connected server as its service source; a Session presentation facet receives the selected Session. A provider facet resolves services within its own host. Transport binding and routing remain host infrastructure rather than facet API. +Every facet uses the same unqualified `env.use()` and `env.observe()` operations. A presentation host combines its local services with connected server and selected-Session services, then routes each token internally. Provider facets resolve services from the same host graph. Transport binding and routing remain host infrastructure rather than facet API. ## What each facet kind grants @@ -462,7 +456,7 @@ const Tui = defineService("pi.local.tui"); `acquireModal()` waits in one presentation-owned queue and holds the modal slot across a multi-step interaction. Its signal removes a queued request or dismisses an active one, and `close()` is idempotent. `select()` is the one-step acquire/select/close convenience. Both return selected values directly, so feature code never recovers identity from a display label. -A server TUI facet and a Session TUI facet can use the same host-local TUI services but run in different facet generations. The server generation binds `env.use(SessionDirectory)` to the connected server. The Session generation binds `env.use(Models)` to the selected Session. While detached, Session calls fail with `not_attached` and replicated state has no value. Connection and attachment health are host-local services because they describe presentation control state. A future web host similarly binds local services for routes, views, and DOM dialogs. Its server and Session facets still use unqualified service operations. +The TUI loads all of its facets into one generation. Its host routes `env.use(SessionDirectory)` to the connected server and `env.use(Models)` to the selected Session. While detached, Session calls fail with `not_attached` and replicated state has no value. Connection and attachment health are host-local services because they describe presentation control state. A future web host similarly binds local services for routes, views, and DOM dialogs. Its server and Session facets still use unqualified service operations. Concretely, the minimum chat slice exposes `prompt(request, context)` returning `{ accepted, operationId, error }`, plus `requestAbort(operationId, context)`, and implements it with direct local lane capabilities. The experimental `Chat` contract also predeclares the later steer, follow-up, next-run, queue cancellation, resume, compaction, and navigation presentation operations; those members throw `ServiceSliceNotImplemented` until their Harness adapter slices land: @@ -480,7 +474,7 @@ session(env) { } ``` -Its Session TUI facet consumes `Chat` through `env.use()` exactly as the model picker consumes `Models`. Neither reveals the Harness object behind them; there is no universal remote Harness for arbitrary plugins. `rpc.md` may still define generic Harness proxies for other trusted integrations (an IDE bridge, an orchestrator) — deliberate, separate exposures, not the plugin boundary. +Its TUI facet consumes `Chat` through `env.use()` exactly as the model picker consumes `Models`. Neither reveals the Harness object behind them; there is no universal remote Harness for arbitrary plugins. `rpc.md` may still define generic Harness proxies for other trusted integrations (an IDE bridge, an orchestrator) — deliberate, separate exposures, not the plugin boundary. ## Local services and narrow remote facades @@ -789,7 +783,7 @@ The TUI facet consumes a service provided by the one connected server. There is 1. the server authorizes the client for one of its managed sessions; 2. it closes the client's previous session-scoped requests, subscriptions, references, and observed instance tasks; -3. it binds the Session facet generation's service source to the selected Session worker; +3. it binds the presentation host's Session services to the selected Session worker; 4. the Session worker hydrates singleton state and current keyed instances from complete fresh snapshots; attachment state becomes `attached`. Session service handles are stable across switches: a proxy returned once by a Session facet's `env.use(Models)` keeps working against the new Session, and `env.observe(QuestionDialogs, ...)` reconciles the new Session's instances. Frames belonging to closed subscriptions or requests are dropped. @@ -999,14 +993,14 @@ Disconnect behavior, from the plugin author's perspective: - **A presentation disconnects.** Its server aborts the client's active requests, closes its observed instance tasks and other session-routed resources, and the selected session's `onClientDetach` fires. Session-owned work continues per application policy. An added question dialog remains Session-owned; the disconnected presentation loses its proxy and any in-flight `submitAnswer()` call fails. - **A session worker disconnects or crashes.** Its server fails routed in-flight calls with `session_unavailable`, closes that worker's observed instance tasks, and updates the managed record's status. Attached presentations see `attachment.status === "degraded"` while the server connection stays healthy, so the picker still works and the user can attach elsewhere. -- **A process loses its server connection.** Its server and Session facet service sources become unavailable. A Session worker loses server services and all attached presentations at once; unattended-Session policy decides whether it exits. +- **A process loses its server connection.** Its connected server and Session services become unavailable. A Session worker loses server services and all attached presentations at once; unattended-Session policy decides whether it exits. - Reconnect and reattach always hydrate from a fresh authoritative snapshot; prior remote references are invalid unless the exposure has explicitly session-stable identity. **Never blindly replay a mutation after an uncertain disconnect** — a replayed `select()` is harmless, a replayed `prompt()` is not. Reconnect, hydrate, and reconcile, or design the operation around a stable operation ID with explicit lookup semantics. Errors cross the wire as a JSON envelope `{ code, message, data? }` with stable codes. Expected service errors use stable codes or result values; unexpected exceptions become an internal error with safe metadata — no stacks or sensitive causes by default. Experimental scaffold members use `service_not_implemented` until their implementation slice lands. Cancellation, disconnect, unknown service/method, invalid arguments/results, stale references, `not_attached`, `not_authorized`, `unknown_session`, and `session_unavailable` need distinct codes. Security rules every providing host (session and server alike) must enforce: -- service IDs are allowlisted by trusted manifests; `provide()`, `provideMany()`, and `ServiceInstances` handles expose only implementation functions and branded replicated-state/event members, instance generations are host-owned, and local services are never discoverable remotely; +- RPC-capable service IDs come from trusted loaded service tokens; the remote boundary accepts only implementation functions and branded replicated-state/event members, instance generations are host-owned, and `{ rpc: false }` services are never discoverable remotely; - business arguments, results, state, and events are validated as JSON; protocol envelopes cannot be forged as ordinary values; - clients cannot choose context position, server typed values, telemetry parents, or cancellation targets other than their own request IDs; - credentials, prompts, completions, tool arguments/results, and filesystem contents are not exposed unless an explicit contract permits them; state snapshots contain only client-safe data; @@ -1025,7 +1019,7 @@ Before this becomes normative: - the exact minimal kernel contract (environment shape, phase ordering, failure policy), the built-in manifest, and the concrete server/Session/TUI host services; - the logical-manifest format, per-host entry-point resolution, and cross-process version pinning; - whether directory state is projected per client (workspace-scoped snapshots) or one presentation-safe value plus method authorization; -- multiple selected Sessions per presentation connection (a multi-pane web UI) — deferred; it changes how Session facet generations select service sources; +- multiple selected Sessions per presentation connection (a multi-pane web UI) — deferred; it requires explicit routing for services from more than one selected Session; - authentication of presentations and session workers, and protocol version negotiation; - the exact lazy `Service` member-discovery and provider-kind-validation API, and the context-position and JSON-safe optional-argument policy from `rpc.md`; - whether `ReplicatedState` is generic RPC infrastructure or host infrastructure; snapshot granularity; exact hydration/delivery-context semantics; diff --git a/packages/agent/docs/rpc.md b/packages/agent/docs/rpc.md index 9719fe8b4..58172969e 100644 --- a/packages/agent/docs/rpc.md +++ b/packages/agent/docs/rpc.md @@ -20,11 +20,11 @@ Do not serialize `Context`, `AbortSignal`, telemetry objects, callbacks, tools, ## Service contracts and typed facades -A remote-service token is a shared TypeScript contract and stable service ID. It is not a generated descriptor and creates no provider. `provide()` exposes one singleton implementation. `provideMany()` registers one keyed-service owner during facet setup and returns a `ServiceInstances` handle whose later `add()` calls expose instances. A token has one mode in one service source: mixing singleton and keyed use is an error. +A service token is a shared TypeScript contract and stable service ID. It is not a generated descriptor and creates no provider. Tokens are RPC-capable by default; process-local tokens declare `{ rpc: false }`. `provide()` adds one singleton implementation to the host graph. `provideMany()` registers one keyed-service owner during facet setup and returns a `ServiceInstances` handle whose later `add()` calls add instances. The host automatically publishes every provided RPC-capable service. A token has one mode in one host service graph: mixing singleton and keyed use is an error. The provider must make enough control-plane information available for the host to validate an accessed member as a method, `ReplicatedState`, or `RemoteEvents`. The consumer must be able to obtain the member name from ordinary property access—for example, a JavaScript `Proxy` receives `"state"` for `models.state` and `"refresh"` for `models.refresh(context)`. How a disconnected lazy proxy represents an unresolved member, and the exact provider-kind validation exchange, are implementation details. -Local and remote `use()` both return a stable, lazy typed facade shared by consumers of that token. During synchronous facet setup the facade is disconnected, so setup can capture it but cannot invoke methods, read state, or register member subscriptions. After assembly, a local facade binds through an in-process connection and a remote facade binds through its host namespace. While disconnected, invoking a method fails and state remains unhydrated; no call is queued merely because it was made through a proxy. +Local and remote `use()` both return a stable, lazy typed facade shared by consumers of that token. During synchronous facet setup the facade is disconnected, so setup can capture it but cannot invoke methods, read state, or register member subscriptions. After assembly, a local facade resolves to its facet-provided implementation and a remote facade binds through the host's connected services. While disconnected, invoking a method fails and state remains unhydrated; no call is queued merely because it was made through a proxy. Remote methods return promises and accept and return strict JSON apart from their declared `Context`; `void` is a successful response without a result field. A private returned reference is a separately validated control envelope, not a business JSON value. The client removes the context before transport and the receiving host constructs a fresh local context. The contract position is host-controlled and must be consistent; the current examples use one required trailing `Context`. Business absence is JSON `null` or an options object, never transported `undefined`. @@ -39,19 +39,19 @@ Type erasure does not hide service identity: every service token retains its sta - `use()` records a singleton requirement; and - `observe()` records a keyed requirement. -The facet kind selects its host and service source. Facets always call unqualified `env.use()` or `env.observe()`; routing is not encoded in the call. The method supplies the mode and the token supplies the ID. The host therefore needs no reflection over the erased `T` and no handwritten parallel dependency list. +Facets always call unqualified `env.use()` or `env.observe()`; routing is not encoded in the call. These operations return source-independent disconnected handles during setup. After setup, each connection returns its provider-generated service catalogue, and the host binds every requirement to its local provision or exactly one connected provider. The method supplies the mode and the token supplies the ID. The host therefore needs no reflection over the erased `T` and no handwritten parallel dependency list. First acquisition or provision is permitted only during facet setup. Later commands, hooks, events, and activation callbacks use setup-acquired singleton facades, observer registrations, or `ServiceInstances` handles. In particular, dynamic instances are added through the handle returned by `provideMany()`; a late direct addition cannot introduce a previously undeclared provision. -After setup, each host privately resolves the recorded requirements and provisions to reject missing or duplicate providers and mode mismatches and to derive lifecycle edges. This internal service graph is distinct from the module loader's source import graph; it is not a plugin-authored or plugin-visible plan. +After setup, each host privately resolves the recorded requirements against local provisions and connection catalogues to reject missing providers, duplicate remote offers, and mode mismatches and to derive lifecycle edges. A selected-Session connection with no live attachment may provisionally accept unresolved requirements as unavailable; attachment validates them against the worker's generated catalogue and caches that catalogue for later detached generations. Connection bindings are generation-owned and include only selected requirements, so failed or retired generations release their subscriptions without disposing the underlying transport connection. This internal service graph is distinct from the module loader's source import graph; it is not a plugin-authored or plugin-visible plan. ## Bindings and identity -Server facets consume services from the connected server generation. Session facets consume services from the selected Session generation. Both use the same unqualified environment API. A Session-service call never accepts a client-selected durable `sessionId`; the server authorizes attachment and routes the selected Session binding to its worker. +A presentation host combines services from its connected server and selected Session in one graph. All of its facets use the same unqualified environment API. A Session-service call never accepts a client-selected durable `sessionId`; the server authorizes attachment and routes the selected Session binding to its worker. ### Server control plane -Session listing and management are ordinary server singleton services, not generic remote `Session` methods. `SessionDirectory` exposes presentation-safe session summaries as replicated state plus semantic directory events. `SessionManagement` exposes `create`, `remove`, `attach`, and `detach` methods. A server presentation facet consumes both through its environment: +Session listing and management are ordinary server singleton services, not generic remote `Session` methods. `SessionDirectory` exposes presentation-safe session summaries as replicated state plus semantic directory events. `SessionManagement` exposes `create`, `remove`, `attach`, and `detach` methods. A TUI or web facet consumes both through its environment: ```ts const directory = env.use(SessionDirectory); @@ -60,7 +60,7 @@ const management = env.use(SessionManagement); The server derives workspace and client authority from its locally authenticated identity, not from the summary or method arguments. It may project directory state per client; either way, summaries never expose server-private fields such as owner IDs or working directories. -`management.attach(sessionId, context)` changes the service binding used by the presentation's Session facets. The server authorizes the requested managed Session, closes the presentation's previous Session-scoped requests, subscriptions, references, and observer tasks, binds the Session service source to the worker, then hydrates its singleton state and keyed-instance directory. Attachment state is host control state reporting this selection and its health; it is not a directory service. `detach()` performs the same cleanup without a replacement. +`management.attach(sessionId, context)` changes the selected-Session services in the presentation host. The server authorizes the requested managed Session, closes the presentation's previous Session-scoped requests, subscriptions, references, and observer tasks, binds the presentation's Session services to the worker, then hydrates their singleton state and keyed-instance directory. Attachment state is host control state reporting this selection and its health; it is not a directory service. `detach()` performs the same cleanup without a replacement. The host needs a private, host-owned binding incarnation for that route. It changes when the presentation attaches, detaches, switches session, or replaces a failed worker. Its representation is deliberately unspecified. The binding prevents a delayed frame for the old selected session from being applied to the new one; it is not a plugin-visible service value or a substitute for authorization. @@ -74,7 +74,7 @@ There is no separately discoverable state ID. An added instance key is a plugin- ## Calls, context, and routing -A call carries enough control-plane information to select a namespace, service, optional keyed instance, and member, plus a request ID, JSON arguments, and trace carrier. The server may parse those control-plane fields to route a session call, but it does not parse plugin business payloads or load plugin contracts. The service endpoint validates the member and values, creates a request-local abort controller and `Context`, installs authenticated client identity as a server-created context value, and invokes the local implementation. +A call carries enough control-plane information to select a provider binding, service, optional keyed instance, and member, plus a request ID, JSON arguments, and trace carrier. The server may parse those control-plane fields to route a session call, but it does not parse plugin business payloads or load plugin contracts. The service endpoint validates the member and values, creates a request-local abort controller and `Context`, installs authenticated client identity as a server-created context value, and invokes the local implementation. The client maps `context.abortSignal` to cancellation of that one request. Disconnect cancels that connection's active calls and closes its subscriptions. Neither action cancels service-owned work or writes durable Harness cancellation. Per-client request correlation reaches the worker so request IDs from different presentations cannot collide. @@ -111,7 +111,7 @@ Three cancellation domains remain separate: aborting one RPC invocation; explici ## Security and lifecycle -Only manifest-allowlisted remote service IDs may be provided or registered as multi-instance services. Only the owning `ServiceInstances` handle may add instances. Local services are never discoverable remotely. Providers validate member kinds and every JSON business value; clients cannot forge control envelopes as ordinary values, choose instance generations, select another session's route, choose server context values, or cancel another client's request. +Only loaded service tokens marked RPC-capable may be registered at the remote boundary. Only the owning `ServiceInstances` handle may add instances. Services marked `{ rpc: false }` are never discoverable remotely. Providers validate member kinds and every JSON business value; clients cannot forge control envelopes as ordinary values, choose instance generations, select another session's route, choose server context values, or cancel another client's request. The server authenticates its connection and authorizes attachment. It reconstructs client identity locally at the service endpoint; ordinary arguments never carry authority. Credentials, prompts, completions, tool data, filesystem contents, and other sensitive values cross only through an explicit presentation-safe contract. @@ -121,7 +121,7 @@ Facet environments own registrations, added service instances, subscriptions, ob Test the plugin-facing semantics over loopback and a real framed transport: -- setup-time dependency-ledger ownership, rejection of late acquisition, local and remote `use()`, keyed-provider ownership, singleton/keyed mode validation, manifest allowlisting, lazy member access, and local services remaining unreachable remotely; +- setup-time dependency-ledger ownership, rejection of late acquisition, local and remote `use()`, keyed-provider ownership, singleton/keyed mode validation, token-driven RPC publication, lazy member access, and `{ rpc: false }` services remaining unreachable remotely; - strict JSON boundaries, method context reconstruction, request cancellation isolation, and trace propagation without serializing context values; - server/Session facet isolation, authorized attach, selected-Session switching, stale-frame rejection, and worker-side per-client request correlation; - cold and hydrated `ReplicatedState`, snapshot/update race freedom, delivery contexts, stale display on reconnect, and clearing on provider switch; @@ -181,7 +181,7 @@ serverContext.provide(SessionManagement, { }); ``` -A server presentation facet renders and selects Sessions: +A presentation facet renders and selects Sessions: ```ts setup(env) { @@ -202,7 +202,7 @@ setup(env) { } ``` -A Session presentation facet independently acquires Session services through the same API: +Another facet in the same presentation acquires Session services through the same API: ```ts setup(env) { @@ -211,4 +211,4 @@ setup(env) { } ``` -The presentation never routes `models` with a selected `sessionId`; the facet generations select the server or Session service source and transport retains the routing state. The server closes the prior Session binding's resources before hydrating the new one. +The presentation never routes `models` with a selected `sessionId`; its host routes each service token and transport retains the selected-Session binding. The server closes the prior Session binding's resources before hydrating the new one. diff --git a/packages/agent/src/plugins/services/index.ts b/packages/agent/src/plugins/services/index.ts index c0cd17c2b..ae3ef9d63 100644 --- a/packages/agent/src/plugins/services/index.ts +++ b/packages/agent/src/plugins/services/index.ts @@ -33,17 +33,19 @@ export { type RemoteServiceConnection, type RemoteServiceContract, type RemoteServiceInstance, - type RemoteServiceNamespaceApi, type RemoteServiceNamespaceOptions, type RemoteServiceSubscription, + type RemoteServices, type ReplicatedState, type Service, type ServiceCall, + type ServiceCatalogueEntry, type ServiceInstanceAddress, type ServiceInstanceSnapshot, type ServiceMemberDescription, type ServiceMemberKind, type ServiceMode, + type ServiceProviderDefinition, type ServiceProviderSubscription, type ServiceProviderUpdate, ServiceSliceNotImplemented, diff --git a/packages/agent/src/plugins/services/namespace.ts b/packages/agent/src/plugins/services/namespace.ts index 3f2eba6b7..4ea743682 100644 --- a/packages/agent/src/plugins/services/namespace.ts +++ b/packages/agent/src/plugins/services/namespace.ts @@ -10,9 +10,9 @@ import { type RemoteEventType, type RemoteServiceConnection, type RemoteServiceInstance, - type RemoteServiceNamespaceApi, type RemoteServiceNamespaceOptions, type RemoteServiceSubscription, + type RemoteServices, type ReplicatedState, type Service, type ServiceInstanceAddress, @@ -648,9 +648,9 @@ class KeyedBinding { } } -export class RemoteServiceNamespace implements RemoteServiceNamespaceApi { +export class RemoteServiceNamespace implements RemoteServices { readonly #connection: RemoteServiceConnection; - readonly #allowlist: ReadonlySet; + readonly #allowlist = new Set(); readonly #reportError: ErrorReporter; readonly #modes = new Map(); #assertAccess: () => void; @@ -665,7 +665,7 @@ export class RemoteServiceNamespace implements RemoteServiceNamespaceApi { this.#connection = options.connection; const ids = options.services.map(({ id }) => id); if (new Set(ids).size !== ids.length) throw new TypeError("Remote service namespace has duplicate service IDs"); - this.#allowlist = new Set(ids); + for (const id of ids) this.#allowlist.add(id); this.#reportError = options.onError ?? (() => {}); this.#assertAccess = options.assertAccess ?? (() => {}); this.#bound = options.bound ?? true; @@ -729,6 +729,10 @@ export class RemoteServiceNamespace implements RemoteServiceNamespaceApi { this.#assertAccess = assertAccess; } + async activate(context: Context): Promise { + await this.ready(context); + } + async ready(context: Context): Promise { if (this.#disposed) throw new Error("Remote service namespace is disposed"); while (true) { diff --git a/packages/agent/src/plugins/services/provider.ts b/packages/agent/src/plugins/services/provider.ts index 2c3db3e24..23d721640 100644 --- a/packages/agent/src/plugins/services/provider.ts +++ b/packages/agent/src/plugins/services/provider.ts @@ -9,10 +9,12 @@ import { type RemoteServiceContract, type Service, type ServiceCall, + type ServiceCatalogueEntry, type ServiceInstanceAddress, type ServiceInstanceSnapshot, type ServiceMemberDescription, type ServiceMode, + type ServiceProviderDefinition, type ServiceProviderSubscription, type ServiceProviderUpdate, type ServiceSubscriptionSnapshot, @@ -72,13 +74,33 @@ interface ServiceRegistration { export class RemoteServiceProvider { readonly #allowlist: ReadonlySet; + readonly #catalogue: readonly ServiceCatalogueEntry[]; readonly #registrations = new Map(); #disposed = false; - constructor(services: readonly { readonly id: string }[]) { - const ids = services.map(({ id }) => id); - if (new Set(ids).size !== ids.length) throw new TypeError("Remote service allowlist contains duplicate IDs"); + constructor(entries: readonly (ServiceProviderDefinition | { readonly id: string })[]) { + const definitions = entries.map( + (entry): ServiceProviderDefinition => ("service" in entry ? entry : { service: entry, mode: "singleton" }), + ); + const ids = definitions.map(({ service }) => service.id); + if (new Set(ids).size !== ids.length) throw new TypeError("Remote service catalogue contains duplicate IDs"); this.#allowlist = new Set(ids); + this.#catalogue = Object.freeze( + definitions.map(({ service, mode }) => Object.freeze({ serviceId: service.id, mode })), + ); + for (const { service, mode } of definitions) { + this.#registrations.set(service.id, { + serviceId: service.id, + mode, + instances: new Map(), + generations: new Map(), + subscribers: new Set(), + }); + } + } + + get catalogue(): readonly ServiceCatalogueEntry[] { + return this.#catalogue; } provide(service: Service, implementation: NoInfer>): void { diff --git a/packages/agent/src/plugins/services/types.ts b/packages/agent/src/plugins/services/types.ts index 61073728a..61ae27698 100644 --- a/packages/agent/src/plugins/services/types.ts +++ b/packages/agent/src/plugins/services/types.ts @@ -6,6 +6,16 @@ const SERVICE_TYPE = Symbol("pi.service.type"); export type ServiceMode = "singleton" | "keyed"; export type ServiceMemberKind = "method" | "state" | "events"; +export interface ServiceCatalogueEntry { + readonly serviceId: string; + readonly mode: ServiceMode; +} + +export interface ServiceProviderDefinition { + readonly service: { readonly id: string }; + readonly mode: ServiceMode; +} + export class ServiceSliceNotImplemented extends Error { readonly code = "service_not_implemented" as const; @@ -88,15 +98,23 @@ export type RemoteServiceContract = InvalidRemoteMemberNames extends never /** Stable identity for one shared TypeScript service contract. */ export interface Service { readonly id: string; + readonly rpc: boolean; readonly [SERVICE_TYPE]?: (value: T) => T; } export type ServiceType = TService extends Service ? T : never; -export function defineService(id: string): Service { - if (id.length === 0) throw new TypeError("Remote service ID must not be empty"); - if (id.startsWith("$pi.")) throw new TypeError("Remote service IDs beginning with $pi. are reserved"); - return Object.freeze({ id }); +export function defineService(id: string, options: { readonly rpc: false }): Service; +export function defineService( + id: string, + ...options: [RemoteServiceContract] extends [never] + ? readonly [options: never] + : readonly [options?: { readonly rpc?: true }] +): Service; +export function defineService(id: string, options?: { readonly rpc?: boolean }): Service { + if (id.length === 0) throw new TypeError("Service ID must not be empty"); + if (id.startsWith("$pi.")) throw new TypeError("Service IDs beginning with $pi. are reserved"); + return Object.freeze({ id, rpc: options?.rpc ?? true }); } export interface ServiceInstanceAddress { @@ -186,12 +204,14 @@ export interface RemoteServiceNamespaceOptions { readonly assertAccess?: () => void; } -export interface RemoteServiceNamespaceApi { +export interface RemoteServices { use(service: Service): T; observe( service: Service, handler: (instance: RemoteServiceInstance, context: Context) => void | Promise, ): () => void; + /** Activate this service set and wait for every acquired service to become ready. */ + activate(context: Context): Promise; /** Wait until every currently acquired service has installed its initial snapshot. */ ready(context: Context): Promise; /** Install a host lifecycle guard for proxy member access. */ diff --git a/packages/agent/test/plugins/services.test.ts b/packages/agent/test/plugins/services.test.ts index 50267b828..058edfa8c 100644 --- a/packages/agent/test/plugins/services.test.ts +++ b/packages/agent/test/plugins/services.test.ts @@ -41,6 +41,12 @@ interface Activity { const Activity = defineService("test.activity"); describe("plugin remote services", () => { + test("marks services RPC-capable by default and permits explicit local services", () => { + const local = defineService<{ readonly value: string }>("test.local", { rpc: false }); + expect(Models.rpc).toBe(true); + expect(local.rpc).toBe(false); + }); + test("does not defensively clone borrowed state values", () => { const initial: ModelsState = { selected: null, revision: 0 }; const state = remoteState(initial); @@ -60,6 +66,7 @@ describe("plugin remote services", () => { test("provides and consumes one singleton with replicated state", async () => { const provider = new RemoteServiceProvider([Models]); + expect(provider.catalogue).toEqual([{ serviceId: Models.id, mode: "singleton" }]); const initialState: ModelsState = { selected: null, revision: 0 }; const state = remoteState(initialState); let publishedState: ModelsState | undefined; @@ -230,7 +237,8 @@ describe("plugin remote services", () => { }); test("hydrates keyed state before observe handlers and fences reused keys", async () => { - const provider = new RemoteServiceProvider([QuestionDialogs]); + const provider = new RemoteServiceProvider([{ service: QuestionDialogs, mode: "keyed" }]); + expect(provider.catalogue).toEqual([{ serviceId: QuestionDialogs.id, mode: "keyed" }]); const connection = createLoopbackServiceConnection(provider); const errors: Error[] = []; const namespace = new RemoteServiceNamespace({ @@ -356,7 +364,7 @@ describe("plugin remote services", () => { }); test("routes keyed RemoteEvents only for the live instance generation", async () => { - const provider = new RemoteServiceProvider([Activity]); + const provider = new RemoteServiceProvider([{ service: Activity, mode: "keyed" }]); const namespace = new RemoteServiceNamespace({ services: [Activity], connection: createLoopbackServiceConnection(provider), @@ -382,7 +390,7 @@ describe("plugin remote services", () => { }); test("rejects mode mixing, unsupported members, and non-JSON values", async () => { - const provider = new RemoteServiceProvider([Models, QuestionDialogs]); + const provider = new RemoteServiceProvider([Models, { service: QuestionDialogs, mode: "keyed" }]); provider.provide(Models, { state: remoteState({ selected: null, revision: 0 }), async select() {}, diff --git a/packages/client/src/client.ts b/packages/client/src/client.ts index f5033c2f8..4b96522ad 100644 --- a/packages/client/src/client.ts +++ b/packages/client/src/client.ts @@ -1,6 +1,7 @@ import { type AttachmentEnvelope, createRpcClient, + createServiceCatalogueCall, createServiceSubscribeCall, createServiceUnsubscribeCall, encodeClientMessage, @@ -12,11 +13,13 @@ import { type PromptMessage, type ProtocolRpcCall, ProtocolValidationError, + parseServiceCatalogue, parseServiceSubscriptionSnapshot, type ResponseEnvelope, type RpcTarget, type ServerEventEnvelope, type ServerHello, + type ServiceCatalogueEntry, type ServiceMode, type ServiceProviderUpdate, ServiceRpc, @@ -220,6 +223,19 @@ export class Client { return this.#rpc.prompt([message]); } + async serviceCatalogue(target: RpcTarget, signal?: AbortSignal): Promise { + const result = await this.#request(target, createServiceCatalogueCall(), signal); + try { + return parseServiceCatalogue(result); + } catch (error) { + const validationError = new ProtocolValidationError( + error instanceof Error ? error.message : "Invalid service catalogue", + ); + this.#connection.fail(validationError); + throw validationError; + } + } + async subscribeService( target: RpcTarget, serviceId: string, diff --git a/packages/coding-agent/src/experimental/client-runtime.ts b/packages/coding-agent/src/experimental/client-runtime.ts index 23b4a1504..021f0ce17 100644 --- a/packages/coding-agent/src/experimental/client-runtime.ts +++ b/packages/coding-agent/src/experimental/client-runtime.ts @@ -1,5 +1,5 @@ import { basename } from "node:path"; -import { BACKGROUND_CONTEXT, type RemoteServiceNamespaceApi } from "@earendil-works/pi-agent-core"; +import { BACKGROUND_CONTEXT } 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"; @@ -7,10 +7,10 @@ import type { ClientCommand } from "../cli/experimental/commands/client.ts"; import { activateServer, ENV_SERVER_ID, resolveServerDirectory, resolveSessionDirectory } from "./server.ts"; import { Chat } from "./services/chat.ts"; import { - createBuiltinServerServiceNamespace, - createBuiltinSessionServiceNamespace, - type ServerServices, - type SessionServices, + createServerServiceConnection, + createSessionServiceConnection, + type ServerServiceConnection, + type SessionServiceConnection, } from "./services/connection.ts"; import { Models } from "./services/models.ts"; import { SessionDirectory, SessionManagement } from "./services/sessions.ts"; @@ -18,8 +18,8 @@ import { SessionDirectory, SessionManagement } from "./services/sessions.ts"; export interface ClientRuntimeServer { readonly route: UnixServerRoute; readonly client: Client; - readonly server: ServerServices; - readonly session: SessionServices; + readonly server: ServerServiceConnection; + readonly session: SessionServiceConnection; } export interface ActivatedClientRuntimeServer extends ClientRuntimeServer { @@ -75,17 +75,17 @@ export async function openClientRuntime( } const clients: Client[] = []; - const namespaces: RemoteServiceNamespaceApi[] = []; + const serviceConnections: Array = []; 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 connectionResults = await Promise.allSettled( + serviceConnections.map((connection) => connection.dispose(BACKGROUND_CONTEXT)), ); const clientResults = await Promise.allSettled(clients.map((client) => client.dispose())); - const errors = [...namespaceResults, ...clientResults].flatMap((result) => + const errors = [...connectionResults, ...clientResults].flatMap((result) => result.status === "rejected" ? [result.reason] : [], ); if (errors.length === 1) throw errors[0]; @@ -102,9 +102,9 @@ export async function openClientRuntime( })); activatedClient = undefined; clients.push(client); - const server = createBuiltinServerServiceNamespace(client, { deferred: true }); - const session = createBuiltinSessionServiceNamespace(client, { deferred: true }); - namespaces.push(server, session); + const server = createServerServiceConnection(client); + const session = createSessionServiceConnection(client); + serviceConnections.push(server, session); servers.push({ route, client, server, session }); } return { servers, dispose }; @@ -122,8 +122,18 @@ export async function openClientRuntime( export async function activateBuiltinClientServices( server: ClientRuntimeServer, ): Promise { - const directory = server.server.use(SessionDirectory); - const remoteManagement = server.server.use(SessionManagement); + const serverServices = server.server.open({ + services: [SessionDirectory, SessionManagement], + assertAccess() {}, + onError() {}, + }); + const sessionServices = server.session.open({ + services: [Models, Chat], + assertAccess() {}, + onError() {}, + }); + const directory = serverServices.use(SessionDirectory); + const remoteManagement = serverServices.use(SessionManagement); const management: SessionManagement = { create: (options, context) => remoteManagement.create(options, context), async remove(sessionId, context) { @@ -140,9 +150,9 @@ export async function activateBuiltinClientServices( await server.session.whenDetached(context); }, }; - const models = server.session.use(Models); - const chat = server.session.use(Chat); - await Promise.all([server.server.activate(BACKGROUND_CONTEXT), server.session.activate(BACKGROUND_CONTEXT)]); + const models = sessionServices.use(Models); + const chat = sessionServices.use(Chat); + await Promise.all([serverServices.activate(BACKGROUND_CONTEXT), sessionServices.activate(BACKGROUND_CONTEXT)]); return { ...server, directory, management, models, chat }; } diff --git a/packages/coding-agent/src/experimental/client-tui.ts b/packages/coding-agent/src/experimental/client-tui.ts index 81737e966..572276ed4 100644 --- a/packages/coding-agent/src/experimental/client-tui.ts +++ b/packages/coding-agent/src/experimental/client-tui.ts @@ -12,25 +12,30 @@ import { import chalk from "chalk"; import type { ClientCommand } from "../cli/experimental/commands/client.ts"; import { type OpenClientRuntimeOptions, openClientRuntime } from "./client-runtime.ts"; -import { activateRemoteFacets, bindService, defineFacet, type RemoteFacetActivation } from "./facets.ts"; -import type { ServerServices, SessionAttachmentState, SessionServices } from "./services/connection.ts"; +import { combineFacetLoaders, createStaticFacetLoader, type FacetLoader, type LoadedFacets } from "./facet-loader.ts"; +import { createFacetHost, defineFacet, type FacetHost } from "./facets.ts"; +import type { + ServerServiceConnection, + SessionAttachmentState, + SessionServiceConnection, +} from "./services/connection.ts"; import { Models, type Models as ModelsService } from "./services/models.ts"; -import { - SessionDirectory, - SessionManagement, - type SessionManagement as SessionManagementService, -} from "./services/sessions.ts"; +import { SessionDirectory, SessionManagement } from "./services/sessions.ts"; + +export interface RunClientTuiOptions extends OpenClientRuntimeOptions { + readonly facetLoader?: FacetLoader; +} export interface ClientTuiServer { readonly serverId: string; - readonly server: ServerServices; - readonly session: SessionServices; + readonly server: ServerServiceConnection; + readonly session: SessionServiceConnection; } interface SessionPickerFeature { readonly serverId: string; readonly directory: SessionDirectory; - readonly management: Pick; + readonly management: SessionManagement; readonly attachment: ReplicatedState; } @@ -49,7 +54,7 @@ interface Tui { setStatus(status: string): void; } -const Tui = defineService("pi.local.tui"); +const Tui = defineService("pi.local.tui", { rpc: false }); type ClientTuiAction = | { readonly kind: "attach"; readonly feature: SessionPickerFeature; readonly sessionId: string } @@ -72,8 +77,9 @@ export const sessionPickerTuiFacet = defineFacet({ const tui = env.use(Tui); const directory = env.use(SessionDirectory); const remoteManagement = env.use(SessionManagement); - const management: SessionPickerFeature["management"] = { + const management: SessionManagement = { create: (options, context) => remoteManagement.create(options, context), + remove: (sessionId, context) => remoteManagement.remove(sessionId, context), async attach(sessionId, context) { await remoteManagement.attach(sessionId, context); await tui.whenAttached(sessionId, context); @@ -112,12 +118,15 @@ export const modelSelectionTuiFacet = defineFacet({ }, }); +const BUILTIN_TUI_FACET_LOADER = createStaticFacetLoader([sessionPickerTuiFacet, modelSelectionTuiFacet]); + /** Minimal service-only presentation used before Harness execution is available. */ export class ExperimentalClientTui implements Component { readonly #requestRender: () => void; readonly #finish: () => void; readonly #container = new Container(); - readonly #remoteFacetActivations: RemoteFacetActivation[] = []; + readonly #facetLoader: FacetLoader; + readonly #facetHosts: FacetHost[] = []; readonly #sessionPickers = new Map(); readonly #modelSelections = new Map(); readonly #actions = new Map(); @@ -126,19 +135,26 @@ export class ExperimentalClientTui implements Component { #selectedServerId: string | undefined; #status = "Select a Session or create one."; #busy = false; + #loadedFacets: LoadedFacets | undefined; #closePromise: Promise | undefined; - private constructor(requestRender: () => void, finish: () => void) { + private constructor(requestRender: () => void, finish: () => void, facetLoader: FacetLoader) { this.#requestRender = requestRender; this.#finish = finish; + this.#facetLoader = facetLoader; } static async create(options: { readonly servers: readonly ClientTuiServer[]; + readonly facetLoader?: FacetLoader; requestRender(): void; finish(): void; }): Promise { - const component = new ExperimentalClientTui(options.requestRender, options.finish); + const facetLoader = combineFacetLoaders([ + BUILTIN_TUI_FACET_LOADER, + ...(options.facetLoader === undefined ? [] : [options.facetLoader]), + ]); + const component = new ExperimentalClientTui(options.requestRender, options.finish, facetLoader); try { await component.#start(options.servers); return component; @@ -169,46 +185,69 @@ export class ExperimentalClientTui implements Component { } close(): Promise { - this.#closePromise ??= disposeRemoteFacetActivations(this.#remoteFacetActivations.splice(0).reverse()); + this.#closePromise ??= this.#close(); return this.#closePromise; } async #start(servers: readonly ClientTuiServer[]): Promise { + const loaded = await this.#facetLoader.load(); + this.#loadedFacets = loaded; for (const server of servers) { - const bindings = [ - bindService(Tui, { - attachment: server.session.attachment, - whenAttached: (sessionId, context) => server.session.whenAttached(sessionId, context), - whenDetached: (context) => server.session.whenDetached(context), - registerSessionPicker: (feature) => - this.#register(this.#sessionPickers, "Session picker", { ...feature, serverId: server.serverId }), - registerModelSelection: (feature) => - this.#register(this.#modelSelections, "Model selection", { ...feature, serverId: server.serverId }), - refresh: () => this.#rebuild(), - setStatus: (status) => { - this.#status = status; - this.#rebuild(); - }, - }), - ]; - const serverActivation = await activateRemoteFacets({ - facets: [sessionPickerTuiFacet], - bindings, - source: server.server, - connect: () => server.server.activate(BACKGROUND_CONTEXT), + const tuiRuntimeFacet = defineFacet({ + id: "@pi/tui-runtime", + setup: (env) => { + env.provide(Tui, { + attachment: server.session.attachment, + whenAttached: (sessionId, context) => server.session.whenAttached(sessionId, context), + whenDetached: (context) => server.session.whenDetached(context), + registerSessionPicker: (feature) => + this.#register(this.#sessionPickers, "Session picker", { + ...feature, + serverId: server.serverId, + }), + registerModelSelection: (feature) => + this.#register(this.#modelSelections, "Model selection", { + ...feature, + serverId: server.serverId, + }), + refresh: () => this.#rebuild(), + setStatus: (status) => { + this.#status = status; + this.#rebuild(); + }, + }); + }, }); - this.#remoteFacetActivations.push(serverActivation); - const sessionActivation = await activateRemoteFacets({ - facets: [modelSelectionTuiFacet], - bindings, - source: server.session, - connect: () => server.session.activate(BACKGROUND_CONTEXT), + const facetHost = await createFacetHost({ + facets: [tuiRuntimeFacet, ...loaded.facets], + connections: [server.server, server.session], }); - this.#remoteFacetActivations.push(sessionActivation); + this.#facetHosts.push(facetHost); } this.#rebuild(); } + async #close(): Promise { + const results = await Promise.allSettled( + this.#facetHosts + .splice(0) + .reverse() + .map((host) => host.dispose()), + ); + const errors = results.flatMap((result) => (result.status === "rejected" ? [result.reason] : [])); + const loaded = this.#loadedFacets; + this.#loadedFacets = undefined; + if (loaded !== undefined) { + try { + await loaded.dispose(); + } catch (error) { + errors.push(error); + } + } + if (errors.length === 1) throw errors[0]; + if (errors.length > 1) throw new AggregateError(errors, "Failed to dispose experimental TUI facets"); + } + #register(features: Map, label: string, feature: T): () => void { if (features.has(feature.serverId)) { throw new Error(`${label} is already registered for server ${feature.serverId}`); @@ -353,7 +392,7 @@ export class ExperimentalClientTui implements Component { } } -export async function runClientTui(command: ClientCommand, options: OpenClientRuntimeOptions = {}): Promise { +export async function runClientTui(command: ClientCommand, options: RunClientTuiOptions = {}): Promise { const runtime = await openClientRuntime(command, options); const terminal = new ProcessTerminal(); const tui = new TuiMainScreen(terminal); @@ -372,6 +411,7 @@ export async function runClientTui(command: ClientCommand, options: OpenClientRu server: server.server, session: server.session, })), + facetLoader: options.facetLoader, requestRender: () => tui.requestRender(), finish, }); @@ -385,13 +425,6 @@ export async function runClientTui(command: ClientCommand, options: OpenClientRu } } -async function disposeRemoteFacetActivations(activations: readonly RemoteFacetActivation[]): Promise { - const results = await Promise.allSettled(activations.map((activation) => activation.dispose())); - const errors = results.flatMap((result) => (result.status === "rejected" ? [result.reason] : [])); - if (errors.length === 1) throw errors[0]; - if (errors.length > 1) throw new AggregateError(errors, "Failed to dispose experimental TUI facets"); -} - function findSession(feature: SessionPickerFeature, sessionId: string): SessionSummary { const session = feature.directory.state.value?.sessions.find((candidate) => candidate.sessionId === sessionId); if (session === undefined) throw new Error(`Unknown Session: ${sessionId}`); diff --git a/packages/coding-agent/src/experimental/facet-loader.ts b/packages/coding-agent/src/experimental/facet-loader.ts new file mode 100644 index 000000000..7c39b1d82 --- /dev/null +++ b/packages/coding-agent/src/experimental/facet-loader.ts @@ -0,0 +1,52 @@ +import type { Facet } from "./facets.ts"; + +export interface LoadedFacets { + readonly facets: readonly Facet[]; + dispose(): Promise; +} + +export interface FacetLoader { + load(): Promise; +} + +export function createStaticFacetLoader(facets: readonly Facet[]): FacetLoader { + const loadedFacets = Object.freeze([...facets]); + return { + async load() { + return { facets: loadedFacets, async dispose() {} }; + }, + }; +} + +export function combineFacetLoaders(loaders: readonly FacetLoader[]): FacetLoader { + return { + async load() { + const loaded: LoadedFacets[] = []; + try { + for (const loader of loaders) loaded.push(await loader.load()); + } catch (error) { + const cleanupErrors = await disposeLoadedFacets(loaded.reverse()); + if (cleanupErrors.length > 0) { + throw new AggregateError([error, ...cleanupErrors], "Facet loading and cleanup failed"); + } + throw error; + } + let disposed = false; + return { + facets: Object.freeze(loaded.flatMap(({ facets }) => facets)), + async dispose() { + if (disposed) return; + disposed = true; + const errors = await disposeLoadedFacets([...loaded].reverse()); + if (errors.length === 1) throw errors[0]; + if (errors.length > 1) throw new AggregateError(errors, "Failed to dispose loaded facets"); + }, + }; + }, + }; +} + +async function disposeLoadedFacets(loaded: readonly LoadedFacets[]): Promise { + const results = await Promise.allSettled(loaded.map((entry) => entry.dispose())); + return results.flatMap((result) => (result.status === "rejected" ? [result.reason] : [])); +} diff --git a/packages/coding-agent/src/experimental/facets.ts b/packages/coding-agent/src/experimental/facets.ts index 5c2d42c64..2e5712397 100644 --- a/packages/coding-agent/src/experimental/facets.ts +++ b/packages/coding-agent/src/experimental/facets.ts @@ -4,15 +4,15 @@ import { createLoopbackServiceConnection, type MutableRemoteEvents, type MutableReplicatedState, - type RemoteServiceConnection, type RemoteServiceContract, type RemoteServiceInstance, RemoteServiceNamespace, - type RemoteServiceNamespaceApi, RemoteServiceProvider, + type RemoteServices, remoteEvents, remoteState, type Service, + type ServiceCatalogueEntry, type ServiceMode, } from "@earendil-works/pi-agent-core"; @@ -20,15 +20,6 @@ export interface ServiceInstances { add(key: string, implementation: RemoteServiceContract): () => void; } -export interface FacetServiceBinding { - readonly service: { readonly id: string }; - readonly implementation: unknown; -} - -export function bindService(service: Service, implementation: NoInfer): FacetServiceBinding { - return Object.freeze({ service, implementation }); -} - interface FacetServiceReference { readonly serviceId: string; readonly mode: ServiceMode; @@ -46,7 +37,7 @@ export interface FacetEnvironment { service: Service, handler: (instance: RemoteServiceInstance, context: Context) => void | Promise, ): () => void; - provide(service: Service, implementation: NoInfer>): void; + provide(service: Service, implementation: NoInfer): void; provideMany(service: Service): ServiceInstances; remoteState(initial: T): MutableReplicatedState; remoteEvents(): MutableRemoteEvents; @@ -172,139 +163,112 @@ class FacetLedger { } } -interface LocalServiceHandle { - readonly mode: ServiceMode; - readonly namespace: RemoteServiceNamespace; +interface SingletonServiceHandle { + bind(implementation: object): void; + bindRemote(services: RemoteServices): void; + readonly proxy: object; } -class LocalServiceHandles { - readonly #assertAccess: () => void; - readonly #onError: (error: Error) => void; - readonly #handles = new Map(); - #connection: RemoteServiceConnection | undefined; +interface StagedObservation { + start(): () => void; + stop(): void; +} - constructor(assertAccess: () => void, onError: (error: Error) => void) { +class FacetServiceHandles { + readonly #assertAccess: () => void; + readonly #singletons = new Map(); + readonly #keyedSources = new Map(); + + constructor(assertAccess: () => void) { this.#assertAccess = assertAccess; - this.#onError = onError; } use(service: Service): T { - return this.#handle(service, "singleton").namespace.use(service); + let handle = this.#singletons.get(service.id); + if (handle === undefined) { + let implementation: object | undefined; + let remote = false; + const methods = new Map(); + const proxy = new Proxy( + {}, + { + get: (_target, property) => { + this.#assertAccess(); + if (implementation === undefined) throw new Error(`Service ${service.id} is disconnected`); + const value: unknown = Reflect.get(implementation, property, implementation); + if (remote || typeof value !== "function") return value; + let method = methods.get(property); + if (method === undefined) { + method = (...args: unknown[]) => Reflect.apply(value, implementation, args); + methods.set(property, method); + } + return method; + }, + }, + ); + handle = { + proxy, + bind: (target) => { + implementation = target; + remote = false; + }, + bindRemote: (services) => { + implementation = services.use(service) as object; + remote = true; + }, + }; + this.#singletons.set(service.id, handle); + } + return handle.proxy as T; } observe( service: Service, handler: (instance: RemoteServiceInstance, context: Context) => void | Promise, - ): () => void { - return this.#handle(service, "keyed").namespace.observe(service, handler); - } - - async connect(provider: RemoteServiceProvider): Promise { - this.#connection = createLoopbackServiceConnection(provider); - await Promise.all( - [...this.#handles.values()].map(async ({ namespace }) => { - await namespace.rebind(true, BACKGROUND_CONTEXT); - await namespace.ready(BACKGROUND_CONTEXT); - }), - ); - } - - async dispose(): Promise { - const results = await Promise.allSettled( - [...this.#handles.values()].map(({ namespace }) => namespace.dispose(BACKGROUND_CONTEXT)), - ); - this.#handles.clear(); - this.#connection = undefined; - const errors = results.flatMap((result) => (result.status === "rejected" ? [result.reason] : [])); - if (errors.length === 1) throw errors[0]; - if (errors.length > 1) throw new AggregateError(errors, "Failed to dispose local facet service handles"); - } - - #handle(service: Service, mode: ServiceMode): LocalServiceHandle { - const existing = this.#handles.get(service.id); - if (existing !== undefined) { - if (existing.mode !== mode) { - throw new Error(`Local service ${service.id} is used as both singleton and keyed`); - } - return existing; - } - const namespace = new RemoteServiceNamespace({ - services: [service], - bound: false, - assertAccess: this.#assertAccess, - onError: this.#onError, - connection: { - invoke: (call, context) => { - if (this.#connection === undefined) throw new Error("Local facet services are not connected"); - return this.#connection.invoke(call, context); - }, - subscribe: (serviceId, serviceMode, listener, context) => { - if (this.#connection === undefined) throw new Error("Local facet services are not connected"); - return this.#connection.subscribe(serviceId, serviceMode, listener, context); - }, + ): StagedObservation { + let close: (() => void) | undefined; + let closed = false; + const stop = (): void => { + if (closed) return; + closed = true; + close?.(); + }; + return { + start: () => { + if (closed) return () => {}; + const source = this.#keyedSources.get(service.id); + if (source === undefined) throw new Error(`Service ${service.id} is disconnected`); + close = source.observe(service, handler); + return stop; }, - }); - const handle = { mode, namespace }; - this.#handles.set(service.id, handle); - return handle; - } -} - -class FacetServiceBindings { - readonly #assertAccess: () => void; - readonly #implementations = new Map(); - readonly #handles = new Map(); - - constructor(bindings: readonly FacetServiceBinding[], assertAccess: () => void) { - this.#assertAccess = assertAccess; - for (const binding of bindings) { - if (this.#implementations.has(binding.service.id)) { - throw new Error(`Facet host service ${binding.service.id} is bound more than once`); - } - if (typeof binding.implementation !== "object" || binding.implementation === null) { - throw new TypeError(`Facet host service ${binding.service.id} implementation must be an object`); - } - this.#implementations.set(binding.service.id, binding.implementation); - } + stop, + }; } - get serviceIds(): ReadonlySet { - return new Set(this.#implementations.keys()); + bindSingleton(serviceId: string, implementation: object): void { + this.#singletons.get(serviceId)?.bind(implementation); } - has(service: { readonly id: string }): boolean { - return this.#implementations.has(service.id); + bindRemoteSingleton(serviceId: string, services: RemoteServices): void { + const handle = this.#singletons.get(serviceId); + if (handle === undefined) throw new Error(`Service ${serviceId} has no singleton handle`); + handle.bindRemote(services); } - use(service: Service): T { - const implementation = this.#implementations.get(service.id); - if (implementation === undefined) throw new Error(`Facet host service ${service.id} is not bound`); - let handle = this.#handles.get(service.id); - if (handle === undefined) { - const methods = new Map(); - handle = new Proxy(implementation, { - get: (target, property) => { - this.#assertAccess(); - const value: unknown = Reflect.get(target, property, target); - if (typeof value !== "function") return value; - let method = methods.get(property); - if (method === undefined) { - method = value.bind(target); - methods.set(property, method); - } - return method; - }, - }); - this.#handles.set(service.id, handle); - } - return handle as T; + bindKeyed(serviceId: string, services: RemoteServices): void { + this.#keyedSources.set(serviceId, services); + } + + dispose(): void { + this.#singletons.clear(); + this.#keyedSources.clear(); } } class StagedServiceInstances implements ServiceInstances { readonly #service: Service; readonly #lifecycle: FacetLifecycle; - #provider: RemoteServiceProvider | undefined; + readonly #providers: RemoteServiceProvider[] = []; constructor(service: Service, lifecycle: FacetLifecycle) { this.#service = service; @@ -312,61 +276,66 @@ class StagedServiceInstances implements ServiceInstances { } connect(provider: RemoteServiceProvider): void { - this.#provider = provider; + this.#providers.push(provider); } add(key: string, implementation: RemoteServiceContract): () => void { this.#lifecycle.assertActive("add service instances"); - if (this.#provider === undefined) throw new Error("Facet service provider is not connected"); - const close = this.#provider.spawn(this.#service, key, implementation); + if (this.#providers.length === 0) throw new Error("Facet service provider is not connected"); + const closes: Array<() => void> = []; + try { + for (const provider of this.#providers) closes.push(provider.spawn(this.#service, key, implementation)); + } catch (error) { + for (const close of closes.reverse()) close(); + throw error; + } + let closed = false; + const close = (): void => { + if (closed) return; + closed = true; + for (const release of closes.reverse()) release(); + }; this.#lifecycle.own(close); return close; } } interface SingletonProvision { - readonly service: { readonly id: string }; + readonly service: { readonly id: string; readonly rpc: boolean }; + readonly implementation: object; install(provider: RemoteServiceProvider): void; } interface KeyedProvision { - readonly service: { readonly id: string }; + readonly service: { readonly id: string; readonly rpc: boolean }; connect(provider: RemoteServiceProvider): void; } +export interface FacetConnection { + /** Whether this currently unavailable route may provisionally own requirements absent from its catalogue. */ + readonly acceptsUnavailableServices: boolean; + catalogue(context: Context): Promise; + open(options: { + readonly services: readonly { readonly id: string }[]; + assertAccess(): void; + onError(error: Error): void; + }): RemoteServices; +} + interface FacetOptions { readonly facets: readonly Facet[]; - readonly bindings?: readonly FacetServiceBinding[]; + readonly connections?: readonly FacetConnection[]; readonly onError?: (error: Error) => void; } -interface RemoteFacetOptions extends FacetOptions { - readonly source: RemoteServiceNamespaceApi; - readonly connect?: () => void | Promise; -} - -type FacetKernelOptions = FacetOptions & - ( - | { readonly kind: "provider" } - | { - readonly kind: "remote"; - readonly source: RemoteServiceNamespaceApi; - readonly connect?: () => void | Promise; - } - ); - export interface FacetHost { readonly services: RemoteServiceProvider; dispose(): Promise; } -export interface RemoteFacetActivation { - dispose(): Promise; -} - -/** Create an active local host for facets that provide and consume services. */ +/** Create an active host for one complete set of facets. */ export async function createFacetHost(options: FacetOptions): Promise { - const kernel = new FacetKernel({ ...options, kind: "provider" }); + const kernel = new FacetKernel(options); await kernel.activate(); return Object.freeze({ services: kernel.provider, @@ -374,39 +343,30 @@ export async function createFacetHost(options: FacetOptions): Promise }); } -/** Activate facets whose services come from an existing remote binding. */ -export async function activateRemoteFacets(options: RemoteFacetOptions): Promise { - const kernel = new FacetKernel({ ...options, kind: "remote" }); - await kernel.activate(); - return Object.freeze({ dispose: () => kernel.dispose() }); -} - -/** Private lifecycle and dependency kernel shared by the two atomic entry points. */ +/** Private lifecycle and dependency kernel behind the atomic host entry point. */ class FacetKernel { readonly #facets: readonly Facet[]; - readonly #bindings: FacetServiceBindings; - readonly #source: RemoteServiceNamespaceApi | undefined; - readonly #connect: (() => void | Promise) | undefined; + readonly #connections: readonly FacetConnection[]; readonly #onError: (error: Error) => void; readonly #lifecycles = new Map(); readonly #singletons: SingletonProvision[] = []; readonly #keyed: KeyedProvision[] = []; - readonly #localHandles: LocalServiceHandles; + readonly #handles: FacetServiceHandles; + readonly #connectionBindings = new Map(); #activationOrder: readonly string[] = []; #provider: RemoteServiceProvider | undefined; + #keyedProvider: RemoteServiceProvider | undefined; + #localKeyedServices: RemoteServiceNamespace | undefined; #phase: GenerationPhase = "setup"; - constructor(options: FacetKernelOptions) { + constructor(options: FacetOptions) { const ids = options.facets.map((facet) => facet.id); if (ids.some((id) => id.length === 0)) throw new Error("Facet ID must not be empty"); if (new Set(ids).size !== ids.length) throw new Error("Facet IDs must be unique within a generation"); this.#facets = options.facets; - this.#bindings = new FacetServiceBindings(options.bindings ?? [], () => this.#assertServiceAccess()); - this.#source = options.kind === "remote" ? options.source : undefined; - this.#connect = options.kind === "remote" ? options.connect : undefined; + this.#connections = options.connections ?? []; this.#onError = options.onError ?? (() => {}); - this.#localHandles = new LocalServiceHandles(() => this.#assertServiceAccess(), this.#onError); - this.#source?.setAccessGuard(() => this.#assertServiceAccess()); + this.#handles = new FacetServiceHandles(() => this.#assertServiceAccess()); } get provider(): RemoteServiceProvider { @@ -431,16 +391,16 @@ class FacetKernel { } this.#phase = "assembling"; - if (this.#source === undefined) { - this.#activationOrder = validateFacets(records, this.#bindings.serviceIds); - this.#assembleProvider(); - } else { - this.#activationOrder = records.map((record) => record.facetId); - } + const externalServices = await this.#resolveExternalServices(records); + this.#activationOrder = validateFacets(records, externalServices); + this.#assembleProviders(); + this.#bindServices(externalServices); this.#phase = "connecting"; - if (this.#source === undefined) await this.#localHandles.connect(this.provider); - else await this.#connect?.(); + await Promise.all([ + this.#localKeyedNamespace.activate(BACKGROUND_CONTEXT), + ...[...this.#connectionBindings.values()].map((services) => services.activate(BACKGROUND_CONTEXT)), + ]); this.#phase = "activating"; for (const id of this.#activationOrder) await this.#lifecycles.get(id)!.activate(); @@ -449,10 +409,13 @@ class FacetKernel { const cleanupErrors = await this.#disposeLifecycles(); this.#phase = "disposing"; try { - await this.#localHandles.dispose(); + await this.#localKeyedServices?.dispose(BACKGROUND_CONTEXT); } catch (cleanupError) { cleanupErrors.push(cleanupError); } + cleanupErrors.push(...(await this.#disposeConnectionBindings())); + this.#handles.dispose(); + this.#keyedProvider?.dispose(); this.#provider?.dispose(); this.#phase = "dead"; if (cleanupErrors.length > 0) { @@ -467,10 +430,13 @@ class FacetKernel { const errors = await this.#disposeLifecycles(); this.#phase = "disposing"; try { - await this.#localHandles.dispose(); + await this.#localKeyedServices?.dispose(BACKGROUND_CONTEXT); } catch (error) { errors.push(error); } + errors.push(...(await this.#disposeConnectionBindings())); + this.#handles.dispose(); + this.#keyedProvider?.dispose(); this.#provider?.dispose(); this.#phase = "dead"; if (errors.length === 1) throw errors[0]; @@ -479,18 +445,20 @@ class FacetKernel { #environment(lifecycle: FacetLifecycle, ledger: FacetLedger): FacetEnvironment { return { - provide: (service: Service, implementation: NoInfer>): void => { + provide: (service: Service, implementation: NoInfer): void => { lifecycle.assertSettingUp("provide services"); - this.#assertProviderGeneration(lifecycle); + if (typeof implementation !== "object" || implementation === null || Array.isArray(implementation)) { + throw new TypeError(`Service ${service.id} implementation must be an object`); + } ledger.provide(service, "singleton"); this.#singletons.push({ service, - install: (provider) => provider.provide(service, implementation), + implementation, + install: (provider) => provider.provide(service, implementation as NoInfer>), }); }, provideMany: (service: Service): ServiceInstances => { lifecycle.assertSettingUp("provide service instances"); - this.#assertProviderGeneration(lifecycle); ledger.provide(service, "keyed"); const instances = new StagedServiceInstances(service, lifecycle); this.#keyed.push({ service, connect: (target) => instances.connect(target) }); @@ -499,19 +467,17 @@ class FacetKernel { use: (service: Service): T => { lifecycle.assertSettingUp("acquire services"); ledger.require(service, "singleton"); - if (this.#bindings.has(service)) return this.#bindings.use(service); - if (this.#source !== undefined) return this.#source.use(service); - return this.#localHandles.use(service); + return this.#handles.use(service); }, observe: ( service: Service, handler: (instance: RemoteServiceInstance, context: Context) => void | Promise, ): (() => void) => { lifecycle.assertSettingUp("observe services"); - const services = this.#source ?? this.#localHandles; - if (this.#source === undefined) ledger.require(service, "keyed"); - const prepare = services.observe(service, () => {}); - return this.#stageObservation(lifecycle, () => services.observe(service, handler), prepare); + ledger.require(service, "keyed"); + const observation = this.#handles.observe(service, handler); + lifecycle.observe(() => observation.start()); + return observation.stop; }, remoteState: (initial: T) => { lifecycle.assertRunning("create replicated state"); @@ -527,39 +493,120 @@ class FacetKernel { }; } - #assertProviderGeneration(lifecycle: FacetLifecycle): void { - if (this.#source !== undefined) { - throw new Error(`Remote facet ${lifecycle.id} cannot provide services`); + async #resolveExternalServices( + records: readonly FacetRecord[], + ): Promise> { + const catalogues = await Promise.all( + this.#connections.map(async (connection) => ({ + connection, + entries: await connection.catalogue(BACKGROUND_CONTEXT), + })), + ); + const offered = new Map(); + for (const { connection, entries } of catalogues) { + for (const { serviceId, mode } of entries) { + if (offered.has(serviceId)) { + throw new Error(`Facet host service ${serviceId} is offered by more than one connection`); + } + offered.set(serviceId, { mode, connection }); + } + } + + const local = new Set(records.flatMap(({ provides }) => provides.map(({ serviceId }) => serviceId))); + const external = new Map(); + for (const { requires } of records) { + for (const requirement of requires) { + if (local.has(requirement.serviceId) || external.has(requirement.serviceId)) continue; + let provider = offered.get(requirement.serviceId); + if (provider === undefined) { + const deferred = this.#connections.filter( + ({ acceptsUnavailableServices }) => acceptsUnavailableServices, + ); + if (deferred.length > 1) { + throw new Error(`Facet host service ${requirement.serviceId} has more than one deferred connection`); + } + if (deferred.length === 1) provider = { mode: requirement.mode, connection: deferred[0]! }; + } + if (provider !== undefined) external.set(requirement.serviceId, provider); + } + } + const serviceIdsByConnection = new Map(); + for (const [serviceId, { connection }] of external) { + let serviceIds = serviceIdsByConnection.get(connection); + if (serviceIds === undefined) { + serviceIds = []; + serviceIdsByConnection.set(connection, serviceIds); + } + serviceIds.push(serviceId); + } + for (const [connection, serviceIds] of serviceIdsByConnection) { + this.#connectionBindings.set( + connection, + connection.open({ + services: serviceIds.map((id) => ({ id })), + assertAccess: () => this.#assertServiceAccess(), + onError: this.#onError, + }), + ); + } + return external; + } + + #assembleProviders(): void { + const rpcServices = [ + ...this.#singletons + .filter(({ service }) => service.rpc) + .map(({ service }) => ({ service, mode: "singleton" as const })), + ...this.#keyed + .filter(({ service }) => service.rpc) + .map(({ service }) => ({ service, mode: "keyed" as const })), + ]; + const provider = new RemoteServiceProvider(rpcServices); + const keyedProvider = new RemoteServiceProvider(this.#keyed.map(({ service }) => ({ service, mode: "keyed" }))); + this.#provider = provider; + this.#keyedProvider = keyedProvider; + this.#localKeyedServices = new RemoteServiceNamespace({ + services: this.#keyed.map(({ service }) => service), + connection: createLoopbackServiceConnection(keyedProvider), + assertAccess: () => this.#assertServiceAccess(), + onError: this.#onError, + }); + for (const provision of this.#singletons) { + if (provision.service.rpc) provision.install(provider); + } + for (const provision of this.#keyed) { + provision.connect(keyedProvider); + if (provision.service.rpc) provision.connect(provider); } } - #stageObservation(lifecycle: FacetLifecycle, start: () => () => void, prepare: () => void): () => void { - let close: (() => void) | undefined; - let closed = false; - const stop = (): void => { - if (closed) return; - closed = true; - close?.(); - prepare(); - }; - lifecycle.own(prepare); - lifecycle.observe(() => { - if (!closed) { - close = start(); - prepare(); - } - return stop; - }); - return stop; + #bindServices( + externalServices: ReadonlyMap, + ): void { + for (const provision of this.#singletons) { + this.#handles.bindSingleton(provision.service.id, provision.implementation); + } + for (const provision of this.#keyed) { + this.#handles.bindKeyed(provision.service.id, this.#localKeyedNamespace); + } + for (const [serviceId, { mode, connection }] of externalServices) { + const services = this.#connectionBindings.get(connection); + if (services === undefined) throw new Error(`Service connection for ${serviceId} is not open`); + if (mode === "singleton") this.#handles.bindRemoteSingleton(serviceId, services); + else this.#handles.bindKeyed(serviceId, services); + } } - #assembleProvider(): void { - const services = new Map(); - for (const { service } of [...this.#singletons, ...this.#keyed]) services.set(service.id, service); - const provider = new RemoteServiceProvider([...services.values()]); - this.#provider = provider; - for (const provision of this.#singletons) provision.install(provider); - for (const provision of this.#keyed) provision.connect(provider); + get #localKeyedNamespace(): RemoteServiceNamespace { + if (this.#localKeyedServices === undefined) throw new Error("Facet keyed services are not assembled"); + return this.#localKeyedServices; + } + + async #disposeConnectionBindings(): Promise { + const bindings = [...this.#connectionBindings.values()]; + this.#connectionBindings.clear(); + const results = await Promise.allSettled(bindings.map((services) => services.dispose(BACKGROUND_CONTEXT))); + return results.flatMap((result) => (result.status === "rejected" ? [result.reason] : [])); } #assertServiceAccess(): void { @@ -588,14 +635,20 @@ class FacetKernel { } } -function validateFacets(records: readonly FacetRecord[], boundServices: ReadonlySet): string[] { - const providers = new Map(); - for (const serviceId of boundServices) providers.set(serviceId, { facetId: undefined, mode: "singleton" }); +function validateFacets( + records: readonly FacetRecord[], + externalServices: ReadonlyMap, +): string[] { + const providers = new Map< + string, + { readonly facetId: string | undefined; readonly mode: ServiceMode | undefined } + >(); + for (const [serviceId, { mode }] of externalServices) providers.set(serviceId, { facetId: undefined, mode }); for (const record of records) { for (const provision of record.provides) { const existing = providers.get(provision.serviceId); if (existing !== undefined) { - if (existing.mode !== provision.mode) { + if (existing.mode !== undefined && existing.mode !== provision.mode) { throw new Error(`Service ${provision.serviceId} is provided as both singleton and keyed`); } if (existing.facetId === undefined) { @@ -619,7 +672,7 @@ function validateFacets(records: readonly FacetRecord[], boundServices: Readonly `Facet ${record.facetId} requires local/${requirement.serviceId}/${requirement.mode}, but no facet provides it`, ); } - if (provider.mode !== requirement.mode) { + if (provider.mode !== undefined && provider.mode !== requirement.mode) { throw new Error( `Facet ${record.facetId} requires ${requirement.serviceId} as ${requirement.mode}, but ${provider.facetId ?? "the host"} provides it as ${provider.mode}`, ); diff --git a/packages/coding-agent/src/experimental/services/README.md b/packages/coding-agent/src/experimental/services/README.md index 2e5406583..7c5bc2d27 100644 --- a/packages/coding-agent/src/experimental/services/README.md +++ b/packages/coding-agent/src/experimental/services/README.md @@ -1,6 +1,6 @@ # Experimental client/server service slices -`server-builtins.ts` and `session-builtins.ts` are the separate provider allowlists; `builtins.ts` combines them only for presentation hosts and tests, so the server router does not load Session contracts. A token in that inventory is either implemented end to end or registered with an explicit `ServiceSliceNotImplemented` provider. Framed callers receive the stable `service_not_implemented` code. Keyed services hydrate as an empty directory until their owning feature spawns an instance. +Facet setup generates each host's RPC service catalogue from its provided RPC-capable tokens. Connections exchange those catalogues and bind only services required by consuming facets; there is no handwritten built-in service inventory. With no selected Session, its deferred connection admits unresolved requirements as unavailable and keeps their handles disconnected. Attachment validates them against the worker's generated catalogue, which is cached for later detached generations. A built-in token is either implemented end to end or registered with an explicit `ServiceSliceNotImplemented` provider. Framed callers receive the stable `service_not_implemented` code. Keyed services hydrate as an empty directory until their owning feature spawns an instance. | Scope | Service | Current slice | Continuation point | |---|---|---|---| @@ -11,10 +11,10 @@ | session | `Accounts` | state declared; mutations throw | connect the local credential service to the presentation-safe facade | | 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` 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. +`RemoteEvents` has local and remote ordered delivery, hydration-race buffering, keyed-instance routing, and no replay for late subscribers. `ServerServiceConnection.connection` and `SessionServiceConnection.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 Session worker creates one active `FacetHost` for its built-in providers and `SessionWorkerRuntime.facets` with `createFacetHost()` from `../facets.ts`. The host exposes the services provided by those facets and owns their lifecycle. The service-only TUI activates its server and Session facets through `activateRemoteFacets()`, which returns a disposable `RemoteFacetActivation`. Hosts bind their concrete local services before activation, and facets declare every local or remote dependency through the same `env.use()` operation. Synchronous setup-time `env.provide()`, `env.provideMany()`, `env.use()`, and `env.observe()` calls produce the internal dependency graph; setup does not repeat a declarative dependency list, and service handles remain disconnected until the complete graph validates. Providers activate before consumers, observations connect with their consuming facet, and disposal runs in reverse dependency order for the complete generation. +The Session worker loads all built-in and configured Session facets through one `FacetLoader`, then creates one active `FacetHost`. The service-only TUI likewise loads all TUI facets once and creates one host that combines its local `Tui` service with connected server and selected-Session services. Runtime facets provide host-created services such as `Harness`, `ModelsRuntime`, and `Tui` through the same `env.provide()` operation used by every other facet. Synchronous setup-time `env.provide()`, `env.provideMany()`, `env.use()`, and `env.observe()` calls produce the internal dependency graph; setup does not repeat a declarative dependency list, and service handles remain disconnected until the complete graph validates. Providers activate before consumers, observations connect with their consuming facet, and disposal runs in reverse dependency order for the complete generation. Services are RPC-capable by default; process-local tokens such as `Harness`, `ModelsRuntime`, and `Tui` declare `{ rpc: false }`. The host automatically publishes every provided RPC-capable service. -A facet always calls unqualified `env.use()` or `env.observe()`. Its facet kind and host select the local, connected-server, or selected-Session service source. Transport namespaces remain internal binding machinery rather than part of the facet environment. TUI facets own subscriptions to their remote replicated state and events; `ExperimentalClientTui` retains only terminal rendering, navigation, and action dispatch. The server provider remains directly assembled until its complete host environment exists. 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, and flow control remain protocol/host infrastructure slices rather than presentation service tokens. +A facet always calls unqualified `env.use()` or `env.observe()`. The host resolves each token across facet-provided and connected services. Transport namespaces remain internal binding machinery rather than part of the facet environment. TUI facets own subscriptions to their remote replicated state and events; `ExperimentalClientTui` retains only terminal rendering, navigation, and action dispatch. The server provider remains directly assembled until its complete host environment exists. 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, and flow control remain protocol/host infrastructure slices rather than presentation service tokens. diff --git a/packages/coding-agent/src/experimental/services/builtins.ts b/packages/coding-agent/src/experimental/services/builtins.ts deleted file mode 100644 index 892ff5a2f..000000000 --- a/packages/coding-agent/src/experimental/services/builtins.ts +++ /dev/null @@ -1,2 +0,0 @@ -export { BUILTIN_SERVER_SERVICES } from "./server-builtins.ts"; -export { BUILTIN_SESSION_SERVICES } from "./session-builtins.ts"; diff --git a/packages/coding-agent/src/experimental/services/connection.ts b/packages/coding-agent/src/experimental/services/connection.ts index ecc21ebcf..09b125c16 100644 --- a/packages/coding-agent/src/experimental/services/connection.ts +++ b/packages/coding-agent/src/experimental/services/connection.ts @@ -7,13 +7,19 @@ import { type MutableReplicatedState, type RemoteServiceConnection, RemoteServiceNamespace, - type RemoteServiceNamespaceApi, + type RemoteServices, type ReplicatedState, remoteState, } from "@earendil-works/pi-agent-core"; import type { Client } from "@earendil-works/pi-client"; -import type { ProtocolRpcCall, RpcTarget, ServiceProviderUpdate, SessionTarget } from "@earendil-works/pi-protocol"; -import { BUILTIN_SERVER_SERVICES, BUILTIN_SESSION_SERVICES } from "./builtins.ts"; +import type { + ProtocolRpcCall, + RpcTarget, + ServiceCatalogueEntry, + ServiceProviderUpdate, + SessionTarget, +} from "@earendil-works/pi-protocol"; +import type { FacetConnection } from "../facets.ts"; export type ServerConnectionState = | { status: "connecting"; attempt: number } @@ -24,44 +30,90 @@ export type SessionAttachmentState = | { status: "detached" } | { status: "attaching" | "attached" | "degraded"; sessionId: string }; -export interface ServerServices extends RemoteServiceNamespaceApi { +export interface ServerServiceConnection extends FacetConnection { readonly connection: ReplicatedState; - activate(context: Context): Promise; + dispose(context: Context): Promise; } -export interface SessionServices extends RemoteServiceNamespaceApi { +export interface SessionServiceConnection extends FacetConnection { readonly attachment: ReplicatedState; - activate(context: Context): Promise; /** Wait for the exact current attachment generation to finish hydrating. */ whenAttached(sessionId: string, context: Context): Promise; - /** Wait for the detached namespace to finish releasing its prior binding. */ + /** Wait for every binding to finish releasing the previous attachment. */ whenDetached(context: Context): Promise; + dispose(context: Context): Promise; } -export interface ServiceNamespaceOptions { - readonly services: readonly { readonly id: string }[]; +export interface ServiceConnectionOptions { readonly onError?: (error: Error) => void; - readonly deferred?: boolean; } -class ServerServiceNamespace extends RemoteServiceNamespace implements ServerServices { - readonly connection: ReplicatedState; - readonly #removeConnectionListener: () => void; - readonly #onError: ((error: Error) => void) | undefined; - #transition = Promise.resolve(); - #connectionAttempt: number; - #enabled: boolean; +export interface ServiceNamespaceOptions extends ServiceConnectionOptions { + readonly services: readonly { readonly id: string }[]; +} - constructor(client: Client, connection: RemoteServiceConnection, options: ServiceNamespaceOptions) { - const enabled = options.deferred !== true; +class RoutedServiceBinding extends RemoteServiceNamespace { + readonly #getBound: () => boolean; + readonly #onActivate: (context: Context) => Promise; + readonly #remove: () => void; + #activated = false; + + constructor(options: { + readonly services: readonly { readonly id: string }[]; + readonly connection: RemoteServiceConnection; + readonly getBound: () => boolean; + readonly assertAccess: () => void; + readonly onError: (error: Error) => void; + readonly onActivate?: (context: Context) => Promise; + readonly remove: () => void; + }) { super({ services: options.services, - connection, - bound: enabled && client.connected, - ...(options.onError === undefined ? {} : { onError: options.onError }), + connection: options.connection, + bound: false, + assertAccess: options.assertAccess, + onError: options.onError, }); - this.#onError = options.onError; - this.#enabled = enabled; + this.#getBound = options.getBound; + this.#onActivate = options.onActivate ?? (() => Promise.resolve()); + this.#remove = options.remove; + } + + override async activate(context: Context): Promise { + if (!this.#activated) { + this.#activated = true; + await this.rebind(this.#getBound(), context); + } + await this.ready(context); + await this.#onActivate(context); + } + + updateBound(bound: boolean, context: Context): Promise { + return this.#activated ? this.rebind(bound, context) : Promise.resolve(); + } + + override async dispose(context: Context): Promise { + this.#remove(); + await super.dispose(context); + } +} + +class ServerServiceConnectionImpl implements ServerServiceConnection { + readonly acceptsUnavailableServices = false; + readonly connection: ReplicatedState; + readonly #client: Client; + readonly #remoteConnection: RemoteServiceConnection; + readonly #bindings = new Set(); + readonly #removeConnectionListener: () => void; + readonly #onError: (error: Error) => void; + #transition = Promise.resolve(); + #connectionAttempt: number; + #disposed = false; + + constructor(client: Client, options: ServiceConnectionOptions) { + this.#client = client; + this.#remoteConnection = createRemoteServiceConnection(client, () => ({ serverId: client.serverId })); + this.#onError = options.onError ?? (() => {}); this.#connectionAttempt = client.connectionState === "connecting" ? 1 : 0; const connectionState = remoteState( toServerConnectionState(client, this.#connectionAttempt), @@ -70,60 +122,76 @@ class ServerServiceNamespace extends RemoteServiceNamespace implements ServerSer this.#removeConnectionListener = client.onConnectionStateChange(({ state, error }) => { if (state === "connecting") this.#connectionAttempt += 1; connectionState.set(toServerConnectionState(client, this.#connectionAttempt, error), BACKGROUND_CONTEXT); - if (!this.#enabled) return; this.#transition = this.#transition - .then(() => this.rebind(state === "connected", BACKGROUND_CONTEXT)) - .catch((transitionError: unknown) => this.#onError?.(toError(transitionError))); + .then(async () => { + const results = await Promise.allSettled( + [...this.#bindings].map((binding) => binding.updateBound(state === "connected", BACKGROUND_CONTEXT)), + ); + const failures = results.filter( + (result): result is PromiseRejectedResult => result.status === "rejected", + ); + if (failures.length > 0) throw new AggregateError(failures.map(({ reason }) => reason)); + }) + .catch((transitionError: unknown) => this.#onError(toError(transitionError))); }); } - async activate(context: Context): Promise { - if (!this.#enabled) { - this.#enabled = true; - this.#transition = this.#transition.then(() => - this.rebind(this.connection.value?.status === "connected", context), - ); - } - await this.ready(context); + catalogue(context: Context) { + return this.#client.serviceCatalogue({ serverId: this.#client.serverId }, context.abortSignal); } - override async ready(context: Context): Promise { - while (true) { - const transition = this.#transition; - await transition; - await super.ready(context); - if (transition === this.#transition) return; - } + open(options: { + readonly services: readonly { readonly id: string }[]; + assertAccess(): void; + onError(error: Error): void; + }): RemoteServices { + if (this.#disposed) throw new Error("Server service connection is disposed"); + let binding!: RoutedServiceBinding; + binding = new RoutedServiceBinding({ + services: options.services, + connection: this.#remoteConnection, + getBound: () => this.#client.connected, + assertAccess: options.assertAccess, + onError: options.onError, + remove: () => this.#bindings.delete(binding), + }); + this.#bindings.add(binding); + return binding; } - override async dispose(context: Context): Promise { + async dispose(context: Context): Promise { + if (this.#disposed) return; + this.#disposed = true; this.#removeConnectionListener(); await this.#transition; - await super.dispose(context); + const bindings = [...this.#bindings]; + this.#bindings.clear(); + const results = await Promise.allSettled(bindings.map((binding) => binding.dispose(context))); + throwFailures(results, "Failed to dispose server service connection"); } } -class SessionServiceNamespace extends RemoteServiceNamespace implements SessionServices { +class SessionServiceConnectionImpl implements SessionServiceConnection { readonly attachment: ReplicatedState; readonly #client: Client; + readonly #remoteConnection: RemoteServiceConnection; readonly #attachmentState: MutableReplicatedState; + readonly #bindings = new Set(); readonly #removeAttachmentListener: () => void; - readonly #onError: ((error: Error) => void) | undefined; + readonly #onError: (error: Error) => void; readonly #transitions = new Set>(); + #catalogue: readonly ServiceCatalogueEntry[] | undefined; #attachmentRevision = 0; - #enabled: boolean; + #disposed = false; - constructor(client: Client, connection: RemoteServiceConnection, options: ServiceNamespaceOptions) { - const enabled = options.deferred !== true; - super({ - services: options.services, - connection, - bound: enabled && client.attachment !== undefined, - ...(options.onError === undefined ? {} : { onError: options.onError }), - }); + get acceptsUnavailableServices(): boolean { + return this.#client.attachment === undefined && this.#catalogue === undefined; + } + + constructor(client: Client, options: ServiceConnectionOptions) { this.#client = client; - this.#onError = options.onError; - this.#enabled = enabled; + this.#remoteConnection = createRemoteServiceConnection(client, () => client.attachment); + this.#onError = options.onError ?? (() => {}); this.#attachmentState = remoteState( client.attachment === undefined ? { status: "detached" } @@ -134,11 +202,16 @@ class SessionServiceNamespace extends RemoteServiceNamespace implements SessionS const revision = ++this.#attachmentRevision; if (attachment !== undefined) { this.#attachmentState.set({ status: "attaching", sessionId: attachment.sessionId }, BACKGROUND_CONTEXT); - } else if (!this.#enabled) { - this.#attachmentState.set({ status: "detached" }, BACKGROUND_CONTEXT); + void this.#client.serviceCatalogue(attachment).then( + (catalogue) => { + if (this.#attachmentRevision === revision && sameAttachment(this.#client.attachment, attachment)) { + this.#catalogue = catalogue; + } + }, + (error: unknown) => this.#onError(toError(error)), + ); } - if (!this.#enabled) return; - const transition = this.rebind(attachment !== undefined, BACKGROUND_CONTEXT); + const transition = this.#rebind(attachment !== undefined, BACKGROUND_CONTEXT); this.#transitions.add(transition); void transition.then( () => { @@ -156,35 +229,48 @@ class SessionServiceNamespace extends RemoteServiceNamespace implements SessionS this.#transitions.delete(transition); if (this.#attachmentRevision !== revision || !sameAttachment(this.#client.attachment, attachment)) return; - if (attachment === undefined) { - this.#attachmentState.set({ status: "detached" }, BACKGROUND_CONTEXT); - } else { - this.#attachmentState.set( - { status: "degraded", sessionId: attachment.sessionId }, - BACKGROUND_CONTEXT, - ); - } - this.#onError?.(toError(error)); + this.#attachmentState.set( + attachment === undefined + ? { status: "detached" } + : { status: "degraded", sessionId: attachment.sessionId }, + BACKGROUND_CONTEXT, + ); + this.#onError(toError(error)); }, ); }); } - async activate(context: Context): Promise { - if (!this.#enabled) { - this.#enabled = true; - const transition = this.rebind(this.#client.attachment !== undefined, context); - this.#transitions.add(transition); - await transition.finally(() => this.#transitions.delete(transition)); - } - await this.ready(context); + async catalogue(context: Context): Promise { + const target = this.#client.attachment; + if (target === undefined) return this.#catalogue ?? []; + const catalogue = await this.#client.serviceCatalogue(target, context.abortSignal); + this.#catalogue = catalogue; + return catalogue; } - override async ready(context: Context): Promise { - const attachment = this.#client.attachment; - const revision = this.#attachmentRevision; - if (attachment === undefined) await this.#whenDetached(revision, context); - else await this.#whenAttached(attachment, revision, context); + open(options: { + readonly services: readonly { readonly id: string }[]; + assertAccess(): void; + onError(error: Error): void; + }): RemoteServices { + if (this.#disposed) throw new Error("Session service connection is disposed"); + let binding!: RoutedServiceBinding; + binding = new RoutedServiceBinding({ + services: options.services, + connection: this.#remoteConnection, + getBound: () => this.#client.attachment !== undefined, + onActivate: async (context) => { + const attachment = this.#client.attachment; + if (attachment === undefined) await this.#whenDetached(this.#attachmentRevision, context); + else await this.#whenAttached(attachment, this.#attachmentRevision, context); + }, + assertAccess: options.assertAccess, + onError: options.onError, + remove: () => this.#bindings.delete(binding), + }); + this.#bindings.add(binding); + return binding; } async whenAttached(sessionId: string, context: Context): Promise { @@ -200,15 +286,27 @@ class SessionServiceNamespace extends RemoteServiceNamespace implements SessionS await this.#whenDetached(this.#attachmentRevision, context); } - override async dispose(context: Context): Promise { + async dispose(context: Context): Promise { + if (this.#disposed) return; + this.#disposed = true; this.#removeAttachmentListener(); await Promise.allSettled(this.#transitions); - await super.dispose(context); + const bindings = [...this.#bindings]; + this.#bindings.clear(); + const results = await Promise.allSettled(bindings.map((binding) => binding.dispose(context))); + throwFailures(results, "Failed to dispose Session service connection"); + } + + async #rebind(bound: boolean, context: Context): Promise { + const results = await Promise.allSettled( + [...this.#bindings].map((binding) => binding.updateBound(bound, context)), + ); + throwFailures(results, "Failed to rebind Session services"); } async #whenAttached(attachment: SessionTarget, revision: number, context: Context): Promise { try { - await super.ready(context); + await Promise.all([...this.#bindings].map((binding) => binding.ready(context))); } catch (error) { if (this.#attachmentRevision === revision && sameAttachment(this.#client.attachment, attachment)) { this.#attachmentState.set({ status: "degraded", sessionId: attachment.sessionId }, context); @@ -225,7 +323,7 @@ class SessionServiceNamespace extends RemoteServiceNamespace implements SessionS } async #whenDetached(revision: number, context: Context): Promise { - await super.ready(context); + await Promise.all([...this.#bindings].map((binding) => binding.ready(context))); if (this.#attachmentRevision !== revision || this.#client.attachment !== undefined) { throw new Error("The Session attachment changed while detaching"); } @@ -235,39 +333,104 @@ class SessionServiceNamespace extends RemoteServiceNamespace implements SessionS } } -/** Bind plugin service facades to the connected server route. */ -export function createServerServiceNamespace(client: Client, options: ServiceNamespaceOptions): ServerServices { - const connection = createServiceConnection(client, () => ({ serverId: client.serverId })); - return new ServerServiceNamespace(client, connection, options); -} - -/** Bind every built-in server service expected by the presentation host. */ -export function createBuiltinServerServiceNamespace( +/** Create an explicitly bound server-service namespace outside a facet host. */ +export function createServerServiceNamespace( client: Client, - options: Omit = {}, -): ServerServices { - return createServerServiceNamespace(client, { ...options, services: BUILTIN_SERVER_SERVICES }); + options: ServiceNamespaceOptions, +): ServerServiceConnection & RemoteServices { + const connection = createServerServiceConnection(client, options); + const services = connection.open({ + services: options.services, + assertAccess() {}, + onError: options.onError ?? (() => {}), + }); + const activation = services.activate(BACKGROUND_CONTEXT); + void activation.catch(options.onError ?? (() => {})); + const ready = async (context: Context): Promise => { + await activation; + await services.ready(context); + }; + return { + acceptsUnavailableServices: connection.acceptsUnavailableServices, + connection: connection.connection, + catalogue: (context) => connection.catalogue(context), + open: (openOptions) => connection.open(openOptions), + use: (service) => services.use(service), + observe: (service, handler) => services.observe(service, handler), + activate: ready, + ready, + setAccessGuard: (assertAccess) => services.setAccessGuard(assertAccess), + async dispose(context) { + const results = await Promise.allSettled([services.dispose(context), connection.dispose(context)]); + throwFailures(results, "Failed to dispose server service namespace"); + }, + }; } -/** Bind plugin service facades to the client's selected Session route. */ -export function createSessionServiceNamespace(client: Client, options: ServiceNamespaceOptions): SessionServices { - const connection = createServiceConnection(client, () => client.attachment); - return new SessionServiceNamespace(client, connection, options); -} - -/** Bind every built-in Session service expected by the presentation host. */ -export function createBuiltinSessionServiceNamespace( +/** Create an explicitly bound selected-Session namespace outside a facet host. */ +export function createSessionServiceNamespace( client: Client, - options: Omit = {}, -): SessionServices { - return createSessionServiceNamespace(client, { ...options, services: BUILTIN_SESSION_SERVICES }); + options: ServiceNamespaceOptions, +): SessionServiceConnection & RemoteServices { + const connection = createSessionServiceConnection(client, options); + const services = connection.open({ + services: options.services, + assertAccess() {}, + onError: options.onError ?? (() => {}), + }); + const activation = services.activate(BACKGROUND_CONTEXT); + void activation.catch(options.onError ?? (() => {})); + const ready = async (context: Context): Promise => { + await activation; + const attachment = connection.attachment.value; + if (attachment?.status === "detached") await connection.whenDetached(context); + else if (attachment !== undefined) await connection.whenAttached(attachment.sessionId, context); + }; + return { + get acceptsUnavailableServices() { + return connection.acceptsUnavailableServices; + }, + attachment: connection.attachment, + catalogue: (context) => connection.catalogue(context), + open: (openOptions) => connection.open(openOptions), + use: (service) => services.use(service), + observe: (service, handler) => services.observe(service, handler), + activate: ready, + ready, + setAccessGuard: (assertAccess) => services.setAccessGuard(assertAccess), + whenAttached: (sessionId, context) => connection.whenAttached(sessionId, context), + whenDetached: (context) => connection.whenDetached(context), + async dispose(context) { + const results = await Promise.allSettled([services.dispose(context), connection.dispose(context)]); + throwFailures(results, "Failed to dispose Session service namespace"); + }, + }; } -function createServiceConnection(client: Client, getTarget: () => RpcTarget | undefined): RemoteServiceConnection { +/** Create the server-scoped service connection for one presentation client. */ +export function createServerServiceConnection( + client: Client, + options: ServiceConnectionOptions = {}, +): ServerServiceConnection { + return new ServerServiceConnectionImpl(client, options); +} + +/** Create the selected-Session service connection for one presentation client. */ +export function createSessionServiceConnection( + client: Client, + options: ServiceConnectionOptions = {}, +): SessionServiceConnection { + return new SessionServiceConnectionImpl(client, options); +} + +function createRemoteServiceConnection( + client: Client, + getTarget: () => RpcTarget | undefined, +): RemoteServiceConnection { return { invoke: async (call, context) => { const target = getTarget(); - if (target === undefined) throw new Error("Service namespace is not routed"); + if (target === undefined) throw new Error("Service connection is not routed"); const wireCall: ProtocolRpcCall = { serviceId: call.serviceId, ...(call.instance === undefined ? {} : { instance: call.instance }), @@ -280,7 +443,7 @@ function createServiceConnection(client: Client, getTarget: () => RpcTarget | un }, subscribe: async (serviceId, mode, listener, context) => { const target = getTarget(); - if (target === undefined) throw new Error("Service namespace is not routed"); + if (target === undefined) throw new Error("Service connection is not routed"); const subscription = await client.subscribeService( target, serviceId, @@ -322,6 +485,12 @@ function sameAttachment(left: SessionTarget | undefined, right: SessionTarget | ); } +function throwFailures(results: readonly PromiseSettledResult[], message: string): void { + const errors = results.flatMap((result) => (result.status === "rejected" ? [result.reason] : [])); + if (errors.length === 1) throw errors[0]; + if (errors.length > 1) throw new AggregateError(errors, message); +} + function toError(error: unknown): Error { return error instanceof Error ? error : new Error(String(error)); } diff --git a/packages/coding-agent/src/experimental/services/harness.ts b/packages/coding-agent/src/experimental/services/harness.ts index 344e6b89a..4f4801edb 100644 --- a/packages/coding-agent/src/experimental/services/harness.ts +++ b/packages/coding-agent/src/experimental/services/harness.ts @@ -1,4 +1,4 @@ import { type AgentHarness, type AgentLane, defineService } from "@earendil-works/pi-agent-core"; -export const Harness = defineService("pi.local.harness"); -export const Lane = defineService("pi.local.lane"); +export const Harness = defineService("pi.local.harness", { rpc: false }); +export const Lane = defineService("pi.local.lane", { rpc: false }); diff --git a/packages/coding-agent/src/experimental/services/models-provider.ts b/packages/coding-agent/src/experimental/services/models-provider.ts index f4edc03e1..f85826495 100644 --- a/packages/coding-agent/src/experimental/services/models-provider.ts +++ b/packages/coding-agent/src/experimental/services/models-provider.ts @@ -16,7 +16,7 @@ export interface ModelsRuntime { refresh(options?: Parameters[0]): ReturnType | undefined; getModel(...args: Parameters): ReturnType; } -export const ModelsRuntime = defineService("pi.local.models-runtime"); +export const ModelsRuntime = defineService("pi.local.models-runtime", { rpc: false }); export function createModelsRuntime(runtime: ModelRuntime | undefined): ModelsRuntime { return ( diff --git a/packages/coding-agent/src/experimental/services/server-builtins.ts b/packages/coding-agent/src/experimental/services/server-builtins.ts deleted file mode 100644 index 5850c1409..000000000 --- a/packages/coding-agent/src/experimental/services/server-builtins.ts +++ /dev/null @@ -1,4 +0,0 @@ -import { SessionDirectory, SessionManagement } from "./sessions.ts"; - -/** Complete documented server-service allowlist for the experimental presentation host. */ -export const BUILTIN_SERVER_SERVICES = [SessionDirectory, SessionManagement] as const; diff --git a/packages/coding-agent/src/experimental/services/server.ts b/packages/coding-agent/src/experimental/services/server.ts index 1e833d8fa..899d928b3 100644 --- a/packages/coding-agent/src/experimental/services/server.ts +++ b/packages/coding-agent/src/experimental/services/server.ts @@ -19,7 +19,6 @@ import { } from "@earendil-works/pi-protocol"; import type { RoutedServerServiceAttachment, RoutedServerServiceHost } from "@earendil-works/pi-server"; import { Check } from "typebox/value"; -import { BUILTIN_SERVER_SERVICES } from "./server-builtins.ts"; import { SessionDirectory, type SessionDirectoryEvent, @@ -66,7 +65,10 @@ export async function createExperimentalServerServices(options: { return { host: { attachClient(presentation) { - const provider = new RemoteServiceProvider(BUILTIN_SERVER_SERVICES); + const provider = new RemoteServiceProvider([ + { service: SessionDirectory, mode: "singleton" }, + { service: SessionManagement, mode: "singleton" }, + ]); provider.provide(SessionDirectory, { state: directory, events: directoryEvents }); provider.provide(SessionManagement, { create: (createOptions, context) => @@ -141,6 +143,7 @@ function createProviderAttachment( async invokeService(call, publish, context) { if (released) throw new Error("Server service attachment is released"); const control = decodeServiceControlCall(call); + if (control?.type === "catalogue") return toProtocolJson(provider.catalogue); if (control?.type === "subscribe") { if (subscriptions.has(control.subscriptionId)) { throw new Error("Service subscription ID is already active"); diff --git a/packages/coding-agent/src/experimental/services/session-builtins.ts b/packages/coding-agent/src/experimental/services/session-builtins.ts deleted file mode 100644 index 4728d2d9a..000000000 --- a/packages/coding-agent/src/experimental/services/session-builtins.ts +++ /dev/null @@ -1,10 +0,0 @@ -import { Accounts } from "./accounts.ts"; -import { Chat } from "./chat.ts"; -import { Models } from "./models.ts"; -import { Transcript } from "./transcript.ts"; - -/** - * Complete built-in session-service allowlist for the experimental presentation host. - * Some providers intentionally throw ServiceSliceNotImplemented until their implementation slice lands. - */ -export const BUILTIN_SESSION_SERVICES = [Accounts, Chat, Models, Transcript] as const; diff --git a/packages/coding-agent/src/experimental/services/worker.ts b/packages/coding-agent/src/experimental/services/worker.ts index 08b72c7d6..5055655c0 100644 --- a/packages/coding-agent/src/experimental/services/worker.ts +++ b/packages/coding-agent/src/experimental/services/worker.ts @@ -3,6 +3,7 @@ import type { AgentLane, Context, ServiceProviderUpdate as CoreServiceProviderUpdate, + ServiceCatalogueEntry, ServiceProviderSubscription, } from "@earendil-works/pi-agent-core"; import { @@ -16,7 +17,8 @@ import { import Type, { type Static } from "typebox"; import { Check } from "typebox/value"; import type { ModelRuntime } from "../../core/model-runtime.ts"; -import { bindService, createFacetHost, type Facet } from "../facets.ts"; +import { combineFacetLoaders, createStaticFacetLoader, type FacetLoader } from "../facet-loader.ts"; +import { createFacetHost, defineFacet, type Facet, type FacetHost } from "../facets.ts"; import { chatServiceFacet } from "./chat-provider.ts"; import { Harness, Lane } from "./harness.ts"; import { createModelsRuntime, ModelsRuntime, modelsServiceFacet } from "./models-provider.ts"; @@ -32,7 +34,7 @@ export interface SessionWorkerRuntime { readonly harness: AgentHarness; readonly lane?: AgentLane; readonly modelRuntime?: ModelRuntime; - readonly facets?: readonly Facet[]; + readonly facetLoader?: FacetLoader; } export interface WorkerServiceScope { @@ -46,33 +48,51 @@ interface WorkerServiceSubscription { } export interface SessionWorkerServices { + readonly catalogue: readonly ServiceCatalogueEntry[]; invoke(call: ProtocolRpcCall, scope: WorkerServiceScope, context: Context): Promise; removeSubscriptions(matches: (scope: WorkerServiceScope) => boolean): void; dispose(): Promise; } -const BUILTIN_SESSION_FACETS = [ +const BUILTIN_SESSION_FACET_LOADER = createStaticFacetLoader([ chatServiceFacet, modelsServiceFacet, accountsServiceFacet, transcriptServiceFacet, -] satisfies readonly Facet[]; +] satisfies readonly Facet[]); export async function createSessionWorkerServices(options: { readonly harness: AgentHarness; readonly lane: AgentLane; readonly modelRuntime: ModelRuntime | undefined; - readonly facets: readonly Facet[]; + readonly facetLoader?: FacetLoader; publish(scope: WorkerServiceScope, subscriptionId: string, update: ProtocolServiceProviderUpdate): Promise; }): Promise { - const facetHost = await createFacetHost({ - facets: [...BUILTIN_SESSION_FACETS, ...options.facets], - bindings: [ - bindService(Harness, options.harness), - bindService(Lane, options.lane), - bindService(ModelsRuntime, createModelsRuntime(options.modelRuntime)), - ], + const sessionRuntimeFacet = defineFacet({ + id: "@pi/session-runtime", + setup(env) { + env.provide(Harness, options.harness); + env.provide(Lane, options.lane); + env.provide(ModelsRuntime, createModelsRuntime(options.modelRuntime)); + }, }); + const loader = combineFacetLoaders([ + createStaticFacetLoader([sessionRuntimeFacet]), + BUILTIN_SESSION_FACET_LOADER, + ...(options.facetLoader === undefined ? [] : [options.facetLoader]), + ]); + const loaded = await loader.load(); + let facetHost: FacetHost; + try { + facetHost = await createFacetHost({ facets: loaded.facets }); + } catch (error) { + try { + await loaded.dispose(); + } catch (cleanupError) { + throw new AggregateError([error, cleanupError], "Session facets failed to start and clean up"); + } + throw error; + } const provider = facetHost.services; const subscriptions = new Map(); @@ -85,8 +105,10 @@ export async function createSessionWorkerServices(options: { }; return { + catalogue: provider.catalogue, async invoke(call, scope, context) { const controlCall = decodeServiceControlCall(call); + if (controlCall?.type === "catalogue") return toProtocolJson(provider.catalogue); if (controlCall?.type === "subscribe") { const key = scopedSubscriptionKey(scope, controlCall.subscriptionId); if (subscriptions.has(key)) throw new Error("Service subscription ID is already active"); @@ -110,7 +132,19 @@ export async function createSessionWorkerServices(options: { removeSubscriptions, async dispose() { removeSubscriptions(() => true); - await facetHost.dispose(); + const errors: unknown[] = []; + try { + await facetHost.dispose(); + } catch (error) { + errors.push(error); + } + try { + await loaded.dispose(); + } catch (error) { + errors.push(error); + } + if (errors.length === 1) throw errors[0]; + if (errors.length > 1) throw new AggregateError(errors, "Failed to dispose Session facets"); }, }; } diff --git a/packages/coding-agent/src/experimental/session-worker.ts b/packages/coding-agent/src/experimental/session-worker.ts index b402f8b92..f5601e3cd 100644 --- a/packages/coding-agent/src/experimental/session-worker.ts +++ b/packages/coding-agent/src/experimental/session-worker.ts @@ -590,7 +590,7 @@ async function run(options: SessionWorkerOptions, createHarness: CreateSessionWo harness, lane, modelRuntime: runtime.modelRuntime, - facets: runtime.facets ?? [], + facetLoader: runtime.facetLoader, publish: (scope, subscriptionId, update) => control.send({ type: "service_event", diff --git a/packages/coding-agent/test/experimental-chat-service.test.ts b/packages/coding-agent/test/experimental-chat-service.test.ts index df292d7c8..f80b6de71 100644 --- a/packages/coding-agent/test/experimental-chat-service.test.ts +++ b/packages/coding-agent/test/experimental-chat-service.test.ts @@ -10,7 +10,7 @@ import { UnknownTemplate, } from "@earendil-works/pi-agent-core"; import { describe, expect, test, vi } from "vitest"; -import { bindService, createFacetHost } from "../src/experimental/facets.ts"; +import { createFacetHost, defineFacet } from "../src/experimental/facets.ts"; import { chatServiceFacet, toChatPromptResponse } from "../src/experimental/services/chat-provider.ts"; import { Lane } from "../src/experimental/services/harness.ts"; @@ -55,10 +55,13 @@ describe("Chat service", () => { }, })); const lane = { prompt, requestAbort } as unknown as AgentLane; - const host = await createFacetHost({ - facets: [chatServiceFacet], - bindings: [bindService(Lane, lane)], + const laneFacet = defineFacet({ + id: "test-lane", + setup(env) { + env.provide(Lane, lane); + }, }); + const host = await createFacetHost({ facets: [laneFacet, chatServiceFacet] }); try { await expect( host.services.invoke( diff --git a/packages/coding-agent/test/experimental-client-tui.test.ts b/packages/coding-agent/test/experimental-client-tui.test.ts index d156ee58f..7816d40d5 100644 --- a/packages/coding-agent/test/experimental-client-tui.test.ts +++ b/packages/coding-agent/test/experimental-client-tui.test.ts @@ -9,11 +9,13 @@ import { 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 { FacetLoader } from "../src/experimental/facet-loader.ts"; +import { defineFacet } from "../src/experimental/facets.ts"; import type { ServerConnectionState, - ServerServices, + ServerServiceConnection, SessionAttachmentState, - SessionServices, + SessionServiceConnection, } from "../src/experimental/services/connection.ts"; import { Models, type ModelsState } from "../src/experimental/services/models.ts"; import { @@ -83,8 +85,15 @@ describe("experimental client TUI", () => { connection: createLoopbackServiceConnection(serverProvider), bound: false, }); - const serverServices: ServerServices = Object.assign(serverNamespace, { + const serverServices: ServerServiceConnection = Object.assign(serverNamespace, { + acceptsUnavailableServices: false, connection: remoteState({ status: "connected", since: "now" }), + async catalogue() { + return serverProvider.catalogue; + }, + open() { + return serverNamespace; + }, async activate() { await serverNamespace.rebind(true, BACKGROUND_CONTEXT); await serverNamespace.ready(BACKGROUND_CONTEXT); @@ -95,8 +104,15 @@ describe("experimental client TUI", () => { connection: createLoopbackServiceConnection(sessionProvider), bound: false, }); - const sessionServices: SessionServices = Object.assign(sessionNamespace, { + const sessionServices: SessionServiceConnection = Object.assign(sessionNamespace, { + acceptsUnavailableServices: true, attachment, + async catalogue() { + return []; + }, + open() { + return sessionNamespace; + }, async activate() { await sessionNamespace.ready(BACKGROUND_CONTEXT); }, @@ -113,8 +129,18 @@ describe("experimental client TUI", () => { const server: ClientTuiServer = { serverId, server: serverServices, session: sessionServices }; let finished = false; const requestRender = vi.fn(); + const disposeLoadedFacets = vi.fn(async () => {}); + const facetLoader: FacetLoader = { + async load() { + return { + facets: [defineFacet({ id: "test-tui-facet", setup() {} })], + dispose: disposeLoadedFacets, + }; + }, + }; const component = await ExperimentalClientTui.create({ servers: [server], + facetLoader, requestRender, finish() { finished = true; @@ -144,6 +170,7 @@ describe("experimental client TUI", () => { expect(finished).toBe(true); await component.close(); + expect(disposeLoadedFacets).toHaveBeenCalledOnce(); const rendersAfterClose = requestRender.mock.calls.length; directoryState.set({ revision: 3, sessions: [] }, BACKGROUND_CONTEXT); directoryEvents.emit({ type: "deleted", sessionId: "two" }, BACKGROUND_CONTEXT); diff --git a/packages/coding-agent/test/experimental-facet-loader.test.ts b/packages/coding-agent/test/experimental-facet-loader.test.ts new file mode 100644 index 000000000..d8a3348dc --- /dev/null +++ b/packages/coding-agent/test/experimental-facet-loader.test.ts @@ -0,0 +1,67 @@ +import { describe, expect, test, vi } from "vitest"; +import { combineFacetLoaders, createStaticFacetLoader, type FacetLoader } from "../src/experimental/facet-loader.ts"; +import { defineFacet } from "../src/experimental/facets.ts"; + +const firstFacet = defineFacet({ id: "first", setup() {} }); +const secondFacet = defineFacet({ id: "second", setup() {} }); + +describe("experimental facet loader", () => { + test("combines loaded facets in loader order and disposes generations in reverse", async () => { + const trace: string[] = []; + const first: FacetLoader = { + async load() { + trace.push("load first"); + return { + facets: [firstFacet], + async dispose() { + trace.push("dispose first"); + }, + }; + }, + }; + const second: FacetLoader = { + async load() { + trace.push("load second"); + return { + facets: [secondFacet], + async dispose() { + trace.push("dispose second"); + }, + }; + }, + }; + + const loaded = await combineFacetLoaders([first, second]).load(); + expect(loaded.facets).toEqual([firstFacet, secondFacet]); + await loaded.dispose(); + await loaded.dispose(); + expect(trace).toEqual(["load first", "load second", "dispose second", "dispose first"]); + }); + + test("cleans up loaded facets when a later loader fails", async () => { + const dispose = vi.fn(async () => {}); + const failure = new Error("load failed"); + const first: FacetLoader = { + async load() { + return { facets: [firstFacet], dispose }; + }, + }; + const second: FacetLoader = { + async load() { + throw failure; + }, + }; + + await expect(combineFacetLoaders([first, second]).load()).rejects.toBe(failure); + expect(dispose).toHaveBeenCalledOnce(); + }); + + test("creates a reusable static loader", async () => { + const loader = createStaticFacetLoader([firstFacet]); + const first = await loader.load(); + const second = await loader.load(); + expect(first.facets).toEqual([firstFacet]); + expect(second.facets).toEqual([firstFacet]); + await Promise.all([first.dispose(), second.dispose()]); + }); +}); diff --git a/packages/coding-agent/test/experimental-facets.test.ts b/packages/coding-agent/test/experimental-facets.test.ts index 5cec64250..0568f605d 100644 --- a/packages/coding-agent/test/experimental-facets.test.ts +++ b/packages/coding-agent/test/experimental-facets.test.ts @@ -1,12 +1,15 @@ import { BACKGROUND_CONTEXT, type Context, + createLoopbackServiceConnection, defineService, type MutableReplicatedState, + RemoteServiceNamespace, + RemoteServiceProvider, type ReplicatedState, } from "@earendil-works/pi-agent-core"; import { describe, expect, test, vi } from "vitest"; -import { bindService, createFacetHost, defineFacet } from "../src/experimental/facets.ts"; +import { createFacetHost, defineFacet, type FacetConnection } from "../src/experimental/facets.ts"; interface Source { read(context: Context): Promise; @@ -29,11 +32,26 @@ interface HostValues { readonly use: string; } +interface LeftValue { + read(context: Context): Promise; +} + +interface RightValue { + read(context: Context): Promise; +} + +interface CombinedValue { + read(context: Context): Promise; +} + const Source = defineService("test.experimental.source"); const Projection = defineService("test.experimental.projection"); const KeyedValue = defineService("test.experimental.keyed-value"); const Watched = defineService("test.experimental.watched"); -const HostValues = defineService("test.experimental.host-values"); +const HostValues = defineService("test.experimental.host-values", { rpc: false }); +const LeftValue = defineService("test.experimental.left-value"); +const RightValue = defineService("test.experimental.right-value"); +const CombinedValue = defineService("test.experimental.combined-value"); describe("experimental facet host", () => { test("discovers setup dependencies before connecting stable service handles", async () => { @@ -152,9 +170,9 @@ describe("experimental facet host", () => { expect(deliveries).toBe(2); }); - test("resolves explicit host service bindings", async () => { + test("provides arbitrary host services through the facet graph", async () => { const values: HostValues = { name: "session", use: "host value" }; - const facet = defineFacet({ + const consumer = defineFacet({ id: "host-service-consumer", setup(env) { const hostValues = env.use(HostValues); @@ -166,10 +184,179 @@ describe("experimental facet host", () => { }); }, }); - const host = await createFacetHost({ facets: [facet], bindings: [bindService(HostValues, values)] }); + const provider = defineFacet({ + id: "host-service-provider", + setup(env) { + env.provide(HostValues, values); + }, + }); + const host = await createFacetHost({ facets: [consumer, provider] }); + expect(() => host.services.use(HostValues)).toThrow( + "Remote service test.experimental.host-values is not allowlisted", + ); await host.dispose(); }); + test("combines connected services and facet-provided services in one host", async () => { + const leftProvider = new RemoteServiceProvider([LeftValue]); + leftProvider.provide(LeftValue, { + async read() { + return "left"; + }, + }); + const rightProvider = new RemoteServiceProvider([RightValue]); + rightProvider.provide(RightValue, { + async read() { + return "right"; + }, + }); + const leftNamespace = new RemoteServiceNamespace({ + services: [LeftValue], + connection: createLoopbackServiceConnection(leftProvider), + bound: false, + }); + const rightNamespace = new RemoteServiceNamespace({ + services: [RightValue], + connection: createLoopbackServiceConnection(rightProvider), + bound: false, + }); + const connections = [ + { namespace: leftNamespace, provider: leftProvider }, + { namespace: rightNamespace, provider: rightProvider }, + ].map(({ namespace, provider }) => + Object.assign(namespace, { + acceptsUnavailableServices: false, + async catalogue() { + return provider.catalogue; + }, + open() { + return namespace; + }, + async activate(context: Context) { + await namespace.rebind(true, context); + await namespace.ready(context); + }, + }), + ); + const facet = defineFacet({ + id: "combined", + setup(env) { + const left = env.use(LeftValue); + const right = env.use(RightValue); + env.provide(CombinedValue, { + async read(context) { + return `${await left.read(context)} ${await right.read(context)}`; + }, + }); + }, + }); + + const host = await createFacetHost({ facets: [facet], connections }); + await expect(host.services.use(CombinedValue).read(BACKGROUND_CONTEXT)).resolves.toBe("left right"); + + await host.dispose(); + await Promise.all([leftNamespace.dispose(BACKGROUND_CONTEXT), rightNamespace.dispose(BACKGROUND_CONTEXT)]); + leftProvider.dispose(); + rightProvider.dispose(); + }); + + test("rejects a service offered by multiple connections", async () => { + const duplicate = { + acceptsUnavailableServices: false, + async catalogue() { + return [{ serviceId: LeftValue.id, mode: "singleton" as const }]; + }, + open() { + throw new Error("Ambiguous connections must not open"); + }, + } satisfies FacetConnection; + const consumer = defineFacet({ + id: "duplicate-consumer", + setup(env) { + env.use(LeftValue); + }, + }); + await expect(createFacetHost({ facets: [consumer], connections: [duplicate, duplicate] })).rejects.toThrow( + `Facet host service ${LeftValue.id} is offered by more than one connection`, + ); + }); + + test("reopens connection bindings from changed catalogues for a replacement generation", async () => { + const leftProvider = new RemoteServiceProvider([LeftValue]); + leftProvider.provide(LeftValue, { + async read() { + return "left"; + }, + }); + const rightProvider = new RemoteServiceProvider([RightValue]); + rightProvider.provide(RightValue, { + async read() { + return "right"; + }, + }); + let currentProvider = leftProvider; + let opened = 0; + let disposed = 0; + const connection = { + acceptsUnavailableServices: false, + async catalogue() { + return currentProvider.catalogue; + }, + open(options) { + opened += 1; + const namespace = new RemoteServiceNamespace({ + services: options.services, + connection: createLoopbackServiceConnection(currentProvider), + bound: false, + assertAccess: options.assertAccess, + onError: options.onError, + }); + const dispose = namespace.dispose.bind(namespace); + return Object.assign(namespace, { + async activate(context: Context) { + await namespace.rebind(true, context); + await namespace.ready(context); + }, + async dispose(context: Context) { + disposed += 1; + await dispose(context); + }, + }); + }, + } satisfies FacetConnection; + const values: string[] = []; + const leftConsumer = defineFacet({ + id: "left-consumer", + setup(env) { + const left = env.use(LeftValue); + env.onActivate(async () => { + values.push(await left.read(BACKGROUND_CONTEXT)); + }); + }, + }); + const first = await createFacetHost({ facets: [leftConsumer], connections: [connection] }); + await first.dispose(); + + currentProvider = rightProvider; + const rightConsumer = defineFacet({ + id: "right-consumer", + setup(env) { + const right = env.use(RightValue); + env.onActivate(async () => { + values.push(await right.read(BACKGROUND_CONTEXT)); + }); + }, + }); + const second = await createFacetHost({ facets: [rightConsumer], connections: [connection] }); + await second.dispose(); + + expect(values).toEqual(["left", "right"]); + expect(opened).toBe(2); + expect(disposed).toBe(2); + leftProvider.dispose(); + rightProvider.dispose(); + }); + test("rejects missing dependencies, cycles, and asynchronous setup", async () => { let activated = false; const missing = defineFacet({ diff --git a/packages/coding-agent/test/experimental-remote-runtime.test.ts b/packages/coding-agent/test/experimental-remote-runtime.test.ts index d756cb9f0..432a30bbc 100644 --- a/packages/coding-agent/test/experimental-remote-runtime.test.ts +++ b/packages/coding-agent/test/experimental-remote-runtime.test.ts @@ -6,10 +6,12 @@ import { createUnixTransportFactory } from "@earendil-works/pi-client/unix"; import { afterEach, beforeEach, describe, expect, test, vi } from "vitest"; import { runClient } from "../src/experimental/client.ts"; import { activateBuiltinClientServices, openClientRuntime } from "../src/experimental/client-runtime.ts"; +import { createFacetHost, defineFacet } from "../src/experimental/facets.ts"; import * as processRuntime from "../src/experimental/process.ts"; import { activateServer, type RunningServer, startServer } from "../src/experimental/server.ts"; import { createServerServiceNamespace, + createSessionServiceConnection, createSessionServiceNamespace, type SessionAttachmentState, } from "../src/experimental/services/connection.ts"; @@ -311,6 +313,10 @@ describe("experimental durable server composition", () => { }); expect(firstServices.connection.value).toMatchObject({ status: "connected" }); expect(secondServices.connection.value).toMatchObject({ status: "connected" }); + await expect(firstServices.catalogue(BACKGROUND_CONTEXT)).resolves.toEqual([ + { serviceId: SessionDirectory.id, mode: "singleton" }, + { serviceId: SessionManagement.id, mode: "singleton" }, + ]); const firstDirectory = firstServices.use(SessionDirectory); const secondDirectory = secondServices.use(SessionDirectory); const firstManagement = firstServices.use(SessionManagement); @@ -523,18 +529,26 @@ describe("experimental durable server composition", () => { const { runtime } = await makeServer(); const client = await attachClient(runtime, "demo-1"); const errors: Error[] = []; - const services = createSessionServiceNamespace(client, { - services: [KeyedProbe], + const services = createSessionServiceConnection(client, { onError: (error) => errors.push(error), }); - const observed: { service: KeyedProbe; context: Context; value: string | undefined }[] = []; - const stop = services.observe(KeyedProbe, (instance, context) => { - observed.push({ service: instance.service, context, value: instance.service.state.value?.value }); + await expect(services.catalogue(BACKGROUND_CONTEXT)).resolves.toContainEqual({ + serviceId: KeyedProbe.id, + mode: "keyed", }); + const observed: { service: KeyedProbe; context: Context; value: string | undefined }[] = []; + const consumer = defineFacet({ + id: "@test/keyed-probe-consumer", + setup(env) { + env.observe(KeyedProbe, (instance, context) => { + observed.push({ service: instance.service, context, value: instance.service.state.value?.value }); + }); + }, + }); + const facetHost = await createFacetHost({ facets: [consumer], connections: [services] }); - await services.ready(BACKGROUND_CONTEXT); expect(services.attachment.value).toEqual({ status: "attached", sessionId: "demo-1" }); - expect(observed).toHaveLength(1); + await vi.waitFor(() => expect(observed).toHaveLength(1)); expect(observed[0]!.value).toBe("first"); const stale = observed[0]!.service; const cancellable = withCancel(BACKGROUND_CONTEXT); @@ -562,7 +576,7 @@ describe("experimental durable server composition", () => { }); expect(errors).toEqual([]); - stop(); + await facetHost.dispose(); await expect(services.dispose(BACKGROUND_CONTEXT)).resolves.toBeUndefined(); }); diff --git a/packages/coding-agent/test/experimental-service-stubs.test.ts b/packages/coding-agent/test/experimental-service-stubs.test.ts index 02714213a..130959e3f 100644 --- a/packages/coding-agent/test/experimental-service-stubs.test.ts +++ b/packages/coding-agent/test/experimental-service-stubs.test.ts @@ -1,20 +1,9 @@ import { BACKGROUND_CONTEXT, ServiceSliceNotImplemented } from "@earendil-works/pi-agent-core"; import { describe, expect, test } from "vitest"; import { createFacetHost } from "../src/experimental/facets.ts"; -import { BUILTIN_SERVER_SERVICES, BUILTIN_SESSION_SERVICES } from "../src/experimental/services/builtins.ts"; import { accountsServiceFacet, transcriptServiceFacet } from "../src/experimental/services/stubs-provider.ts"; describe("experimental built-in service surface", () => { - test("inventories every built-in server and Session service", () => { - expect(BUILTIN_SERVER_SERVICES.map(({ id }) => id)).toEqual(["pi.session-directory", "pi.session-management"]); - expect(BUILTIN_SESSION_SERVICES.map(({ id }) => id)).toEqual([ - "pi.accounts", - "pi.chat", - "pi.models", - "pi.transcript", - ]); - }); - test("exposes later singleton slices as explicit unimplemented providers", async () => { const host = await createFacetHost({ facets: [accountsServiceFacet, transcriptServiceFacet] }); try { diff --git a/packages/coding-agent/test/fixtures/faux-session-worker.ts b/packages/coding-agent/test/fixtures/faux-session-worker.ts index 765642f66..521d70a5a 100644 --- a/packages/coding-agent/test/fixtures/faux-session-worker.ts +++ b/packages/coding-agent/test/fixtures/faux-session-worker.ts @@ -2,6 +2,7 @@ import { resolve } from "node:path"; import { fileURLToPath } from "node:url"; import { AgentHarness, BACKGROUND_CONTEXT } from "@earendil-works/pi-agent-core"; import { createModels, fauxAssistantMessage, fauxProvider } from "@earendil-works/pi-ai"; +import { createStaticFacetLoader } from "../../src/experimental/facet-loader.ts"; import { defineFacet } from "../../src/experimental/facets.ts"; import { consumeInternalProcessRole } from "../../src/experimental/process.ts"; import { runSessionWorkerWithHarness } from "../../src/experimental/session-worker.ts"; @@ -56,7 +57,7 @@ if (process.argv[1] !== undefined && resolve(process.argv[1]) === fileURLToPath( env.onActivate(() => spawn("first")); }, }); - return { harness, facets: [keyedProbeFacet] }; + return { harness, facetLoader: createStaticFacetLoader([keyedProbeFacet]) }; }).catch((error: unknown) => { console.error(error); process.exit(1); diff --git a/packages/protocol/src/protocol.ts b/packages/protocol/src/protocol.ts index 77e327750..5278420b3 100644 --- a/packages/protocol/src/protocol.ts +++ b/packages/protocol/src/protocol.ts @@ -91,6 +91,20 @@ export const ServiceRpcResultSchema = Type.Unsafe(createR export const ServiceModeSchema = Type.Union([Type.Literal("singleton"), Type.Literal("keyed")]); export type ServiceMode = Static; +export const ServiceCatalogueEntrySchema = StrictObject({ + serviceId: IdSchema, + mode: ServiceModeSchema, +}); +export const ServiceCatalogueSchema = Type.Array(ServiceCatalogueEntrySchema); +export type ServiceCatalogueEntry = Static; + +export function parseServiceCatalogue(value: unknown): readonly ServiceCatalogueEntry[] { + if (!Check(ServiceCatalogueSchema, value)) throw new TypeError("Invalid service catalogue"); + const ids = value.map(({ serviceId }) => serviceId); + if (new Set(ids).size !== ids.length) throw new TypeError("Service catalogue contains duplicate IDs"); + return value; +} + export const ServiceInstanceAddressSchema = StrictObject({ key: IdSchema, generation: Type.Integer({ minimum: 1 }), @@ -166,10 +180,12 @@ export const ServiceProviderUpdateSchema = Type.Union([ export type ServiceProviderUpdate = Static; export const SERVICE_CONTROL_ID = "$pi.service"; +export const SERVICE_CATALOGUE_MEMBER = "catalogue"; export const SERVICE_SUBSCRIBE_MEMBER = "subscribe"; export const SERVICE_UNSUBSCRIBE_MEMBER = "unsubscribe"; export type ServiceControlCall = + | { readonly type: "catalogue" } | { readonly type: "subscribe"; readonly subscriptionId: string; @@ -178,6 +194,10 @@ export type ServiceControlCall = } | { readonly type: "unsubscribe"; readonly subscriptionId: string }; +export function createServiceCatalogueCall(): ProtocolRpcCall { + return { serviceId: SERVICE_CONTROL_ID, member: SERVICE_CATALOGUE_MEMBER, args: [] }; +} + export function createServiceSubscribeCall( subscriptionId: string, serviceId: string, @@ -192,6 +212,7 @@ export function createServiceUnsubscribeCall(subscriptionId: string): ProtocolRp export function decodeServiceControlCall(call: ProtocolRpcCall): ServiceControlCall | undefined { if (call.serviceId !== SERVICE_CONTROL_ID || call.instance !== undefined) return undefined; + if (call.member === SERVICE_CATALOGUE_MEMBER && call.args.length === 0) return { type: "catalogue" }; if ( call.member === SERVICE_SUBSCRIBE_MEMBER && call.args.length === 3 && diff --git a/packages/protocol/test/protocol.test.ts b/packages/protocol/test/protocol.test.ts index ba5a6dee9..d2d9cd89b 100644 --- a/packages/protocol/test/protocol.test.ts +++ b/packages/protocol/test/protocol.test.ts @@ -6,6 +6,7 @@ import { ClientMessageDecoder, createRpcClient, createRpcDispatcher, + createServiceCatalogueCall, createServiceSubscribeCall, createServiceUnsubscribeCall, decodeCbor, @@ -27,6 +28,7 @@ import { ProtocolValidationError, parseClientMessage, parseServerMessage, + parseServiceCatalogue, parseServiceSubscriptionSnapshot, type RunResult, RunResultSchema, @@ -245,6 +247,16 @@ describe("protocol validation", () => { }); test("encodes service control and keyed instance addresses", () => { + expect(decodeServiceControlCall(createServiceCatalogueCall())).toEqual({ type: "catalogue" }); + expect( + parseServiceCatalogue([ + { serviceId: "pi.models", mode: "singleton" }, + { serviceId: "pi.dialogs", mode: "keyed" }, + ]), + ).toEqual([ + { serviceId: "pi.models", mode: "singleton" }, + { serviceId: "pi.dialogs", mode: "keyed" }, + ]); const subscribe = createServiceSubscribeCall("subscription-1", "pi.models", "singleton"); expect(decodeServiceControlCall(subscribe)).toEqual({ type: "subscribe",