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(chord): keep services available during facet reloads

Browse source
This commit is contained in:
Armin Ronacher 2026-09-01 10:47:08 +02:00
parent 7cf456aa1c
commit c4b0e35abe
13 changed files with 386 additions and 163 deletions
Download patch file Download diff file Expand all files Collapse all files

31
packages/chord/PLANNING.md
View file

@ -207,25 +207,25 @@ The sequence is:
load replacement module generation
→ run replacement setup
→ validate replacement shape and remote implementations
→ withdraw replaced providers
→ activate replacements in dependency order while the old providers remain routed
→ atomically replace local targets and publish complete remote singleton replacement snapshots
→ drain calls admitted by the retired singleton and keyed instances
→ deactivate replaced plugins in reverse dependency order
→ activate replacements in dependency order
→ reconnect local provider slots
→ publish complete remote singleton replacement snapshots
→ dispose the retired loaded generation
```
Properties:
- existing local and remote singleton facades keep object identity;
- captured local methods dispatch to the replacement;
- captured remote methods and replicated-state facades address the replacement;
- while withdrawn, calls fail rather than queue and replicated state is unhydrated;
- keyed instances from the old plugin close; instances reconstructed by the replacement use fresh generations;
- setup or shape-validation failure leaves the active generation unchanged; and
- a failure after provider withdrawal begins disposes the complete host instead of exposing a partially active graph.
- captured local and remote methods dispatch to the old provider before cutover and the replacement after cutover;
- replicated-state facades install the replacement snapshot without becoming unhydrated;
- calls admitted before cutover finish against the retired provider while later calls use the replacement;
- keyed instances from the old plugin close; instances staged by the replacement use fresh generations;
- setup, shape-validation, or replacement-activation failure leaves the active generation unchanged;
- named host resources acquired during activation support overlapping staged replacements, so candidate cleanup restores the old registration and retired cleanup cannot remove the replacement; and
- cutover performs no asynchronous work or facet callbacks.
There is no rollback guarantee after old-plugin deactivation begins. Committed application effects are outside the reload transaction.
There is no rollback after cutover. Retired-generation cleanup failures are reported without disconnecting the replacement. Committed application effects are outside the reload transaction.
#### Structural replacement
@ -253,14 +253,9 @@ This full-generation path supplies actual plugin load and unload semantics witho
### 6.2 Calls during unload
Provider withdrawal rejects new calls. The runtime must generation-fence admitted calls and state publication so old completions cannot update a replacement provider.
Replacement atomically stops new admission to the retired service target. Calls already admitted to that target retain it until they settle, while new calls through the same stable facade use the replacement. Old state publication is detached at cutover and cannot update the replacement replica.
The initial implementation must explicitly choose and test one admitted-call drain policy before module-loader disposal:
1. wait for admitted calls to settle while signalling plugin deactivation separately; or
2. cancel them through a provider-generation signal, then wait for settlement.
Silently abandoning admitted promises is not acceptable. Transport disconnect still cancels only connection-owned invocations; it does not imply application-level cancellation or rollback.
Facet deactivation and loaded-module disposal wait for admitted calls to drain. A future killable-isolate host may force termination during abnormal shutdown, but ordinary reload does not reject admitted calls. Transport disconnect still cancels only connection-owned invocations; it does not imply application-level cancellation or rollback.
### 6.3 Update failure reporting

8
packages/chord/README.md
View file

@ -193,8 +193,12 @@ receiving host.
To reload, load a candidate, pass its facets to `FacetHost.reload()`, dispose the
candidate on failure, and dispose the retired `LoadedFacets` only after a
successful cutover. Reload retains the host's existing shape-preservation rules.
The bundler writes a complete temporary directory before replacing the previous
successful cutover. The host activates and validates the candidate while the old
providers remain routed, atomically replaces singleton targets, then drains calls
admitted by the retired targets before resolving reload. Stable service handles
therefore do not become unavailable during an ordinary reload. Keyed instances
remain incarnation-specific and replacements receive fresh generations. The
bundler writes a complete temporary directory before replacing the previous
output, so loaders do not observe partially built generations.
See [PLANNING.md](PLANNING.md) for the broader RPC and generation-loading

184
packages/chord/src/facets/host.ts
View file

@ -3,7 +3,7 @@ import { RemoteServiceBindingImpl } from "../services/consumer.ts";
import { ServiceSlot } from "../services/handle.ts";
import { InstanceDirectory, type InstanceDirectoryEntry } from "../services/instances.ts";
import { createLoopbackServiceTransport } from "../services/loopback.ts";
import { RemoteServiceProvider } from "../services/provider.ts";
import { RemoteServiceProvider, validateRemoteServiceImplementation } from "../services/provider.ts";
import { MutableReplicatedStateImpl } from "../services/state.ts";
import type {
Context,
@ -174,18 +174,23 @@ class LocalKeyedServiceRegistry implements KeyedServiceSource {
}
const generation = (registration.generations.get(key) ?? 0) + 1;
registration.generations.set(key, generation);
const slot = new ServiceSlot(service.id, false);
void slot.bind(implementation as object);
const instance: InstanceDirectoryEntry = {
key,
generation,
service: implementation as object,
service: slot.view<T>(() => {}) as object,
deactivate() {},
};
registration.directory.insert(instance);
let closed = false;
let retired: Promise<void> | undefined;
return () => {
if (closed) return;
if (closed) return retired;
closed = true;
registration.directory.remove(instance);
retired = slot.unbind();
return retired;
};
}
@ -239,7 +244,7 @@ class HostServiceSlots {
let stopped = false;
const stop = source.observe(service, (target, context) => {
const slot = new ServiceSlot(service.id, !service.local);
slot.bind(target as object);
void slot.bind(target as object);
return handler(
slot.view<T>(() => {
assertAccess();
@ -257,12 +262,8 @@ class HostServiceSlots {
};
}
bindSingleton(serviceId: string, target: object): void {
this.#singletons.get(serviceId)?.bind(target);
}
unbindSingleton(serviceId: string): void {
this.#singletons.get(serviceId)?.unbind();
bindSingleton(serviceId: string, target: object): Promise<void> {
return this.#singletons.get(serviceId)?.bind(target) ?? Promise.resolve();
}
bindKeyed(serviceId: string, services: KeyedServiceSource): void {
@ -270,37 +271,53 @@ class HostServiceSlots {
}
dispose(): void {
for (const slot of this.#singletons.values()) slot.unbind();
for (const slot of this.#singletons.values()) void slot.unbind();
this.#singletons.clear();
this.#keyedSources.clear();
}
}
type ServiceInstanceInstaller<T> = (key: string, implementation: T) => () => void;
type ServiceInstanceInstaller<T> = (key: string, implementation: T) => Disposal;
interface StagedServiceInstance<T> {
readonly key: string;
readonly implementation: T;
release?: Disposal;
retirement?: Promise<void>;
}
class StagedServiceSpawner<T> implements ServiceSpawner<T> {
readonly #lifecycle: FacetLifecycle;
readonly #validate: (key: string, implementation: T) => void;
readonly #instances = new Map<string, StagedServiceInstance<T>>();
#installer: ServiceInstanceInstaller<T> | undefined;
constructor(lifecycle: FacetLifecycle) {
constructor(lifecycle: FacetLifecycle, validate: (key: string, implementation: T) => void) {
this.#lifecycle = lifecycle;
this.#validate = validate;
}
connect(installer: ServiceInstanceInstaller<T>): void {
if (this.#installer !== undefined) throw new Error("Facet service provider is already connected");
this.#installer = installer;
for (const instance of this.#instances.values()) {
instance.release = installer(instance.key, instance.implementation);
}
}
spawn(key: string, implementation: T): () => void {
this.#lifecycle.assertActive("spawn service instances");
const installer = this.#installer;
if (installer === undefined) throw new Error("Facet service provider is not connected");
const release = installer(key, implementation);
let closed = false;
const close = (): void => {
if (closed) return;
closed = true;
release();
this.#validate(key, implementation);
if (this.#instances.has(key)) throw new Error(`Facet service already has a live instance with key ${key}`);
const instance: StagedServiceInstance<T> = { key, implementation };
this.#instances.set(key, instance);
if (this.#installer !== undefined) instance.release = this.#installer(key, implementation);
const close = (): void | Promise<void> => {
if (this.#instances.get(key) !== instance) return instance.retirement;
this.#instances.delete(key);
const result = instance.release?.();
if (isPromiseLike(result)) instance.retirement = Promise.resolve(result);
return instance.retirement;
};
this.#lifecycle.own(close);
return close;
@ -314,8 +331,7 @@ type FacetProvision =
readonly implementation: object;
install(provider: RemoteServiceProvider): void;
validateReplacement(provider: RemoteServiceProvider): void;
withdraw(provider: RemoteServiceProvider): void;
replace(provider: RemoteServiceProvider): void;
replace(provider: RemoteServiceProvider): Promise<void>;
}
| {
readonly kind: "keyed";
@ -433,14 +449,7 @@ export class FacetKernel {
candidates.push(record);
}
} catch (error) {
const cleanupErrors: unknown[] = [];
for (const record of staged.reverse()) {
try {
await record.lifecycle.dispose();
} catch (cleanupError) {
cleanupErrors.push(cleanupError);
}
}
const cleanupErrors = await disposeFacetRecords(staged.reverse());
this.#phase = "active";
if (cleanupErrors.length > 0) {
throw new AggregateError([error, ...cleanupErrors], "Facet reload setup and cleanup failed");
@ -449,61 +458,50 @@ export class FacetKernel {
}
const replacements = new Map(candidates.map((record) => [record.facetId, record]));
let candidatesInstalled = false;
try {
for (const candidate of candidates) {
for (const provision of candidate.provisions) {
if (provision.kind !== "singleton") continue;
if (provision.service.local) this.#serviceSlots.unbindSingleton(provision.service.id);
else provision.withdraw(this.provider);
}
}
const disposalErrors: unknown[] = [];
for (const id of [...this.#activationOrder].reverse()) {
if (!replacements.has(id)) continue;
const record = this.#facets.get(id)!;
this.#facets.delete(id);
try {
await record.lifecycle.dispose();
} catch (error) {
disposalErrors.push(error);
}
}
if (disposalErrors.length > 0) {
throw new AggregateError(disposalErrors, "Failed to deactivate reloaded facets");
}
for (const candidate of candidates) this.#facets.set(candidate.facetId, candidate);
candidatesInstalled = true;
for (const candidate of candidates) {
for (const provision of candidate.provisions) {
if (provision.kind !== "keyed") continue;
if (provision.service.local) provision.connectLocal(this.#localKeyedRegistry);
else provision.connectRemote(this.provider);
}
}
for (const id of this.#activationOrder) {
const candidate = replacements.get(id);
if (candidate === undefined) continue;
await candidate.lifecycle.activate();
for (const provision of candidate.provisions) {
if (provision.kind !== "singleton") continue;
if (provision.service.local) {
this.#serviceSlots.bindSingleton(provision.service.id, provision.implementation);
} else {
provision.replace(this.provider);
}
}
await replacements.get(id)?.lifecycle.activate();
}
this.#phase = "active";
} catch (error) {
const cleanupErrors = await this.#terminate(candidatesInstalled ? [] : candidates);
const cleanupErrors = await disposeFacetRecords([...candidates].reverse());
this.#phase = "active";
if (cleanupErrors.length > 0) {
throw new AggregateError([error, ...cleanupErrors], "Facet reload failed and host cleanup failed");
throw new AggregateError([error, ...cleanupErrors], "Facet reload activation and cleanup failed");
}
throw error;
}
const previous = new Map(candidates.map(({ facetId }) => [facetId, this.#facets.get(facetId)!] as const));
const retirements = candidates.flatMap(({ provisions }) =>
provisions.flatMap((provision) => {
if (provision.kind !== "singleton") return [];
return [
provision.service.local
? this.#serviceSlots.bindSingleton(provision.service.id, provision.implementation)
: provision.replace(this.provider),
];
}),
);
for (const candidate of candidates) this.#facets.set(candidate.facetId, candidate);
await Promise.all(retirements);
const retirementErrors = await disposeFacetRecords(
[...this.#activationOrder].reverse().flatMap((id) => {
const record = previous.get(id);
return record === undefined ? [] : [record];
}),
);
for (const candidate of candidates) {
for (const provision of candidate.provisions) {
if (provision.kind !== "keyed") continue;
if (provision.service.local) provision.connectLocal(this.#localKeyedRegistry);
else provision.connectRemote(this.provider);
}
}
this.#phase = "active";
if (retirementErrors.length === 1) throw retirementErrors[0];
if (retirementErrors.length > 1) {
throw new AggregateError(retirementErrors, "Failed to retire replaced facets");
}
}
async dispose(): Promise<void> {
@ -537,14 +535,19 @@ export class FacetKernel {
install: (provider) => provider.provide(service, implementation as NoInfer<RemoteServiceContract<T>>),
validateReplacement: (provider) =>
provider.validateReplacement(service, implementation as NoInfer<RemoteServiceContract<T>>),
withdraw: (provider) => provider.withdraw(service),
replace: (provider) => provider.replace(service, implementation as NoInfer<RemoteServiceContract<T>>),
});
},
provideMany: <T>(service: Service<T>): ServiceSpawner<T> => {
lifecycle.assertSettingUp("provide service instances");
recordServiceReference(runtime.provides, service, "keyed");
const instances = new StagedServiceSpawner<T>(lifecycle);
const instances = new StagedServiceSpawner<T>(lifecycle, (key, implementation) => {
if (key.length === 0) throw new TypeError("Facet service instance key must not be empty");
if (typeof implementation !== "object" || implementation === null || Array.isArray(implementation)) {
throw new TypeError(`Facet service ${service.id} implementation must be an object`);
}
if (!service.local) validateRemoteServiceImplementation(service.id, implementation);
});
provisions.push({
kind: "keyed",
service,
@ -735,16 +738,9 @@ export class FacetKernel {
}
}
async #terminate(extraRecords: readonly FacetRuntime[] = []): Promise<unknown[]> {
async #terminate(): Promise<unknown[]> {
this.#phase = "disposing";
const errors = await this.#disposeLifecycles();
for (const record of [...extraRecords].reverse()) {
try {
await record.lifecycle.dispose();
} catch (error) {
errors.push(error);
}
}
try {
this.#localKeyedServices?.dispose();
} catch (error) {
@ -784,6 +780,18 @@ export class FacetKernel {
}
}
async function disposeFacetRecords(records: readonly FacetRuntime[]): Promise<unknown[]> {
const errors: unknown[] = [];
for (const record of records) {
try {
await record.lifecycle.dispose();
} catch (error) {
errors.push(error);
}
}
return errors;
}
function validateFacets(
records: readonly FacetShape[],
externalServices: ReadonlyMap<string, { readonly mode: ServiceMode }>,

60
packages/chord/src/services/calls.ts Normal file
View file

@ -0,0 +1,60 @@
type Resolve = () => void;
/** Tracks calls admitted to one service target so replacement can drain it without disconnecting consumers. */
export class CallTracker {
#active = 0;
#retired = false;
#settled: Promise<void> | undefined;
#resolve: Resolve | undefined;
run<T>(callback: () => T): T {
if (this.#retired) throw new Error("Service target is retired");
this.#active += 1;
let result: T;
try {
result = callback();
} catch (error) {
this.#release();
throw error;
}
if (!isPromiseLike(result)) {
this.#release();
return result;
}
return Promise.resolve(result).then(
(value) => {
this.#release();
return value;
},
(error: unknown) => {
this.#release();
throw error;
},
) as T;
}
retire(): Promise<void> {
if (this.#retired) return this.#settled ?? Promise.resolve();
this.#retired = true;
if (this.#active === 0) return Promise.resolve();
this.#settled = new Promise<void>((resolve) => {
this.#resolve = resolve;
});
return this.#settled;
}
#release(): void {
this.#active -= 1;
if (!this.#retired || this.#active !== 0) return;
this.#resolve?.();
this.#resolve = undefined;
}
}
function isPromiseLike(value: unknown): value is PromiseLike<unknown> {
return (
((typeof value === "object" && value !== null) || typeof value === "function") &&
"then" in value &&
typeof value.then === "function"
);
}

2
packages/chord/src/services/consumer.ts
View file

@ -279,7 +279,7 @@ class KeyedBinding<T> {
let stopped = false;
const stop = this.#instances.observe<T>((service, context) => {
const slot = new ServiceSlot(this.#service.id, true);
slot.bind(service as object);
void slot.bind(service as object);
return handler(
slot.view<T>(() => {
this.#assertAccess();

36
packages/chord/src/services/handle.ts
View file

@ -1,6 +1,14 @@
import { CallTracker } from "./calls.ts";
interface ResolvedValue {
readonly value: unknown;
readonly receiver: object;
readonly calls: CallTracker;
}
interface ServiceTarget {
readonly implementation: object;
readonly calls: CallTracker;
}
type ValueResolver = () => ResolvedValue;
@ -9,7 +17,7 @@ type ValueResolver = () => ResolvedValue;
export class ServiceSlot {
readonly #serviceId: string;
readonly #wrapObjects: boolean;
#implementation: object | undefined;
#target: ServiceTarget | undefined;
constructor(serviceId: string, wrapObjects: boolean) {
this.#serviceId = serviceId;
@ -20,21 +28,26 @@ export class ServiceSlot {
return new ServiceView(this, assertAccess, this.#wrapObjects).proxy as T;
}
bind(implementation: object): void {
this.#implementation = implementation;
bind(implementation: object): Promise<void> {
const previous = this.#target;
this.#target = { implementation, calls: new CallTracker() };
return previous?.calls.retire() ?? Promise.resolve();
}
unbind(): void {
this.#implementation = undefined;
unbind(): Promise<void> {
const previous = this.#target;
this.#target = undefined;
return previous?.calls.retire() ?? Promise.resolve();
}
resolve(property: PropertyKey, assertAccess: () => void): ResolvedValue {
assertAccess();
const implementation = this.#implementation;
if (implementation === undefined) throw new Error(`Service ${this.#serviceId} is disconnected`);
const target = this.#target;
if (target === undefined) throw new Error(`Service ${this.#serviceId} is disconnected`);
return {
value: Reflect.get(implementation, property, implementation),
receiver: implementation,
value: Reflect.get(target.implementation, property, target.implementation),
receiver: target.implementation,
calls: target.calls,
};
}
}
@ -83,9 +96,9 @@ class ValueView {
}
#invoke(args: unknown[]): unknown {
const { value, receiver } = this.#resolve();
const { value, receiver, calls } = this.#resolve();
if (typeof value !== "function") throw new TypeError("Service member is not callable");
return Reflect.apply(value, receiver, args);
return calls.run(() => Reflect.apply(value, receiver, args));
}
#get(property: PropertyKey): unknown {
@ -95,6 +108,7 @@ class ValueView {
return {
value: Reflect.get(parent.value, property, parent.value),
receiver: parent.value,
calls: parent.calls,
};
};
const current = resolve().value;

