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

feat(agent): remove remote service events

Browse source
This commit is contained in:
Armin Ronacher 2026-08-26 13:46:19 +02:00
parent 6f0ea81e8b
commit a17ead3292
37 changed files with 223 additions and 631 deletions
Download patch file Download diff file Expand all files Collapse all files

2
packages/agent/README.md
View file

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

16
packages/agent/docs/plugin-reloading.md
View file

@ -203,26 +203,22 @@ The experimental core services already illustrate the reconstruction rules:
| Service | Current owner | Reconstruction after reload |
|---|---|---|
| `SessionDirectory` | Server, with one provider per presentation over shared repository-backed state | List the repository again and publish a complete directory snapshot. Directory events during the gap are not replayed. |
| `SessionDirectory` | Server, with one provider per presentation over shared repository-backed state | List the repository again and publish a complete directory snapshot. |
| `SessionManagement` | Server provider closure bound to one presentation | Recreate the per-presentation implementation after reconnect. In-flight management calls follow ordinary uncertain-disconnect semantics. |
| `Models` | Session worker over Harness and `ModelRuntime` | Read selected model and thinking level from the Harness and rebuild the catalog from the model runtime. Process-local refresh status is discarded. |
| `Chat` | Session worker over the Harness | Durable accepted work remains in the Session. A call interrupted before receiving its operation ID has an uncertain outcome and must not be blindly retried. |
| `Accounts` | Session worker facade over credential authority | Rebuild presentation-safe account summaries from the credential source. Credentials never live in `ReplicatedState`. |
| `Transcript` | Session worker projection of durable lane state | Hydrate an authoritative revisioned snapshot, then resume semantic event delivery. Snapshot/event gap handling is required before removing the compatibility watch. |
| `Transcript` | Session worker projection of durable lane state | Implement an authoritative revisioned snapshot before removing the compatibility watch. |
`Accounts` and `Transcript` are currently incomplete scaffolds, but their reload ownership follows the same rule.
Provider-local counters are not automatically durable. `ReplicatedState` transport sequences, keyed generations, model-catalog revisions, and directory revisions may restart with a new provider binding unless their application contract explicitly makes them durable. Consumers must not compare binding-scoped revisions across a rebind as if they were globally monotonic.
## Replicated state and events
## Replicated state and keyed services
`ReplicatedState` is a projection, not storage. Every state needed after reload must be reconstructible from durable authority, configuration, or another owned source. On provider replacement, the consumer becomes unready and installs a fresh complete snapshot before later updates.
`RemoteEvents` is non-durable and non-replayed. Events emitted while no subscription is active are lost. A consumer that must recover after reload needs one of:
- a `ReplicatedState` snapshot;
- a pull method returning current authority; or
- an application-level revisioned snapshot/event protocol such as `Transcript`.
A consumer that must recover after reload needs a `ReplicatedState` snapshot or a pull method returning current authority.
Keyed services are live projections. If an instance must return after worker replacement, its owner reconstructs it from durable invocation or plugin records. The new provider may reuse the logical key, but it creates a fresh live generation under the new attachment binding. A stale proxy never locates durable work by key alone.
@ -269,7 +265,6 @@ This also follows Cordis's system-boundary caveat: acquiring and releasing a pro
- Reloading one facet while another active facet still imports the same changed source bytes.
- Opening the same Session in old and replacement workers concurrently.
- Treating `ReplicatedState` as the durable source rather than a reconstructible projection.
- Depending on `RemoteEvents` to repair missed state after restart.
- Reusing an attachment ID across worker generations and weakening stale-route fencing.
- Automatically retrying a call after reload when its durable outcome is uncertain.
- Deleting namespaced durable data when a plugin leaves the manifest.
@ -298,8 +293,7 @@ A later implementation handoff should cover at least:
- a committed durable operation remains discoverable after its initiating response is lost;
- logical Session selection survives the route gap, reports `attaching` or `degraded`, and receives a fresh attachment ID on success;
- old attachment calls, service updates, and keyed proxies are rejected after replacement;
- singleton state rehydrates from a complete new snapshot before updates and events flow;
- non-durable events are not replayed, while snapshot-backed services recover current state;
- singleton state rehydrates from a complete new snapshot before updates flow;
- durable keyed interactions reconstruct from records with fresh live generations;
- removed plugin data remains stored and is readable when the plugin returns;
- server replacement reads desired manifest and worker intentions from the server directory;

65
packages/agent/docs/plugins.md
View file

