From c4b0e35abe631bc830190fa6cafbe81b098b97d7 Mon Sep 17 00:00:00 2001 From: Armin Ronacher Date: Tue, 1 Sep 2026 10:47:08 +0200 Subject: [PATCH] feat(chord): keep services available during facet reloads --- packages/chord/PLANNING.md | 31 ++- packages/chord/README.md | 8 +- packages/chord/src/facets/host.ts | 184 +++++++++--------- packages/chord/src/services/calls.ts | 60 ++++++ packages/chord/src/services/consumer.ts | 2 +- packages/chord/src/services/handle.ts | 36 ++-- packages/chord/src/services/provider.ts | 28 ++- packages/chord/src/types.ts | 2 +- packages/chord/test/facet-loader.test.ts | 88 ++++++++- .../plugins/pi-example-plugin/src/tui.ts | 7 +- .../services/slash-commands-provider.ts | 60 ++++-- .../experimental/services/slash-commands.ts | 2 + .../test/experimental-slash-commands.test.ts | 41 +++- 13 files changed, 386 insertions(+), 163 deletions(-) create mode 100644 packages/chord/src/services/calls.ts diff --git a/packages/chord/PLANNING.md b/packages/chord/PLANNING.md index 5933a1b68..73e1cfd13 100644 --- a/packages/chord/PLANNING.md +++ b/packages/chord/PLANNING.md @@ -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 diff --git a/packages/chord/README.md b/packages/chord/README.md index ce0a1aa65..7615d6d92 100644 --- a/packages/chord/README.md +++ b/packages/chord/README.md @@ -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 diff --git a/packages/chord/src/facets/host.ts b/packages/chord/src/facets/host.ts index 9decac82c..75557480a 100644 --- a/packages/chord/src/facets/host.ts +++ b/packages/chord/src/facets/host.ts @@ -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(() => {}) as object, deactivate() {}, }; registration.directory.insert(instance); let closed = false; + let retired: Promise | 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(() => { 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 { + 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 = (key: string, implementation: T) => () => void; +type ServiceInstanceInstaller = (key: string, implementation: T) => Disposal; + +interface StagedServiceInstance { + readonly key: string; + readonly implementation: T; + release?: Disposal; + retirement?: Promise; +} class StagedServiceSpawner implements ServiceSpawner { readonly #lifecycle: FacetLifecycle; + readonly #validate: (key: string, implementation: T) => void; + readonly #instances = new Map>(); #installer: ServiceInstanceInstaller | undefined; - constructor(lifecycle: FacetLifecycle) { + constructor(lifecycle: FacetLifecycle, validate: (key: string, implementation: T) => void) { this.#lifecycle = lifecycle; + this.#validate = validate; } connect(installer: ServiceInstanceInstaller): 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 = { key, implementation }; + this.#instances.set(key, instance); + if (this.#installer !== undefined) instance.release = this.#installer(key, implementation); + const close = (): void | Promise => { + 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; } | { 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 { @@ -537,14 +535,19 @@ export class FacetKernel { install: (provider) => provider.provide(service, implementation as NoInfer>), validateReplacement: (provider) => provider.validateReplacement(service, implementation as NoInfer>), - withdraw: (provider) => provider.withdraw(service), replace: (provider) => provider.replace(service, implementation as NoInfer>), }); }, provideMany: (service: Service): ServiceSpawner => { lifecycle.assertSettingUp("provide service instances"); recordServiceReference(runtime.provides, service, "keyed"); - const instances = new StagedServiceSpawner(lifecycle); + const instances = new StagedServiceSpawner(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 { + async #terminate(): Promise { 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 { + 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, diff --git a/packages/chord/src/services/calls.ts b/packages/chord/src/services/calls.ts new file mode 100644 index 000000000..883bd8dcb --- /dev/null +++ b/packages/chord/src/services/calls.ts @@ -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 | undefined; + #resolve: Resolve | undefined; + + run(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 { + if (this.#retired) return this.#settled ?? Promise.resolve(); + this.#retired = true; + if (this.#active === 0) return Promise.resolve(); + this.#settled = new Promise((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 { + return ( + ((typeof value === "object" && value !== null) || typeof value === "function") && + "then" in value && + typeof value.then === "function" + ); +} diff --git a/packages/chord/src/services/consumer.ts b/packages/chord/src/services/consumer.ts index ab6a9b9f9..7c2388faa 100644 --- a/packages/chord/src/services/consumer.ts +++ b/packages/chord/src/services/consumer.ts @@ -279,7 +279,7 @@ class KeyedBinding { let stopped = false; const stop = this.#instances.observe((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(() => { this.#assertAccess(); diff --git a/packages/chord/src/services/handle.ts b/packages/chord/src/services/handle.ts index 129cc887c..af2b54a5f 100644 --- a/packages/chord/src/services/handle.ts +++ b/packages/chord/src/services/handle.ts @@ -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 { + const previous = this.#target; + this.#target = { implementation, calls: new CallTracker() }; + return previous?.calls.retire() ?? Promise.resolve(); } - unbind(): void { - this.#implementation = undefined; + unbind(): Promise { + 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; diff --git a/packages/chord/src/services/provider.ts b/packages/chord/src/services/provider.ts index 385f1f5b4..3e3c85eb9 100644 --- a/packages/chord/src/services/provider.ts +++ b/packages/chord/src/services/provider.ts @@ -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; 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(service: Service): void { + /** Disconnect one singleton and resolve after calls admitted by the retired target settle. */ + withdraw(service: Service): Promise { 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(service: Service, implementation: NoInfer>): void { + /** Replace one singleton immediately and resolve after calls admitted by the retired target settle. */ + replace(service: Service, implementation: NoInfer>): Promise { 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(service: Service): 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 | 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) { diff --git a/packages/chord/src/types.ts b/packages/chord/src/types.ts index 2c76b296f..5b73078d6 100644 --- a/packages/chord/src/types.ts +++ b/packages/chord/src/types.ts @@ -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; dispose(): Promise; } diff --git a/packages/chord/test/facet-loader.test.ts b/packages/chord/test/facet-loader.test.ts index 9b79f1c9e..1632d4951 100644 --- a/packages/chord/test/facet-loader.test.ts +++ b/packages/chord/test/facet-loader.test.ts @@ -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((resolve) => { + reportOldCallsStarted = resolve; + }); + let finishOldCalls!: () => void; + const oldCallsCanFinish = new Promise((resolve) => { + finishOldCalls = resolve; + }); + let reportReplacementActivated!: () => void; + const replacementActivated = new Promise((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 () => { diff --git a/packages/coding-agent/examples/plugins/pi-example-plugin/src/tui.ts b/packages/coding-agent/examples/plugins/pi-example-plugin/src/tui.ts index 2158827ab..9ba3d2775 100644 --- a/packages/coding-agent/examples/plugins/pi-example-plugin/src/tui.ts +++ b/packages/coding-agent/examples/plugins/pi-example-plugin/src/tui.ts @@ -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: "", 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); }, }), diff --git a/packages/coding-agent/src/experimental/services/slash-commands-provider.ts b/packages/coding-agent/src/experimental/services/slash-commands-provider.ts index e2ae7189f..68ddd1f88 100644 --- a/packages/coding-agent/src/experimental/services/slash-commands-provider.ts +++ b/packages/coding-agent/src/experimental/services/slash-commands-provider.ts @@ -16,30 +16,28 @@ const THINKING_DESCRIPTIONS: Record = { max: "Maximum reasoning", }; +interface RegisteredSlashCommand { + readonly command: SlashCommandContribution; + closed: boolean; +} + export class SlashCommandRegistry implements SlashCommands { - readonly #commands = new Map(); + readonly #commands = new Map(); 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) { diff --git a/packages/coding-agent/src/experimental/services/slash-commands.ts b/packages/coding-agent/src/experimental/services/slash-commands.ts index 4fb80f1cc..62927dbb5 100644 --- a/packages/coding-agent/src/experimental/services/slash-commands.ts +++ b/packages/coding-agent/src/experimental/services/slash-commands.ts @@ -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; } diff --git a/packages/coding-agent/test/experimental-slash-commands.test.ts b/packages/coding-agent/test/experimental-slash-commands.test.ts index 9e353cebe..3fa75fd74 100644 --- a/packages/coding-agent/test/experimental-slash-commands.test.ts +++ b/packages/coding-agent/test/experimental-slash-commands.test.ts @@ -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 })), ); }, }),