28
packages/chord/src/services/provider.ts
View file

@ -13,6 +13,7 @@ import type {
ServiceSubscription,
ServiceSubscriptionSnapshot,
} from "../types.ts";
import { CallTracker } from "./calls.ts";
import { RemoteServiceError } from "./errors.ts";
import { serviceDeliveryContext } from "./state.ts";
import { getReplicatedStateInternals, type ReplicatedStateInternals } from "./state-internals.ts";
@ -36,6 +37,7 @@ interface ProviderInstance {
readonly implementation: object;
readonly members: ReadonlyMap<string, InstanceMember>;
readonly removeMemberListeners: readonly (() => void)[];
readonly calls: CallTracker;
active: boolean;
}
@ -123,18 +125,20 @@ export class RemoteServiceProvider {
registration.singletonShape = shape;
}
/** Disconnect one singleton while preserving its active subscriptions and remote facades. */
withdraw<T>(service: Service<T>): void {
/** Disconnect one singleton and resolve after calls admitted by the retired target settle. */
withdraw<T>(service: Service<T>): Promise<void> {
this.#assertActive();
this.#assertRemotable(service);
this.#assertAllowed(service.id);
const registration = this.#registration(service.id, "singleton");
const previous = registration.singleton;
if (previous === undefined) return;
if (previous === undefined) return Promise.resolve();
previous.active = false;
for (const remove of previous.removeMemberListeners) remove();
delete registration.singleton;
const retired = previous.calls.retire();
this.#emit(registration, { type: "unavailable" });
return retired;
}
/** Check a singleton replacement without changing the active provider. */
@ -147,8 +151,8 @@ export class RemoteServiceProvider {
this.#assertSingletonShape(registration, serviceMemberShape(classified.members));
}
/** Replace one singleton while preserving its active subscriptions and remote facades. */
replace<T>(service: Service<T>, implementation: NoInfer<RemoteServiceContract<T>>): void {
/** Replace one singleton immediately and resolve after calls admitted by the retired target settle. */
replace<T>(service: Service<T>, implementation: NoInfer<RemoteServiceContract<T>>): Promise<void> {
this.#assertActive();
this.#assertRemotable(service);
this.#assertAllowed(service.id);
@ -164,7 +168,9 @@ export class RemoteServiceProvider {
}
registration.singleton = replacement;
registration.singletonShape = shape;
const retired = previous?.calls.retire() ?? Promise.resolve();
this.#emit(registration, { type: "replaced", snapshot: this.#snapshotInstance(replacement) });
return retired;
}
use<T>(service: Service<T>): T {
@ -198,14 +204,17 @@ export class RemoteServiceProvider {
registration.instances.set(key, instance);
this.#emit(registration, { type: "spawned", instance: this.#snapshotInstance(instance) });
let closed = false;
let retired: Promise<void> | undefined;
return () => {
if (closed) return;
if (closed) return retired;
closed = true;
if (registration.instances.get(key) !== instance) return;
if (registration.instances.get(key) !== instance) return undefined;
instance.active = false;
for (const remove of instance.removeMemberListeners) remove();
registration.instances.delete(key);
retired = instance.calls.retire();
this.#emit(registration, { type: "closed", instance: address });
return retired;
};
}
@ -230,7 +239,9 @@ export class RemoteServiceProvider {
`Remote service member ${call.serviceId}.${call.member} is not a method`,
);
}
const result: unknown = await Reflect.apply(member.method, instance.implementation, [...call.args, context]);
const result: unknown = await instance.calls.run(() =>
Reflect.apply(member.method, instance.implementation, [...call.args, context]),
);
return result as JsonValue | undefined;
}
@ -341,6 +352,7 @@ export class RemoteServiceProvider {
implementation: classified.implementation,
members: classified.members,
removeMemberListeners,
calls: new CallTracker(),
active: true,
};
for (const [name, member] of classified.members) {

2
packages/chord/src/types.ts
View file

@ -246,7 +246,7 @@ export interface FacetOptions {
export interface FacetHost {
readonly services: RemoteServiceProvider;
/** Replace active facets with matching IDs while preserving consumer service handles. */
/** Activate and atomically replace facets with matching IDs without disconnecting consumer service handles. */
reload(facets: readonly Facet[]): Promise<void>;
dispose(): Promise<void>;
}

88
packages/chord/test/facet-loader.test.ts
View file

@ -171,8 +171,8 @@ describe("facet loader", () => {
const loadedB = await loader.load();
const reload = host.reload(loadedB.facets);
await replacementStarted;
expect(() => localRead(BACKGROUND_CONTEXT)).toThrow(`Service ${LocalGenerationValue.id} is disconnected`);
await expect(remoteRead(BACKGROUND_CONTEXT)).rejects.toMatchObject({ code: "service_not_found" });
await expect(localRead(BACKGROUND_CONTEXT)).resolves.toBe("A");
await expect(remoteRead(BACKGROUND_CONTEXT)).resolves.toBe("A");
continueReplacement();
await reload;
await loadedA.dispose();
@ -193,8 +193,8 @@ describe("facet loader", () => {
"activate consumer:A",
"load B",
"setup provider B",
"deactivate provider A",
"activate provider B",
"deactivate provider A",
"unload A",
"deactivate consumer",
"deactivate provider B",
@ -202,6 +202,76 @@ describe("facet loader", () => {
]);
});
test("drains calls admitted by the old provider without blocking calls to the replacement", async () => {
let localValue: GenerationValue | undefined;
let oldCalls = 0;
let reportOldCallsStarted!: () => void;
const oldCallsStarted = new Promise<void>((resolve) => {
reportOldCallsStarted = resolve;
});
let finishOldCalls!: () => void;
const oldCallsCanFinish = new Promise<void>((resolve) => {
finishOldCalls = resolve;
});
let reportReplacementActivated!: () => void;
const replacementActivated = new Promise<void>((resolve) => {
reportReplacementActivated = resolve;
});
const provider = (name: string, block: boolean) =>
defineFacet({
id: "draining-provider",
setup(env) {
const implementation: GenerationValue = {
async read() {
if (block) {
oldCalls += 1;
if (oldCalls === 2) reportOldCallsStarted();
await oldCallsCanFinish;
}
return name;
},
};
env.provide(LocalGenerationValue, implementation);
env.provide(RemoteGenerationValue, implementation);
if (!block) env.onActivate(reportReplacementActivated);
},
});
const consumer = defineFacet({
id: "draining-consumer",
setup(env) {
localValue = env.use(LocalGenerationValue);
},
});
const host = await createFacetHost({ facets: [consumer, provider("A", true)] });
const remoteServices = createRemoteServiceBinding({
services: [RemoteGenerationValue],
transport: createLoopbackServiceTransport(host.services),
});
const remoteValue = remoteServices.use(RemoteGenerationValue);
await remoteServices.ready(BACKGROUND_CONTEXT);
const oldLocal = localValue!.read(BACKGROUND_CONTEXT);
const oldRemote = remoteValue.read(BACKGROUND_CONTEXT);
await oldCallsStarted;
let reloadSettled = false;
const reload = host.reload([provider("B", false)]).finally(() => {
reloadSettled = true;
});
await replacementActivated;
await new Promise((resolve) => setTimeout(resolve, 0));
await expect(localValue!.read(BACKGROUND_CONTEXT)).resolves.toBe("B");
await expect(remoteValue.read(BACKGROUND_CONTEXT)).resolves.toBe("B");
expect(reloadSettled).toBe(false);
finishOldCalls();
await expect(oldLocal).resolves.toBe("A");
await expect(oldRemote).resolves.toBe("A");
await reload;
await remoteServices.dispose(BACKGROUND_CONTEXT);
await host.dispose();
});
test("rejects remote singleton member shape changes before reload cutover", async () => {
let retained: GenerationValue | undefined;
let providerDisposed = false;
@ -247,7 +317,7 @@ describe("facet loader", () => {
expect(providerDisposed).toBe(true);
});
test("terminates the host when replacement activation fails after cutover", async () => {
test("keeps the old generation active when replacement activation fails before cutover", async () => {
const failure = new Error("replacement activation failed");
const trace: string[] = [];
let retained: GenerationValue | undefined;
@ -282,13 +352,11 @@ describe("facet loader", () => {
await expect(retained!.read(BACKGROUND_CONTEXT)).resolves.toBe("A");
await expect(host.reload([provider("B", true)])).rejects.toBe(failure);
expect(trace).toEqual(["activate A", "deactivate A", "activate B", "deactivate consumer", "deactivate B"]);
expect(() => retained!.read(BACKGROUND_CONTEXT)).toThrow(
"Facet terminal-consumer service handles cannot be used while dead",
);
await expect(host.reload([])).rejects.toThrow("Facet host cannot reload while dead");
expect(() => host.services.use(RemoteGenerationValue)).toThrow("Remote service provider is disposed");
expect(trace).toEqual(["activate A", "activate B", "deactivate B"]);
await expect(retained!.read(BACKGROUND_CONTEXT)).resolves.toBe("A");
await host.reload([]);
await host.dispose();
expect(trace).toEqual(["activate A", "activate B", "deactivate B", "deactivate consumer", "deactivate A"]);
});
test("creates a reusable static loader", async () => {

7
packages/coding-agent/examples/plugins/pi-example-plugin/src/tui.ts
View file

@ -11,14 +11,17 @@ export default defineFacet({
const ui = env.use(PresentationUI);
env.onActivate(() => {
env.own(
commands.register({
commands.replace({
name: "hello",
description: "Call the bundled Session worker facet",
argumentHint: "<name>",
async run(args, context) {
const message = args.length === 0 ? "from the TUI" : args;
const reply = await example.greet({ name: message }, context);
ui.showStatus(`${reply.message} Worker activations: ${reply.workerActivations}.`, context);
ui.showStatus(
`We got this: ${reply.message} Worker activations: ${reply.workerActivations}.`,
context,
);
return controller.prompt({ message: `pong: ${message}`, images: null }, context);
},
}),

60
packages/coding-agent/src/experimental/services/slash-commands-provider.ts
View file

@ -16,30 +16,28 @@ const THINKING_DESCRIPTIONS: Record<ThinkingLevel, string> = {
max: "Maximum reasoning",
};
interface RegisteredSlashCommand {
readonly command: SlashCommandContribution;
closed: boolean;
}
export class SlashCommandRegistry implements SlashCommands {
readonly #commands = new Map<string, SlashCommandContribution>();
readonly #commands = new Map<string, RegisteredSlashCommand[]>();
readonly #listeners = new Set<(commands: readonly SlashCommandContribution[]) => void>();
register(command: SlashCommandContribution): () => void {
if (!/^[a-z0-9][a-z0-9:-]*$/u.test(command.name)) {
throw new TypeError(`Invalid slash command name: ${command.name}`);
}
this.#validate(command);
if (this.#commands.has(command.name)) throw new Error(`Slash command /${command.name} is already registered`);
const registered = Object.freeze({ ...command });
this.#commands.set(command.name, registered);
this.#publish();
let closed = false;
return () => {
if (closed) return;
closed = true;
if (this.#commands.get(command.name) !== registered) return;
this.#commands.delete(command.name);
this.#publish();
};
return this.#add(command);
}
replace(command: SlashCommandContribution): () => void {
this.#validate(command);
return this.#add(command);
}
list(): readonly SlashCommandContribution[] {
return Object.freeze([...this.#commands.values()]);
return Object.freeze([...this.#commands.values()].map((entries) => entries[0]!.command));
}
subscribe(listener: (commands: readonly SlashCommandContribution[]) => void): () => void {
@ -48,6 +46,28 @@ export class SlashCommandRegistry implements SlashCommands {
return () => this.#listeners.delete(listener);
}
#add(command: SlashCommandContribution): () => void {
const entry: RegisteredSlashCommand = { command: Object.freeze({ ...command }), closed: false };
const entries = this.#commands.get(command.name) ?? [];
entries.push(entry);
this.#commands.set(command.name, entries);
if (entries.length === 1) this.#publish();
return () => {
if (entry.closed) return;
entry.closed = true;
if (entries[0] !== entry) return;
while (entries[0]?.closed) entries.shift();
if (entries.length === 0) this.#commands.delete(command.name);
this.#publish();
};
}
#validate(command: SlashCommandContribution): void {
if (!/^[a-z0-9][a-z0-9:-]*$/u.test(command.name)) {
throw new TypeError(`Invalid slash command name: ${command.name}`);
}
}
#publish(): void {
const commands = this.list();
for (const listener of this.#listeners) listener(commands);
@ -76,11 +96,11 @@ export function createBuiltInSlashCommandsFacet(options: {
const presentationPlugins = env.use(PresentationPlugins);
const sessionPlugins = env.use(SessionPlugins);
env.onActivate(() => {
env.own(commands.register(modelCommand(models, ui)));
env.own(commands.register(thinkingCommand(models, ui)));
env.own(commands.register(compactCommand(controller, ui)));
env.own(commands.replace(modelCommand(models, ui)));
env.own(commands.replace(thinkingCommand(models, ui)));
env.own(commands.replace(compactCommand(controller, ui)));
env.own(
commands.register({
commands.replace({
name: "reload",
description: "Reload server-selected plugins",
async run(_args, context) {

2
packages/coding-agent/src/experimental/services/slash-commands.ts
View file

@ -21,6 +21,8 @@ export interface SlashCommandContribution {
export interface SlashCommands {
register(command: SlashCommandContribution): () => void;
/** Stage a same-name replacement while the previous facet generation retires. */
replace(command: SlashCommandContribution): () => void;
list(): readonly SlashCommandContribution[];
subscribe(listener: (commands: readonly SlashCommandContribution[]) => void): () => void;
}

41
packages/coding-agent/test/experimental-slash-commands.test.ts
View file

@ -21,8 +21,26 @@ describe("experimental slash command facets", () => {
unsubscribe();
});
test("stages replacements until the previous registration retires", () => {
const registry = new SlashCommandRegistry();
const first = { name: "hello", description: "First", run: () => undefined };
const second = { name: "hello", description: "Second", run: () => undefined };
const closeFirst = registry.register(first);
const closeSecond = registry.replace(second);
expect(registry.list()).toEqual([first]);
closeSecond();
expect(registry.list()).toEqual([first]);
const closeReplacement = registry.replace(second);
closeFirst();
expect(registry.list()).toEqual([second]);
closeReplacement();
expect(registry.list()).toEqual([]);
});
test("tracks plugin facet reload and unload", async () => {
const registry = new SlashCommandRegistry();
const originalRun = vi.fn();
const host = await createFacetHost({
facets: [
createSlashCommandsRuntimeFacet(registry),
@ -30,13 +48,32 @@ describe("experimental slash command facets", () => {
id: "@test/example-hello",
setup(env) {
const commands = env.use(SlashCommands);
env.onActivate(() => env.own(commands.register({ name: "hello", run: () => undefined })));
env.onActivate(() =>
env.own(commands.replace({ name: "hello", description: "Original", run: originalRun })),
);
},
}),
],
});
expect(registry.list().map(({ name }) => name)).toEqual(["hello"]);
const failure = new Error("replacement failed");
await expect(
host.reload([
defineFacet({
id: "@test/example-hello",
setup(env) {
const commands = env.use(SlashCommands);
env.onActivate(() => {
env.own(commands.replace({ name: "hello", description: "Failing", run: () => undefined }));
throw failure;
});
},
}),
]),
).rejects.toBe(failure);
expect(registry.list()).toEqual([expect.objectContaining({ name: "hello", run: originalRun })]);
const replacementRun = vi.fn();
await host.reload([
defineFacet({
@ -44,7 +81,7 @@ describe("experimental slash command facets", () => {
setup(env) {
const commands = env.use(SlashCommands);
env.onActivate(() =>
env.own(commands.register({ name: "hello", description: "Replacement", run: replacementRun })),
env.own(commands.replace({ name: "hello", description: "Replacement", run: replacementRun })),
);
},
}),