From 18224d47ffa9cd65e11597c4bc6e4e2a7239cdcf Mon Sep 17 00:00:00 2001 From: "haozhe.yang" Date: Tue, 9 Jun 2026 20:09:29 +0800 Subject: [PATCH] refactor(services,daemon,agent-core): align service registry bootstrap and expand DI lifecycle primitives - remove defaultServicesModule() and services/src/module.ts; consume getSingletonServiceDescriptors() directly\n- update daemon service registrations and bootstrap to use registry descriptors\n- add DisposableMap, DisposableSet, disposable tracking, and disposeOnReturn to agent-core DI\n- update AGENTS.md with new registration patterns --- .../align-service-registry-bootstrap.md | 6 + packages/agent-core/src/di/index.ts | 13 +- packages/agent-core/src/di/lifecycle.ts | 780 ++++++++++++++---- packages/agent-core/test/di/child.test.ts | 9 +- packages/agent-core/test/di/lifecycle.test.ts | 47 +- .../src/services/approval/approvalService.ts | 109 ++- .../services/fileStore/fileStoreService.ts | 2 +- .../src/services/fs/fsWatcherService.ts | 95 ++- .../gateway/connectionRegistryService.ts | 2 +- .../services/gateway/restGatewayService.ts | 2 +- .../services/gateway/sessionClientsService.ts | 2 +- .../services/gateway/wsBroadcastService.ts | 4 +- .../src/services/gateway/wsGatewayService.ts | 2 +- .../src/services/question/questionService.ts | 116 ++- .../daemon/src/services/serviceCollection.ts | 102 +-- .../services/workspace/workspaceFsService.ts | 2 +- .../workspace/workspaceRegistryService.ts | 2 +- packages/services/AGENTS.md | 29 +- .../src/authSummary/authSummaryService.ts | 2 +- .../src/coreProcess/coreProcessService.ts | 4 +- packages/services/src/event/eventService.ts | 2 +- packages/services/src/index.ts | 65 -- packages/services/src/module.ts | 66 -- packages/services/src/oauth/oauthService.ts | 94 ++- packages/services/src/prompt/promptService.ts | 2 +- packages/services/src/session/session.ts | 143 ---- .../services/src/session/sessionService.ts | 94 +-- .../services/test/coreProcessService.test.ts | 91 +- packages/services/test/interfaces.test.ts | 130 +-- .../services/test/session-service.test.ts | 82 +- packages/services/vitest.config.ts | 9 +- 31 files changed, 1059 insertions(+), 1049 deletions(-) create mode 100644 .changeset/align-service-registry-bootstrap.md delete mode 100644 packages/services/src/module.ts diff --git a/.changeset/align-service-registry-bootstrap.md b/.changeset/align-service-registry-bootstrap.md new file mode 100644 index 000000000..59cc13714 --- /dev/null +++ b/.changeset/align-service-registry-bootstrap.md @@ -0,0 +1,6 @@ +--- +"@moonshot-ai/services": patch +"@moonshot-ai/daemon": patch +--- + +Use singleton service descriptors directly for daemon service bootstrap. diff --git a/packages/agent-core/src/di/index.ts b/packages/agent-core/src/di/index.ts index 020493234..bb3e544d3 100644 --- a/packages/agent-core/src/di/index.ts +++ b/packages/agent-core/src/di/index.ts @@ -29,11 +29,22 @@ export { InstantiationService } from './instantiationService'; export { Disposable, DisposableStore, + DisposableMap, + DisposableSet, MutableDisposable, + DisposableTracker, combinedDisposable, toDisposable, + dispose, + disposeIfDisposable, + disposeOnReturn, + isDisposable, + markAsSingleton, + setDisposableTracker, + trackDisposable, + markAsDisposed, } from './lifecycle'; -export type { IDisposable } from './lifecycle'; +export type { IDisposable, IDisposableTracker } from './lifecycle'; export { CyclicDependencyError } from './errors'; export { InstantiationType, diff --git a/packages/agent-core/src/di/lifecycle.ts b/packages/agent-core/src/di/lifecycle.ts index c67afe0ad..07a764fd1 100644 --- a/packages/agent-core/src/di/lifecycle.ts +++ b/packages/agent-core/src/di/lifecycle.ts @@ -1,68 +1,435 @@ /** * Lifecycle primitives for DI-managed services: `IDisposable` interface, a * `Disposable` base class, plus the helper primitives (`DisposableStore`, - * `MutableDisposable`, `toDisposable`, `combinedDisposable`, `Disposable.None`) - * that `Event` / `Emitter` (in `base/common/event.ts`) build on top of. + * `MutableDisposable`, `DisposableMap`, `DisposableSet`, `toDisposable`, + * `combinedDisposable`, `dispose`, `disposeOnReturn`, `Disposable.None`) that + * `Event` / `Emitter` (in `base/common/event.ts`) build on top of. * - * Modelled after VSCode's `base/common/lifecycle.ts`. Children disposed by - * `Disposable` / `DisposableStore` are torn down in reverse register order - * (LIFO) and each `dispose()` is wrapped in try/catch so one failing child - * does not skip its siblings. + * Modelled after VSCode's `base/common/lifecycle.ts`. Two intentional + * deviations from upstream: + * + * 1. **Error policy** — VSCode's iterable `dispose(...)` collects per-child + * errors and throws an `AggregateError` at the end. Here every dispose + * path routes failures through `onUnexpectedError` and continues, so a + * single misbehaving child cannot abort sibling teardown. Same policy + * `Emitter.fire()` uses; keep it consistent. + * 2. **Disposable tracker** — `IDisposableTracker` interface and the + * plumbing (`trackDisposable` / `markAsDisposed` / `setParentOfDisposable`) + * are present, but `disposableTracker` defaults to `null` so there is + * zero overhead. Install a `DisposableTracker` in test setup (or a custom + * one in a debug build) when leak hunting. We do not ship the + * `GCBasedDisposableTracker` variant — `FinalizationRegistry` timing is + * non-deterministic enough to produce noisy reports. + * + * Sibling teardown order is **insertion order** (`Set` iteration). Subclasses + * that need a specific order must sequence teardown explicitly inside + * `override dispose()` before calling `super.dispose()`. */ import { onUnexpectedError } from '../errors/unexpectedError'; +// #region Disposable Tracking + +/** + * Hook surface for tracking living disposables, parentage, and lifecycle + * transitions. Install via `setDisposableTracker`. Defaults to `null` + * (zero overhead). + */ +export interface IDisposableTracker { + /** Called on construction of every disposable. */ + trackDisposable(disposable: IDisposable): void; + /** + * Called when a disposable is registered as child of another. If `parent` + * is `null`, the disposable was detached from its former parent (e.g. by + * `DisposableStore.deleteAndLeak`). + */ + setParent(child: IDisposable, parent: IDisposable | null): void; + /** Called after a disposable's `dispose()` runs. */ + markAsDisposed(disposable: IDisposable): void; + /** + * Mark a disposable as a singleton (lives for the lifetime of the process) + * so it isn't reported as a leak. + */ + markAsSingleton(disposable: IDisposable): void; +} + +interface DisposableInfo { + value: IDisposable; + source: string | null; + parent: IDisposable | null; + isSingleton: boolean; + idx: number; +} + +/** + * Default tracker for test / dev use. Records a constructor stack on each + * `trackDisposable` call and removes the entry on `markAsDisposed`. After a + * suspected leak window (e.g. an `afterAll` hook), call + * `getTrackedDisposables()` to inspect what's still alive. + * + * Roots whose ancestor was marked as a singleton are filtered out (they + * intentionally live for the process). + */ +export class DisposableTracker implements IDisposableTracker { + private static idx = 0; + private readonly livingDisposables = new Map(); + + private getDisposableData(d: IDisposable): DisposableInfo { + let val = this.livingDisposables.get(d); + if (!val) { + val = { + parent: null, + source: null, + isSingleton: false, + value: d, + idx: DisposableTracker.idx++, + }; + this.livingDisposables.set(d, val); + } + return val; + } + + trackDisposable(d: IDisposable): void { + const data = this.getDisposableData(d); + if (!data.source) { + data.source = new Error().stack ?? null; + } + } + + setParent(child: IDisposable, parent: IDisposable | null): void { + this.getDisposableData(child).parent = parent; + } + + markAsDisposed(x: IDisposable): void { + this.livingDisposables.delete(x); + } + + markAsSingleton(d: IDisposable): void { + this.getDisposableData(d).isSingleton = true; + } + + private getRootParent( + data: DisposableInfo, + cache: Map, + ): DisposableInfo { + const cached = cache.get(data); + if (cached) return cached; + const result = data.parent + ? this.getRootParent(this.getDisposableData(data.parent), cache) + : data; + cache.set(data, result); + return result; + } + + /** + * All currently-living disposables whose root ancestor is NOT a singleton. + * Use as the post-condition assertion in test teardown. + */ + getTrackedDisposables(): IDisposable[] { + const cache = new Map(); + return [...this.livingDisposables.entries()] + .filter( + ([, v]) => v.source !== null && !this.getRootParent(v, cache).isSingleton, + ) + .map(([k]) => k); + } +} + +let disposableTracker: IDisposableTracker | null = null; + +export function setDisposableTracker(tracker: IDisposableTracker | null): void { + disposableTracker = tracker; +} + +export function trackDisposable(x: T): T { + disposableTracker?.trackDisposable(x); + return x; +} + +export function markAsDisposed(disposable: IDisposable): void { + disposableTracker?.markAsDisposed(disposable); +} + +function setParentOfDisposable( + child: IDisposable, + parent: IDisposable | null, +): void { + disposableTracker?.setParent(child, parent); +} + +function setParentOfDisposables( + children: IDisposable[], + parent: IDisposable | null, +): void { + if (!disposableTracker) return; + for (const child of children) { + disposableTracker.setParent(child, parent); + } +} + +/** + * Indicates that the given object lives for the lifetime of the process so + * the tracker should not report it as a leak. + */ +export function markAsSingleton(singleton: T): T { + disposableTracker?.markAsSingleton(singleton); + return singleton; +} + +// #endregion + export interface IDisposable { dispose(): void; } /** - * Base class for services that own other disposables. Subclasses call - * `this._register(child)` to take ownership; `dispose()` tears children down - * in reverse register order (LIFO) and is idempotent. + * Type guard for heterogeneous collections. Matches VSCode `isDisposable`: + * accepts any object with a zero-arg `dispose()` method. */ -export abstract class Disposable implements IDisposable { - private _disposed = false; - protected _toDispose: IDisposable[] = []; +export function isDisposable(thing: E): thing is E & IDisposable { + return ( + typeof thing === 'object' && + thing !== null && + typeof (thing as unknown as IDisposable).dispose === 'function' && + (thing as unknown as IDisposable).dispose.length === 0 + ); +} - /** - * Take ownership of a child disposable. Returns the child for ergonomic - * one-liner chaining (`const x = this._register(new Foo())`). - */ - protected _register(d: T): T { - if (this._disposed) { - // Don't silently hold a reference after disposal; tear down immediately - // so we don't leak the child if someone calls `_register` post-dispose. +/** + * Dispose one or many `IDisposable`s. Per-child errors are routed through + * `onUnexpectedError` and do not abort the loop (kimi-code policy — VSCode + * collects errors into an `AggregateError` and throws; we do not, to keep + * a single misbehaving child from breaking sibling teardown). + * + * Overloads mirror VSCode for ergonomic call sites. + */ +export function dispose(disposable: T): T; +export function dispose( + disposable: T | undefined, +): T | undefined; +export function dispose = Iterable>( + disposables: A, +): A; +export function dispose(disposables: Array): Array; +export function dispose( + disposables: ReadonlyArray, +): ReadonlyArray; +export function dispose( + arg: T | Iterable | undefined, +): unknown { + if (arg === undefined || arg === null) return arg; + if (isIterable(arg)) { + for (const d of arg) { + if (d) { + try { + d.dispose(); + } catch (err) { + onUnexpectedError(err); + } + } + } + return Array.isArray(arg) ? [] : arg; + } + try { + (arg as T).dispose(); + } catch (err) { + onUnexpectedError(err); + } + return arg; +} + +function isIterable(arg: unknown): arg is Iterable { + return ( + typeof arg === 'object' && + arg !== null && + typeof (arg as { [Symbol.iterator]?: unknown })[Symbol.iterator] === 'function' + ); +} + +/** + * Dispose only the entries in `disposables` that pass the `isDisposable` + * type guard. Mirrors VSCode helper of the same name; useful when holding + * mixed collections (e.g. legacy code that may or may not implement the + * interface). + */ +export function disposeIfDisposable( + disposables: Array, +): Array { + for (const d of disposables) { + if (isDisposable(d)) { try { d.dispose(); - } catch { - // Swallow: dispose() must be idempotent / forgiving. + } catch (err) { + onUnexpectedError(err); } - return d; } - this._toDispose.push(d); - return d; + } + return []; +} + +/** + * Wrap a function as an `IDisposable`. The returned object's `dispose()` + * invokes `fn` at most once — repeated calls are a no-op (idempotent). + * + * Implemented as a class so the returned object has a stable shape for + * debuggers / V8 hidden-class optimisation, and so the tracker can record + * a construction stack. + */ +class FunctionDisposable implements IDisposable { + private _isDisposed = false; + private readonly _fn: () => void; + + constructor(fn: () => void) { + this._fn = fn; + trackDisposable(this); } dispose(): void { - if (this._disposed) { - return; + if (this._isDisposed) return; + this._isDisposed = true; + markAsDisposed(this); + try { + this._fn(); + } catch (err) { + onUnexpectedError(err); } - this._disposed = true; - // Reverse order: most-recently-registered tears down first (LIFO). - while (this._toDispose.length > 0) { - const child = this._toDispose.pop(); - if (!child) continue; + } +} + +export function toDisposable(fn: () => void): IDisposable { + return new FunctionDisposable(fn); +} + +/** + * Aggregate multiple disposables into a single `IDisposable`. Children are + * disposed in insertion order via the iterable `dispose(...)` helper, so + * one throwing child does not skip its siblings. + */ +export function combinedDisposable(...disposables: IDisposable[]): IDisposable { + const parent = toDisposable(() => dispose(disposables)); + setParentOfDisposables(disposables, parent); + return parent; +} + +/** + * Container that owns multiple `IDisposable`s. Iteration / disposal order is + * **insertion order** (`Set` semantics). Mirrors VSCode + * `base/common/lifecycle.ts DisposableStore`. + */ +export class DisposableStore implements IDisposable { + private readonly _toDispose = new Set(); + private _isDisposed = false; + + constructor() { + trackDisposable(this); + } + + /** + * Take ownership of `d`. Returns `d` for ergonomic chaining + * (`const x = store.add(new Foo())`). After the store has been disposed, + * `add` disposes the incoming child immediately and still returns it. + * Adding the store to itself throws. + */ + add(d: T): T { + if ((d as unknown as DisposableStore) === this) { + throw new Error('Cannot register a disposable on itself!'); + } + setParentOfDisposable(d, this); + if (this._isDisposed) { try { - child.dispose(); - } catch { - // Continue tearing down siblings even if one throws. + d.dispose(); + } catch (err) { + onUnexpectedError(err); } + return d; + } + this._toDispose.add(d); + return d; + } + + /** + * Remove `d` from the store AND dispose it. Matches VSCode + * `DisposableStore.delete`. Use `deleteAndLeak` to detach without + * disposing. + */ + delete(d: T): void { + if (this._isDisposed) return; + if ((d as unknown as DisposableStore) === this) { + throw new Error('Cannot dispose a disposable on itself!'); + } + this._toDispose.delete(d); + try { + d.dispose(); + } catch (err) { + onUnexpectedError(err); } } - protected get _isDisposed(): boolean { - return this._disposed; + /** + * Remove `d` from the store WITHOUT disposing. Caller takes ownership of + * `d`'s lifetime. Matches VSCode `DisposableStore.deleteAndLeak`. + */ + deleteAndLeak(d: T): void { + if (this._isDisposed) return; + if (this._toDispose.delete(d)) { + setParentOfDisposable(d, null); + } + } + + /** + * Dispose every currently-held child but keep the store usable. + */ + clear(): void { + if (this._isDisposed) return; + if (this._toDispose.size === 0) return; + const items = Array.from(this._toDispose); + this._toDispose.clear(); + dispose(items); + } + + /** + * Dispose every currently-held child and mark the store as disposed. + * Idempotent. + */ + dispose(): void { + if (this._isDisposed) return; + this._isDisposed = true; + markAsDisposed(this); + const items = Array.from(this._toDispose); + this._toDispose.clear(); + dispose(items); + } + + get isDisposed(): boolean { + return this._isDisposed; + } +} + +/** + * Base class for services that own other disposables. Subclasses call + * `this._register(child)` to take ownership; `dispose()` tears children down + * in **insertion order** (matching VSCode) and is idempotent. + * + * Subclasses inspect "have I been disposed yet?" via `this._store.isDisposed`. + */ +export abstract class Disposable implements IDisposable { + protected readonly _store = new DisposableStore(); + + constructor() { + trackDisposable(this); + setParentOfDisposable(this._store, this); + } + + protected _register(d: T): T { + if ((d as unknown as Disposable) === this) { + throw new Error('Cannot register a disposable on itself!'); + } + return this._store.add(d); + } + + dispose(): void { + markAsDisposed(this); + this._store.dispose(); } } @@ -70,6 +437,10 @@ export abstract class Disposable implements IDisposable { * Static zero-value disposable. `Disposable.None.dispose()` is a no-op and * is safe to call repeatedly. The object is frozen so callers can't mutate * the shared instance. Modelled after VSCode `base/common/lifecycle.ts`. + * + * Declared as a namespace merger rather than a static class property so we + * don't pull `DisposableStore` allocation into module load just to read + * `Disposable.None`. */ // eslint-disable-next-line @typescript-eslint/no-namespace export namespace Disposable { @@ -80,52 +451,11 @@ export namespace Disposable { }); } -/** - * Wrap a function as an `IDisposable`. The returned object's `dispose()` - * invokes `fn` at most once — repeated calls are a no-op (idempotent). - */ -export function toDisposable(fn: () => void): IDisposable { - let called = false; - return { - dispose(): void { - if (called) return; - called = true; - fn(); - }, - }; -} - -/** - * Aggregate multiple disposables into a single `IDisposable`. The returned - * object's `dispose()` invokes each child's `dispose()` — each call wrapped - * in try/catch so one throwing child does not skip its siblings (mirrors - * `Disposable.dispose()` semantics above). Idempotent: a second `dispose()` - * is a no-op. - */ -export function combinedDisposable(...disposables: IDisposable[]): IDisposable { - let disposed = false; - return { - dispose(): void { - if (disposed) return; - disposed = true; - for (const child of disposables) { - try { - child.dispose(); - } catch (err) { - // Route to onUnexpectedError so the failure is visible, but keep - // iterating siblings (consistent with Disposable.dispose()). - onUnexpectedError(err); - } - } - }, - }; -} - /** * Mutable slot that owns a single `IDisposable`. Assigning a new value * disposes the previous one; assigning `undefined` disposes the current * value. After this store has itself been disposed any subsequent value - * is disposed immediately on assignment (mirrors `Disposable._register`). + * is disposed immediately on assignment. * * Mirrors VSCode `base/common/lifecycle.ts MutableDisposable`. */ @@ -133,14 +463,16 @@ export class MutableDisposable implements IDisposable { private _value: T | undefined; private _isDisposed = false; + constructor() { + trackDisposable(this); + } + get value(): T | undefined { return this._isDisposed ? undefined : this._value; } set value(value: T | undefined) { if (this._isDisposed) { - // Once disposed, the slot can no longer hold a reference — tear - // down the new value immediately and keep the slot empty. if (value !== undefined) { try { value.dispose(); @@ -150,11 +482,10 @@ export class MutableDisposable implements IDisposable { } return; } - if (this._value === value) { - return; - } + if (this._value === value) return; const prev = this._value; this._value = value; + if (value) setParentOfDisposable(value, this); if (prev !== undefined) { try { prev.dispose(); @@ -164,14 +495,10 @@ export class MutableDisposable implements IDisposable { } } - /** - * Dispose the held value (if any) and mark the slot as disposed. - * Subsequent `value = ...` assignments dispose the incoming value - * immediately. Idempotent. - */ dispose(): void { if (this._isDisposed) return; this._isDisposed = true; + markAsDisposed(this); const prev = this._value; this._value = undefined; if (prev !== undefined) { @@ -199,87 +526,240 @@ export class MutableDisposable implements IDisposable { } } } + + /** + * Clear the slot WITHOUT disposing the current value; returns the old + * value. Caller takes ownership of its lifetime. + */ + clearAndLeak(): T | undefined { + if (this._isDisposed) return undefined; + const prev = this._value; + this._value = undefined; + if (prev !== undefined) setParentOfDisposable(prev, null); + return prev; + } } /** - * Container that owns multiple `IDisposable`s. `add` returns the child for - * chaining; `delete` removes (without disposing); `clear` disposes every - * currently-held child; `dispose` disposes every child and marks the store - * as disposed. Once disposed, subsequent `add(child)` calls dispose the - * incoming child immediately (mirrors `Disposable._register`). + * Map whose values are `IDisposable`. Overwriting a key disposes the previous + * value; `deleteAndDispose(key)` removes and disposes; `dispose()` disposes + * every value and marks the map as disposed. Mirrors VSCode + * `base/common/lifecycle.ts DisposableMap`. * - * Mirrors VSCode `base/common/lifecycle.ts DisposableStore`. + * Use this to collapse the "Map of per-entity state + manual teardown loop in + * `override dispose()`" pattern that recurs across daemon services. */ -export class DisposableStore implements IDisposable { - private _toDispose = new Set(); +export class DisposableMap + implements IDisposable +{ + private readonly _store: Map; private _isDisposed = false; - /** - * Take ownership of `d`. Returns `d` for ergonomic chaining - * (`const x = store.add(new Foo())`). After the store has been disposed, - * `add` disposes the incoming child immediately and still returns it. - */ - add(d: T): T { - if (this._isDisposed) { - try { - d.dispose(); - } catch (err) { - onUnexpectedError(err); - } - return d; - } - this._toDispose.add(d); - return d; + constructor(store: Map = new Map()) { + this._store = store; + trackDisposable(this); } /** - * Remove `d` from the store WITHOUT disposing it. No-op if `d` is not - * currently tracked. - */ - delete(d: T): void { - if (this._isDisposed) return; - this._toDispose.delete(d); - } - - /** - * Dispose every currently-held child but keep the store usable - * (subsequent `add` calls still work). - */ - clear(): void { - if (this._isDisposed) return; - if (this._toDispose.size === 0) return; - const items = Array.from(this._toDispose); - this._toDispose.clear(); - // Iterate in reverse so most-recently-added tears down first (LIFO), - // matching `Disposable.dispose()` semantics. - for (let i = items.length - 1; i >= 0; i--) { - try { - items[i]!.dispose(); - } catch (err) { - onUnexpectedError(err); - } - } - } - - /** - * Dispose every currently-held child and mark the store as disposed. - * Idempotent. + * Dispose every stored value and mark this object as disposed. Subsequent + * mutation (`set`) is a no-op + warning. */ dispose(): void { if (this._isDisposed) return; this._isDisposed = true; - const items = Array.from(this._toDispose); - this._toDispose.clear(); - for (let i = items.length - 1; i >= 0; i--) { + markAsDisposed(this); + this.clearAndDisposeAll(); + } + + /** + * Dispose every stored value and clear the map, but DO NOT mark the map + * itself as disposed (subsequent `set` calls still work). + */ + clearAndDisposeAll(): void { + if (this._store.size === 0) return; + try { + dispose(this._store.values()); + } finally { + this._store.clear(); + } + } + + has(key: K): boolean { + return this._store.has(key); + } + + get size(): number { + return this._store.size; + } + + get(key: K): V | undefined { + return this._store.get(key); + } + + /** + * Insert `value` at `key`. If `key` already has a value, that previous + * value is disposed unless `skipDisposeOnOverwrite` is set. + */ + set(key: K, value: V, skipDisposeOnOverwrite = false): void { + if (this._isDisposed) { + // eslint-disable-next-line no-console + console.warn( + new Error( + 'Trying to add a disposable to a DisposableMap that has already been disposed of. The added object will be leaked!', + ).stack, + ); + return; + } + if (!skipDisposeOnOverwrite) { + const prev = this._store.get(key); + if (prev !== undefined && prev !== value) { + try { + prev.dispose(); + } catch (err) { + onUnexpectedError(err); + } + } + } + this._store.set(key, value); + setParentOfDisposable(value, this); + } + + /** + * Remove the value stored for `key` AND dispose it. + */ + deleteAndDispose(key: K): void { + const value = this._store.get(key); + if (value !== undefined) { try { - items[i]!.dispose(); + value.dispose(); + } catch (err) { + onUnexpectedError(err); + } + } + this._store.delete(key); + } + + /** + * Remove the value stored for `key` and return it. Caller takes + * ownership of the lifetime. + */ + deleteAndLeak(key: K): V | undefined { + const value = this._store.get(key); + if (value !== undefined) setParentOfDisposable(value, null); + this._store.delete(key); + return value; + } + + keys(): IterableIterator { + return this._store.keys(); + } + + values(): IterableIterator { + return this._store.values(); + } + + [Symbol.iterator](): IterableIterator<[K, V]> { + return this._store[Symbol.iterator](); + } +} + +/** + * Set whose values are `IDisposable`. `add(v)` takes ownership; + * `deleteAndDispose(v)` removes and disposes; `dispose()` disposes every + * value. Mirrors VSCode `base/common/lifecycle.ts DisposableSet`. + */ +export class DisposableSet + implements IDisposable +{ + private readonly _store: Set; + private _isDisposed = false; + + constructor(store: Set = new Set()) { + this._store = store; + trackDisposable(this); + } + + dispose(): void { + if (this._isDisposed) return; + this._isDisposed = true; + markAsDisposed(this); + this.clearAndDisposeAll(); + } + + clearAndDisposeAll(): void { + if (this._store.size === 0) return; + try { + dispose(this._store.values()); + } finally { + this._store.clear(); + } + } + + has(value: V): boolean { + return this._store.has(value); + } + + get size(): number { + return this._store.size; + } + + add(value: V): void { + if (this._isDisposed) { + // eslint-disable-next-line no-console + console.warn( + new Error( + 'Trying to add a disposable to a DisposableSet that has already been disposed of. The added object will be leaked!', + ).stack, + ); + return; + } + this._store.add(value); + setParentOfDisposable(value, this); + } + + deleteAndDispose(value: V): void { + if (this._store.delete(value)) { + try { + value.dispose(); } catch (err) { onUnexpectedError(err); } } } - get isDisposed(): boolean { - return this._isDisposed; + deleteAndLeak(value: V): V | undefined { + if (this._store.delete(value)) { + setParentOfDisposable(value, null); + return value; + } + return undefined; + } + + values(): IterableIterator { + return this._store.values(); + } + + [Symbol.iterator](): IterableIterator { + return this._store[Symbol.iterator](); + } +} + +/** + * Scoped `using` helper: construct a `DisposableStore`, run `fn(store)`, then + * tear the store down in a `finally`. Lets callers register transient + * disposables for the duration of a function without writing + * `try { ... } finally { store.dispose(); }` by hand. + * + * disposeOnReturn(store => { + * const a = store.add(new Foo()); + * doStuff(a); + * }); + */ +export function disposeOnReturn(fn: (store: DisposableStore) => void): void { + const store = new DisposableStore(); + try { + fn(store); + } finally { + store.dispose(); } } diff --git a/packages/agent-core/test/di/child.test.ts b/packages/agent-core/test/di/child.test.ts index f421f201b..93b987419 100644 --- a/packages/agent-core/test/di/child.test.ts +++ b/packages/agent-core/test/di/child.test.ts @@ -292,7 +292,7 @@ describe('InstantiationService.createChild', () => { }); describe('Disposable base class', () => { - it('reverse register order on dispose', () => { + it('insertion order on dispose', () => { const events: string[] = []; class Child implements IDisposable { constructor(public readonly label: string) {} @@ -310,7 +310,7 @@ describe('Disposable base class', () => { } const o = new Owner(); o.dispose(); - expect(events).toEqual(['disposed third', 'disposed second', 'disposed first']); + expect(events).toEqual(['disposed first', 'disposed second', 'disposed third']); }); it('idempotent dispose on the base class', () => { @@ -372,7 +372,8 @@ describe('Disposable base class', () => { } const o = new Owner(); expect(() => o.dispose()).not.toThrow(); - // BadChild is registered last so it tears down first (LIFO). - expect(events).toEqual(['bad-attempted', 'good']); + // GoodChild is registered first, so it tears down first (insertion order); + // BadChild throws but the loop continues. + expect(events).toEqual(['good', 'bad-attempted']); }); }); diff --git a/packages/agent-core/test/di/lifecycle.test.ts b/packages/agent-core/test/di/lifecycle.test.ts index 05fd039fd..be97b3641 100644 --- a/packages/agent-core/test/di/lifecycle.test.ts +++ b/packages/agent-core/test/di/lifecycle.test.ts @@ -41,7 +41,7 @@ describe('toDisposable', () => { }); describe('combinedDisposable', () => { - it('disposes all children', () => { + it('disposes all children in insertion order', () => { const order: string[] = []; const d = combinedDisposable( makeRecorder('a', order), @@ -142,14 +142,14 @@ describe('DisposableStore', () => { store.dispose(); }); - it('dispose tears down children in LIFO order', () => { + it('dispose tears down children in insertion order', () => { const order: string[] = []; const store = new DisposableStore(); store.add(makeRecorder('a', order)); store.add(makeRecorder('b', order)); store.add(makeRecorder('c', order)); store.dispose(); - expect(order).toEqual(['c', 'b', 'a']); + expect(order).toEqual(['a', 'b', 'c']); }); it('clear disposes children but keeps the store usable', () => { @@ -158,18 +158,30 @@ describe('DisposableStore', () => { store.add(makeRecorder('a', order)); store.add(makeRecorder('b', order)); store.clear(); - expect(order).toEqual(['b', 'a']); + expect(order).toEqual(['a', 'b']); store.add(makeRecorder('c', order)); store.dispose(); - expect(order).toEqual(['b', 'a', 'c']); + expect(order).toEqual(['a', 'b', 'c']); }); - it('delete removes a child without disposing it', () => { + it('delete removes a child AND disposes it', () => { const order: string[] = []; const store = new DisposableStore(); const rec = makeRecorder('a', order); store.add(rec); store.delete(rec); + expect(order).toEqual(['a']); + // No second dispose when the store itself is later torn down. + store.dispose(); + expect(order).toEqual(['a']); + }); + + it('deleteAndLeak removes a child WITHOUT disposing it', () => { + const order: string[] = []; + const store = new DisposableStore(); + const rec = makeRecorder('a', order); + store.add(rec); + store.deleteAndLeak(rec); store.dispose(); expect(order).toEqual([]); }); @@ -205,16 +217,16 @@ describe('DisposableStore', () => { }); store.add(makeRecorder('c', order)); store.dispose(); - // LIFO: c, throwing-child, a - expect(order).toEqual(['c', 'a']); + // Insertion order: a, throwing-child, c + expect(order).toEqual(['a', 'c']); expect(captured).toHaveLength(1); expect((captured[0] as Error).message).toBe('store-child-boom'); resetUnexpectedErrorHandler(); }); }); -describe('Disposable base class (unchanged behaviour)', () => { - it('LIFO teardown still holds after the lifecycle expansion', () => { +describe('Disposable base class', () => { + it('insertion-order teardown', () => { const order: string[] = []; class Owner extends Disposable { add(label: string): void { @@ -226,6 +238,19 @@ describe('Disposable base class (unchanged behaviour)', () => { owner.add('b'); owner.add('c'); owner.dispose(); - expect(order).toEqual(['c', 'b', 'a']); + expect(order).toEqual(['a', 'b', 'c']); + }); + + it('registering self throws', () => { + class Owner extends Disposable { + registerSelf(): void { + this._register(this); + } + } + const owner = new Owner(); + expect(() => owner.registerSelf()).toThrow( + /Cannot register a disposable on itself/, + ); + owner.dispose(); }); }); diff --git a/packages/daemon/src/services/approval/approvalService.ts b/packages/daemon/src/services/approval/approvalService.ts index a7dcb2dbf..341a1bc67 100644 --- a/packages/daemon/src/services/approval/approvalService.ts +++ b/packages/daemon/src/services/approval/approvalService.ts @@ -48,7 +48,7 @@ import { ulid } from 'ulid'; -import { Disposable } from '@moonshot-ai/agent-core'; +import { Disposable, DisposableMap, type IDisposable } from '@moonshot-ai/agent-core'; import type { ApprovalRequest as ProtocolApprovalRequest, Event, @@ -86,23 +86,64 @@ export class ApprovalExpiredError extends Error { } } -interface PendingApproval { - readonly approvalId: string; - readonly sessionId: string; - readonly toolCallId: string; - readonly createdAt: string; - readonly expiresAt: string; - readonly protocolRequest: ProtocolApprovalRequest; - resolve: (r: ApprovalResponse) => void; - reject: (e: Error) => void; - timer: NodeJS.Timeout; +/** + * One pending approval. Owns the 60s timer + promise callbacks. Two ways out: + * - `markSettled()` — the happy path called by resolve/expire AFTER they've + * decided the outcome; just stops the timer. + * - `dispose()` — the shutdown fallback. If the entry hasn't been settled + * yet, clears the timer AND rejects with "daemon shutting down" so the + * awaiter doesn't dangle. + * + * Stored in `ApprovalService._pending` (`DisposableMap`) so service-wide + * teardown via `super.dispose()` triggers each entry's `dispose()` + * automatically. + */ +class PendingApproval implements IDisposable { + private _settled = false; + + constructor( + readonly approvalId: string, + readonly sessionId: string, + readonly toolCallId: string, + readonly createdAt: string, + readonly expiresAt: string, + readonly protocolRequest: ProtocolApprovalRequest, + private readonly _resolveFn: (r: ApprovalResponse) => void, + private readonly _rejectFn: (e: Error) => void, + private readonly _timer: NodeJS.Timeout, + ) {} + + markSettled(): void { + if (this._settled) return; + this._settled = true; + clearTimeout(this._timer); + } + + resolve(r: ApprovalResponse): void { + this._resolveFn(r); + } + + reject(e: Error): void { + this._rejectFn(e); + } + + dispose(): void { + if (this._settled) return; + this._settled = true; + clearTimeout(this._timer); + try { + this._rejectFn(new Error('daemon shutting down')); + } catch { + // awaiter may not have a catch handler attached yet + } + } } export class ApprovalService extends Disposable implements IApprovalService { readonly _serviceBrand: undefined; /** Indexed by daemon-minted `approval_id` (REST path key). */ - private readonly _pending = new Map(); + private readonly _pending: DisposableMap; /** Reverse lookup for `toolCallId` (legacy in-process interface compatibility). */ private readonly _byToolCallId = new Map(); /** @@ -119,12 +160,13 @@ export class ApprovalService extends Disposable implements IApprovalService { @IEventService private readonly eventService: IEventService, ) { super(); + this._pending = this._register(new DisposableMap()); } async request( req: ApprovalRequest & { sessionId: string; agentId: string }, ): Promise { - if (this._isDisposed) { + if (this._store.isDisposed) { throw new Error('approval service disposed'); } @@ -169,17 +211,20 @@ export class ApprovalService extends Disposable implements IApprovalService { return await new Promise((resolve, reject) => { const timer = setTimeout(() => this._expire(approvalId), this._timeoutMs); timer.unref?.(); - this._pending.set(approvalId, { + this._pending.set( approvalId, - sessionId: req.sessionId, - toolCallId: req.toolCallId, - createdAt, - expiresAt, - protocolRequest, - resolve, - reject, - timer, - }); + new PendingApproval( + approvalId, + req.sessionId, + req.toolCallId, + createdAt, + expiresAt, + protocolRequest, + resolve, + reject, + timer, + ), + ); this._byToolCallId.set(req.toolCallId, approvalId); }); } @@ -194,8 +239,8 @@ export class ApprovalService extends Disposable implements IApprovalService { resolve(id: string, response: ApprovalResponse): void { const p = this._pending.get(id); if (!p) return; - clearTimeout(p.timer); - this._pending.delete(id); + p.markSettled(); + this._pending.deleteAndLeak(id); this._byToolCallId.delete(p.toolCallId); this.markResolved(p.approvalId); @@ -275,7 +320,8 @@ export class ApprovalService extends Disposable implements IApprovalService { private _expire(approvalId: string): void { const p = this._pending.get(approvalId); if (!p) return; - this._pending.delete(approvalId); + p.markSettled(); + this._pending.deleteAndLeak(approvalId); this._byToolCallId.delete(p.toolCallId); // Mark as resolved-style for idempotency — a late REST resolve on this id // gets 40902 rather than 40404 (matches "expired ≈ already_resolved" UX). @@ -293,16 +339,7 @@ export class ApprovalService extends Disposable implements IApprovalService { } override dispose(): void { - if (this._isDisposed) return; - for (const [, p] of this._pending) { - clearTimeout(p.timer); - try { - p.reject(new Error('daemon shutting down')); - } catch { - // ignore — the awaiter may not have a catch handler attached yet. - } - } - this._pending.clear(); + if (this._store.isDisposed) return; this._byToolCallId.clear(); this._recentlyResolved.clear(); super.dispose(); diff --git a/packages/daemon/src/services/fileStore/fileStoreService.ts b/packages/daemon/src/services/fileStore/fileStoreService.ts index d9ec6cbf9..095e5e22e 100644 --- a/packages/daemon/src/services/fileStore/fileStoreService.ts +++ b/packages/daemon/src/services/fileStore/fileStoreService.ts @@ -222,7 +222,7 @@ export class FileStore extends Disposable implements IFileStore { } override dispose(): void { - if (this._isDisposed) return; + if (this._store.isDisposed) return; this.indexCache = undefined; super.dispose(); } diff --git a/packages/daemon/src/services/fs/fsWatcherService.ts b/packages/daemon/src/services/fs/fsWatcherService.ts index 9f60189ea..bc90d791e 100644 --- a/packages/daemon/src/services/fs/fsWatcherService.ts +++ b/packages/daemon/src/services/fs/fsWatcherService.ts @@ -6,7 +6,7 @@ import nodePath from 'node:path'; import { FSWatcher } from 'chokidar'; -import { Disposable } from '@moonshot-ai/agent-core'; +import { Disposable, DisposableMap, type IDisposable } from '@moonshot-ai/agent-core'; import { ISessionService } from '@moonshot-ai/services'; import type { FsChangeEntry, FsChangeAction, FsChangeKind } from '@moonshot-ai/protocol'; @@ -42,25 +42,51 @@ interface PendingChange { kind: FsChangeKind; } -interface SessionEntry { - /** Live chokidar watcher; closed + dropped on last unref. */ - watcher: FSWatcher; - /** `sessionId.cwd` (absolute, post-realpath) for relative-path mapping. */ - cwd: string; +/** + * Per-session watcher state. Owns the chokidar `FSWatcher` + active debounce + * timer; `dispose()` clears the timer and fire-and-forgets `watcher.close()`. + * Stored in `FsWatcherService.sessions` (a `DisposableMap`) so removal via + * `sessions.deleteAndDispose(sessionId)` and service-wide teardown via + * `super.dispose()` both flow through `dispose()` automatically. + */ +class SessionEntry implements IDisposable { /** `absPath → refCount` across all connections subscribed to this session. */ - pathRefs: Map; + readonly pathRefs = new Map(); /** `connectionId → Set` for overlap filtering on emit. */ - connectionPaths: Map>; + readonly connectionPaths = new Map>(); /** Accumulating changes for the current 200ms window. */ - pendingChanges: PendingChange[]; + pendingChanges: PendingChange[] = []; /** Raw event count (used for `truncated.count`). */ - pendingRawCount: number; + pendingRawCount = 0; /** True once `pendingChanges.length > maxChangesPerWindow`. */ - truncated: boolean; + truncated = false; /** Timer for the active debounce window; `undefined` between windows. */ - debounceTimer: NodeJS.Timeout | undefined; + debounceTimer: NodeJS.Timeout | undefined = undefined; /** Per-session seq counter, monotonic, starts at 1. */ - seq: number; + seq = 0; + private _disposed = false; + + constructor( + public readonly sessionId: string, + public readonly watcher: FSWatcher, + public cwd: string, + private readonly logger: ILogService, + ) {} + + dispose(): void { + if (this._disposed) return; + this._disposed = true; + if (this.debounceTimer) { + clearTimeout(this.debounceTimer); + this.debounceTimer = undefined; + } + void this.watcher.close().catch((err) => { + this.logger.warn( + { sessionId: this.sessionId, err: String(err) }, + 'fs-watcher close failed', + ); + }); + } } export class FsWatcherService extends Disposable implements IFsWatcher { @@ -70,7 +96,7 @@ export class FsWatcherService extends Disposable implements IFsWatcher { private readonly maxChangesPerWindow: number; private readonly maxPathsPerConnection: number; private readonly makeWatcher: () => FSWatcher; - private readonly sessions = new Map(); + private readonly sessions: DisposableMap; /** `connectionId → Map>`. */ private readonly connections = new Map>>(); @@ -89,6 +115,7 @@ export class FsWatcherService extends Disposable implements IFsWatcher { @ISessionService _sessionService: ISessionService, ) { super(); + this.sessions = this._register(new DisposableMap()); this.debounceMs = options.debounceMs ?? DEFAULT_DEBOUNCE_MS; this.maxChangesPerWindow = options.maxChangesPerWindow ?? DEFAULT_MAX_CHANGES_PER_WINDOW; @@ -111,7 +138,7 @@ export class FsWatcherService extends Disposable implements IFsWatcher { connectionId: string, absPaths: readonly string[], ): readonly string[] { - if (this._isDisposed) return []; + if (this._store.isDisposed) return []; // Project the new total for this connection (assuming all `absPaths` are // additions). Dedup against existing first. @@ -184,7 +211,7 @@ export class FsWatcherService extends Disposable implements IFsWatcher { connectionId: string, absPaths: readonly string[], ): readonly string[] { - if (this._isDisposed) return []; + if (this._store.isDisposed) return []; const entry = this.sessions.get(sessionId); if (!entry) return []; const connSessions = this.connections.get(connectionId); @@ -218,7 +245,7 @@ export class FsWatcherService extends Disposable implements IFsWatcher { } // Per-session cleanup: if no path references remain, close the watcher. if (entry.pathRefs.size === 0) { - this.disposeSessionEntry(sessionId, entry); + this.sessions.deleteAndDispose(sessionId); } return connSessionPaths ? Array.from(connSessionPaths) : []; } @@ -285,17 +312,7 @@ export class FsWatcherService extends Disposable implements IFsWatcher { private createSessionEntry(sessionId: string, cwd: string): SessionEntry { const watcher = this.makeWatcher(); - const entry: SessionEntry = { - watcher, - cwd, - pathRefs: new Map(), - connectionPaths: new Map(), - pendingChanges: [], - pendingRawCount: 0, - truncated: false, - debounceTimer: undefined, - seq: 0, - }; + const entry = new SessionEntry(sessionId, watcher, cwd, this.logger); watcher.on( 'all', (eventName: string, absPath: string) => { @@ -311,27 +328,13 @@ export class FsWatcherService extends Disposable implements IFsWatcher { return entry; } - private disposeSessionEntry(sessionId: string, entry: SessionEntry): void { - if (entry.debounceTimer) { - clearTimeout(entry.debounceTimer); - entry.debounceTimer = undefined; - } - void entry.watcher.close().catch((err) => { - this.logger.warn( - { sessionId, err: String(err) }, - 'fs-watcher close failed', - ); - }); - this.sessions.delete(sessionId); - } - private onRawChange( sessionId: string, entry: SessionEntry, eventName: string, absPath: string, ): void { - if (this._isDisposed) return; + if (this._store.isDisposed) return; const action = mapChokidarEventToAction(eventName); if (action === undefined) return; // 'ready', 'raw', 'all', 'error' const kind = mapChokidarEventToKind(eventName); @@ -414,11 +417,7 @@ export class FsWatcherService extends Disposable implements IFsWatcher { } override dispose(): void { - if (this._isDisposed) return; - const entries = Array.from(this.sessions.entries()); - for (const [sid, e] of entries) { - this.disposeSessionEntry(sid, e); - } + if (this._store.isDisposed) return; this.connections.clear(); super.dispose(); } diff --git a/packages/daemon/src/services/gateway/connectionRegistryService.ts b/packages/daemon/src/services/gateway/connectionRegistryService.ts index f757c1b84..4a1ed16d1 100644 --- a/packages/daemon/src/services/gateway/connectionRegistryService.ts +++ b/packages/daemon/src/services/gateway/connectionRegistryService.ts @@ -47,7 +47,7 @@ export class ConnectionRegistry extends Disposable implements IConnectionRegistr } override dispose(): void { - if (this._isDisposed) return; + if (this._store.isDisposed) return; // Belt-and-suspenders: WSGateway.dispose() already called closeAll(). // Idempotent. this.closeAll(); diff --git a/packages/daemon/src/services/gateway/restGatewayService.ts b/packages/daemon/src/services/gateway/restGatewayService.ts index e4ab60041..be77cdd93 100644 --- a/packages/daemon/src/services/gateway/restGatewayService.ts +++ b/packages/daemon/src/services/gateway/restGatewayService.ts @@ -18,7 +18,7 @@ export class FastifyRestGateway extends Disposable implements IRestGateway { } override dispose(): void { - if (this._isDisposed) return; + if (this._store.isDisposed) return; // Fire-and-forget — Fastify's close is async but the DI dispose contract is sync. // The daemon's RunningDaemon.close() awaits `app.close()` explicitly before // calling ix.dispose(), so by the time we get here the listener is already diff --git a/packages/daemon/src/services/gateway/sessionClientsService.ts b/packages/daemon/src/services/gateway/sessionClientsService.ts index 4d47479b2..e771605fb 100644 --- a/packages/daemon/src/services/gateway/sessionClientsService.ts +++ b/packages/daemon/src/services/gateway/sessionClientsService.ts @@ -80,7 +80,7 @@ export class SessionClientsService extends Disposable implements ISessionClients } override dispose(): void { - if (this._isDisposed) return; + if (this._store.isDisposed) return; this._bySession.clear(); super.dispose(); } diff --git a/packages/daemon/src/services/gateway/wsBroadcastService.ts b/packages/daemon/src/services/gateway/wsBroadcastService.ts index c8759bb04..a3e3f08bb 100644 --- a/packages/daemon/src/services/gateway/wsBroadcastService.ts +++ b/packages/daemon/src/services/gateway/wsBroadcastService.ts @@ -77,7 +77,7 @@ export class WSBroadcastService extends Disposable implements IWSBroadcastServic } private _onEvent(event: Event): void { - if (this._isDisposed) return; + if (this._store.isDisposed) return; const sid = extractSessionId(event); const evType = (event as { type?: string }).type ?? ''; if (!sid) { @@ -157,7 +157,7 @@ export class WSBroadcastService extends Disposable implements IWSBroadcastServic } override dispose(): void { - if (this._isDisposed) return; + if (this._store.isDisposed) return; this._sessions.clear(); super.dispose(); } diff --git a/packages/daemon/src/services/gateway/wsGatewayService.ts b/packages/daemon/src/services/gateway/wsGatewayService.ts index d9b6e5d7a..1f1ea6367 100644 --- a/packages/daemon/src/services/gateway/wsGatewayService.ts +++ b/packages/daemon/src/services/gateway/wsGatewayService.ts @@ -92,7 +92,7 @@ export class WSGateway extends Disposable implements IWSGateway { } override dispose(): void { - if (this._isDisposed) return; + if (this._store.isDisposed) return; // 1. Close every attached connection (WS code 1001 = going away). try { this.registry.closeAll('daemon shutting down'); diff --git a/packages/daemon/src/services/question/questionService.ts b/packages/daemon/src/services/question/questionService.ts index 111c2f20c..bae3ae5b3 100644 --- a/packages/daemon/src/services/question/questionService.ts +++ b/packages/daemon/src/services/question/questionService.ts @@ -31,7 +31,7 @@ import { ulid } from 'ulid'; -import { Disposable } from '@moonshot-ai/agent-core'; +import { Disposable, DisposableMap, type IDisposable } from '@moonshot-ai/agent-core'; import type { Event, QuestionRequest as ProtocolQuestionRequest, @@ -66,23 +66,67 @@ export class QuestionExpiredError extends Error { } } -interface PendingQuestion { - readonly questionId: string; - readonly sessionId: string; - readonly toolCallId: string | undefined; - readonly createdAt: string; - readonly expiresAt: string; - readonly protocolRequest: ProtocolQuestionRequest; - resolve: (r: QuestionResult) => void; - reject: (e: Error) => void; - timer: NodeJS.Timeout; +/** + * One pending question. Owns the 60s timer + promise callbacks. Two ways out: + * - `markSettled()` — the happy path called by resolve/dismiss/expire AFTER + * they've decided the outcome; just stops the timer. + * - `dispose()` — the shutdown fallback. If the entry hasn't been settled + * yet, clears the timer AND rejects with "daemon shutting down" so the + * awaiter doesn't dangle. + * + * Stored in `QuestionService._pending` (`DisposableMap`) so service-wide + * teardown via `super.dispose()` triggers each entry's `dispose()` + * automatically, replacing the old hand-written for-loop in `override + * dispose()`. + */ +class PendingQuestion implements IDisposable { + private _settled = false; + + constructor( + readonly questionId: string, + readonly sessionId: string, + readonly toolCallId: string | undefined, + readonly createdAt: string, + readonly expiresAt: string, + readonly protocolRequest: ProtocolQuestionRequest, + private readonly _resolveFn: (r: QuestionResult) => void, + private readonly _rejectFn: (e: Error) => void, + private readonly _timer: NodeJS.Timeout, + ) {} + + /** Happy-path settle (resolve/dismiss/expire decided the outcome). */ + markSettled(): void { + if (this._settled) return; + this._settled = true; + clearTimeout(this._timer); + } + + resolve(r: QuestionResult): void { + this._resolveFn(r); + } + + reject(e: Error): void { + this._rejectFn(e); + } + + /** Shutdown fallback: reject any unsettled awaiter. Idempotent. */ + dispose(): void { + if (this._settled) return; + this._settled = true; + clearTimeout(this._timer); + try { + this._rejectFn(new Error('daemon shutting down')); + } catch { + // swallow — the awaiter side may have already gone away + } + } } export class QuestionService extends Disposable implements IQuestionService { readonly _serviceBrand: undefined; /** Indexed by daemon-minted `question_id` (REST path key). */ - private readonly _pending = new Map(); + private readonly _pending: DisposableMap; /** Bounded set of resolved/dismissed ids for idempotency. */ private readonly _recentlyResolved = new Set(); private _timeoutMs = QUESTION_DEFAULT_TIMEOUT_MS; @@ -93,12 +137,13 @@ export class QuestionService extends Disposable implements IQuestionService { @IEventService private readonly eventService: IEventService, ) { super(); + this._pending = this._register(new DisposableMap()); } async request( req: QuestionRequest & { sessionId: string; agentId: string }, ): Promise { - if (this._isDisposed) { + if (this._store.isDisposed) { throw new Error('question service disposed'); } @@ -135,17 +180,20 @@ export class QuestionService extends Disposable implements IQuestionService { return await new Promise((resolve, reject) => { const timer = setTimeout(() => this._expire(questionId), this._timeoutMs); timer.unref?.(); - this._pending.set(questionId, { + this._pending.set( questionId, - sessionId: req.sessionId, - toolCallId: req.toolCallId, - createdAt, - expiresAt, - protocolRequest, - resolve, - reject, - timer, - }); + new PendingQuestion( + questionId, + req.sessionId, + req.toolCallId, + createdAt, + expiresAt, + protocolRequest, + resolve, + reject, + timer, + ), + ); }); } @@ -157,8 +205,8 @@ export class QuestionService extends Disposable implements IQuestionService { resolve(id: string, response: QuestionResult): void { const p = this._pending.get(id); if (!p) return; - clearTimeout(p.timer); - this._pending.delete(id); + p.markSettled(); + this._pending.deleteAndLeak(id); this.markResolved(p.questionId); const resolvedAt = new Date().toISOString(); @@ -187,8 +235,8 @@ export class QuestionService extends Disposable implements IQuestionService { dismiss(id: string): void { const p = this._pending.get(id); if (!p) return; - clearTimeout(p.timer); - this._pending.delete(id); + p.markSettled(); + this._pending.deleteAndLeak(id); this.markResolved(p.questionId); const dismissedAt = new Date().toISOString(); @@ -253,7 +301,8 @@ export class QuestionService extends Disposable implements IQuestionService { private _expire(questionId: string): void { const p = this._pending.get(questionId); if (!p) return; - this._pending.delete(questionId); + p.markSettled(); + this._pending.deleteAndLeak(questionId); this.markResolved(p.questionId); const expiredEvent: Event = { @@ -268,16 +317,7 @@ export class QuestionService extends Disposable implements IQuestionService { } override dispose(): void { - if (this._isDisposed) return; - for (const [, p] of this._pending) { - clearTimeout(p.timer); - try { - p.reject(new Error('daemon shutting down')); - } catch { - // ignore - } - } - this._pending.clear(); + if (this._store.isDisposed) return; this._recentlyResolved.clear(); super.dispose(); } diff --git a/packages/daemon/src/services/serviceCollection.ts b/packages/daemon/src/services/serviceCollection.ts index eaf6505cc..24d9b4d91 100644 --- a/packages/daemon/src/services/serviceCollection.ts +++ b/packages/daemon/src/services/serviceCollection.ts @@ -1,70 +1,9 @@ -/** - * `createDaemonServiceCollection` — central wiring for the daemon's DI graph. - * Mirrors the VSCode `electron-main/main.ts:162-233` - * pattern: hybrid `ServiceCollection`, with: - * - * - **Prebuilt** `services.set(I, new C(...))` for services that capture - * runtime handles or closures the container can't synthesize - * (`PinoLogger` wraps the Fastify-shared `pino.Logger`; - * `FastifyRestGateway` wraps the `FastifyLike` instance; - * `IEnvironmentService` carries CLI-resolved `homeDir` / `configPath`). - * - **Descriptor** `services.set(I, new SyncDescriptor(C, [], false))` for - * services whose ctor is pure `@I…` injection. The container drives - * construction via `_createAndCacheServiceInstance` so decorators auto-inject. - * A handful of services still take a leading options bag (e.g. - * `CoreProcessService` with `coreProcessOptions`, `WSGateway` with - * `wsGatewayOptions`); those use `new SyncDescriptor(C, [options], false)`. - * - * `supportsDelayedInstantiation = false` for every descriptor here to preserve - * the `_constructionOrder` discipline that powers reverse-dispose order. The - * `a.get(IX)` touch sequence in `start.ts` still pins the ordering. - * - * # Why a helper (not inline in start.ts)? - * - * Centralizing all `services.set(... new SyncDescriptor(...))` in one place - * keeps the wiring shape auditable in a single file while `start.ts` retains - * the construction-order touch list + post-collection adapters - * (`IFsWatcher` closure construction, - * `setUnexpectedErrorHandler`, WS abort + fs-watch handler wiring). - * - * # Why `IFsWatcher` stays in start.ts - * - * `FsWatcherService` ctor takes a `connection-lookup` closure built from - * `IConnectionRegistry.get` at runtime. That closure isn't serializable - * into a `SyncDescriptor` static-arg slot, so we keep its construction - * inside the `ix.invokeFunction` block in `start.ts` post-collection. - */ - import { + getSingletonServiceDescriptors, ServiceCollection, SyncDescriptor, } from '@moonshot-ai/agent-core'; -import { - AuthSummaryService, - CoreProcessService, - defaultServicesModule, - EventService, - IApprovalService, - IAuthSummaryService, - IEnvironmentService, - IEventService, - ICoreProcessService, - IMcpService, - IMessageService, - IOAuthService, - IPromptService, - IQuestionService, - ISessionService, - ITaskService, - IToolService, - McpService, - MessageService, - OAuthService, - PromptService, - SessionService, - TaskService, - ToolService, -} from '@moonshot-ai/services'; +import * as Services from '@moonshot-ai/services'; import type { Logger as PinoLogger } from 'pino'; import type { FastifyLike } from '#/services/gateway/restGateway'; @@ -100,66 +39,43 @@ import { IWSBroadcastService } from '#/services/gateway/wsBroadcast'; import { WSBroadcastService } from '#/services/gateway/wsBroadcastService'; export interface DaemonServiceCollectionOptions { - /** Original `startDaemon` options bag — carries the per-service tunables. */ readonly daemon: DaemonStartOptions; - /** Resolved Fastify instance (`app`) — needed by `FastifyRestGateway`. */ readonly app: FastifyLike; - /** Fastify-shared pino logger — wrapped by `PinoLoggerAdapter`. */ readonly pinoLogger: PinoLogger; - /** Pre-resolved environment paths (homeDir / configPath). */ - readonly envService: IEnvironmentService; + readonly envService: Services.IEnvironmentService; } -/** - * Assemble the daemon's `ServiceCollection`. The returned collection has - * EVERY singleton seeded — either as a prebuilt instance (runtime-handle - * services) or as a `SyncDescriptor` (descriptor-first singletons). - * - * One singleton NOT registered here, by design: - * - `IFsWatcher` — needs a closure over `IConnectionRegistry.get` at - * construction time; built inline in `start.ts` (see file header). - */ export function createDaemonServiceCollection( input: DaemonServiceCollectionOptions, ): ServiceCollection { const { daemon, app, pinoLogger, envService } = input; const services = new ServiceCollection( - // Registry entries from `@moonshot-ai/services` (self-registered by each - // impl file at module-load time). These supply the default descriptors - // for services whose ctor is pure `@I…` injection. - ...defaultServicesModule(), - // Daemon-only services not shipped by `@moonshot-ai/services`. + ...getSingletonServiceDescriptors(), [IConnectionRegistry, new SyncDescriptor(ConnectionRegistry, [], false)], [ISessionClientsService, new SyncDescriptor(SessionClientsService, [], false)], [IWSBroadcastService, new SyncDescriptor(WSBroadcastService, [], false)], - [IApprovalService, new SyncDescriptor(ApprovalService, [], false)], - [IQuestionService, new SyncDescriptor(QuestionService, [], false)], + [Services.IApprovalService, new SyncDescriptor(ApprovalService, [], false)], + [Services.IQuestionService, new SyncDescriptor(QuestionService, [], false)], [IFsService, new SyncDescriptor(FsService, [], false)], [IFsSearchService, new SyncDescriptor(FsSearchService, [], false)], [IFsGitService, new SyncDescriptor(FsGitService, [], false)], [IWorkspaceFsService, new SyncDescriptor(WorkspaceFsService, [], false)], ); - // -- Prebuilt: services that need runtime handles / external closures ------ services.set(ILogService, new PinoLoggerAdapter(pinoLogger)); services.set(IRestGateway, new FastifyRestGateway(app)); - services.set(IEnvironmentService, envService); + services.set(Services.IEnvironmentService, envService); - // -- Override registry entries with runtime static args -------------------- services.set( IWSGateway, new SyncDescriptor(WSGateway, [daemon.wsGatewayOptions ?? {}], false), ); services.set( - ICoreProcessService, - new SyncDescriptor(CoreProcessService, [daemon.coreProcessOptions ?? {}], false), + Services.ICoreProcessService, + new SyncDescriptor(Services.CoreProcessService, [daemon.coreProcessOptions ?? {}], false), ); - // `IFileStore` + `IWorkspaceRegistry` derive their on-disk base from - // `IEnvironmentService.homeDir` (which itself reads - // `opts.coreProcessOptions?.homeDir`). No static args here — the impls - // inject `IEnvironmentService` directly. services.set(IFileStore, new SyncDescriptor(FileStore, [], false)); services.set(IWorkspaceRegistry, new SyncDescriptor(WorkspaceRegistryService, [], false)); diff --git a/packages/daemon/src/services/workspace/workspaceFsService.ts b/packages/daemon/src/services/workspace/workspaceFsService.ts index 30be9a800..cb1b07fd8 100644 --- a/packages/daemon/src/services/workspace/workspaceFsService.ts +++ b/packages/daemon/src/services/workspace/workspaceFsService.ts @@ -92,7 +92,7 @@ export class WorkspaceFsService extends Disposable implements IWorkspaceFsServic } override dispose(): void { - if (this._isDisposed) return; + if (this._store.isDisposed) return; super.dispose(); } } diff --git a/packages/daemon/src/services/workspace/workspaceRegistryService.ts b/packages/daemon/src/services/workspace/workspaceRegistryService.ts index 35228f98f..76aa0975e 100644 --- a/packages/daemon/src/services/workspace/workspaceRegistryService.ts +++ b/packages/daemon/src/services/workspace/workspaceRegistryService.ts @@ -278,7 +278,7 @@ export class WorkspaceRegistryService extends Disposable implements IWorkspaceRe } override dispose(): void { - if (this._isDisposed) return; + if (this._store.isDisposed) return; super.dispose(); } } diff --git a/packages/services/AGENTS.md b/packages/services/AGENTS.md index ee45639cd..26252519f 100644 --- a/packages/services/AGENTS.md +++ b/packages/services/AGENTS.md @@ -96,13 +96,13 @@ no new suffixes get reintroduced. Adding a new service: create the folder + contracts + impl pair, add a bottom-of-file `registerSingleton(IXxxService, XxxService, -InstantiationType.Delayed)` in the impl, add the corresponding side-effect -import to `module.ts`, re-export from `index.ts`. The daemon's `start.ts` -consumes `defaultServicesModule()` for descriptor-only services; only override -the registry entry (via `services.set(I, prebuiltInstance)` or -`services.set(I, new SyncDescriptor(C, [runtimeArgs], false))`) when the -service needs an external handle or runtime static args that the registry -can't supply. +InstantiationType.Delayed)` in the impl, then re-export the contracts and impl +from `index.ts` so importing `@moonshot-ai/services` runs the registration +side effect. Daemon bootstrap consumes `getSingletonServiceDescriptors()` for +descriptor-only services; only override the registry entry (via +`services.set(I, prebuiltInstance)` or `services.set(I, new SyncDescriptor(C, +[runtimeArgs], false))`) when the service needs an external handle or runtime +static args that the registry can't supply. ## Service registration (normative) @@ -127,11 +127,9 @@ can't supply. `CoreProcessService`'s `options`), fall back to the descriptor overload: `registerSingleton(IXxxService, new SyncDescriptor(XxxService, [optionsBag]))`. -2. **`defaultServicesModule()` is a thin projection** of - `getSingletonServiceDescriptors()`. It does NOT maintain a separate - list — `module.ts`'s only responsibility is the side-effect import - list that populates the registry plus the - `InstantiationType.Delayed | Eager` projection. +2. **Consumers seed from `getSingletonServiceDescriptors()` directly**. + Importing `@moonshot-ai/services` loads the package barrel, whose impl + re-exports run the `registerSingleton(...)` side effects. 3. **Daemon-side `services.set(...)` may override** the registry-derived entry for services that need runtime static args (e.g. @@ -142,9 +140,9 @@ can't supply. from `registerSingleton` (plan §158); the later registration wins at every layer. -The legacy "hand-built array in `module.ts`" pattern that lived here -through Phase 2 is gone. Do NOT reintroduce it — extending the array in -`module.ts` no longer has any effect on what the daemon resolves. +The legacy "hand-built array" and `defaultServicesModule()` wrapper patterns +are gone. Do NOT reintroduce them — the registry is the source of truth, and +bootstrap code should read it with `getSingletonServiceDescriptors()`. ## Comments (normative) @@ -178,4 +176,3 @@ Existing files in this package over-comment by historical accident. file, prefer leaving the surrounding comments alone — large comment deletions belong in their own dedicated cleanup pass, not bundled into behavior changes. - diff --git a/packages/services/src/authSummary/authSummaryService.ts b/packages/services/src/authSummary/authSummaryService.ts index 3148815ea..d2890d6f9 100644 --- a/packages/services/src/authSummary/authSummaryService.ts +++ b/packages/services/src/authSummary/authSummaryService.ts @@ -106,7 +106,7 @@ export class AuthSummaryService } override dispose(): void { - if (this._isDisposed) return; + if (this._store.isDisposed) return; super.dispose(); } diff --git a/packages/services/src/coreProcess/coreProcessService.ts b/packages/services/src/coreProcess/coreProcessService.ts index 6a09cb1f6..27ed2fb2b 100644 --- a/packages/services/src/coreProcess/coreProcessService.ts +++ b/packages/services/src/coreProcess/coreProcessService.ts @@ -142,7 +142,7 @@ export class CoreProcessService extends Disposable implements ICoreProcessServic } override dispose(): void { - if (this._isDisposed) return; + if (this._store.isDisposed) return; // KimiCore does not currently expose a dispose() — when it does, we'll // await/call it here BEFORE super.dispose(). For now, disposing the // service flips _disposed, which makes future rpc.* invocations reject @@ -152,7 +152,7 @@ export class CoreProcessService extends Disposable implements ICoreProcessServic private _buildRpcProxy(): CoreRPC { const rpcPromise = this._coreRpcPromise; - const isDisposedRef = () => this._isDisposed; + const isDisposedRef = () => this._store.isDisposed; // We don't know the concrete method set at compile time here (CoreAPI is // a structural interface; `RPCMethods` is a mapped type). diff --git a/packages/services/src/event/eventService.ts b/packages/services/src/event/eventService.ts index 0ba364f75..4766cc27f 100644 --- a/packages/services/src/event/eventService.ts +++ b/packages/services/src/event/eventService.ts @@ -35,7 +35,7 @@ export class EventService extends Disposable implements IEventService { readonly onDidPublish = this._onDidPublish.event; publish(event: ProtocolEvent): void { - if (this._isDisposed) return; + if (this._store.isDisposed) return; this._onDidPublish.fire(event); } } diff --git a/packages/services/src/index.ts b/packages/services/src/index.ts index 93a55463b..eefaac80a 100644 --- a/packages/services/src/index.ts +++ b/packages/services/src/index.ts @@ -1,42 +1,3 @@ -/** - * `@moonshot-ai/services` — in-process service container for the kimi-code - * daemon. Houses every `IXxxService` decorator (per-domain folder), the - * `CoreProcessService` that owns the in-process `KimiCore` instance, and the - * adapters that translate `KimiCore` shapes into protocol-shaped data. - * - * Naming convention is encoded in `packages/services/AGENTS.md` — every - * injectable uses the `Service` suffix, contracts live in `.ts`, - * impl lives in `Service.ts`, folder names are camelCase. - * - * Per-domain layout: - * coreProcess/coreProcess.ts — ICoreProcessService + CoreProcessServiceOptions - * coreProcess/coreProcessService.ts — CoreProcessService (self-registers via registerSingleton) - * coreProcess/coreProcessClient.ts — BridgeClientAPI (SDK-side of the RPC pair) - * event/event.ts — IEventService - * event/eventService.ts — EventService (pure in-process Emitter wrapper) - * approval/approval.ts — IApprovalService + protocol adapter - * question/question.ts — IQuestionService + protocol adapter - * environment/environment.ts — IEnvironmentService - * session/session.ts — ISessionService + toProtocolSession - * session/sessionService.ts — SessionService - * message/message.ts — IMessageService + toProtocolMessage - * message/messageService.ts — MessageService - * prompt/prompt.ts — IPromptService + SyntheticPrompt* events - * prompt/promptService.ts — PromptService - * tool/tool.ts — IToolService + toProtocolTool - * tool/toolService.ts — ToolService - * mcp/mcp.ts — IMcpService + toProtocolMcpServer - * mcp/mcpService.ts — McpService - * task/task.ts — ITaskService + toProtocolTask - * task/taskService.ts — TaskService - * oauth/oauth.ts — IOAuthService - * oauth/oauthService.ts — OAuthService - * authSummary/authSummary.ts — IAuthSummaryService + sentinel errors - * authSummary/authSummaryService.ts — AuthSummaryService - * modelCatalog/modelCatalog.ts — IModelCatalogService + config adapters - * modelCatalog/modelCatalogService.ts — ModelCatalogService - */ - export { BridgeClientAPI } from './coreProcess/coreProcessClient'; export type { CoreProcessClientDeps } from './coreProcess/coreProcessClient'; export { @@ -44,18 +5,10 @@ export { type CoreProcessServiceOptions, } from './coreProcess/coreProcess'; export { CoreProcessService } from './coreProcess/coreProcessService'; -export { - defaultServicesModule, - type ServiceModuleEntry, -} from './module'; -// --- per-domain exports --------------------------------------------------- - -// event service export { IEventService } from './event/event'; export { EventService } from './event/eventService'; -// approval service + adapter export { IApprovalService } from './approval/approval'; export type { ApprovalRequest, ApprovalResponse } from './approval/approval'; export { @@ -64,7 +17,6 @@ export { type ToBrokerRequestParams as ApprovalToBrokerRequestParams, } from './approval/approval'; -// question service + adapter export { IQuestionService } from './question/question'; export type { QuestionRequest, QuestionResult } from './question/question'; export { @@ -74,10 +26,8 @@ export { type QuestionToBrokerRequestParams, } from './question/question'; -// environment service export { IEnvironmentService } from './environment/environment'; -// authSummary service export { IAuthSummaryService, AuthProvisioningRequiredError, @@ -87,11 +37,9 @@ export { } from './authSummary/authSummary'; export { AuthSummaryService } from './authSummary/authSummaryService'; -// oauth service export { IOAuthService } from './oauth/oauth'; export { OAuthService } from './oauth/oauthService'; -// model catalog service + adapter export { IModelCatalogService, ModelNotFoundError, @@ -103,7 +51,6 @@ export { export type { ProviderCredentialState } from './modelCatalog/modelCatalog'; export { ModelCatalogService } from './modelCatalog/modelCatalogService'; -// session service + adapter export { ISessionService, SessionNotFoundError, @@ -113,7 +60,6 @@ export { export type { SessionListQuery } from './session/session'; export { SessionService } from './session/sessionService'; -// message service + adapter export { IMessageService, MessageNotFoundError, @@ -124,7 +70,6 @@ export { export type { MessageListQuery } from './message/message'; export { MessageService } from './message/messageService'; -// prompt service export { IPromptService, PromptAlreadyCompletedError, @@ -141,7 +86,6 @@ export type { } from './prompt/prompt'; export { PromptService } from './prompt/promptService'; -// tool service + adapter export { IToolService, toProtocolTool, @@ -149,7 +93,6 @@ export { } from './tool/tool'; export { ToolService } from './tool/toolService'; -// mcp service + adapter export { IMcpService, McpServerNotFoundError, @@ -157,7 +100,6 @@ export { } from './mcp/mcp'; export { McpService } from './mcp/mcpService'; -// task service + adapter export { ITaskService, TaskAlreadyFinishedError, @@ -167,10 +109,3 @@ export { } from './task/task'; export type { TaskListQuery } from './task/task'; export { TaskService } from './task/taskService'; - -// NOTE: every `Service.ts` impl file self-registers at the bottom via -// `registerSingleton(IXxx, XxxService, InstantiationType.Delayed)` (or the -// descriptor overload when a leading options bag is required, e.g. -// `CoreProcessService`). `defaultServicesModule()` is a thin projection of -// that global registry. Consumers override entries with `services.set(...)` -// for runtime static args or prebuilt instances. diff --git a/packages/services/src/module.ts b/packages/services/src/module.ts deleted file mode 100644 index d1a4fb631..000000000 --- a/packages/services/src/module.ts +++ /dev/null @@ -1,66 +0,0 @@ -/** - * `defaultServicesModule()` — DI entries shipped by `@moonshot-ai/services`. - * - * Thin projection of the global singleton registry maintained by - * `@moonshot-ai/agent-core`. Each service impl file self-registers at - * module-load time via `registerSingleton` (ctor overload for pure `@I…` - * injection, descriptor overload when a leading options bag is required). - * Importing this `module.ts` triggers the side-effect imports below, which - * populate the registry; `defaultServicesModule()` snapshots the registry - * via `getSingletonServiceDescriptors()`. - * - * Usage: - * - * const services = new ServiceCollection(...defaultServicesModule()); - * - * Each entry is `[ServiceIdentifier, SyncDescriptor]`. The descriptor carries - * the `supportsDelayedInstantiation` flag; no extra projection is needed. - * - * Consumers (e.g. the daemon) override entries with `services.set(...)` for - * runtime static args (`CoreProcessService` with real `coreProcessOptions`) - * or prebuilt instances (`PinoLogger`). Later registrations win. - * - * Per-domain layout: see `packages/services/AGENTS.md`. Classes live in - * per-domain folders (`session/`, `message/`, …) with one `.ts` - * contracts file and one `Service.ts` impl file each. - */ - -import { - getSingletonServiceDescriptors, - SyncDescriptor, - type ServiceIdentifier, -} from '@moonshot-ai/agent-core'; - -// Side-effect imports — each impl file calls `registerSingleton(...)` at -// file bottom. Ordering matters: it determines the order entries surface from -// `getSingletonServiceDescriptors()`, which the daemon's reverse-dispose -// semantics piggy-back on. CoreProcessService MUST register first — the -// existing `defaultServicesModule()` test -// (`packages/services/test/coreProcessService.test.ts:315`) asserts it -// sits at index 0, and downstream `a.get(...)` "touch" ordering in -// `packages/daemon/src/start.ts` assumes the bridge is the first -// service-package entry into the construction-order list. -import './coreProcess/coreProcessService'; -import './event/eventService'; -import './session/sessionService'; -import './message/messageService'; -import './prompt/promptService'; -import './tool/toolService'; -import './mcp/mcpService'; -import './task/taskService'; -import './authSummary/authSummaryService'; -import './oauth/oauthService'; -import './modelCatalog/modelCatalogService'; - -export type ServiceModuleEntry = readonly [ - // eslint-disable-next-line @typescript-eslint/no-explicit-any - ServiceIdentifier, - // eslint-disable-next-line @typescript-eslint/no-explicit-any - SyncDescriptor, -]; - -export function defaultServicesModule(): ReadonlyArray { - return getSingletonServiceDescriptors().map( - ([id, descriptor]) => [id, descriptor] as const, - ); -} diff --git a/packages/services/src/oauth/oauthService.ts b/packages/services/src/oauth/oauthService.ts index 176d3d9d5..7da0b4133 100644 --- a/packages/services/src/oauth/oauthService.ts +++ b/packages/services/src/oauth/oauthService.ts @@ -2,7 +2,7 @@ * `OAuthService` — implementation of `IOAuthService`. */ -import { Disposable, InstantiationType, registerSingleton } from '@moonshot-ai/agent-core'; +import { Disposable, DisposableMap, InstantiationType, registerSingleton, type IDisposable } from '@moonshot-ai/agent-core'; import { DeviceCodeTimeoutError, KIMI_CODE_PROVIDER_NAME, @@ -22,19 +22,52 @@ import { ulid } from 'ulid'; import { IEnvironmentService } from '../environment/environment'; import { IOAuthService } from './oauth'; -interface FlowState { - readonly flowId: string; - readonly provider: string; - readonly deviceAuth: DeviceAuthorization; - /** Resolved seconds-until-expiry (may differ from `deviceAuth.expiresIn` if that was null). */ - readonly expiresInSec: number; - readonly startedAt: number; - readonly expiresAt: number; - status: OAuthFlowStatus; - readonly controller: AbortController; +/** + * One in-flight (or recently-completed) device-code flow. Stored in + * `OAuthService._flows` (a `DisposableMap`) so: + * - `_flows.set(provider, newState)` auto-disposes the supersedee + * - `_flows.deleteAndDispose(provider)` (called from the GC timer) tears + * down a terminal entry's leftover state + * - service-wide `super.dispose()` walks every entry's `dispose()` + * instead of the old hand-written for-loop in `override dispose()`. + * + * `dispose()` is idempotent and aborts the controller only when the flow is + * still pending — terminal flows have already returned from their underlying + * promise, so a second abort would be a noisy no-op. + */ +class FlowState implements IDisposable { + status: OAuthFlowStatus = 'pending'; resolvedAt?: number; errorMessage?: string; gcTimer?: NodeJS.Timeout; + private _disposed = false; + + constructor( + readonly flowId: string, + readonly provider: string, + readonly deviceAuth: DeviceAuthorization, + /** Resolved seconds-until-expiry (may differ from `deviceAuth.expiresIn` if that was null). */ + readonly expiresInSec: number, + readonly startedAt: number, + readonly expiresAt: number, + readonly controller: AbortController, + ) {} + + dispose(): void { + if (this._disposed) return; + this._disposed = true; + if (this.gcTimer !== undefined) { + clearTimeout(this.gcTimer); + this.gcTimer = undefined; + } + if (this.status === 'pending') { + try { + this.controller.abort(); + } catch { + // ignore + } + } + } } /** Terminal flows live this long after resolution before GC. */ @@ -44,10 +77,11 @@ export class OAuthService extends Disposable implements IOAuthService { readonly _serviceBrand: undefined; private readonly _authFacade: KimiAuthFacade; - private readonly _flows = new Map(); + private readonly _flows: DisposableMap; constructor(@IEnvironmentService private readonly env: IEnvironmentService) { super(); + this._flows = this._register(new DisposableMap()); this._authFacade = new KimiAuthFacade({ homeDir: env.homeDir, configPath: env.configPath, @@ -117,16 +151,15 @@ export class OAuthService extends Disposable implements IOAuthService { // `OAuthManager.login`, so the `expires_at` we surface to clients is // never further out than the deadline that's actually being enforced. const expiresInSec = deviceAuth.expiresIn ?? 15 * 60; - const state: FlowState = { + const state = new FlowState( flowId, - provider: name, + name, deviceAuth, expiresInSec, startedAt, - expiresAt: startedAt + expiresInSec * 1000, - status: 'pending', + startedAt + expiresInSec * 1000, controller, - }; + ); this._flows.set(name, state); // Wire the background promise's terminal transition. We branch on error @@ -185,18 +218,7 @@ export class OAuthService extends Disposable implements IOAuthService { } override dispose(): void { - if (this._isDisposed) return; - for (const state of this._flows.values()) { - if (state.gcTimer !== undefined) clearTimeout(state.gcTimer); - if (state.status === 'pending') { - try { - state.controller.abort(); - } catch { - // ignore - } - } - } - this._flows.clear(); + if (this._store.isDisposed) return; super.dispose(); } @@ -220,14 +242,18 @@ export class OAuthService extends Disposable implements IOAuthService { if (state.status === status) return; state.status = status; state.resolvedAt = Date.now(); - // Schedule GC. If a new flow supersedes this entry first, the new flow - // replaces the map entry and this timer just no-ops on the stale state. + // Schedule GC. If a new flow supersedes this entry first, the supersede + // path runs `_flows.set(name, newState)` which auto-disposes this entry — + // its `dispose()` clears `gcTimer` before the timer can fire, so the + // callback can rely on "this state is still the current map entry" + // without the equality check needing a stale-guard. if (state.gcTimer !== undefined) clearTimeout(state.gcTimer); state.gcTimer = setTimeout(() => { const current = this._flows.get(state.provider); - // Only GC if this state IS the current map entry. A newer flow may - // have already overwritten the slot. - if (current === state) this._flows.delete(state.provider); + // Belt-and-suspenders: even though dispose() clears the timer on + // overwrite, keep the identity check in case the timer was queued + // before the clearTimeout took effect. + if (current === state) this._flows.deleteAndDispose(state.provider); }, TERMINAL_RETENTION_MS); // Don't keep the process alive solely for GC. state.gcTimer.unref?.(); diff --git a/packages/services/src/prompt/promptService.ts b/packages/services/src/prompt/promptService.ts index d3233bfb3..371b889dc 100644 --- a/packages/services/src/prompt/promptService.ts +++ b/packages/services/src/prompt/promptService.ts @@ -819,7 +819,7 @@ export class PromptService } override dispose(): void { - if (this._isDisposed) return; + if (this._store.isDisposed) return; this._active.clear(); this._queued.clear(); this._agentState.clear(); diff --git a/packages/services/src/session/session.ts b/packages/services/src/session/session.ts index 3de280f85..749f4ccfc 100644 --- a/packages/services/src/session/session.ts +++ b/packages/services/src/session/session.ts @@ -1,39 +1,3 @@ -/** - * `ISessionService` — daemon-facing session CRUD interface. - * - * Wraps `ICoreProcessService.rpc.{createSession, listSessions, closeSession, - * updateSessionMetadata}` and adapts agent-core's camelCase + number - * timestamps to the protocol's snake_case + ISO 8601 `Z` shape (see SCHEMAS.md - * §2). Other services in `@moonshot-ai/services` (messages, prompts, ...) - * inherit this camelCase ↔ snake_case + number ↔ ISO pattern. - * - * **Why a service layer**: REST handlers in `@moonshot-ai/daemon` are - * disallowed from importing `@moonshot-ai/kimi-code-sdk` (anti-corruption - * test). Routes call `accessor.get(ISessionService).(...)`; the - * adapter is here. - * - * **CoreAPI shape gap**: agent-core does NOT expose `getSession(id)` returning - * a full `SessionSummary` — `getSessionMetadata` returns the smaller - * `SessionMeta` shape. `get(id)` is implemented via `listSessions({})` + - * filter, throwing `SessionNotFoundError` (→ 40401) when the id is absent. - * See `SessionService` for details + the gap documentation. - * - * **Adapter helpers**: `toProtocolSession` is co-located here. - * - * **DI wiring**: this class takes `ICoreProcessService` via ctor positional - * arg. `defaultServicesModule()` adds a `SyncDescriptor(SessionService)` - * entry, but the container has no ctor-arg DI, so the daemon's `start.ts` - * wires it via - * `ix.createInstance(SessionService, a.get(ICoreProcessService))` then - * `services.set(ISessionService, instance)` — same pattern as - * `CoreProcessService` itself. The descriptor entry is the canonical - * declaration; the daemon's manual wiring is the runtime path. - * - * **Anti-corruption**: this file imports from `@moonshot-ai/agent-core` only - * for type-only `SessionSummary` / `SessionMeta`. Runtime calls go through - * `ICoreProcessService.rpc.`, not direct CoreAPI consumption. - */ - import { createDecorator } from '@moonshot-ai/agent-core'; import { encodeWorkDirKey } from '@moonshot-ai/agent-core/session/store'; import type { Event } from '@moonshot-ai/agent-core/base/common/event'; @@ -54,116 +18,41 @@ import { type UndoSessionResponse, } from '@moonshot-ai/protocol'; -/** - * Listing query — `before_id`/`after_id` + `page_size` mutual exclusivity is - * already enforced by `cursorQuerySchema`. The service layer adds an optional - * status filter the daemon layer parses out of the REST query string, and an - * optional `workDir` filter for the `?workspace_id=` fast path (the daemon - * route layer resolves `workspace_id → workspace.root` and sets `workDir`). - */ export interface SessionListQuery extends CursorQuery { status?: import('@moonshot-ai/protocol').SessionStatus; - /** - * When set, the underlying `core.rpc.listSessions({workDir})` path uses - * agent-core's `listWorkDir` (readdir-based) instead of a full `listAll`. - * Daemon-route caller is responsible for resolving the workspace_id to its - * registered root before populating this. - */ workDir?: string; } export interface ISessionService { readonly _serviceBrand: undefined; - /** - * `POST /v1/sessions` — create a new session. Requires `metadata.cwd` - * (agent-core's `createSession` calls `requiredWorkDir`; missing cwd ⇒ throw). - */ create(input: SessionCreate): Promise; - /** - * `GET /v1/sessions` — list sessions. Cursor pagination is applied - * client-side over `core.rpc.listSessions({})` (the CoreAPI surface - * doesn't take a cursor today). Default - * `page_size = 20` per REST.md §1.6 is applied at the route layer, not here. - */ list(query: SessionListQuery): Promise>; - /** - * `GET /v1/sessions/{id}` and `GET /v1/sessions/{id}/profile` — single - * session by id. Implemented as `listSessions({}) + .find(id)`; throws - * `SessionNotFoundError` (→ 40401) when not found. - */ get(id: string): Promise; - /** - * `POST /v1/sessions/{id}/profile` — update session mutable properties. - * Backed by `updateSessionMetadata` for metadata changes; `title` writes - * through the same path (mapped onto agent-core's `SessionMeta.title`). - * `agent_config.model` is dispatched to `core.rpc.setModel` when present. - * Returns the post-update Session. - */ update(id: string, input: SessionUpdate): Promise; - /** - * `POST /v1/sessions/{id}:fork` — create a new persisted session from an - * idle source session and return the fork. - */ fork(id: string, input: SessionFork): Promise; - /** - * `GET /v1/sessions/{id}/children` — list direct child sessions whose - * metadata points at the parent session. - */ listChildren(id: string, query: SessionListQuery): Promise>; - /** - * `POST /v1/sessions/{id}/children` — create a persisted child session from - * the parent session and return the child. - */ createChild(id: string, input: SessionChildCreate): Promise; - /** - * `DELETE /v1/sessions/{id}` — close (= soft-delete in v1) the session. - * Backed by `bridge.rpc.closeSession({sessionId})`. CoreAPI does not - * surface a hard delete; the daemon currently conflates close == delete. - * - * Returns `{ deleted: true }` envelope shape per REST §3.3. - */ getStatus(id: string): Promise; compact(id: string, input: CompactSessionRequest): Promise; undo(id: string, input: UndoSessionRequest): Promise; - /** - * `DELETE /v1/sessions/{id}` — close (= soft-delete in v1) the session. - * Backed by `bridge.rpc.closeSession({sessionId})`. CoreAPI does not - * surface a hard delete; the daemon currently conflates close == delete. - * - * Returns `{ deleted: true }` envelope shape per REST §3.3. - */ delete(id: string): Promise<{ deleted: true }>; - /** - * VSCode-style accessor for session-creation events. The listener fires - * synchronously after the bridge RPC returns a new `Session`. - * - * Subscribing returns an `IDisposable`. Owners stash it via - * `Disposable._register(svc.onDidCreate(handler))` so it tears down - * with the owning service. - */ readonly onDidCreate: Event<{ session: Session }>; - /** - * VSCode-style accessor for session-close events. The listener fires - * synchronously after `bridge.rpc.closeSession` resolves. Same - * `IDisposable` contract as `onDidCreate`. - */ readonly onDidClose: Event<{ sessionId: string }>; } -// eslint-disable-next-line @typescript-eslint/no-redeclare export const ISessionService = createDecorator('sessionService'); export class SessionUndoUnavailableError extends Error { @@ -175,11 +64,6 @@ export class SessionUndoUnavailableError extends Error { } } -/** - * Sentinel error class — daemon's route layer catches this and maps to - * `code: 40401` (session.not_found). Other errors fall through to - * `installErrorHandler` (→ 50001 internal). - */ export class SessionNotFoundError extends Error { readonly sessionId: string; constructor(sessionId: string) { @@ -189,27 +73,6 @@ export class SessionNotFoundError extends Error { } } -/** - * Convert agent-core's `SessionSummary` + optional `SessionMeta` into the - * protocol-level `Session` shape. The optional `meta` argument is the result - * of `getSessionMetadata` — when present, its `title` / `custom` enrich the - * baseline summary; when absent, defaults are used. - * - * `cwd` overrides apply in this priority order: - * 1. `meta.custom.cwd` (set by daemon when update wrote a new cwd). - * 2. `summary.metadata.cwd` (when caller-supplied during create). - * 3. `summary.workDir` (agent-core canonical field). - * - * `workspace_id` is ALWAYS derived from `summary.workDir` via - * `encodeWorkDirKey`, so every session round-trips to a stable workspace - * key. If the daemon has never seen a `POST /workspaces` for that wd-key - * the id simply won't appear in the workspaces list; the session still has - * an id the front-end can group on. - * - * The merged `Session.metadata` keeps `cwd` plus persistent custom metadata - * from the summary and live `meta.custom` when available (excluding - * daemon-internal `goal` plumbing — that's not protocol surface). - */ export function toProtocolSession( summary: SessionSummary, meta?: SessionMeta | undefined, @@ -221,8 +84,6 @@ export function toProtocolSession( (typeof summaryMetadata['cwd'] === 'string' && summaryMetadata['cwd']) || summary.workDir; - // Strip the internal "goal" key — that's daemon-side runtime state, not - // protocol surface (SCHEMAS §2 doesn't expose it). const { goal: _dropSummaryGoal, ...summaryWithoutGoal } = summaryMetadata; const { goal: _dropCustomGoal, ...customWithoutGoal } = customMetadata; @@ -244,10 +105,6 @@ export function toProtocolSession( status: 'idle', metadata: mergedMetadata, agent_config: { - // CoreAPI doesn't surface a session's effective model on the listSessions - // path; we leave it empty because there is no current source for the - // effective model on this path. Empty string keeps the schema valid for - // consumers that only inspect known keys. model: '', }, usage: emptySessionUsage(), diff --git a/packages/services/src/session/sessionService.ts b/packages/services/src/session/sessionService.ts index 73ea1e8c5..d3232b280 100644 --- a/packages/services/src/session/sessionService.ts +++ b/packages/services/src/session/sessionService.ts @@ -1,7 +1,3 @@ -/** - * `SessionService` — implementation of `ISessionService`. - */ - import { Disposable, Emitter, @@ -50,13 +46,6 @@ const DEFAULT_UNDO_MESSAGE_PAGE_SIZE = 50; const MAX_UNDO_MESSAGE_PAGE_SIZE = 100; const CHILD_SESSION_KIND = 'child'; -/** - * Treat the incoming `metadata` object — schema-validated by zod as - * `{cwd: string}` plus arbitrary `unknown` keys — as a JSON-safe object for - * agent-core's `JsonObject` slot. We don't deep-validate here; clients can - * send non-JSON-serializable values and agent-core will reject at the RPC - * boundary. This cast keeps the adapter narrow and the wire stable. - */ function asJsonObject(value: Record): JsonObject { return value as unknown as JsonObject; } @@ -111,17 +100,8 @@ function pageContextMessages( export class SessionService extends Disposable implements ISessionService { readonly _serviceBrand: undefined; - /** - * VSCode-style Emitter for session-creation events. Listener exceptions - * route to `onUnexpectedError` inside `Emitter.fire()`. Owned via - * `_register(...)` so it disposes when the service is torn down. - */ private readonly _onDidCreate = this._register(new Emitter<{ session: Session }>()); readonly onDidCreate = this._onDidCreate.event; - /** - * VSCode-style Emitter for session-close events. Same ownership + - * exception-routing semantics as `_onDidCreate`. - */ private readonly _onDidClose = this._register(new Emitter<{ sessionId: string }>()); readonly onDidClose = this._onDidClose.event; @@ -134,13 +114,6 @@ export class SessionService extends Disposable implements ISessionService { } async create(input: SessionCreate): Promise { - // The protocol schema now allows `metadata` to be omitted (so callers can - // send `{ workspace_id }` only). The daemon route layer is responsible - // for resolving workspace_id → workspace.root → metadata.cwd BEFORE - // calling this method; by the time we land here `metadata.cwd` must be - // set. agent-core's `createSession` calls `requiredWorkDir(...)` and - // throws if cwd is missing, so a missing-cwd bug surfaces as a - // descriptive error rather than silently picking the daemon's cwd. if (input.metadata === undefined || typeof input.metadata.cwd !== 'string') { throw new Error('SessionService.create: metadata.cwd is required'); } @@ -150,39 +123,25 @@ export class SessionService extends Disposable implements ISessionService { metadata: metadataForCore, ...(input.agent_config?.model !== undefined ? { model: input.agent_config.model } : {}), }); - // agent-core's createSession ignores any caller-supplied title — newly - // created sessions get the default `SessionMeta.title = 'New Session'`. - // When the caller supplied a title we apply it via `renameSession` so the - // post-create get reflects it. if (input.title !== undefined) { try { await this.core.rpc.renameSession({ sessionId: summary.id, title: input.title }); } catch { - // If rename fails (e.g. session closed/race), continue with the - // default — the response shape is unchanged. } } const meta = await this.tryGetMeta(summary.id); const session = toProtocolSession(summary, meta); - // Fire onDidCreate listeners after the core RPC resolves. this._onDidCreate.fire({ session }); return session; } async list(query: SessionListQuery): Promise> { - // Fast path: when caller supplies a workDir (typically from `?workspace_id=` - // resolving to a workspace.root), agent-core's `listSessions({workDir})` - // walks a single wd-key bucket via readdir instead of scanning every - // workdir. Otherwise we list everything and apply downstream filters. const all = query.workDir !== undefined ? await this.core.rpc.listSessions({ workDir: query.workDir }) : await this.core.rpc.listSessions({}); - // Sort by createdAt desc per REST §1.6 "最近 N 条（按 created_at desc）". - const sorted = [...all].sort((a, b) => b.createdAt - a.createdAt); + const sorted = all.toSorted((a, b) => b.createdAt - a.createdAt); - // Cursor: anchor on id. before_id = older than that id; after_id = newer. - // Because the underlying list is desc, "older" = AFTER in the array. let pivotIndex = -1; if (query.before_id !== undefined) { pivotIndex = sorted.findIndex((s) => s.id === query.before_id); @@ -192,10 +151,8 @@ export class SessionService extends Disposable implements ISessionService { let slice: typeof sorted; if (query.before_id !== undefined && pivotIndex >= 0) { - // before_id = older entries → tail of the desc array, exclusive of pivot slice = sorted.slice(pivotIndex + 1); } else if (query.after_id !== undefined && pivotIndex >= 0) { - // after_id = newer entries → head of the desc array, exclusive of pivot slice = sorted.slice(0, pivotIndex); } else { slice = sorted; @@ -206,16 +163,10 @@ export class SessionService extends Disposable implements ISessionService { const pageSummaries = slice.slice(0, pageSize); const hasMore = slice.length > pageSize; - // Hydrate each summary with its metadata. We do these in parallel — - // `getSessionMetadata` is in-memory once the session is loaded, so the - // round-trip count is what matters, not bandwidth. const items = await Promise.all( pageSummaries.map(async (s) => toProtocolSession(s, await this.tryGetMeta(s.id))), ); - // Apply post-hydration status filter if requested. Today all sessions - // are mapped to 'idle' (see header note); the filter is wired now so the - // wire contract is stable when agent-core surfaces a real status enum. const filtered = query.status !== undefined ? items.filter((s) => s.status === query.status) : items; @@ -233,21 +184,16 @@ export class SessionService extends Disposable implements ISessionService { } async update(id: string, input: SessionUpdate): Promise { - // Existence check first — gives a deterministic 40401 if the id is wrong. const all = await this.core.rpc.listSessions({}); const summary = all.find((s) => s.id === id); if (summary === undefined) { throw new SessionNotFoundError(id); } - // 1) title goes through renameSession. if (input.title !== undefined) { await this.core.rpc.renameSession({ sessionId: id, title: input.title }); } - // 2) metadata patches go through updateSessionMetadata. agent-core's - // SessionMeta has top-level `title` + `custom`; we route protocol's - // `metadata` (catchall) into `custom` so it round-trips on the next get. const metadataPatch = input.metadata; if (metadataPatch !== undefined && Object.keys(metadataPatch).length > 0) { await this.core.rpc.updateSessionMetadata({ @@ -256,19 +202,9 @@ export class SessionService extends Disposable implements ISessionService { }); } - // 3) agent_config runtime controls — route the four fields (model, - // thinking, permission_mode, plan_mode) through the per-session - // shadow on `IPromptService.applyAgentState`. The helper diff- - // dispatches against the shadow and writes a dispatch-log entry - // with `source='meta'` for any setter that actually fires. - // Resolution is lazy via `IInstantiationService` to break the - // ctor cycle (`PromptService` already injects `ISessionService`). const ac = input.agent_config; if (ac !== undefined) { const patch: AgentStatePatch = {}; - // SessionService.update has long accepted `agent_config.model` and - // treated empty string as "no change" — we preserve that quirk by - // dropping the empty case before forwarding to the shadow. if (ac.model !== undefined && ac.model !== '') patch.model = ac.model; if (ac.thinking !== undefined) patch.thinking = ac.thinking; if (ac.permission_mode !== undefined) patch.permission_mode = ac.permission_mode; @@ -286,10 +222,6 @@ export class SessionService extends Disposable implements ISessionService { } } - // 4) permission_rules: no CoreAPI surface yet — we accept the input - // (schema-validated) but no CoreAPI surface exists to persist them yet. - - // Re-fetch to return the post-update Session. const allAfter = await this.core.rpc.listSessions({}); const summaryAfter = allAfter.find((s) => s.id === id) ?? summary; const meta = await this.tryGetMeta(id); @@ -374,7 +306,6 @@ export class SessionService extends Disposable implements ISessionService { } async getStatus(id: string): Promise { - // Existence check — same pattern as get() / update() / delete(). const all = await this.core.rpc.listSessions({}); const summary = all.find((s) => s.id === id); if (summary === undefined) { @@ -448,14 +379,12 @@ export class SessionService extends Disposable implements ISessionService { } async delete(id: string): Promise<{ deleted: true }> { - // Existence check — deterministic 40401 even on close. const all = await this.core.rpc.listSessions({}); const summary = all.find((s) => s.id === id); if (summary === undefined) { throw new SessionNotFoundError(id); } await this.core.rpc.closeSession({ sessionId: id }); - // Fire onDidClose listeners after the core RPC resolves. this._onDidClose.fire({ sessionId: id }); return { deleted: true }; } @@ -469,11 +398,6 @@ export class SessionService extends Disposable implements ISessionService { return summary; } - /** - * Pull a session's metadata; swallow errors (session may not be loaded into - * the active session map yet, in which case `sessionApi(id)` throws). The - * caller falls back to defaults from the summary alone. - */ private async tryGetMeta(id: string): Promise { try { const meta = await this.core.rpc.getSessionMetadata({ sessionId: id }); @@ -483,24 +407,10 @@ export class SessionService extends Disposable implements ISessionService { } } - // --- Per-domain event accessors ------------------------------------------- - // - // `onDidCreate` / `onDidClose` are declared above as - // `Emitter.event` getters; consumers subscribe via - // `svc.onDidCreate(handler)` (returns IDisposable) and own the - // detach lifetime through `Disposable._register(...)`. - override dispose(): void { - if (this._isDisposed) return; - // `_onDidCreate` and `_onDidClose` are registered via `this._register(...)`, - // so `super.dispose()` flushes their listeners. + if (this._store.isDisposed) return; super.dispose(); } } -// Self-register under the global singleton registry. Daemon-side bootstrap -// projects this through `defaultServicesModule()` / -// `getSingletonServiceDescriptors()`. All ctor deps are `@I…`-injected, so -// `staticArguments` is `[]`. `supportsDelayedInstantiation = false` preserves -// current reverse-dispose semantics. registerSingleton(ISessionService, SessionService, InstantiationType.Delayed); diff --git a/packages/services/test/coreProcessService.test.ts b/packages/services/test/coreProcessService.test.ts index 7780482e3..0f86a6130 100644 --- a/packages/services/test/coreProcessService.test.ts +++ b/packages/services/test/coreProcessService.test.ts @@ -1,18 +1,3 @@ -/** - * Acceptance: `CoreProcessService` wires peer services + KimiCore + RPC pair; - * `ready()` settles; `dispose()` short-circuits RPC; `defaultServicesModule()` - * composes with the DI container. - * - * Hermetic strategy: KimiCore wants a real HOME dir / config / Git Bash. We - * point it at an isolated tmp dir per test so it doesn't touch the user's - * `~/.kimi`. The `rpc` smoke uses a single round-trip (`getCoreInfo`) that - * doesn't require any external state — exercises the full RPC plumbing (core - * ← createRPC → BridgeClientAPI binding) without touching session/plugin/MCP - * code paths. createSession() smoke is harder to make hermetic because it - * spins up Kaos, hooks, and plugin discovery — we leave that to integration - * suites with daemon-side mocks. - */ - import { mkdtempSync, rmSync } from 'node:fs'; import { tmpdir } from 'node:os'; import { join } from 'node:path'; @@ -20,16 +5,16 @@ import { join } from 'node:path'; import { afterEach, beforeEach, describe, expect, it } from 'vitest'; import { - InstantiationService, - ServiceCollection, SyncDescriptor, Emitter, + getSingletonServiceDescriptors, type ApprovalRequest, type ApprovalResponse, type Event, type QuestionRequest, type QuestionResult, } from '@moonshot-ai/agent-core'; +import { TestInstantiationService } from '@moonshot-ai/agent-core/di/test'; import { BridgeClientAPI, @@ -39,11 +24,8 @@ import { IEventService, ICoreProcessService, IQuestionService, - defaultServicesModule, } from '../src'; -// --- Mock peer-service impls (per-test fresh instances) ---------------------- - class RecordingEventService implements IEventService { readonly _serviceBrand: undefined; @@ -92,8 +74,6 @@ class RecordingQuestionService implements IQuestionService { } } -// --- Sandbox HOME setup ------------------------------------------------------ - let tmpHome: string; let prevHome: string | undefined; @@ -112,7 +92,6 @@ afterEach(() => { try { rmSync(tmpHome, { recursive: true, force: true }); } catch { - // Best-effort cleanup; tmp dirs are auto-pruned. } }); @@ -132,8 +111,6 @@ function makeEnv(homeDir: string): IEnvironmentService { }; } -// --- Tests ------------------------------------------------------------------- - describe('BridgeClientAPI', () => { it('routes emitEvent / requestApproval / requestQuestion / toolCall to peer services', async () => { const { eventService, approvalService, questionService } = makePeers(); @@ -191,7 +168,6 @@ describe('CoreProcessService direct construction', () => { questionService, ); try { - // ready() resolves once the SDK side of the RPC pair has bound. await expect(core.ready()).resolves.toBeUndefined(); expect(typeof core.rpc.getCoreInfo).toBe('function'); } finally { @@ -210,9 +186,6 @@ describe('CoreProcessService direct construction', () => { ); try { await core.ready(); - // getCoreInfo is a pure read on KimiCore (no session/plugin state). It - // round-trips through the full createRPC pair (serialize → core → - // serialize back) — that's the in-process adapter smoke we care about. const info = await core.rpc.getCoreInfo({}); expect(info).toHaveProperty('version'); expect(typeof info.version).toBe('string'); @@ -232,40 +205,19 @@ describe('CoreProcessService direct construction', () => { ); await core.ready(); core.dispose(); - core.dispose(); // second call must be a no-op + core.dispose(); await expect(core.rpc.getCoreInfo({})).rejects.toThrow(/disposed/); }); - // Regression: prior to the BLOCKER fix the in-process adapter never - // forwarded a `resolveOAuthTokenProvider` into KimiCore. - // ProviderManager.resolveAuth then synthesized a closure that ALWAYS - // threw `auth.login_required` even after a successful device-code login. - // The daemon's `/auth` readiness probe (file-existence check) still said - // `ready:true`, so the failure only surfaced inside the prompt turn. - // Lock down that the adapter default-wires a resolver from the same - // home/config paths KimiCore consumes. it('default-wires a resolveOAuthTokenProvider when caller omits one', () => { const resolver = CoreProcessService._defaultOAuthTokenResolver(tmpHome, join(tmpHome, 'config.toml')); expect(typeof resolver).toBe('function'); - // Calling the resolver with the managed-kimi-code provider name must - // return an object exposing `getAccessToken`. We don't invoke it — - // there's no token on disk in this hermetic test — but the shape is - // sufficient to prove the adapter wired a real BearerTokenProvider - // factory (not the always-throw sentinel). const tokenProvider = resolver('managed:kimi-code'); expect(tokenProvider).toBeDefined(); expect(typeof tokenProvider?.getAccessToken).toBe('function'); }); - // Regression: prior to the identity-wiring fix the adapter never - // forwarded `kimiRequestHeaders` into KimiCore — the daemon-hosted - // KimiCore made upstream fetches with the Node default User-Agent, and - // the managed Kimi-for-Coding endpoint rejected with 40340 ("only - // available for Coding Agents such as Kimi CLI, …"). The in-process TUI - // path (`createKimiHarness`) was unaffected because `SDKRpcClient` - // already built these headers from `identity`. Lock down that the - // adapter does the same when given an `identity`. it('default-wires kimiRequestHeaders from identity when caller omits headers', () => { const headers = CoreProcessService._defaultKimiRequestHeaders( tmpHome, @@ -275,10 +227,6 @@ describe('CoreProcessService direct construction', () => { expect(headers!['User-Agent']).toMatch(/^kimi-code-cli\/9\.9\.9/); expect(headers!['X-Msh-Platform']).toBe('kimi_code_cli'); expect(headers!['X-Msh-Version']).toBe('9.9.9'); - // `createKimiDeviceId` mints + caches a per-machine UUID under - // `/device_id`. Assert the header exists (UUID shape, not a - // literal value — we tmp-isolate the home, so the value differs every - // run). expect(headers!['X-Msh-Device-Id']).toMatch( /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/, ); @@ -290,10 +238,6 @@ describe('CoreProcessService direct construction', () => { }); it('caller-supplied kimiRequestHeaders win over identity-derived defaults', () => { - // Sanity: when both are present the ctor takes `options.kimiRequestHeaders` - // first. We can't observe the KimiCore ctor arg directly without exposing - // it, so we just lock down the helper's precedence contract — the ctor's - // `??` chain depends on it. const explicit = { 'User-Agent': 'override/1.0' }; const picked = explicit ?? CoreProcessService._defaultKimiRequestHeaders( @@ -304,33 +248,24 @@ describe('CoreProcessService direct construction', () => { }); }); -describe('defaultServicesModule() composition', () => { +describe('singleton registry composition', () => { it('returns a CoreProcessService descriptor that composes with the DI container', async () => { const { eventService, approvalService, questionService } = makePeers(); - const moduleEntries = defaultServicesModule(); - // ICoreProcessService is the FIRST entry (its position matters because - // the daemon's start.ts uses createInstance + services.set on top of the - // descriptor — the order documents the canonical construction sequence). + const moduleEntries = getSingletonServiceDescriptors(); expect(moduleEntries.length).toBeGreaterThanOrEqual(1); expect(moduleEntries[0]![0]).toBe(ICoreProcessService); expect(moduleEntries[0]![1]).toBeInstanceOf(SyncDescriptor); - const services = new ServiceCollection( - // Spread module entries first so the test's explicit per-decorator - // overrides below win (last-write-wins in ServiceCollection). The - // module now self-registers `IEventService` too, so we need the - // fake `eventService` to land AFTER that descriptor entry. - ...moduleEntries.map(([id, desc]) => [id, desc] as const), - [IEventService, eventService], - [IApprovalService, approvalService], - [IQuestionService, questionService], - [IEnvironmentService, makeEnv(tmpHome)], - ); - const ix = new InstantiationService(services); + const ix = new TestInstantiationService(); + for (const [id, desc] of moduleEntries) { + ix.set(id, desc); + } + ix.stub(IEventService, eventService); + ix.stub(IApprovalService, approvalService); + ix.stub(IQuestionService, questionService); + ix.stub(IEnvironmentService, makeEnv(tmpHome)); try { - // createInstance with static options prefix; @IEnvironmentService and - // the three peer services auto-inject from the container. const core = ix.createInstance(CoreProcessService, {}); try { await core.ready(); diff --git a/packages/services/test/interfaces.test.ts b/packages/services/test/interfaces.test.ts index cbd48aedb..4d57065af 100644 --- a/packages/services/test/interfaces.test.ts +++ b/packages/services/test/interfaces.test.ts @@ -1,18 +1,9 @@ -/** - * Acceptance: the three peer-service decorators (`IEventService`, - * `IApprovalService`, `IQuestionService`) are typed correctly, can be - * registered in a `ServiceCollection`, resolved through - * `InstantiationService`, and surface their diagnostic names in - * not-registered errors. - */ - import { describe, expect, it, vi } from 'vitest'; import { - InstantiationService, - ServiceCollection, Emitter, } from '@moonshot-ai/agent-core'; +import { TestInstantiationService } from '@moonshot-ai/agent-core/di/test'; import type { ApprovalRequest, Event, QuestionRequest } from '@moonshot-ai/agent-core'; import { @@ -72,7 +63,6 @@ class FakeQuestionService implements IQuestionService { } function makeFakeEvent(): Event { - // Minimal AgentStatusUpdatedEvent shape — the union narrows by `type`. return { type: 'agent_status_updated', sessionId: 'sess-1', @@ -106,117 +96,81 @@ function makeFakeQuestion(): QuestionRequest & { sessionId: string; agentId: str } describe('@moonshot-ai/services · interfaces', () => { - it('registers all three peer services in a ServiceCollection and resolves them through InstantiationService', () => { + it('registers all three peer services in a test instantiation service', () => { const events = new FakeEventService(); const approvals = new FakeApprovalService(); const questions = new FakeQuestionService(); - const services = new ServiceCollection( - [IEventService, events], - [IApprovalService, approvals], - [IQuestionService, questions], - ); - const ix = new InstantiationService(services); + const ix = new TestInstantiationService(); + ix.stub(IEventService, events); + ix.stub(IApprovalService, approvals); + ix.stub(IQuestionService, questions); - try { - ix.invokeFunction((accessor) => { - expect(accessor.get(IEventService)).toBe(events); - expect(accessor.get(IApprovalService)).toBe(approvals); - expect(accessor.get(IQuestionService)).toBe(questions); - }); - } finally { - ix.dispose(); - } + expect(ix.get(IEventService)).toBe(events); + expect(ix.get(IApprovalService)).toBe(approvals); + expect(ix.get(IQuestionService)).toBe(questions); }); - it('end-to-end smoke: invokes service methods via the accessor', async () => { + it('end-to-end smoke: invokes service methods through the test container', async () => { const events = new FakeEventService(); const approvals = new FakeApprovalService(); const questions = new FakeQuestionService(); - const services = new ServiceCollection( - [IEventService, events], - [IApprovalService, approvals], - [IQuestionService, questions], - ); - const ix = new InstantiationService(services); + const ix = new TestInstantiationService(); + ix.stub(IEventService, events); + ix.stub(IApprovalService, approvals); + ix.stub(IQuestionService, questions); - try { - const event = makeFakeEvent(); - ix.invokeFunction((a) => a.get(IEventService).publish(event)); - expect(events.events).toEqual([event]); + const event = makeFakeEvent(); + ix.get(IEventService).publish(event); + expect(events.events).toEqual([event]); - const approval = makeFakeApproval(); - const approvalResp = await ix.invokeFunction((a) => - a.get(IApprovalService).request(approval), - ); - expect(approvalResp).toEqual({ decision: 'approved' }); - expect(approvals.received).toHaveLength(1); + const approval = makeFakeApproval(); + const approvalResp = await ix.get(IApprovalService).request(approval); + expect(approvalResp).toEqual({ decision: 'approved' }); + expect(approvals.received).toHaveLength(1); - const question = makeFakeQuestion(); - const questionResp = await ix.invokeFunction((a) => - a.get(IQuestionService).request(question), - ); - expect(questionResp).toBeNull(); - expect(questions.received).toHaveLength(1); - } finally { - ix.dispose(); - } + const question = makeFakeQuestion(); + const questionResp = await ix.get(IQuestionService).request(question); + expect(questionResp).toBeNull(); + expect(questions.received).toHaveLength(1); }); it('resolve/dismiss service methods are wired through the same DI value', () => { const approvals = new FakeApprovalService(); const questions = new FakeQuestionService(); - const services = new ServiceCollection( - [IApprovalService, approvals], - [IQuestionService, questions], - ); - const ix = new InstantiationService(services); + const ix = new TestInstantiationService(); + ix.stub(IApprovalService, approvals); + ix.stub(IQuestionService, questions); - try { - ix.invokeFunction((a) => { - a.get(IApprovalService).resolve('tc-1', { decision: 'rejected', feedback: 'no' }); - a.get(IQuestionService).resolve('q-1', { answers: { q_1: 'A' } }); - a.get(IQuestionService).dismiss('q-2'); - }); + ix.get(IApprovalService).resolve('tc-1', { decision: 'rejected', feedback: 'no' }); + ix.get(IQuestionService).resolve('q-1', { answers: { q_1: 'A' } }); + ix.get(IQuestionService).dismiss('q-2'); - expect(approvals.resolveCalls).toEqual([ - { id: 'tc-1', response: { decision: 'rejected', feedback: 'no' } }, - ]); - expect(questions.resolveCalls).toEqual([ - { id: 'q-1', response: { answers: { q_1: 'A' } } }, - ]); - expect(questions.dismissCalls).toEqual(['q-2']); - } finally { - ix.dispose(); - } + expect(approvals.resolveCalls).toEqual([ + { id: 'tc-1', response: { decision: 'rejected', feedback: 'no' } }, + ]); + expect(questions.resolveCalls).toEqual([ + { id: 'q-1', response: { answers: { q_1: 'A' } } }, + ]); + expect(questions.dismissCalls).toEqual(['q-2']); }); it('looking up an unregistered service returns undefined in non-strict mode', () => { - const ix = new InstantiationService(new ServiceCollection()); - try { - expect(ix.invokeFunction((a) => a.get(IEventService))).toBeUndefined(); - expect(ix.invokeFunction((a) => a.get(IApprovalService))).toBeUndefined(); - expect(ix.invokeFunction((a) => a.get(IQuestionService))).toBeUndefined(); - } finally { - ix.dispose(); - } + const ix = new TestInstantiationService(); + expect(ix.get(IEventService)).toBeUndefined(); + expect(ix.get(IApprovalService)).toBeUndefined(); + expect(ix.get(IQuestionService)).toBeUndefined(); }); it('IEventService / IApprovalService / IQuestionService are callable ServiceIdentifiers (compile-time guard)', () => { - // The const half of the dual export must be usable as a ServiceCollection - // key and as a `createDecorator` brand value. We exercise both at runtime - // to also catch any accidental swap of the value with the type. expect(typeof IEventService).toBe('function'); expect(typeof IApprovalService).toBe('function'); expect(typeof IQuestionService).toBe('function'); - // Avoid an unused-import warning on the type-only re-export. const _typeProbe: ApprovalResponse | QuestionResult = null; void _typeProbe; - // And use vi to keep the import surface (helpful when running with strict - // unused-imports lints in the future). vi.fn(); }); }); diff --git a/packages/services/test/session-service.test.ts b/packages/services/test/session-service.test.ts index ac8256a3f..ca3f730b6 100644 --- a/packages/services/test/session-service.test.ts +++ b/packages/services/test/session-service.test.ts @@ -1,19 +1,3 @@ -/** - * `SessionService` (Chain 2 / P1.2) unit tests. - * - * Hermetic: we mock `ICoreProcessService` with an in-memory `rpc` proxy whose - * methods return controllable promises. No KimiCore, no agent-core RPC pair - * — the adapter is exercised against a fake bridge. - * - * Test cases cover: - * - create → toProtocolSession (camelCase ↔ snake_case + number → ISO) - * - list pagination (default/before_id/after_id/page_size; has_more) - * - get + SessionNotFoundError → 40401 mapping at the daemon layer - * - update (title-only / metadata-only / both / empty) - * - delete returning {deleted: true} - * - toProtocolSession field defaults for fields agent-core doesn't surface - */ - import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; import { @@ -23,15 +7,13 @@ import { type CreateSessionPayload, Emitter, type ForkSessionPayload, - type IInstantiationService, type RenameSessionPayload, type ResumeSessionResult, - type ServiceIdentifier, - type ServicesAccessor, type SessionMeta, type SessionSummary, type UpdateSessionMetadataPayload, } from '@moonshot-ai/agent-core'; +import { TestInstantiationService } from '@moonshot-ai/agent-core/di/test'; import { emptySessionUsage, type Session } from '@moonshot-ai/protocol'; import { @@ -63,11 +45,6 @@ interface FakeBridgeState { postUndoContexts: Map; } -/** - * Build a tiny fake `ICoreProcessService` whose `rpc` proxy implements just the - * five session methods the impl uses. Each method delegates to an in-memory - * state object the test owns. - */ function makeFakeBridge(state: FakeBridgeState): ICoreProcessService { const rpc: Partial = { createSession: vi @@ -152,8 +129,6 @@ function makeFakeBridge(state: FakeBridgeState): ICoreProcessService { .fn() .mockImplementation(async (payload: WithSessionId) => { state.renamedTitles.set(payload.sessionId, payload.title); - // Reflect into the metadata map so subsequent `getSessionMetadata` - // returns the updated title (mirrors real KimiCore behavior). const existing = state.metas.get(payload.sessionId); if (existing !== undefined) { state.metas.set(payload.sessionId, { ...existing, title: payload.title }); @@ -257,15 +232,8 @@ function textMessage( let state: FakeBridgeState; let svc: SessionService; let promptStub: ReturnType; +let instantiation: TestInstantiationService; -/** - * Stub `IPromptService` for hermetic SessionService tests. Records every - * `applyAgentState(sid, patch, source)` call so tests can assert that - * `SessionService.update` forwards `agent_config` runtime fields through - * the shared shadow-aware helper rather than dispatching `core.rpc.*` - * directly. The other `IPromptService` methods aren't reachable from - * SessionService and are stubbed to throw on access. - */ function makePromptServiceStub(): { promptService: IPromptService; calls: Array<{ sid: string; patch: Record; source: string; promptId: string | undefined }>; @@ -290,45 +258,27 @@ function makePromptServiceStub(): { return { promptService, calls }; } -/** - * Fake `IInstantiationService` that only resolves the one service - * `SessionService.update` reaches for (`IPromptService`). Other lookups - * throw — they would indicate an unintended dependency creeping in. - */ -function makeFakeInstantiation(stubs: { +function makeTestInstantiation(stubs: { promptService: IPromptService; -}): IInstantiationService { - const accessor: ServicesAccessor = { - get: (id: ServiceIdentifier): T => { - if ((id as unknown) === (IPromptService as unknown)) { - return stubs.promptService as unknown as T; - } - throw new Error(`unexpected service lookup: ${String((id as unknown as { toString(): string }).toString())}`); - }, - }; - return { - _serviceBrand: undefined, - invokeFunction: (fn: (a: ServicesAccessor) => R): R => fn(accessor), - createInstance: (() => { - throw new Error('createInstance not supported in this test stub'); - }) as IInstantiationService['createInstance'], - createChild: () => { - throw new Error('createChild not supported in this test stub'); - }, - } as unknown as IInstantiationService; +}): TestInstantiationService { + const ix = new TestInstantiationService(undefined, true); + ix.stub(IPromptService, stubs.promptService); + return ix; } beforeEach(() => { state = freshState(); promptStub = makePromptServiceStub(); + instantiation = makeTestInstantiation({ promptService: promptStub.promptService }); svc = new SessionService( makeFakeBridge(state), - makeFakeInstantiation({ promptService: promptStub.promptService }), + instantiation, ); }); afterEach(() => { svc.dispose(); + instantiation.dispose(); }); describe('toProtocolSession adapter', () => { @@ -458,7 +408,6 @@ describe('SessionService.create', () => { expect(state.sessions).toHaveLength(1); expect(state.sessions[0]!.workDir).toBe('/tmp/foo'); expect(session.metadata.cwd).toBe('/tmp/foo'); - // title is echoed back even when CoreAPI doesn't reflect it (gap doc). expect(session.title).toBe('My session'); expect(session.created_at.endsWith('Z')).toBe(true); }); @@ -468,9 +417,7 @@ describe('SessionService.create', () => { metadata: { cwd: '/tmp/x' }, agent_config: { model: 'moonshot-v1-128k' }, }); - const created = state.sessions[0]!; - expect((state.sessions as SessionSummary[])[0]!.metadata?.['cwd']).toBe('/tmp/x'); - void created; + expect(state.sessions[0]!.metadata?.['cwd']).toBe('/tmp/x'); }); it('rejects when metadata.cwd is absent (daemon route must pre-resolve workspace_id → cwd)', async () => { @@ -482,7 +429,6 @@ describe('SessionService.create', () => { describe('SessionService.list', () => { beforeEach(async () => { - // Seed 3 sessions in increasing createdAt order. await svc.create({ metadata: { cwd: '/tmp/a' } }); await svc.create({ metadata: { cwd: '/tmp/b' } }); await svc.create({ metadata: { cwd: '/tmp/c' } }); @@ -504,20 +450,19 @@ describe('SessionService.list', () => { it('before_id returns older sessions only', async () => { const all = await svc.list({}); - const pivotId = all.items[0]!.id; // newest + const pivotId = all.items[0]!.id; const olderPage = await svc.list({ before_id: pivotId }); expect(olderPage.items.map((s) => s.metadata.cwd)).toEqual(['/tmp/b', '/tmp/a']); }); it('after_id returns newer sessions only', async () => { const all = await svc.list({}); - const pivotId = all.items[2]!.id; // oldest + const pivotId = all.items[2]!.id; const newerPage = await svc.list({ after_id: pivotId }); expect(newerPage.items.map((s) => s.metadata.cwd)).toEqual(['/tmp/c', '/tmp/b']); }); it('status filter applies post-hydration', async () => { - // Today everything maps to 'idle'; non-matching filter returns [] const empty = await svc.list({ status: 'running' }); expect(empty.items).toEqual([]); const idle = await svc.list({ status: 'idle' }); @@ -569,7 +514,6 @@ describe('SessionService.update', () => { it('routes title through bridge.rpc.renameSession', async () => { await svc.update(created.id, { title: 'Renamed' }); expect(state.renamedTitles.get(created.id)).toBe('Renamed'); - // Title is reflected via the next get (impl re-fetches metadata). expect(state.metadataPatches.has(created.id)).toBe(false); }); diff --git a/packages/services/vitest.config.ts b/packages/services/vitest.config.ts index faf44f122..31d5dce74 100644 --- a/packages/services/vitest.config.ts +++ b/packages/services/vitest.config.ts @@ -8,9 +8,6 @@ export default defineConfig({ plugins: [rawTextPlugin()], resolve: { alias: [ - // Order matters — list MORE specific entries first so prefix matching - // doesn't route them through the bare `@moonshot-ai/agent-core` alias - // (which points at agent-core/src/index.ts, breaking subpath imports). { find: /^@moonshot-ai\/agent-core\/session\/store$/, replacement: fileURLToPath( @@ -23,6 +20,12 @@ export default defineConfig({ new URL('../agent-core/src/base/common/event.ts', import.meta.url), ), }, + { + find: /^@moonshot-ai\/agent-core\/di\/test$/, + replacement: fileURLToPath( + new URL('../agent-core/src/di/test.ts', import.meta.url), + ), + }, { find: '@moonshot-ai/kimi-code-sdk', replacement: fileURLToPath(