feat(llm): topology event and configurable-provider directory

ctx.llm gains 'llm/adapters-updated' — a payload-free registry notification
emitted at every topology commit point (adapter routes registering or
disposing, directory entries appearing or withdrawing) with contained
observers and INVARIANT rethrow — plus registerConfigurableProviders/
listConfigurableProviders, the directory of routes an adapter plugin can
activate through configuration. llm-deepseek declares deepseek-official
(whole llm-deepseek section as profile); llm-pi-ai declares the full
installed catalog under providers.<provider> even while dormant, so the
web settings surface can offer every provider before any route exists.
The invariant companion asserts the registry stays readable at each
notification.
This commit is contained in:
Yichen Jiang
2026-07-29 16:45:06 +08:00
parent fee12f1af0
commit 4989494e75
12 changed files with 335 additions and 2 deletions
+2
View File
@@ -54,6 +54,8 @@ Connection facts are not frozen at load. `resolveAdapterOptions` is the one expl
The one registration-captured fact is the retry policy: when its resolved value changes, the plugin re-registers the route in place (same adapter instance, one synchronous section), so `ctx.llm.providerRetryPolicy('deepseek-official')` always reports the current policy.
The plugin also declares its route in the configurable-provider directory (`ctx.llm.listConfigurableProviders()`): provider `deepseek-official`, settings namespace `llm-deepseek`, empty settings path — the whole section is the profile. Configuration surfaces use that entry to offer this adapter alongside dormant pi-ai providers.
## App attribution
Every request carries the shared attribution header from dsh-llm's `attributionHeaders()` - the mandatory `User-Agent` baseline identifying the harness (see [dsh-llm § App attribution](../llm/README.md#app-attribution-attributionts)). Direct DeepSeek requests and OpenAI-compatible gateway requests get no provider-specific app-attribution headers under this adapter contract; OpenRouter app attribution is deferred to a future explicit OpenRouter adapter or mode. A request whose `GenerateOptions.purpose` is `compaction` (dsh-compact-basic's auxiliary summarization call) additionally carries `x-deepseek-harness-compact: 1`, so the host can separate compaction traffic from conversation requests.
+3
View File
@@ -209,6 +209,9 @@ export function apply(ctx: Context, config: Config): void {
}
const adapter = new DeepSeekAdapter({ options, resolveApiKey })
ctx.llm.registerConfigurableProviders([
{ provider: 'deepseek-official', displayName: 'DeepSeek', settingsNs: NS, settingsPath: [] },
])
// Route effects bind to this apply fiber via the stable `ctx` reference,
// even when a swap runs inside the scoped settings callback below.
let disposeRoute = ctx.llm.registerAdapter(['deepseek-official'], adapter)
@@ -555,8 +555,15 @@ describe('plugin registration and config', () => {
baseURL: server.url,
})
expect(ctx.llm.listProviders()).toEqual([{ id: 'deepseek-official', name: 'DeepSeek' }])
expect(ctx.llm.listConfigurableProviders()).toEqual([{
provider: 'deepseek-official',
displayName: 'DeepSeek',
settingsNs: 'llm-deepseek',
settingsPath: [],
}])
await fiber.dispose()
expect(ctx.llm.listProviders()).toEqual([])
expect(ctx.llm.listConfigurableProviders()).toEqual([])
})
it('registers retryPolicy from the provider config', async () => {
+1 -1
View File
@@ -35,7 +35,7 @@ Configure credentials and deployment-specific transport settings per provider, k
X-Deployment: production
```
Each dict key must exist in pi-ai's installed catalog; the dict shape makes duplicates unrepresentable, and the pre-release array shape (with per-profile `provider` fields) fails load with migration directions. `providers` may also be empty or omitted entirely: the adapter then mounts **dormant** — zero routes, no extra catalog entries — and registers routes the moment the `llm-pi-ai:` settings section supplies profiles, dropping them again when it empties. Which adapters exist is composition; which providers run can be entirely the user's settings document. Registration with `ctx.llm` is atomic: a collision with any provider route already owned by another adapter fails plugin loading without registering the remaining routes. Model ids are not lifecycle config; an unknown model fails before any provider request with `LlmError('UNKNOWN_MODEL')`.
Each dict key must exist in pi-ai's installed catalog; the dict shape makes duplicates unrepresentable, and the pre-release array shape (with per-profile `provider` fields) fails load with migration directions. `providers` may also be empty or omitted entirely: the adapter then mounts **dormant** — zero routes, no extra catalog entries — and registers routes the moment the `llm-pi-ai:` settings section supplies profiles, dropping them again when it empties. Dormant or not, the plugin declares every installed catalog provider in the configurable-provider directory (`ctx.llm.listConfigurableProviders()`, settings path `providers.<provider>`), so configuration surfaces can offer the full catalog before any route exists. Which adapters exist is composition; which providers run can be entirely the user's settings document. Registration with `ctx.llm` is atomic: a collision with any provider route already owned by another adapter fails plugin loading without registering the remaining routes. Model ids are not lifecycle config; an unknown model fails before any provider request with `LlmError('UNKNOWN_MODEL')`.
## Dynamic configuration (settings + credentials)
+10
View File
@@ -29,6 +29,7 @@
*/
import type { Context } from 'cordis'
import { getBuiltinProviders } from '@earendil-works/pi-ai/providers/all'
import type {} from '@deepseek-ai/dsh-llm'
import { deepEqualJson, installSettingsSection, settingsNamespace } from '@deepseek-ai/dsh-settings'
import { PiAiAdapter } from './adapter.ts'
@@ -90,6 +91,15 @@ export function apply(ctx: Context, config: Config): void {
}
const adapter = new PiAiAdapter({ profiles, resolveApiKey })
// The full installed catalog is configurable from the moment the plugin
// mounts — dormant or not — so configuration surfaces can offer every
// pi-ai provider before any route exists.
ctx.llm.registerConfigurableProviders(getBuiltinProviders().map(provider => ({
provider,
displayName: provider,
settingsNs: NS,
settingsPath: ['providers', provider],
})))
// Route effects bind to this apply fiber via the stable `ctx` reference,
// even when a swap runs inside the scoped settings callback below. A bare
// mount (zero routes) is the dormant posture: nothing registers until a
@@ -51,6 +51,16 @@ describe('request-level dynamic profiles', () => {
const ctx = await boot(dir, {})
expect(ctx.llm.listProviders()).toEqual([])
// Dormant ≠ invisible: every installed catalog provider is configurable
// before any route exists, each addressed inside the providers dict.
const directory = ctx.llm.listConfigurableProviders()
expect(directory.length).toBeGreaterThan(30)
expect(directory).toContainEqual({
provider: 'openai',
displayName: 'openai',
settingsNs: 'llm-pi-ai',
settingsPath: ['providers', 'openai'],
})
await ctx.settings.update(NS, {
providers: { deepseek: { apiKeyEnv: 'PI_DYNAMIC_KEY', baseURL: server.url } },
})
+4
View File
@@ -12,6 +12,8 @@ An adapter registry plus a single streaming call surface, interceptable via a wa
- `ctx.llm.registerAdapter(providers: string[], adapter: LlmAdapter): () => void` Register one adapter instance for the given provider routes. Registration is all-or-nothing, and is disposed with the calling fiber.
- `ctx.llm.listProviders(): LlmProviderInfo[]` Describe registered provider routes in registration order.
- `ctx.llm.registerConfigurableProviders(entries: readonly LlmConfigurableProvider[]): () => void` Declare provider routes an adapter plugin can activate through configuration — registered or dormant — each naming its owning settings namespace and the path to its profile inside that section. All-or-nothing (`INVALID_DIRECTORY`/`DUPLICATE_DIRECTORY`), disposed with the calling fiber.
- `ctx.llm.listConfigurableProviders(): LlmConfigurableProvider[]` List the declared directory in declaration order; configuration surfaces merge it with `listProviders()` to mark each entry live or dormant.
- `ctx.llm.providerRetryPolicy(provider: string): ResolvedRetryPolicy` Return the provider-owned retry policy captured during registration, with normal defaults resolved.
- `ctx.llm.listModels(provider: string): Promise<LlmModelInfo[]>` Discover the models one registered provider currently advertises.
- `ctx.llm.resolveModelInfo(provider: string, model: string, signal?: AbortSignal): Promise<LlmResolvedModelInfo>` Resolve validated exact-model identity plus available context and reasoning metadata from the owning adapter, with optional cancellation for asynchronous adapters.
@@ -23,6 +25,8 @@ An adapter registry plus a single streaming call surface, interceptable via a wa
Provider and model metadata is a discovery surface, not a routing whitelist. `registerAdapter()` still owns provider exclusivity and captures the adapter's retry policy for each route, while an adapter may accept model ids absent from `listModels()`; consumers must not reject a request because its model is unlisted. Returned selector metadata is detached and invalid or duplicate adapter entries fail with `INVALID_ADAPTER` or `INVALID_CATALOG`.
Every topology commit point — adapter routes registering or disposing, directory entries appearing or withdrawing — emits the payload-free `llm/adapters-updated` event after the mutation, so consumers re-read `listProviders()`/`listModels()`/`listConfigurableProviders()` instead of polling. Observer failures are contained (logged, non-vetoing); only `INVARIANT`-coded failures rethrow after the fan-out.
Exact-model metadata is a separate correctness query, not a catalog decoration or global LLM setting. `resolveModelInfo()` asks the adapter that owns the exact provider/model route once; an adapter can describe an unlisted dynamic model, and absent `context` or `reasoning` fields mean only that those capabilities are unavailable. Invalid identity, context, or reasoning metadata fails with `INVALID_MODEL_INFO`, `INVALID_MODEL_CONTEXT`, or `INVALID_MODEL_REASONING`.
Reasoning identifiers are opaque adapter-owned strings rather than a core enum. An adapter publishes its ordered selectable list, including an `off` id when that model's capability API exposes one. `resolveCallConfig()` accepts only an exact advertised identifier, materializes `defaultEffort` when present, and otherwise preserves the provider default. Asynchronous model resolvers receive the caller's signal and must settle promptly after cancellation. `prepareCall()` additionally retains the exact adapter registration through header logging and terminal dispatch, so HMR cannot combine one adapter's capability result with another adapter's request; reusing its one-shot handle or changing its call-config fields fails with `INVALID_PREPARED_CALL`. An unsupported explicit or configured effort fails with `UNSUPPORTED_REASONING_EFFORT` before provider I/O.
+80
View File
@@ -9,6 +9,7 @@
import { Context, Service } from 'cordis'
import type {
GenerateOptions,
LlmConfigurableProvider,
LlmFailure,
LlmModelInfo,
LlmResolvedModelInfo,
@@ -56,6 +57,17 @@ declare module 'cordis' {
* @mode waterfall
*/
'llm/stream'(this: LlmService, options: GenerateOptions, next: () => AsyncIterable<StreamChunk>): AsyncIterable<StreamChunk>
/**
* The provider topology changed: an adapter registered or unregistered
* routes, or the configurable-provider directory gained or lost entries.
* This is a payload-free registry notification fired at each commit point
* (including registration disposal); consumers re-read `listProviders()`,
* `listModels()`, or `listConfigurableProviders()` for the new state.
* Observer failures are contained and cannot veto the registry mutation.
* @mode emit
*/
'llm/adapters-updated'(): void
}
}
@@ -190,11 +202,33 @@ export abstract class LlmAdapter {
*/
export class LlmService extends Service {
private adapters = new Map<string, AdapterRegistration>()
private directory = new Map<string, LlmConfigurableProvider>()
constructor(ctx: Context) {
super(ctx, 'llm')
}
/** Notify topology observers without letting one broken listener veto the commit. */
private emitAdaptersUpdated(): void {
// Cordis emit uses Array.map: one synchronous throw starves later
// listeners. Registry notifications are non-vetoing, so contain each
// callback independently; INVARIANT-coded failures still surface.
let invariantFailure: unknown
for (const listener of this.ctx.events.dispatch('emit', ['llm/adapters-updated']) as Array<() => unknown>) {
try {
listener()
} catch (error) {
if ((error as { code?: unknown } | null)?.code === 'INVARIANT') {
invariantFailure ??= error
continue
}
this.ctx.logger.warn('llm: an llm/adapters-updated listener failed')
this.ctx.logger.warn(error)
}
}
if (invariantFailure !== undefined) throw invariantFailure as Error
}
/**
* Register an adapter for the given provider routes. Throws `LlmError` with code
* `DUPLICATE_ADAPTER` if any provider already has an adapter (all-or-nothing).
@@ -227,8 +261,10 @@ export class LlmService extends Service {
})
}
for (const registration of registrations) this.adapters.set(registration.provider.id, registration)
this.emitAdaptersUpdated()
yield () => {
for (const provider of providers) this.adapters.delete(provider)
this.emitAdaptersUpdated()
}
}.bind(this), 'llm.registerAdapter()')
// ctx.effect's disposer returns Promise<void>; our disposer API is
@@ -244,6 +280,50 @@ export class LlmService extends Service {
return [...this.adapters.values()].map(({ provider }) => ({ ...provider }))
}
/**
* Declare provider routes an adapter plugin can activate through
* configuration. Registration is all-or-nothing: an empty list, invalid
* entry, or a provider already declared by any registration throws
* `LlmError` without registering the rest. Disposed with the fiber.
* @param entries - every configurable provider this plugin owns.
* @returns the disposer that withdraws all of them.
*/
registerConfigurableProviders(entries: readonly LlmConfigurableProvider[]): () => void {
const dispose = this.ctx.effect(function* (this: LlmService) {
if (entries.length === 0) {
throw new LlmError('a configurable-provider registration must declare at least one provider', 'INVALID_DIRECTORY')
}
const detached: LlmConfigurableProvider[] = []
for (const entry of entries) {
if (entry.provider.length === 0 || entry.displayName.length === 0 || entry.settingsNs.length === 0) {
throw new LlmError('configurable providers need a non-empty provider, displayName, and settingsNs', 'INVALID_DIRECTORY')
}
if (entry.settingsPath.some(segment => segment.length === 0)) {
throw new LlmError(`configurable provider "${entry.provider}" has an empty settingsPath segment`, 'INVALID_DIRECTORY')
}
if (this.directory.has(entry.provider) || detached.some(seen => seen.provider === entry.provider)) {
throw new LlmError(`configurable provider "${entry.provider}" is already declared`, 'DUPLICATE_DIRECTORY')
}
detached.push({ ...entry, settingsPath: [...entry.settingsPath] })
}
for (const entry of detached) this.directory.set(entry.provider, entry)
this.emitAdaptersUpdated()
yield () => {
for (const entry of detached) this.directory.delete(entry.provider)
this.emitAdaptersUpdated()
}
}.bind(this), 'llm.registerConfigurableProviders()')
return () => void dispose()
}
/**
* List every declared configurable provider, registered or dormant.
* @returns detached directory entries in declaration order.
*/
listConfigurableProviders(): LlmConfigurableProvider[] {
return [...this.directory.values()].map(entry => ({ ...entry, settingsPath: [...entry.settingsPath] }))
}
/**
* Resolve the retry policy captured when one provider route was registered.
* @param provider - registered provider route to inspect.
+15
View File
@@ -84,6 +84,21 @@ async function* validateStream(
/** Install validation around every provider stream. */
const install: InvariantInstaller = (ctx, fail) => {
ctx.on('llm/stream', (_options, next) => validateStream(next(), fail), { global: true, prepend: true })
ctx.on('llm/adapters-updated', () => {
// A disposer-time emit can outlive the service-store entry during whole-
// context teardown; only a live service promises a readable registry.
const llm = ctx.get('llm')
if (llm === undefined) return
for (const provider of llm.listProviders()) {
try {
llm.providerRetryPolicy(provider.id)
} catch {
// Reaching here IS the violation: the notification promised a readable
// registry, and only that broken promise can make the lookup throw.
fail(`llm/adapters-updated fired while provider "${provider.id}" has no readable registration`)
}
}
}, { global: true })
}
/**
+20
View File
@@ -119,6 +119,26 @@ export interface LlmProviderInfo {
name: string
}
/**
* One provider route an adapter plugin can activate through configuration,
* whether or not the route is currently registered. Configuration surfaces
* merge this directory with `listProviders()` to offer every configurable
* provider alongside its live/dormant state.
*/
export interface LlmConfigurableProvider {
/** Provider route key this entry activates when configured. */
provider: string
/** Human-readable provider name for configuration surfaces. */
displayName: string
/** User-settings namespace whose section configures this provider. */
settingsNs: string
/**
* Path from that namespace's section root to this provider's profile
* object; empty when the whole section is the profile.
*/
settingsPath: readonly string[]
}
/** One adapter-discovered model; catalog membership is advisory, not request validation. */
export interface LlmModelInfo {
/** Provider route that owns this model entry. */
+38 -1
View File
@@ -1,6 +1,6 @@
import { describe, expect, it } from 'vitest'
import { Context } from 'cordis'
import { CallId } from '@deepseek-ai/dsh-llm'
import LlmService, { CallId, LlmAdapter } from '@deepseek-ai/dsh-llm'
import type { GenerateOptions, StreamChunk } from '@deepseek-ai/dsh-llm'
import * as LlmInvariant from '@deepseek-ai/dsh-llm/invariant'
import InvariantService from '@deepseek-ai/dsh-invariants'
@@ -84,3 +84,40 @@ describe('LLM stream invariants', () => {
})()).rejects.toThrow('provider failed')
})
})
describe('adapters-updated invariants', () => {
class NoopAdapter extends LlmAdapter {
async * stream(_options: GenerateOptions): AsyncIterable<StreamChunk> {
throw new Error('not exercised')
}
}
it('accepts a coherent registry at every topology notification', async () => {
const ctx = await setup()
await ctx.plugin(LlmService)
const dispose = ctx.llm.registerAdapter(['coherent'], new NoopAdapter())
ctx.llm.registerConfigurableProviders([
{ provider: 'dormant', displayName: 'Dormant', settingsNs: 'ns', settingsPath: [] },
])
dispose()
expect(ctx.llm.listProviders()).toEqual([])
})
it('skips the check when the service store has no llm entry', async () => {
const ctx = await setup()
expect(() => { ctx.emit('llm/adapters-updated') }).not.toThrow()
})
it('reports a notification whose registry cannot be re-read', async () => {
class BrokenLlm extends LlmService {
override providerRetryPolicy(_provider: string): never {
throw new Error('registration vanished')
}
}
const ctx = await setup()
await ctx.plugin(BrokenLlm)
expect(() => ctx.llm.registerAdapter(['ghost'], new NoopAdapter()))
.toThrow(/no readable registration/)
})
})
+145
View File
@@ -0,0 +1,145 @@
import { describe, expect, it, vi } from 'vitest'
import { Context } from 'cordis'
import LlmService, { LlmAdapter, LlmError } from '@deepseek-ai/dsh-llm'
import type { GenerateOptions, LlmConfigurableProvider, StreamChunk } from '@deepseek-ai/dsh-llm'
class NoopAdapter extends LlmAdapter {
async * stream(_options: GenerateOptions): AsyncIterable<StreamChunk> {
throw new Error('not exercised')
}
}
async function setup(): Promise<Context> {
const ctx = new Context()
await ctx.plugin(LlmService)
return ctx
}
function entry(overrides: Partial<LlmConfigurableProvider> = {}): LlmConfigurableProvider {
return {
provider: 'openai',
displayName: 'OpenAI',
settingsNs: 'llm-pi-ai',
settingsPath: ['providers', 'openai'],
...overrides,
}
}
describe('llm/adapters-updated', () => {
it('fires at both adapter registration commit points with the registry already readable', async () => {
const ctx = await setup()
const observed: string[][] = []
ctx.on('llm/adapters-updated', () => {
observed.push(ctx.llm.listProviders().map(provider => provider.id))
})
const dispose = ctx.llm.registerAdapter(['a', 'b'], new NoopAdapter())
expect(observed).toEqual([['a', 'b']])
dispose()
expect(observed).toEqual([['a', 'b'], []])
})
it('contains a throwing listener without vetoing registration or starving later listeners', async () => {
const ctx = await setup()
const warn = vi.spyOn(ctx.logger, 'warn').mockImplementation(() => undefined)
const later = vi.fn()
ctx.on('llm/adapters-updated', () => {
throw new Error('broken observer')
})
ctx.on('llm/adapters-updated', later)
ctx.llm.registerAdapter(['a'], new NoopAdapter())
expect(ctx.llm.listProviders().map(provider => provider.id)).toEqual(['a'])
expect(later).toHaveBeenCalledTimes(1)
expect(warn).toHaveBeenCalledWith('llm: an llm/adapters-updated listener failed')
})
it('rethrows the first INVARIANT-coded listener failure after notifying the rest', async () => {
const ctx = await setup()
const later = vi.fn()
ctx.on('llm/adapters-updated', () => {
throw Object.assign(new Error('registry incoherent'), { code: 'INVARIANT' })
})
ctx.on('llm/adapters-updated', later)
expect(() => ctx.llm.registerAdapter(['a'], new NoopAdapter())).toThrow('registry incoherent')
expect(later).toHaveBeenCalledTimes(1)
})
})
describe('configurable-provider directory', () => {
it('registers entries, lists detached copies in order, and fires the topology event', async () => {
const ctx = await setup()
const events = vi.fn()
ctx.on('llm/adapters-updated', events)
ctx.llm.registerConfigurableProviders([
entry({ provider: 'deepseek-official', displayName: 'DeepSeek', settingsNs: 'llm-deepseek', settingsPath: [] }),
entry(),
])
expect(events).toHaveBeenCalledTimes(1)
const listed = ctx.llm.listConfigurableProviders()
expect(listed).toEqual([
{ provider: 'deepseek-official', displayName: 'DeepSeek', settingsNs: 'llm-deepseek', settingsPath: [] },
{ provider: 'openai', displayName: 'OpenAI', settingsNs: 'llm-pi-ai', settingsPath: ['providers', 'openai'] },
])
listed[0]!.displayName = 'mutated'
;(listed[1]!.settingsPath as string[]).push('mutated')
expect(ctx.llm.listConfigurableProviders()[0]!.displayName).toBe('DeepSeek')
expect(ctx.llm.listConfigurableProviders()[1]!.settingsPath).toEqual(['providers', 'openai'])
})
it('detaches stored entries from caller-owned objects', async () => {
const ctx = await setup()
const source = entry()
ctx.llm.registerConfigurableProviders([source])
source.displayName = 'mutated'
expect(ctx.llm.listConfigurableProviders()[0]!.displayName).toBe('OpenAI')
})
it('withdraws every entry when the registration disposes', async () => {
const ctx = await setup()
const dispose = ctx.llm.registerConfigurableProviders([entry()])
const events = vi.fn()
ctx.on('llm/adapters-updated', events)
dispose()
expect(ctx.llm.listConfigurableProviders()).toEqual([])
expect(events).toHaveBeenCalledTimes(1)
})
it('withdraws entries when the contributing fiber disposes', async () => {
const ctx = await setup()
const fiber = await ctx.plugin({
inject: ['llm'],
apply: (child: Context) => {
child.llm.registerConfigurableProviders([entry()])
},
})
expect(ctx.llm.listConfigurableProviders()).toHaveLength(1)
await fiber.dispose()
expect(ctx.llm.listConfigurableProviders()).toEqual([])
})
it('rejects an empty registration', async () => {
const ctx = await setup()
expect(() => ctx.llm.registerConfigurableProviders([])).toThrow(LlmError)
expect(() => ctx.llm.registerConfigurableProviders([])).toThrow(/at least one provider/)
})
it.each([
[entry({ provider: '' }), /non-empty provider/],
[entry({ displayName: '' }), /non-empty provider/],
[entry({ settingsNs: '' }), /non-empty provider/],
[entry({ settingsPath: ['providers', ''] }), /empty settingsPath segment/],
])('rejects invalid entries all-or-nothing', async (invalid, message) => {
const ctx = await setup()
expect(() => ctx.llm.registerConfigurableProviders([entry({ provider: 'valid-first' }), invalid])).toThrow(message)
expect(ctx.llm.listConfigurableProviders()).toEqual([])
})
it('rejects duplicates within one registration and across registrations', async () => {
const ctx = await setup()
expect(() => ctx.llm.registerConfigurableProviders([entry(), entry()])).toThrow(/already declared/)
ctx.llm.registerConfigurableProviders([entry()])
expect(() => ctx.llm.registerConfigurableProviders([entry({ displayName: 'Other' }), entry({ provider: 'unseen' })]))
.toThrow(/already declared/)
expect(ctx.llm.listConfigurableProviders()).toHaveLength(1)
})
})