@ -1,6 +1,6 @@
# Coding-Agent Application Hosts and Plugin Facets
> **Status:** Tentative design input, not a normative contract or implementation handoff. The transport-neutral service token, provider, singleton `use()`, keyed `observe()`, and `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.
> **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. 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.
@ -199,7 +199,7 @@ interface ServiceInstances<T> {
}
```
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.
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` 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.
@ -267,7 +267,7 @@ export const providersBuiltin = definePlugin({
session(env) {
const providers = new ProviderRegistry(); // process-local, non-JSON
const state = env.remoteState<ModelsState>(initialModelsState());
const state = env.replicatedState<ModelsState>(initialModelsState());
env.provide(Models, {
state,
@ -421,8 +421,7 @@ interface FacetEnvironment extends FacetLifecycle {
): () => void;
provide<T>(service: Service<T>, implementation: T): void;
provideMany<T>(service: Service<T>): ServiceInstances<T>;
remoteState<T extends JsonValue>(initial: T): MutableReplicatedState<T>;
remoteEvents<T extends JsonValue>(): MutableRemoteEvents<T>;
replicatedState<T extends JsonValue>(initial: T): MutableReplicatedState<T>;
}
type AttachmentState = { status: "detached" } | { status: "attaching" | "attached" | "degraded"; sessionId: string };
@ -524,7 +523,7 @@ Required behavior:
Anything a consumer must recover after reconnect is exposed as replicated state or pulled through a remote method. Replicated state is latest-value replication, not by itself durable session storage; the providing facet must reconstruct its authoritative value after a worker restart.
Prefer several coarse independent cells over one giant value or a universal patch language, so a catalogue refresh does not retransmit unrelated configuration. High-frequency data such as transcript streaming should use semantic deltas plus a final authoritative replacement. Revision metadata, gap recovery, unchanged-value suppression, and demand-driven subscription are protocol concerns, not plugin-author concerns.
Prefer several coarse independent cells over one giant value or a universal patch language, so a catalogue refresh does not retransmit unrelated configuration. High-frequency data such as transcript streaming needs a future snapshot-and-delta design rather than overloading `ReplicatedState`. Revision metadata, gap recovery, unchanged-value suppression, and demand-driven subscription belong to that future protocol, not individual plugin authors.
## Contribution registries: many contributors, one result
@ -590,21 +589,6 @@ Three cancellation domains must never blur:
A transport disconnect performs only the first for active requests and closes that client's subscriptions. It must not silently cancel service-owned work or write durable cancellation. Work intended to outlive its initiating request must deliberately detach into a service-owned task with its own controller and telemetry root.
## Events
A remotely exposed event source is a projection of host-local events, not a remotely invoked callback. A Git plugin:
```ts
interface Git {
readonly status: ReplicatedState<GitStatus>;
readonly events: RemoteEvents<GitEvent>; // { type: "status_changed" | "head_changed" }
refresh(context: Context): Promise<void>;
}
const Git = defineService<Git>("pi.git");
```
Server-side, the exposure adapter subscribes to the host-local source once per remote subscription and forwards frames. Client-side, `events` is a local facade whose listeners run in the client process — callbacks never cross the wire; `on(type, listener)` filters by discriminator, `subscribe(listener)` observes everything. Each event carries source trace metadata from the providing host's context, so client-side handling stays correlated with the originating operation. The adapter owns subscribe/unsubscribe frames, sequencing, buffering, flow control, and disconnect cleanup (`rpc.md`). Remote events are non-durable and never replayed: a new or reconnected consumer sees only events emitted after its subscription becomes active. Anything needed to reconstruct current behavior after reconnect belongs in replicated state or a pull method.
## Service-owned jobs
Long-running work should return a capability rather than one method that blocks indefinitely:
@ -672,13 +656,8 @@ export interface SessionRecordSummary {
updatedAt: string;
}
export type SessionDirectoryEvent =
| { type: "created" | "changed"; session: SessionRecordSummary }
| { type: "deleted"; sessionId: string };
export interface SessionDirectory {
readonly state: ReplicatedState<{ revision: number; sessions: SessionRecordSummary[] }>;
readonly events: RemoteEvents<SessionDirectoryEvent>;
}
export const SessionDirectory = defineService<SessionDirectory>("pi.session-directory");
@ -699,12 +678,10 @@ export const SessionManagement = defineService<SessionManagement>("pi.session-ma
// server.ts
export function sessionDirectoryServerFacet(env: FacetEnvironment) {
const { managed, attachments } = env.use(Fleet);
const state = env.remoteState({ revision: 0, sessions: [] as SessionRecordSummary[] });
const events = env.remoteEvents<SessionDirectoryEvent>();
const state = env.replicatedState({ revision: 0, sessions: [] as SessionRecordSummary[] });
function publish(change: ManagedSessionChange, context: Context) {
function publish(_change: ManagedSessionChange, context: Context) {
state.set({ revision: state.value.revision + 1, sessions: managed.snapshot().map(toSummary) }, context);
events.emit(toDirectoryEvent(change), context);
}
env.own(managed.onChanged(publish));
@ -712,7 +689,7 @@ export function sessionDirectoryServerFacet(env: FacetEnvironment) {
state.set({ revision: 1, sessions: managed.snapshot().map(toSummary) }, BACKGROUND_CONTEXT),
);
env.provide(SessionDirectory, { state, events });
env.provide(SessionDirectory, { state });
env.provide(SessionManagement, {
async create(options, context) {
const client = requireClientIdentity(context);
@ -884,7 +861,7 @@ export function questionSessionFacet(env: FacetEnvironment) {
if (response === undefined) {
const completion = Promise.withResolvers<QuestionResponse>();
const request = env.remoteState<QuestionRequest>(params);
const request = env.replicatedState<QuestionRequest>(params);
const close = dialogs.add(invocation.invocationId, {
request,
async submitAnswer(candidate, _answerContext) {
@ -1000,7 +977,7 @@ Errors cross the wire as a JSON envelope `{ code, message, data? }` with stable
Security rules every providing host (session and server alike) must enforce:
- 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;
- RPC-capable service IDs come from trusted loaded service tokens; the remote boundary accepts only implementation functions and branded replicated-state 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;
@ -1008,7 +985,7 @@ Security rules every providing host (session and server alike) must enforce:
## The coding agent as a plugin manifest
The coding agent is one manifest of built-in and third-party plugins; each app host loads its facets through the same kernel in deterministic order. Server facets cover the directory/management services, authentication and client authorization, and worker spawn/stop policy and health. Session facets cover session creation/restoration and the scoped Harness facade; providers, model selection, and authentication; tools, wrappers, and permissions; prompt/steer/abort/compaction services; transcript persistence and event projection; filesystem and subprocess effects; temporary service instances for deferred interactions; unattended-session policy; and server service consumption where a feature needs server-wide data. TUI facets cover the chat screen, transcript rendering, editor, commands, keybindings, pickers, renderers, screens/slots/dialogs, and themes; a web host carries analogous browser facets. Shared contracts are service tokens with JSON-safe DTOs, latest-value state and semantic event types, presentation-local screen/slot tokens, stable renderer discriminators, and portable structured errors.
The coding agent is one manifest of built-in and third-party plugins; each app host loads its facets through the same kernel in deterministic order. Server facets cover the directory/management services, authentication and client authorization, and worker spawn/stop policy and health. Session facets cover session creation/restoration and the scoped Harness facade; providers, model selection, and authentication; tools, wrappers, and permissions; prompt/steer/abort/compaction services; transcript persistence and snapshot projection; filesystem and subprocess effects; temporary service instances for deferred interactions; unattended-session policy; and server service consumption where a feature needs server-wide data. TUI facets cover the chat screen, transcript rendering, editor, commands, keybindings, pickers, renderers, screens/slots/dialogs, and themes; a web host carries analogous browser facets. Shared contracts are service tokens with JSON-safe DTOs, latest-value state, presentation-local screen/slot tokens, stable renderer discriminators, and portable structured errors.
The generic kernel knows none of these domain concepts; each app host knows only its own.
@ -1023,7 +1000,7 @@ Before this becomes normative:
- 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;
- state/event flow control and per-client buffering at the server; reference lifetime and garbage collection;
- replicated-state flow control and per-client buffering at the server; reference lifetime and garbage collection;
- the stable error envelope and expected-error registration; activation/disposal ordering, optional dependencies, and shared-proxy lifetime;
- the exact singleton/keyed mode validation, keyed provider/observe APIs, instance-key and generation rules, instance hydration protocol, and answer authorization; general memo-name ownership remains with the invocation-memo design;
- package boundaries between the generic kernel, agent plugin contracts, generic protocol machinery, and coding-agent host integration.
@ -1223,8 +1200,8 @@ export function diffReviewSessionFacet(bindings: FacetEnvironment) {
const prompts = bindings.use(PromptQueue);
function exposeReview(initial: DiffReviewRecord) {
const document = bindings.remoteState({ reviewId: initial.reviewId, patch: initial.patch });
const activity = bindings.remoteState(toDiffReviewActivity(initial));
const document = bindings.replicatedState({ reviewId: initial.reviewId, patch: initial.patch });
const activity = bindings.replicatedState(toDiffReviewActivity(initial));
function publish(next: DiffReviewRecord, context: Context) {
if (next.revision > activity.value.revision) activity.set(toDiffReviewActivity(next), context);
@ -1355,19 +1332,9 @@ This is a shared review, not a generic room primitive. Keyed services provide di
> **Deferred:** `DeltaState` is not part of the initial plugin or RPC contract. Add it only after a concrete feature demonstrates that full-value `ReplicatedState` updates are too expensive and the same pattern appears in more than one feature.
A real replication gap remains. Some authoritative values are large, change frequently, and must support late joiners. `ReplicatedState` hydrates and reconnects correctly but sends a complete value on every update. `RemoteEvents` can send small deltas but has no snapshot, replay, or automatic gap recovery.
A real replication gap remains. Some authoritative values are large, change frequently, and must support late joiners. `ReplicatedState` hydrates and reconnects correctly but sends a complete value on every update.
A canvas is one possible example: joining requires the complete document, while dragging a shape should send only that operation. With today's primitives, the plugin can expose `getSnapshot()` plus revisioned `RemoteEvents` and implement the join protocol itself:
```text
subscribe and buffer deltas
→ fetch and install a snapshot
→ discard buffered revisions covered by the snapshot
→ apply consecutive later deltas
→ on a gap or reconnect, discard the replica and start over
```
That feature-local protocol is the preferred initial solution. It keeps a hypothetical optimization out of the generic host and reveals whether flow control, authoritative replacement, or other requirements actually matter.
A canvas is one possible example: joining requires the complete document, while dragging a shape should ideally send only that operation. With today's primitives, the plugin must accept complete `ReplicatedState` updates or keep the high-frequency projection process-local. A concrete feature should establish the snapshot, delta, gap-recovery, and flow-control requirements before another remote primitive is added.
### Possible future primitive

28
packages/agent/docs/rpc.md
View file

@ -12,7 +12,7 @@ session/server facet: provide() / provideMany().add()
presentation/session facet: use() / observe()
```
The service system is the plugin boundary. Presentation facets receive semantic services, replicated state, and semantic events; they never receive a raw Harness, Session, tool registry, hook registry, credential store, or storage handle.
The service system is the plugin boundary. Presentation facets receive semantic services and replicated state; they never receive a raw Harness, Session, tool registry, hook registry, credential store, or storage handle.
## Non-goals
@ -22,13 +22,13 @@ Do not serialize `Context`, `AbortSignal`, telemetry objects, callbacks, tools,
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.
The provider must make enough control-plane information available for the host to validate an accessed member as a method or `ReplicatedState`. 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 resolves to its facet-provided implementation and a remote facade binds through the host's connected services. Reloading the providing facet temporarily marks that same facade unavailable, then swaps its target; RPC singletons clear readiness and install a complete replacement snapshot on their existing subscription so captured methods and member facades address the replacement. While no provider is bound, 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`.
Use static assertions and runtime validation. Static checks constrain remote methods and state/event members; runtime boundaries reject unsupported members and non-JSON arguments, results, state, and event values. TypeScript supplies typed facades but does not authenticate a peer or create runtime metadata.
Use static assertions and runtime validation. Static checks constrain remote methods and replicated-state members; runtime boundaries reject unsupported members and non-JSON arguments, results, and state values. TypeScript supplies typed facades but does not authenticate a peer or create runtime metadata.
## Dependency ledger
@ -41,7 +41,7 @@ Type erasure does not hide service identity: every service token retains its sta
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.
First acquisition or provision is permitted only during facet setup. Later commands, hooks, 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 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.
@ -51,7 +51,7 @@ A presentation host combines services from its connected server and selected Ses
### 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 TUI or web 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. `SessionManagement` exposes `create`, `remove`, `attach`, and `detach` methods. A TUI or web facet consumes both through its environment:
```ts
const directory = env.use(SessionDirectory);
@ -64,7 +64,7 @@ The server derives workspace and client authority from its locally authenticated
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.
A state or event source has structural identity:
A replicated-state source has structural identity:
```text
(provider binding, service ID, optional instance key + generation, member name)
@ -78,14 +78,12 @@ A call carries enough control-plane information to select a provider binding, se
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.
## Replicated state, events, and keyed instances
## Replicated state and keyed instances
`ReplicatedState` is authoritative latest-value replication, not event history, durable storage, a CRDT, or multi-writer state. A cold replica has `value === undefined`; subscribing before hydration records a listener without invoking it. Hydration installs a complete snapshot atomically before later updates are delivered, so there is no snapshot/update gap. Once hydrated, `value` is synchronous and subscribing reports the current value followed by later updates. The first snapshot callback uses a fresh hydration context; an already hydrated replica uses a fresh local delivery context rather than retaining the original write context. State values are borrowed immutable JSON and are not defensively cloned; callers must not mutate or retain them.
The providing host attaches source trace metadata to updates. The consumer reconstructs fresh delivery contexts; hydration uses a fresh context parented to the subscription. Disconnect may retain a value as stale display data. Reconnect replaces it from a complete snapshot, while a switch to a different provider clears readiness and disposes binding-specific resources. The precise subscription, buffering, sequence, acknowledgement, and flow-control frames are transport mechanics.
`RemoteEvents` is a projection of provider-local events. Listeners run only in the consuming process; callbacks never cross the wire. Events are non-durable and never replayed after reconnect. Event delivery carries source trace metadata. State subscriptions, event subscriptions, and their cleanup are host-owned resources, not retained caller contexts.
`observe()` is keyed-instance discovery, not a `ReplicatedState` containing proxies. It reconciles a complete initial directory with ordered additions, replacements, and removals. Each instance's initial state members hydrate before its observer task starts. Closing an instance rejects new calls, aborts only that instance's observer task, and allows admitted calls to settle. A Session facet's `env.observe()` registration aborts old tasks when that facet generation closes; the replacement generation reconciles the fresh directory.
## Private returned references
@ -105,7 +103,7 @@ caller
└─ service implementation
```
State and event delivery independently carry source trace metadata. The consumer reconstructs a delivery context from it rather than retaining the context that established the subscription.
State delivery carries source trace metadata. The consumer reconstructs a delivery context from it rather than retaining the context that established the subscription.
Three cancellation domains remain separate: aborting one RPC invocation; explicitly cancelling service-owned work such as `job.cancel()`; and durable Harness cancellation such as `requestAbort()`. Transport cancellation and disconnect perform only the first. Work that outlives a call must detach into a service-owned task with its own controller and telemetry root.
@ -125,14 +123,13 @@ Test the plugin-facing semantics over loopback and a real framed transport:
- 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;
- event subscription setup, non-replay, ordering, cleanup, and bounded flow control;
- instance directory hydration, ordered reconciliation, state hydration before observer tasks, generation-based stale rejection, and task cleanup on close or switch;
- private returned-reference lifetime distinct from keyed-service discovery; and
- the question and shared-review patterns: concurrent instances, late presentation attachment, disconnect without cancelling session work, worker replacement, and durable application-level settlement.
## Open protocol mechanics
The following are intentionally not specified here: most exact control-frame schemas; member-kind discovery and lazy-proxy implementation; subscription sequencing, acknowledgements, buffering, and flow control; reference collection; and how a future multi-pane presentation represents more than one selected session. Singleton provider replacement requires a complete replacement snapshot on the existing subscription so method, state, and event member slots retain identity. The remaining mechanics must preserve the semantics above without changing the plugin contract.
The following are intentionally not specified here: most exact control-frame schemas; member-kind discovery and lazy-proxy implementation; subscription sequencing, acknowledgements, buffering, and flow control; reference collection; and how a future multi-pane presentation represents more than one selected session. Singleton provider replacement requires a complete replacement snapshot on the existing subscription so method and state member slots retain identity. The remaining mechanics must preserve the semantics above without changing the plugin contract.
## Example: directory and selected session
@ -147,13 +144,8 @@ interface SessionSummary {
updatedAt: string;
}
type SessionDirectoryEvent =
| { type: "created" | "changed"; session: SessionSummary }
| { type: "deleted"; sessionId: string };
interface SessionDirectory {
readonly state: ReplicatedState<{ revision: number; sessions: SessionSummary[] }>;
readonly events: RemoteEvents<SessionDirectoryEvent>;
}
interface SessionManagement {
@ -168,7 +160,7 @@ const SessionManagement = defineService<SessionManagement>("pi.session-managemen
A server facet supplies the services. The host-local attachment capability derives the client from `Context`, authorizes the requested session, and performs the binding transition:
```ts
serverContext.provide(SessionDirectory, { state: directoryState, events: directoryEvents });
serverContext.provide(SessionDirectory, { state: directoryState });
serverContext.provide(SessionManagement, {
async attach(sessionId, context) {
const client = requireClientIdentity(context);

71
packages/agent/src/plugins/services/events.ts
View file

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

23
packages/agent/src/plugins/services/index.ts
View file

@ -1,10 +1,3 @@
export {
getRemoteEventsInternals,
REMOTE_EVENTS_INTERNALS,
type RemoteEventsInternals,
type RemoteEventsSource,
remoteEvents,
} from "./events.ts";
export { RemoteServiceNamespace } from "./namespace.ts";
export {
createLoopbackServiceConnection,
@ -14,21 +7,17 @@ export {
} from "./provider.ts";
export {
freshDeliveryContext,
getRemoteStateInternals,
REMOTE_STATE_INTERNALS,
type RemoteStateInternals,
type RemoteStateSource,
remoteState,
} from "./state.ts";
getReplicatedStateInternals,
REPLICATED_STATE_INTERNALS,
type ReplicatedStateInternals,
type ReplicatedStateSource,
replicatedState,
} from "./replicated-state.ts";
export {
cloneJson,
defineService,
isJsonValue,
type MutableRemoteEvents,
type MutableReplicatedState,
type RemoteEventListener,
type RemoteEvents,
type RemoteEventType,
type RemoteJson,
type RemoteServiceConnection,
type RemoteServiceContract,

129
packages/agent/src/plugins/services/namespace.ts
View file

@ -1,13 +1,10 @@
import { awaitWithContext, BACKGROUND_CONTEXT, type Context, withCancel } from "../../harness/context.ts";
import type { JsonValue } from "../../harness/session/types.ts";
import { RemoteServiceError } from "./provider.ts";
import { freshDeliveryContext } from "./state.ts";
import { freshDeliveryContext } from "./replicated-state.ts";
import {
cloneJson,
isJsonValue,
type RemoteEventListener,
type RemoteEvents,
type RemoteEventType,
type RemoteServiceConnection,
type RemoteServiceInstance,
type RemoteServiceNamespaceOptions,
@ -25,7 +22,7 @@ import {
type ErrorReporter = (error: Error) => void;
class RemoteStateReplica implements ReplicatedState<JsonValue> {
class ReplicatedStateReplica implements ReplicatedState<JsonValue> {
readonly #listeners = new Set<(value: JsonValue, context: Context) => void>();
readonly #reportError: ErrorReporter;
#value: JsonValue | undefined;
@ -79,55 +76,13 @@ class RemoteStateReplica implements ReplicatedState<JsonValue> {
}
}
class RemoteEventsReplica implements RemoteEvents<JsonValue> {
readonly #listeners = new Set<RemoteEventListener<JsonValue>>();
readonly #reportError: ErrorReporter;
constructor(reportError: ErrorReporter) {
this.#reportError = reportError;
}
subscribe(listener: RemoteEventListener<JsonValue>): () => void {
this.#listeners.add(listener);
return () => this.#listeners.delete(listener);
}
on<TType extends RemoteEventType<JsonValue>>(
type: TType,
listener: RemoteEventListener<Extract<JsonValue, { readonly type: TType }>>,
): () => void {
return this.subscribe((event, context) => {
if (hasEventType(event, type)) listener(event, context);
});
}
deliver(event: JsonValue, context: Context): void {
for (const listener of this.#listeners) {
try {
listener(event, context);
} catch (error) {
this.#reportError(toError(error));
}
}
}
}
interface PendingMemberSubscription {
readonly listener: RemoteEventListener<JsonValue>;
remove: (() => void) | undefined;
closed: boolean;
}
class MemberSlot {
readonly #serviceId: string;
readonly #member: string;
readonly #invoke: (args: readonly JsonValue[], context: Context) => Promise<JsonValue | undefined>;
readonly #state: RemoteStateReplica;
readonly #events: RemoteEventsReplica;
readonly #state: ReplicatedStateReplica;
readonly #isActive: () => boolean;
readonly #assertAccess: () => void;
readonly #reportError: ErrorReporter;
readonly #pendingSubscriptions = new Set<PendingMemberSubscription>();
readonly value: unknown;
#kind: ServiceMemberKind | undefined;
#expectedKind: ServiceMemberKind | undefined;
@ -145,9 +100,7 @@ class MemberSlot {
this.#invoke = invoke;
this.#isActive = isActive;
this.#assertAccess = assertAccess;
this.#reportError = reportError;
this.#state = new RemoteStateReplica(reportError);
this.#events = new RemoteEventsReplica(reportError);
this.#state = new ReplicatedStateReplica(reportError);
const callable = (): void => {};
this.value = new Proxy(callable, {
apply: (_target, _thisArg, args) => this.#call(args),
@ -158,7 +111,6 @@ class MemberSlot {
return this.#state.value;
}
if (property === "subscribe") return this.#subscribe.bind(this);
if (property === "on") return this.#on.bind(this);
if (property === Symbol.toStringTag) return "RemoteServiceMember";
if (property === "then") return undefined;
return undefined;
@ -177,8 +129,6 @@ class MemberSlot {
`Remote service member ${this.#serviceId}.${this.#member} is ${kind}, not ${this.#expectedKind}`,
);
}
for (const pending of this.#pendingSubscriptions) this.#activateSubscription(pending, kind);
this.#pendingSubscriptions.clear();
}
hydrate(snapshot: ServiceStateSnapshot, context: Context): void {
@ -191,58 +141,17 @@ class MemberSlot {
this.#state.update(sequence, value, context);
}
deliverEvent(event: JsonValue, context: Context): void {
this.setDescription("events");
this.#events.deliver(event, context);
}
clear(): void {
this.#state.clear();
}
#subscribe(listener: RemoteEventListener<JsonValue>): () => void {
#subscribe(listener: (value: JsonValue, context: Context) => void): () => void {
this.#assertAccess();
if (typeof listener !== "function")
throw new TypeError("Remote service subscription listener must be a function");
if (this.#kind === "state" || this.#expectedKind === "state") {
this.#expect("state");
return this.#state.subscribe(listener);
}
if (this.#kind === "events" || this.#expectedKind === "events") {
this.#expect("events");
return this.#events.subscribe(listener);
}
const pending: PendingMemberSubscription = { listener, remove: undefined, closed: false };
this.#pendingSubscriptions.add(pending);
return () => {
if (pending.closed) return;
pending.closed = true;
pending.remove?.();
this.#pendingSubscriptions.delete(pending);
};
}
#on(type: string, listener: RemoteEventListener<JsonValue>): () => void {
this.#assertAccess();
this.#expect("events");
return this.#events.subscribe((event, context) => {
if (hasEventType(event, type)) listener(event, context);
});
}
#activateSubscription(pending: PendingMemberSubscription, kind: ServiceMemberKind): void {
if (pending.closed) return;
if (kind === "state") pending.remove = this.#state.subscribe(pending.listener);
else if (kind === "events") pending.remove = this.#events.subscribe(pending.listener);
else {
pending.closed = true;
this.#reportError(
new RemoteServiceError(
"service_member_mismatch",
`Remote service member ${this.#serviceId}.${this.#member} is a method, not subscribable`,
),
);
if (typeof listener !== "function") {
throw new TypeError("Replicated state subscription listener must be a function");
}
this.#expect("state");
return this.#state.subscribe(listener);
}
#expect(kind: ServiceMemberKind): void {
@ -366,13 +275,6 @@ class ServiceFacade {
this.#slot(member).update(sequence, value, context);
}
deliverEvent(member: string, event: JsonValue, context: Context): void {
if (this.#descriptions.get(member) !== "events") {
throw new Error(`Remote service event targets non-event member ${this.#serviceId}.${member}`);
}
this.#slot(member).deliverEvent(event, context);
}
clear(): void {
for (const slot of this.#slots.values()) slot.clear();
this.#stateSnapshots.clear();
@ -579,13 +481,6 @@ class KeyedBinding<T> {
instance.facade.update(update.member, update.sequence, update.value, context);
break;
}
case "event": {
if (update.instance === undefined) throw new Error("Keyed event has no instance address");
const instance = this.#instances.get(update.instance.key);
if (instance?.address.generation !== update.instance.generation) return;
instance.facade.deliverEvent(update.member, update.event, context);
break;
}
}
} catch (error) {
this.#reportError(toError(error));
@ -826,8 +721,6 @@ export class RemoteServiceNamespace implements RemoteServices {
binding.facade.install(update.snapshot, context);
} else if (update.type === "state" && update.instance === undefined) {
binding.facade.update(update.member, update.sequence, update.value, context);
} else if (update.type === "event" && update.instance === undefined) {
binding.facade.deliverEvent(update.member, update.event, context);
}
} catch (error) {
this.#reportError(toError(error));
@ -890,10 +783,6 @@ function isContext(value: unknown): value is Context {
return typeof candidate.value === "function" && typeof candidate.toString === "function";
}
function hasEventType<T, TType extends string>(event: T, type: TType): event is Extract<T, { readonly type: TType }> {
return typeof event === "object" && event !== null && "type" in event && event.type === type;
}
function toError(error: unknown): Error {
return error instanceof Error ? error : new Error(String(error));
}

35
packages/agent/src/plugins/services/provider.ts
View file

@ -1,7 +1,10 @@
import type { Context } from "../../harness/context.ts";
import type { JsonValue } from "../../harness/session/types.ts";
import { getRemoteEventsInternals, type RemoteEventsInternals } from "./events.ts";
import { freshDeliveryContext, getRemoteStateInternals, type RemoteStateInternals } from "./state.ts";
import {
freshDeliveryContext,
getReplicatedStateInternals,
type ReplicatedStateInternals,
} from "./replicated-state.ts";
import {
cloneJson,
isJsonValue,
@ -45,8 +48,7 @@ type RemoteMethod = (...args: unknown[]) => unknown;
type InstanceMember =
| { readonly kind: "method"; readonly method: RemoteMethod }
| { readonly kind: "state"; readonly state: RemoteStateInternals }
| { readonly kind: "events"; readonly events: RemoteEventsInternals };
| { readonly kind: "state"; readonly state: ReplicatedStateInternals };
interface ProviderInstance {
readonly address?: ServiceInstanceAddress;
@ -314,7 +316,7 @@ export class RemoteServiceProvider {
members.set(name, { kind: "method", method: descriptor.value as RemoteMethod });
continue;
}
const state = getRemoteStateInternals(descriptor.value);
const state = getReplicatedStateInternals(descriptor.value);
if (state !== undefined) {
members.set(name, { kind: "state", state });
removeMemberListeners.push(
@ -341,29 +343,6 @@ export class RemoteServiceProvider {
);
continue;
}
const events = getRemoteEventsInternals(descriptor.value);
if (events !== undefined) {
members.set(name, { kind: "events", events });
removeMemberListeners.push(
events.subscribe((event, context) => {
if (!instance.active) return;
if (!isJsonValue(event)) {
throw new RemoteServiceError("service_invalid_value", "Remote event must be strict JSON");
}
this.#emit(
registration,
{
type: "event",
...(address === undefined ? {} : { instance: address }),
member: name,
event,
},
context,
);
}),
);
continue;
}
throw new TypeError(`Remote service member ${registration.serviceId}.${name} is not remotely exposable`);
}
if (members.size === 0) throw new TypeError(`Remote service ${registration.serviceId} has no members`);

26
packages/agent/src/plugins/services/state.ts → packages/agent/src/plugins/services/replicated-state.ts
View file

@ -1,19 +1,19 @@
import { BACKGROUND_CONTEXT, type Context, withCancel } from "../../harness/context.ts";
import { isJsonValue, type MutableReplicatedState, type ReplicatedState } from "./types.ts";
export const REMOTE_STATE_INTERNALS = Symbol("pi.remoteState.internals");
export const REPLICATED_STATE_INTERNALS = Symbol("pi.replicatedState.internals");
export interface RemoteStateInternals<T = unknown> {
export interface ReplicatedStateInternals<T = unknown> {
readonly sequence: number;
readonly value: T;
subscribe(listener: (value: T, sequence: number, context: Context) => void): () => void;
}
export interface RemoteStateSource<T = unknown> extends MutableReplicatedState<T> {
readonly [REMOTE_STATE_INTERNALS]: RemoteStateInternals<T>;
export interface ReplicatedStateSource<T = unknown> extends MutableReplicatedState<T> {
readonly [REPLICATED_STATE_INTERNALS]: ReplicatedStateInternals<T>;
}
class MutableRemoteStateImpl<T> implements RemoteStateSource<T> {
class MutableReplicatedStateImpl<T> implements ReplicatedStateSource<T> {
readonly #listeners = new Set<(value: T, context: Context) => void>();
readonly #sourceListeners = new Set<(value: T, sequence: number, context: Context) => void>();
#value: T;
@ -42,7 +42,7 @@ class MutableRemoteStateImpl<T> implements RemoteStateSource<T> {
return () => this.#listeners.delete(listener);
}
get [REMOTE_STATE_INTERNALS](): RemoteStateInternals<T> {
get [REPLICATED_STATE_INTERNALS](): ReplicatedStateInternals<T> {
const source = this;
return {
get sequence() {
@ -59,16 +59,16 @@ class MutableRemoteStateImpl<T> implements RemoteStateSource<T> {
}
}
export function remoteState<T>(initial: T): MutableReplicatedState<T> {
return new MutableRemoteStateImpl(initial);
export function replicatedState<T>(initial: T): MutableReplicatedState<T> {
return new MutableReplicatedStateImpl(initial);
}
export function getRemoteStateInternals(value: unknown): RemoteStateInternals | undefined {
if (typeof value !== "object" || value === null || !(REMOTE_STATE_INTERNALS in value)) return undefined;
const candidate = value as { readonly [REMOTE_STATE_INTERNALS]?: unknown };
const internals = candidate[REMOTE_STATE_INTERNALS];
export function getReplicatedStateInternals(value: unknown): ReplicatedStateInternals | undefined {
if (typeof value !== "object" || value === null || !(REPLICATED_STATE_INTERNALS in value)) return undefined;
const candidate = value as { readonly [REPLICATED_STATE_INTERNALS]?: unknown };
const internals = candidate[REPLICATED_STATE_INTERNALS];
if (typeof internals !== "object" || internals === null) return undefined;
return internals as RemoteStateInternals;
return internals as ReplicatedStateInternals;
}
export function freshDeliveryContext(): Context {

44
packages/agent/src/plugins/services/types.ts
View file

@ -4,7 +4,7 @@ import type { JsonValue } from "../../harness/session/types.ts";
const SERVICE_TYPE = Symbol("pi.service.type");
export type ServiceMode = "singleton" | "keyed";
export type ServiceMemberKind = "method" | "state" | "events";
export type ServiceMemberKind = "method" | "state";
export interface ServiceCatalogueEntry {
readonly serviceId: string;
@ -38,22 +38,6 @@ export interface MutableReplicatedState<T> extends ReplicatedState<T> {
set(value: T, context: Context): void;
}
export type RemoteEventType<T> = T extends { readonly type: infer TType extends string } ? TType : never;
export type RemoteEventListener<T> = (event: T, context: Context) => void;
/** Non-durable semantic events. Event values are borrowed immutable JSON and are never replayed. */
export interface RemoteEvents<T> {
subscribe(listener: RemoteEventListener<T>): () => void;
on<TType extends RemoteEventType<T>>(
type: TType,
listener: RemoteEventListener<Extract<T, { readonly type: TType }>>,
): () => void;
}
export interface MutableRemoteEvents<T> extends RemoteEvents<T> {
emit(event: T, context: Context): void;
}
/** Marks an existing schema-validated wire DTO as JSON-safe for static service-contract checking. */
declare const REMOTE_JSON_TYPE: unique symbol;
export type RemoteJson<T> = T & { readonly [REMOTE_JSON_TYPE]?: true };
@ -75,19 +59,15 @@ type InvalidRemoteMember<T> = T extends ReplicatedState<infer TValue>
? InvalidJsonPart<TValue> extends never
? never
: "state value is not JSON"
: T extends RemoteEvents<infer TEvent>
? InvalidJsonPart<TEvent> extends never
? never
: "event value is not JSON"
: T extends (...args: [...infer TArgs, Context]) => Promise<infer TResult>
? InvalidJsonPart<TArgs[number]> extends never
? TResult extends void
: T extends (...args: [...infer TArgs, Context]) => Promise<infer TResult>
? InvalidJsonPart<TArgs[number]> extends never
? TResult extends void
? never
: InvalidJsonPart<TResult> extends never
? never
: InvalidJsonPart<TResult> extends never
? never
: "method result is not JSON or void"
: "method argument is not JSON"
: "member is not a remote method, ReplicatedState, or RemoteEvents";
: "method result is not JSON or void"
: "method argument is not JSON"
: "member is not a remote method or ReplicatedState";
type InvalidRemoteMemberNames<T> = {
[TKey in keyof T]-?: InvalidRemoteMember<T[TKey]> extends never ? never : TKey;
@ -152,12 +132,6 @@ export type ServiceProviderUpdate =
readonly sequence: number;
readonly value: JsonValue;
}
| {
readonly type: "event";
readonly instance?: ServiceInstanceAddress;
readonly member: string;
readonly event: JsonValue;
}
| { readonly type: "unavailable" }
| { readonly type: "replaced"; readonly snapshot: ServiceInstanceSnapshot }
| { readonly type: "spawned"; readonly instance: ServiceInstanceSnapshot }

119
packages/agent/test/plugins/services.test.ts
View file

@ -4,13 +4,11 @@ import {
BACKGROUND_CONTEXT,
createLoopbackServiceConnection,
defineService,
type RemoteEvents,
type RemoteServiceConnection,
RemoteServiceNamespace,
RemoteServiceProvider,
type ReplicatedState,
remoteEvents,
remoteState,
replicatedState,
} from "../../src/index.ts";
type ModelRef = { provider: string; modelId: string };
@ -33,13 +31,6 @@ interface QuestionDialogs {
const QuestionDialogs = defineService<QuestionDialogs>("test.question-dialog");
type ActivityEvent = { type: "changed"; revision: number } | { type: "ignored"; revision: number };
interface Activity {
readonly events: RemoteEvents<ActivityEvent>;
}
const Activity = defineService<Activity>("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 });
@ -49,7 +40,7 @@ describe("plugin remote services", () => {
test("does not defensively clone borrowed state values", () => {
const initial: ModelsState = { selected: null, revision: 0 };
const state = remoteState(initial);
const state = replicatedState(initial);
let delivered: ModelsState | undefined;
const unsubscribe = state.subscribe((value) => {
delivered = value;
@ -68,7 +59,7 @@ describe("plugin remote services", () => {
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);
const state = replicatedState(initialState);
let publishedState: ModelsState | undefined;
provider.provide(Models, {
state,
@ -121,7 +112,7 @@ describe("plugin remote services", () => {
test("keeps singleton facades stable when their provider is replaced", async () => {
const provider = new RemoteServiceProvider([Models]);
provider.provide(Models, {
state: remoteState<ModelsState>({ selected: null, revision: 1 }),
state: replicatedState<ModelsState>({ selected: null, revision: 1 }),
async select() {},
});
const namespace = new RemoteServiceNamespace({
@ -142,7 +133,7 @@ describe("plugin remote services", () => {
const replacementSelect = vi.fn(async () => {});
provider.replace(Models, {
state: remoteState<ModelsState>({ selected: null, revision: 2 }),
state: replicatedState<ModelsState>({ selected: null, revision: 2 }),
select: replacementSelect,
});
@ -159,7 +150,7 @@ describe("plugin remote services", () => {
test("keeps deferred service handles inaccessible until host activation", async () => {
const provider = new RemoteServiceProvider([Models]);
provider.provide(Models, {
state: remoteState<ModelsState>({ selected: null, revision: 0 }),
state: replicatedState<ModelsState>({ selected: null, revision: 0 }),
async select() {},
});
let active = false;
@ -209,7 +200,7 @@ describe("plugin remote services", () => {
test("buffers state updates that race subscription hydration", async () => {
const provider = new RemoteServiceProvider([Models]);
const state = remoteState<ModelsState>({ selected: null, revision: 0 });
const state = replicatedState<ModelsState>({ selected: null, revision: 0 });
provider.provide(Models, {
state,
async select() {},
@ -239,7 +230,7 @@ describe("plugin remote services", () => {
test("hydrates cold ReplicatedState replicas and replaces them across rebinds", async () => {
const provider = new RemoteServiceProvider([Models]);
const state = remoteState<ModelsState>({ selected: null, revision: 0 });
const state = replicatedState<ModelsState>({ selected: null, revision: 0 });
provider.provide(Models, {
state,
async select() {},
@ -300,7 +291,7 @@ describe("plugin remote services", () => {
});
await vi.waitFor(() => expect(errors).toEqual([]));
const firstRequest = remoteState<Question>({ question: "First?" });
const firstRequest = replicatedState<Question>({ question: "First?" });
const firstSubmit = vi.fn(async () => ({ accepted: true }));
const closeFirst = provider.spawn(QuestionDialogs, "invocation-1", {
request: firstRequest,
@ -321,7 +312,7 @@ describe("plugin remote services", () => {
code: "service_stale_instance",
});
const secondRequest = remoteState<Question>({ question: "Again?" });
const secondRequest = replicatedState<Question>({ question: "Again?" });
const closeSecond = provider.spawn(QuestionDialogs, "invocation-1", {
request: secondRequest,
async submit() {
@ -339,98 +330,10 @@ describe("plugin remote services", () => {
provider.dispose();
});
test("delivers filtered RemoteEvents without replay and buffers hydration races", async () => {
const provider = new RemoteServiceProvider([Activity]);
const events = remoteEvents<ActivityEvent>();
provider.provide(Activity, { events });
const local: ActivityEvent[] = [];
const localChanged: ActivityEvent[] = [];
const removeLocal = events.subscribe((event) => local.push(event));
const removeLocalChanged = events.on("changed", (event) => localChanged.push(event));
const beforeSubscription = { type: "changed", revision: 0 } as const;
events.emit(beforeSubscription, BACKGROUND_CONTEXT);
expect(local[0]).toBe(beforeSubscription);
expect(localChanged).toEqual([beforeSubscription]);
const connection: RemoteServiceConnection = {
invoke: (call, context) => provider.invoke(call, context),
subscribe: async (serviceId, mode, listener) => {
const subscription = provider.subscribe(serviceId, mode, listener);
events.emit({ type: "changed", revision: 1 }, BACKGROUND_CONTEXT);
return {
snapshot: subscription.snapshot,
activate: () => subscription.activate(),
close: () => subscription.close(),
};
},
};
const errors: Error[] = [];
const namespace = new RemoteServiceNamespace({
services: [Activity],
connection,
onError: (error) => errors.push(error),
});
const activity = namespace.use(Activity);
const received: ActivityEvent[] = [];
const changed: Extract<ActivityEvent, { type: "changed" }>[] = [];
const unsubscribe = activity.events.subscribe((event) => received.push(event));
const unsubscribeChanged = activity.events.on("changed", (event) => changed.push(event));
await vi.waitFor(() => expect(received).toEqual([{ type: "changed", revision: 1 }]));
events.emit({ type: "ignored", revision: 2 }, BACKGROUND_CONTEXT);
events.emit({ type: "changed", revision: 3 }, BACKGROUND_CONTEXT);
expect(received).toEqual([
{ type: "changed", revision: 1 },
{ type: "ignored", revision: 2 },
{ type: "changed", revision: 3 },
]);
expect(changed).toEqual([
{ type: "changed", revision: 1 },
{ type: "changed", revision: 3 },
]);
expect(received).not.toContainEqual(beforeSubscription);
expect(errors).toEqual([]);
expect(() => events.emit({ type: "changed", revision: Number.NaN }, BACKGROUND_CONTEXT)).toThrow(
"Remote event value must be strict JSON",
);
removeLocal();
removeLocalChanged();
unsubscribe();
unsubscribeChanged();
await namespace.dispose(BACKGROUND_CONTEXT);
provider.dispose();
});
test("routes keyed RemoteEvents only for the live instance generation", async () => {
const provider = new RemoteServiceProvider([{ service: Activity, mode: "keyed" }]);
const namespace = new RemoteServiceNamespace({
services: [Activity],
connection: createLoopbackServiceConnection(provider),
});
let observed: Activity | undefined;
const stop = namespace.observe(Activity, (instance) => {
observed = instance.service;
});
const events = remoteEvents<ActivityEvent>();
const close = provider.spawn(Activity, "activity-1", { events });
await vi.waitFor(() => expect(observed).toBeDefined());
const received: ActivityEvent[] = [];
observed!.events.subscribe((event) => received.push(event));
events.emit({ type: "changed", revision: 1 }, BACKGROUND_CONTEXT);
expect(received).toEqual([{ type: "changed", revision: 1 }]);
close();
events.emit({ type: "changed", revision: 2 }, BACKGROUND_CONTEXT);
expect(received).toEqual([{ type: "changed", revision: 1 }]);
stop();
await namespace.dispose(BACKGROUND_CONTEXT);
provider.dispose();
});
test("rejects mode mixing, unsupported members, and non-JSON values", async () => {
const provider = new RemoteServiceProvider([Models, { service: QuestionDialogs, mode: "keyed" }]);
provider.provide(Models, {
state: remoteState<ModelsState>({ selected: null, revision: 0 }),
state: replicatedState<ModelsState>({ selected: null, revision: 0 }),
async select() {},
});
expect(() => provider.spawn(Models, "wrong", {} as Models)).toThrow(/singleton/);

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

@ -420,7 +420,7 @@ export class Client {
.catch((error: unknown) => this.#reportListenerError(error));
return;
}
if (message.type === "service_event") {
if (message.type === "service_update") {
const active = this.#serviceListeners.get(message.subscriptionId);
if (active === undefined) return;
if (active.ready) this.#deliverServiceUpdate(active, message.update);

2
packages/client/test/client.test.ts
View file

@ -377,7 +377,7 @@ describe("Client service operations", () => {
},
});
server.send({
type: "service_event",
type: "service_update",
subscriptionId: "service-1",
update: { type: "state", member: "state", sequence: 1, value: { revision: 1 } },
});

9
packages/coding-agent/src/experimental/client-tui.ts
View file

@ -65,15 +65,6 @@ export const sessionPickerTuiFacet = defineFacet({
env.onActivate(() => {
env.own(tui.registerSessionPicker(directory, management));
env.own(directory.state.subscribe(tui.refresh));
env.own(
directory.events.subscribe((event) => {
tui.setStatus(
event.type === "deleted"
? `Session removed: ${event.sessionId}`
: `Session ${event.type}: ${event.session.sessionId}`,
);
}),
);
});
},
});

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

@ -72,7 +72,7 @@ export async function runClient(command: ClientCommand, options: RunClientOption
const chat = match.chat;
const completedText = new Map<string, string>();
// Chat deliberately returns no transcript content. Keep the compatibility watch until Transcript uses RemoteEvents.
// Chat deliberately returns no transcript content. Keep the compatibility watch until Transcript is implemented.
const watch = await match.client.watchSession(sessionId);
await watch.start(async (event) => {
if (event.type === "message_end" && event.runId !== undefined && event.message.role === "assistant") {

79
packages/coding-agent/src/experimental/facets.ts
View file

@ -2,15 +2,13 @@ import {
BACKGROUND_CONTEXT,
type Context,
createLoopbackServiceConnection,
type MutableRemoteEvents,
type MutableReplicatedState,
type RemoteServiceContract,
type RemoteServiceInstance,
RemoteServiceNamespace,
RemoteServiceProvider,
type RemoteServices,
remoteEvents,
remoteState,
replicatedState,
type Service,
type ServiceCatalogueEntry,
type ServiceMode,
@ -32,17 +30,80 @@ interface FacetRecord {
}
export interface FacetEnvironment {
/**
* Declare a hard dependency on one singleton service and return its stable handle.
*
* Call this synchronously during setup so the host can validate and order the complete
* dependency graph before activation. The handle cannot be accessed during setup. It
* binds when the host activates and follows later provider-facet replacements, whether
* the implementation is process-local or reached through RPC.
*/
use<T>(service: Service<T>): T;
/**
* Declare a hard dependency on a keyed service and observe each live instance.
*
* The host starts the handler after the instance and its replicated state have hydrated.
* The handler receives a context that is aborted when that instance closes or is replaced.
* The returned idempotent function stops this observation early; otherwise the facet owns
* it automatically. Use keyed services when instances appear and disappear at runtime.
*/
observe<T>(
service: Service<T>,
handler: (instance: RemoteServiceInstance<T>, context: Context) => void | Promise<void>,
): () => void;
/**
* Declare and install this facet's singleton implementation of a service.
*
* Call this synchronously during setup. The declaration lets the host reject duplicate
* providers and order consumers after this facet. RPC-capable tokens are published
* automatically; `{ rpc: false }` tokens remain process-local. Reloading this facet swaps
* the implementation behind existing consumer handles.
*/
provide<T>(service: Service<T>, implementation: NoInfer<T>): void;
/**
* Declare ownership of a keyed service and return the handle used to add live instances.
*
* Declaring ownership during setup makes the service discoverable even while it has no
* instances. Instances may be added only while the facet is active and are closed
* automatically when the facet deactivates.
*/
provideMany<T>(service: Service<T>): ServiceInstances<T>;
remoteState<T>(initial: T): MutableReplicatedState<T>;
remoteEvents<T>(): MutableRemoteEvents<T>;
/**
* Create initialized mutable state suitable for exposing through a service implementation.
*
* Remote consumers receive a stable read-only replica that hydrates from this value and
* follows later updates. This is live projection state, not durable storage; the facet must
* be able to reconstruct it after reload or process restart.
*/
replicatedState<T>(initial: T): MutableReplicatedState<T>;
/**
* Give the facet ownership of a resource cleanup function.
*
* Cleanups run once in reverse registration order during deactivation and startup-failure
* cleanup. Register every subscription, timer, process, file, or contribution here so a
* failed or reloaded facet cannot leak resources into its replacement.
*/
own(disposal: () => void | Promise<void>): void;
/**
* Register asynchronous initialization to run after dependencies are bound and ready.
*
* Setup itself must stay synchronous because it declares the graph. Put I/O, hydration,
* and other effects here. Provider facets activate before facets that consume them.
*/
onActivate(callback: () => void | Promise<void>): void;
/**
* Register final facet teardown.
*
* The callback runs after resources acquired during activation have been released. Use it
* for shutdown work that is not naturally represented by an individual `own()` cleanup.
*/
onDeactivate(callback: () => void | Promise<void>): void;
}
@ -657,13 +718,9 @@ class FacetKernel {
lifecycle.observe(() => observation.start());
return observation.stop;
},
remoteState: <T>(initial: T) => {
replicatedState: <T>(initial: T) => {
lifecycle.assertRunning("create replicated state");
return remoteState(initial);
},
remoteEvents: <T>() => {
lifecycle.assertRunning("create remote events");
return remoteEvents<T>();
return replicatedState(initial);
},
own: (disposal) => lifecycle.own(disposal),
onActivate: (callback) => lifecycle.onActivate(callback),

8
packages/coding-agent/src/experimental/services/README.md
View file

@ -4,17 +4,17 @@ Facet setup generates each host's RPC service catalogue from its provided RPC-ca
| Scope | Service | Current slice | Continuation point |
|---|---|---|---|
| server | `SessionDirectory` | replicated state and semantic directory events implemented | add authenticated per-client projection when identity lands |
| server | `SessionDirectory` | replicated state implemented | add authenticated per-client projection when identity lands |
| server | `SessionManagement` | create, remove, attach, detach implemented | add authenticated workspace authorization |
| session | `Models` | state, selection, thinking, refresh implemented | move provider/auth composition behind plugin facets |
| session | `Chat` | prompt and durable `requestAbort` implemented; queue, resume, compaction, and navigation contracts declared | replace each explicit throwing provider member with its Harness adapter slice |
| session | `Accounts` | state declared; mutations throw | connect the local credential service to the presentation-safe facade |
| session | `Transcript` | revisioned events surface declared; snapshot throws and no Harness event producer is attached | implement snapshot/event gap handling and remove the compatibility lane watch |
| session | `Transcript` | authoritative snapshot declared but throws | implement snapshot projection 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. `ServerServiceConnection.connection` and `SessionServiceConnection.attachment` are implemented local control states.
`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 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. A host may reload selected facets when their declared service shape is unchanged; retained consumers keep the same local and RPC service facades while replacement implementations and singleton snapshots are installed behind them. 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 facet replacement or host shutdown disposes affected lifecycles in reverse dependency order. 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()`. 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.
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 replicated state; `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.

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

@ -9,7 +9,7 @@ import {
RemoteServiceNamespace,
type RemoteServices,
type ReplicatedState,
remoteState,
replicatedState,
} from "@earendil-works/pi-agent-core";
import type { Client } from "@earendil-works/pi-client";
import type {
@ -115,7 +115,7 @@ class ServerServiceConnectionImpl implements ServerServiceConnection {
this.#remoteConnection = createRemoteServiceConnection(client, () => ({ serverId: client.serverId }));
this.#onError = options.onError ?? (() => {});
this.#connectionAttempt = client.connectionState === "connecting" ? 1 : 0;
const connectionState = remoteState<ServerConnectionState>(
const connectionState = replicatedState<ServerConnectionState>(
toServerConnectionState(client, this.#connectionAttempt),
);
this.connection = connectionState;
@ -192,7 +192,7 @@ class SessionServiceConnectionImpl implements SessionServiceConnection {
this.#client = client;
this.#remoteConnection = createRemoteServiceConnection(client, () => client.attachment);
this.#onError = options.onError ?? (() => {});
this.#attachmentState = remoteState<SessionAttachmentState>(
this.#attachmentState = replicatedState<SessionAttachmentState>(
client.attachment === undefined
? { status: "detached" }
: { status: "attaching", sessionId: client.attachment.sessionId },

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

@ -125,7 +125,7 @@ export function createModelsService(
export const modelsServiceFacet = defineFacet({
id: "@pi/models",
setup(env) {
const runtime = createModelsService(env.use(Lane), env.use(ModelsRuntime), env.remoteState);
const runtime = createModelsService(env.use(Lane), env.use(ModelsRuntime), env.replicatedState);
env.provide(Models, runtime.service);
env.onActivate(() => runtime.activate(BACKGROUND_CONTEXT));
},

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

@ -2,10 +2,8 @@ import {
BACKGROUND_CONTEXT,
type Context,
type ServiceProviderUpdate as CoreServiceProviderUpdate,
type MutableRemoteEvents,
RemoteServiceProvider,
remoteEvents,
remoteState,
replicatedState,
type ServiceProviderSubscription,
} from "@earendil-works/pi-agent-core";
import {
@ -19,12 +17,7 @@ import {
} from "@earendil-works/pi-protocol";
import type { RoutedServerServiceAttachment, RoutedServerServiceHost } from "@earendil-works/pi-server";
import { Check } from "typebox/value";
import {
SessionDirectory,
type SessionDirectoryEvent,
type SessionDirectoryState,
SessionManagement,
} from "./sessions.ts";
import { SessionDirectory, type SessionDirectoryState, SessionManagement } from "./sessions.ts";
export interface ExperimentalServerServices {
readonly host: RoutedServerServiceHost;
@ -38,20 +31,17 @@ export async function createExperimentalServerServices(options: {
remove(sessionId: string, context: Context): Promise<void>;
}): Promise<ExperimentalServerServices> {
let revision = 1;
const directory = remoteState<SessionDirectoryState>({
const directory = replicatedState<SessionDirectoryState>({
revision,
sessions: await options.list(BACKGROUND_CONTEXT),
});
const directoryEvents = remoteEvents<SessionDirectoryEvent>();
const attachments = new Set<RoutedServerServiceAttachment>();
let mutationTail = Promise.resolve();
const refreshNow = async (context: Context): Promise<void> => {
const previous = directory.value.sessions;
const sessions = await options.list(context);
revision += 1;
directory.set({ revision, sessions }, context);
publishDirectoryChanges(previous, sessions, directoryEvents, context);
};
const serialize = <T>(operation: () => Promise<T>): Promise<T> => {
const result = mutationTail.catch(() => {}).then(operation);
@ -69,7 +59,7 @@ export async function createExperimentalServerServices(options: {
{ service: SessionDirectory, mode: "singleton" },
{ service: SessionManagement, mode: "singleton" },
]);
provider.provide(SessionDirectory, { state: directory, events: directoryEvents });
provider.provide(SessionDirectory, { state: directory });
provider.provide(SessionManagement, {
create: (createOptions, context) =>
serialize(async () => {
@ -111,28 +101,6 @@ export async function createExperimentalServerServices(options: {
};
}
function publishDirectoryChanges(
previous: readonly SessionSummary[],
next: readonly SessionSummary[],
events: MutableRemoteEvents<SessionDirectoryEvent>,
context: Context,
): void {
const previousById = new Map(previous.map((session) => [session.sessionId, session]));
const nextById = new Map(next.map((session) => [session.sessionId, session]));
for (const session of previous) {
if (!nextById.has(session.sessionId)) events.emit({ type: "deleted", sessionId: session.sessionId }, context);
}
for (const session of next) {
const existing = previousById.get(session.sessionId);
if (existing === undefined) events.emit({ type: "created", session }, context);
else if (!sameSessionSummary(existing, session)) events.emit({ type: "changed", session }, context);
}
}
function sameSessionSummary(left: SessionSummary, right: SessionSummary): boolean {
return left.serverId === right.serverId && left.sessionId === right.sessionId && left.createdAt === right.createdAt;
}
function createProviderAttachment(
provider: RemoteServiceProvider,
onRelease: () => void,

7
packages/coding-agent/src/experimental/services/sessions.ts
View file

@ -1,4 +1,4 @@
import { type Context, defineService, type RemoteEvents, type ReplicatedState } from "@earendil-works/pi-agent-core";
import { type Context, defineService, type ReplicatedState } from "@earendil-works/pi-agent-core";
import type { SessionCreateOptions, SessionSummary } from "@earendil-works/pi-protocol";
export interface SessionDirectoryState {
@ -6,13 +6,8 @@ export interface SessionDirectoryState {
sessions: SessionSummary[];
}
export type SessionDirectoryEvent =
| { type: "created" | "changed"; session: SessionSummary }
| { type: "deleted"; sessionId: string };
export interface SessionDirectory {
readonly state: ReplicatedState<SessionDirectoryState>;
readonly events: RemoteEvents<SessionDirectoryEvent>;
}
export const SessionDirectory = defineService<SessionDirectory>("pi.session-directory");

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

@ -1,14 +1,14 @@
import { ServiceSliceNotImplemented } from "@earendil-works/pi-agent-core";
import { defineFacet } from "../facets.ts";
import { Accounts, type AccountsState } from "./accounts.ts";
import { Transcript, type TranscriptEvent } from "./transcript.ts";
import { Transcript } from "./transcript.ts";
/** Documented built-in surfaces whose implementations belong to later service slices. */
export const accountsServiceFacet = defineFacet({
id: "@pi/accounts",
setup(env) {
env.provide(Accounts, {
state: env.remoteState<AccountsState>({ providers: [] }),
state: env.replicatedState<AccountsState>({ providers: [] }),
async remove() {
throw new ServiceSliceNotImplemented("Accounts.remove");
},
@ -20,7 +20,6 @@ export const transcriptServiceFacet = defineFacet({
id: "@pi/transcript",
setup(env) {
env.provide(Transcript, {
events: env.remoteEvents<TranscriptEvent>(),
async snapshot() {
throw new ServiceSliceNotImplemented("Transcript.snapshot");
},

12
packages/coding-agent/src/experimental/services/transcript.ts
View file

@ -1,22 +1,14 @@
import { type Context, defineService, type RemoteEvents, type RemoteJson } from "@earendil-works/pi-agent-core";
import type { LaneEvent, LaneSnapshot } from "@earendil-works/pi-protocol";
import { type Context, defineService, type RemoteJson } from "@earendil-works/pi-agent-core";
import type { LaneSnapshot } from "@earendil-works/pi-protocol";
export interface TranscriptSnapshot {
revision: number;
lane: RemoteJson<LaneSnapshot>;
}
export interface TranscriptEvent {
type: "lane_event";
revision: number;
event: RemoteJson<LaneEvent>;
}
export interface Transcript {
/** Authoritative snapshot at one transcript revision. */
snapshot(context: Context): Promise<TranscriptSnapshot>;
/** Ordered semantic deltas; final message events replace streamed projections authoritatively. */
readonly events: RemoteEvents<TranscriptEvent>;
}
export const Transcript = defineService<Transcript>("pi.transcript");

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

@ -608,7 +608,7 @@ export class SessionWorkerManager {
);
return;
}
if (message.type === "service_event") {
if (message.type === "service_update") {
this.#handleServiceEvent(
event.from,
message.token,

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

@ -221,7 +221,7 @@ export const SessionWorkerEventSchema = Type.Union([
event: LaneEventSchema,
}),
Type.Object({
type: Type.Literal("service_event"),
type: Type.Literal("service_update"),
token: Type.String(),
sessionKey: Type.String(),
scope: WorkerOperationScopeSchema,
@ -593,7 +593,7 @@ async function run(options: SessionWorkerOptions, createHarness: CreateSessionWo
facetLoader: runtime.facetLoader,
publish: (scope, subscriptionId, update) =>
control.send({
type: "service_event",
type: "service_update",
token,
sessionKey,
scope,

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

@ -3,8 +3,7 @@ import {
createLoopbackServiceConnection,
RemoteServiceNamespace,
RemoteServiceProvider,
remoteEvents,
remoteState,
replicatedState,
} from "@earendil-works/pi-agent-core";
import type { SessionSummary } from "@earendil-works/pi-protocol";
import { describe, expect, test, vi } from "vitest";
@ -20,7 +19,6 @@ import type {
import { Models, type ModelsState } from "../src/experimental/services/models.ts";
import {
SessionDirectory,
type SessionDirectoryEvent,
type SessionDirectoryState,
SessionManagement,
} from "../src/experimental/services/sessions.ts";
@ -34,10 +32,9 @@ function session(sessionId: string, createdAt: number): SessionSummary {
describe("experimental client TUI", () => {
test("creates and switches Sessions, then selects a model through facet service handles", async () => {
const directoryState = remoteState<SessionDirectoryState>({ revision: 1, sessions: [session("one", 1)] });
const directoryEvents = remoteEvents<SessionDirectoryEvent>();
const attachment = remoteState<SessionAttachmentState>({ status: "detached" });
const modelsState = remoteState<ModelsState>({
const directoryState = replicatedState<SessionDirectoryState>({ revision: 1, sessions: [session("one", 1)] });
const attachment = replicatedState<SessionAttachmentState>({ status: "detached" });
const modelsState = replicatedState<ModelsState>({
catalog: {
revision: 1,
availableModels: [
@ -51,7 +48,6 @@ describe("experimental client TUI", () => {
const create = vi.fn(async () => {
const created = session("two", 2);
directoryState.set({ revision: 2, sessions: [...directoryState.value.sessions, created] }, BACKGROUND_CONTEXT);
directoryEvents.emit({ type: "created", session: created }, BACKGROUND_CONTEXT);
return created;
});
const select = vi.fn(async (model: { provider: string; modelId: string }) => {
@ -62,7 +58,7 @@ describe("experimental client TUI", () => {
});
const serverProvider = new RemoteServiceProvider([SessionDirectory, SessionManagement]);
serverProvider.provide(SessionDirectory, { state: directoryState, events: directoryEvents });
serverProvider.provide(SessionDirectory, { state: directoryState });
serverProvider.provide(SessionManagement, {
create,
async remove() {},
@ -88,7 +84,7 @@ describe("experimental client TUI", () => {
});
const serverServices: ServerServiceConnection = Object.assign(serverNamespace, {
acceptsUnavailableServices: false,
connection: remoteState<ServerConnectionState>({ status: "connected", since: "now" }),
connection: replicatedState<ServerConnectionState>({ status: "connected", since: "now" }),
async catalogue() {
return serverProvider.catalogue;
},
@ -158,8 +154,6 @@ describe("experimental client TUI", () => {
try {
expect(component.render(80).join("\n")).toContain("one");
expect(component.render(80).join("\n")).toContain("Loaded test TUI facet");
directoryEvents.emit({ type: "deleted", sessionId: "removed" }, BACKGROUND_CONTEXT);
expect(component.render(80).join("\n")).toContain("Session removed: removed");
component.handleInput("\r");
await vi.waitFor(() => expect(component.render(80).join("\n")).toContain("Experimental Models"));
@ -183,7 +177,6 @@ describe("experimental client TUI", () => {
expect(disposeLoadedFacets).toHaveBeenCalledOnce();
const rendersAfterClose = requestRender.mock.calls.length;
directoryState.set({ revision: 3, sessions: [] }, BACKGROUND_CONTEXT);
directoryEvents.emit({ type: "deleted", sessionId: "two" }, BACKGROUND_CONTEXT);
attachment.set({ status: "detached" }, BACKGROUND_CONTEXT);
modelsState.set({ ...modelsState.value, refresh: { status: "refreshing" } }, BACKGROUND_CONTEXT);
expect(requestRender).toHaveBeenCalledTimes(rendersAfterClose);

2
packages/coding-agent/test/experimental-facets.test.ts
View file

@ -156,7 +156,7 @@ describe("experimental facet host", () => {
const provider = defineFacet({
id: "provider",
setup(env) {
state = env.remoteState({ value: 0 });
state = env.replicatedState({ value: 0 });
env.provide(Watched, { state });
},
});

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

@ -16,11 +16,7 @@ import {
type SessionAttachmentState,
} from "../src/experimental/services/connection.ts";
import { Models } from "../src/experimental/services/models.ts";
import {
SessionDirectory,
type SessionDirectoryEvent,
SessionManagement,
} from "../src/experimental/services/sessions.ts";
import { SessionDirectory, SessionManagement } from "../src/experimental/services/sessions.ts";
import { Transcript } from "../src/experimental/services/transcript.ts";
import {
configureExperimentalWorkerModel,
@ -321,10 +317,6 @@ describe("experimental durable server composition", () => {
const secondDirectory = secondServices.use(SessionDirectory);
const firstManagement = firstServices.use(SessionManagement);
const secondManagement = secondServices.use(SessionManagement);
const firstEvents: SessionDirectoryEvent[] = [];
const secondEvents: SessionDirectoryEvent[] = [];
const removeFirstEvents = firstDirectory.events.subscribe((event) => firstEvents.push(event));
const removeSecondEvents = secondDirectory.events.subscribe((event) => secondEvents.push(event));
await Promise.all([firstServices.ready(BACKGROUND_CONTEXT), secondServices.ready(BACKGROUND_CONTEXT)]);
expect(firstDirectory.state.value?.sessions.map(({ sessionId }) => sessionId)).toEqual(["demo-1", "demo-2"]);
@ -333,11 +325,6 @@ describe("experimental durable server composition", () => {
await vi.waitFor(() => {
expect(firstDirectory.state.value?.sessions.map(({ sessionId }) => sessionId)).toContain("demo-3");
expect(secondDirectory.state.value).toEqual(firstDirectory.state.value);
expect(firstEvents).toContainEqual({
type: "created",
session: expect.objectContaining({ sessionId: "demo-3" }),
});
expect(secondEvents).toEqual(firstEvents);
});
await Promise.all([
@ -354,14 +341,8 @@ describe("experimental durable server composition", () => {
expect(secondDirectory.state.value).toEqual(firstDirectory.state.value);
});
await secondManagement.detach(BACKGROUND_CONTEXT);
await vi.waitFor(() => {
expect(firstEvents).toContainEqual({ type: "deleted", sessionId: "demo-1" });
expect(secondEvents).toEqual(firstEvents);
});
expect(errors).toEqual([]);
removeFirstEvents();
removeSecondEvents();
await Promise.all([firstServices.dispose(BACKGROUND_CONTEXT), secondServices.dispose(BACKGROUND_CONTEXT)]);
});

5
packages/coding-agent/test/experimental-service-stubs.test.ts
View file

@ -16,11 +16,6 @@ describe("experimental built-in service surface", () => {
),
).rejects.toBeInstanceOf(ServiceSliceNotImplemented);
const transcript = host.services.subscribe("pi.transcript", "singleton", () => {}).snapshot;
expect(transcript.instances[0]?.members).toEqual([
{ name: "events", kind: "events" },
{ name: "snapshot", kind: "method" },
]);
await expect(
host.services.invoke({ serviceId: "pi.transcript", member: "snapshot", args: [] }, BACKGROUND_CONTEXT),
).rejects.toBeInstanceOf(ServiceSliceNotImplemented);

2
packages/coding-agent/test/fixtures/faux-session-worker.ts vendored
View file

@ -36,7 +36,7 @@ if (process.argv[1] !== undefined && resolve(process.argv[1]) === fileURLToPath(
setup(env) {
const probes = env.provideMany(KeyedProbe);
const spawn = (value: string): void => {
const state = env.remoteState({ value });
const state = env.replicatedState({ value });
let close = (): void => {};
close = probes.add("probe", {
state,

8
packages/coding-agent/test/fixtures/plugin-app/lib/client.ts vendored
View file

@ -12,7 +12,7 @@ export type ConnectionState =
| { status: "connected" }
| { status: "disconnected"; reason: string };
class ClientRemoteState<T> implements ReplicatedState<T> {
class ClientReplicatedState<T> implements ReplicatedState<T> {
private readonly listeners = new Set<(value: T) => void>();
private current: T;
@ -37,10 +37,10 @@ class ClientRemoteState<T> implements ReplicatedState<T> {
}
export class ClientStateStore {
readonly connection = new ClientRemoteState<ConnectionState>({ status: "connecting" });
readonly connection = new ClientReplicatedState<ConnectionState>({ status: "connecting" });
updates = 0;
private readonly listeners = new Set<() => void>();
private readonly states = new Map<string, ClientRemoteState<unknown>>();
private readonly states = new Map<string, ClientReplicatedState<unknown>>();
apply(message: ServerWireMessage): void {
if (message.type === "response") return;
@ -85,7 +85,7 @@ export class ClientStateStore {
const key = `${service}.${property}`;
const current = this.states.get(key);
if (current) current.set(value);
else this.states.set(key, new ClientRemoteState(value));
else this.states.set(key, new ClientReplicatedState(value));
if (count) this.updates++;
}
}

10
packages/coding-agent/test/fixtures/plugin-app/lib/session.ts vendored
View file

@ -1,7 +1,7 @@
import { type AppPlugin, type MutableReplicatedState, rpcOptions, type Service } from "./api.ts";
import type { ServerWireMessage, SessionRequest, StateSnapshot } from "./protocol.ts";
class SessionRemoteState<T> implements MutableReplicatedState<T> {
class SessionReplicatedState<T> implements MutableReplicatedState<T> {
private readonly listeners = new Set<(value: T) => void>();
private current: T;
@ -55,7 +55,7 @@ export interface SessionContext {
onClientDisconnect(callback: (clientId: string) => void): void;
onClose(callback: () => void): void;
provide<T>(service: Service<T>, implementation: T): void;
remoteState<T>(initial: T): MutableReplicatedState<T>;
replicatedState<T>(initial: T): MutableReplicatedState<T>;
use<T>(service: Service<T>): T;
}
@ -111,7 +111,7 @@ export class SessionRuntime<ClientContext> {
onClientDisconnect: (callback) => this.clientDisconnectCallbacks.push(callback),
onClose: (callback) => this.closeCallbacks.push(callback),
provide: (service, implementation) => this.provide(service, implementation),
remoteState: (initial) => new SessionRemoteState(initial),
replicatedState: (initial) => new SessionReplicatedState(initial),
use: (service) => this.use(service),
};
for (const plugin of this.plugins) await plugin.session?.(context);
@ -126,7 +126,7 @@ export class SessionRuntime<ClientContext> {
if (this.services.has(service.id)) throw new Error(`Service already provided: ${service.id}`);
this.services.set(service.id, implementation as object);
for (const [property, value] of Object.entries(implementation as object)) {
if (!(value instanceof SessionRemoteState)) continue;
if (!(value instanceof SessionReplicatedState)) continue;
value.subscribe((next) => {
this.hub.broadcast({
type: "state_update",
@ -149,7 +149,7 @@ export class SessionRuntime<ClientContext> {
for (const [serviceId, implementation] of this.services) {
const states: Record<string, unknown> = {};
for (const [property, value] of Object.entries(implementation)) {
if (value instanceof SessionRemoteState) states[property] = structuredClone(value.value);
if (value instanceof SessionReplicatedState) states[property] = structuredClone(value.value);
}
if (Object.keys(states).length > 0) snapshot[serviceId] = states;
}

2
packages/coding-agent/test/fixtures/plugin-app/model-app/plugins.ts vendored
View file

@ -23,7 +23,7 @@ export const providersBuiltin = definePlugin<SessionContext, TuiContext>({
values.set(key, structuredClone(value));
},
};
const state = context.remoteState<ModelsState>({
const state = context.replicatedState<ModelsState>({
catalog: providers.snapshot(),
configuration: { model: undefined, thinkingLevel: "high" },
refresh: { status: "idle" },

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

@ -123,7 +123,7 @@ export type ProtocolRpcResult = JsonValue | undefined;
export const ServiceMemberDescriptionSchema = StrictObject({
name: IdSchema,
kind: Type.Union([Type.Literal("method"), Type.Literal("state"), Type.Literal("events")]),
kind: Type.Union([Type.Literal("method"), Type.Literal("state")]),
});
export type ServiceMemberDescription = Static<typeof ServiceMemberDescriptionSchema>;
@ -162,12 +162,6 @@ export const ServiceProviderUpdateSchema = Type.Union([
sequence: Type.Integer({ minimum: 1 }),
value: JsonValueSchema,
}),
StrictObject({
type: Type.Literal("event"),
instance: Type.Optional(ServiceInstanceAddressSchema),
member: IdSchema,
event: JsonValueSchema,
}),
StrictObject({
type: Type.Literal("unavailable"),
}),
@ -381,7 +375,7 @@ export const EventEnvelopeSchema = StrictObject({
event: LaneEventSchema,
});
export const ServiceEventEnvelopeSchema = StrictObject({
type: Type.Literal("service_event"),
type: Type.Literal("service_update"),
subscriptionId: IdSchema,
update: ServiceProviderUpdateSchema,
});

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

@ -270,36 +270,47 @@ describe("protocol validation", () => {
});
expect(
parseServiceSubscriptionSnapshot({
serviceId: "pi.transcript",
serviceId: "pi.models",
mode: "singleton",
instances: [{ members: [{ name: "events", kind: "events" }], states: {} }],
instances: [
{
members: [{ name: "state", kind: "state" }],
states: { state: { sequence: 0, value: { revision: 1 } } },
},
],
}),
).toEqual({
serviceId: "pi.transcript",
serviceId: "pi.models",
mode: "singleton",
instances: [{ members: [{ name: "events", kind: "events" }], states: {} }],
instances: [
{
members: [{ name: "state", kind: "state" }],
states: { state: { sequence: 0, value: { revision: 1 } } },
},
],
});
expect(
parseServerMessage({
type: "service_event",
type: "service_update",
subscriptionId: "subscription-1",
update: {
type: "event",
member: "events",
event: { type: "changed", revision: 1 },
type: "state",
member: "state",
sequence: 1,
value: { revision: 2 },
},
}),
).toMatchObject({ update: { type: "event", member: "events" } });
).toMatchObject({ update: { type: "state", member: "state" } });
expect(
parseServerMessage({
type: "service_event",
type: "service_update",
subscriptionId: "subscription-1",
update: { type: "unavailable" },
}),
).toMatchObject({ update: { type: "unavailable" } });
expect(
parseServerMessage({
type: "service_event",
type: "service_update",
subscriptionId: "subscription-1",
update: {
type: "replaced",

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

@ -326,7 +326,7 @@ export class Server<TMetadata extends SessionMetadata = SessionMetadata> {
result = await state.serverServices.invokeService(
envelope.call,
async (subscriptionId, update) => {
await this.sendMessage(state, { type: "service_event", subscriptionId, update });
await this.sendMessage(state, { type: "service_update", subscriptionId, update });
},
context,
);

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

@ -325,7 +325,7 @@ export class SessionRouter<TMetadata extends SessionMetadata = SessionMetadata>
attachment.lease,
call,
(subscriptionId, update, updateContext) =>
publish({ type: "service_event", subscriptionId, update }, updateContext),
publish({ type: "service_update", subscriptionId, update }, updateContext),
context,
);
this.trackOperation(attachment, result);