diff --git a/packages/host/apiproxy/src/api-proxy.ts b/packages/host/apiproxy/src/api-proxy.ts index 7eec1d8bb6..b1c079a6d9 100644 --- a/packages/host/apiproxy/src/api-proxy.ts +++ b/packages/host/apiproxy/src/api-proxy.ts @@ -6,6 +6,7 @@ import { randomUUID } from 'node:crypto' import { mkdir, stat } from 'node:fs/promises' import { join } from 'node:path' +import { FiberState } from 'cordis' import type { Context } from 'cordis' import { installAgentLlmTarget } from '@deepseek-ai/dsh-agent' import type { @@ -483,6 +484,16 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro }), ctx.on('agent/created', (agent: Agent) => { scheduleMetrics(agent.session) }), ctx.on('session/disposed', (session: Session) => { pendingMetricSessions.delete(session) }), + ctx.on('internal/status', (fiber) => { + if (metricsDisposed) return + if (fiber.state !== FiberState.ACTIVE + && fiber.state !== FiberState.FAILED + && fiber.state !== FiberState.DISPOSED) return + const sessions = ctx.get('sessions') + if (sessions === undefined) return + metricsProjector.invalidateCapacities() + for (const session of sessions.list()) scheduleMetrics(session) + }, { global: true }), ] return () => { metricsDisposed = true diff --git a/packages/host/apiproxy/src/session-metrics.ts b/packages/host/apiproxy/src/session-metrics.ts index 161f92941e..1ad1f488af 100644 --- a/packages/host/apiproxy/src/session-metrics.ts +++ b/packages/host/apiproxy/src/session-metrics.ts @@ -23,7 +23,8 @@ interface UsageState { interface CapacityState { routeKey: string | undefined generation: number - status: 'pending' | 'ready' + epoch: number + status: 'pending' | 'ready' | 'retryable' contextWindow?: number } @@ -88,6 +89,7 @@ function routeKeyFor(target: CapacityTarget | undefined): string | undefined { export class SessionMetricsProjector { private readonly usage = new WeakMap() private readonly capacities = new WeakMap() + private capacityEpoch = 0 /** * @param ctx - Host context providing optional token-meter and LLM services. @@ -100,6 +102,11 @@ export class SessionMetricsProjector { private readonly onCapacityResolved: (agent: Agent) => void, ) {} + /** Retire adapter-owned metadata and fence every resolution already in flight. */ + invalidateCapacities(): void { + this.capacityEpoch++ + } + /** * Read a fresh detached projection through the session's durable tail. * @param session - authoritative durable log owner. @@ -159,10 +166,14 @@ export class SessionMetricsProjector { const target = this.targetFor(agent) const routeKey = routeKeyFor(target) let state = this.capacities.get(agent) - if (state === undefined || state.routeKey !== routeKey) { + if (state === undefined + || state.routeKey !== routeKey + || state.epoch !== this.capacityEpoch + || state.status === 'retryable') { state = { routeKey, generation: (state?.generation ?? 0) + 1, + epoch: this.capacityEpoch, status: target === undefined ? 'ready' : 'pending', } this.capacities.set(agent, state) @@ -178,7 +189,7 @@ export class SessionMetricsProjector { ): void { const llm = this.ctx.get('llm') as LlmLike | undefined if (llm === undefined) { - pending.status = 'ready' + pending.status = 'retryable' return } void Promise.resolve() @@ -191,12 +202,13 @@ export class SessionMetricsProjector { this.onCapacityResolved(agent) }, () => { - if (!this.capacityResolutionIsStale(agent, pending)) pending.status = 'ready' + if (!this.capacityResolutionIsStale(agent, pending)) pending.status = 'retryable' }, ) } private capacityResolutionIsStale(agent: Agent, pending: CapacityState): boolean { + if (pending.epoch !== this.capacityEpoch) return true if (this.capacities.get(agent)?.generation !== pending.generation) return true if (routeKeyFor(this.targetFor(agent)) === pending.routeKey) return false // Unknown is the neutral generation; the next observed concrete route @@ -204,6 +216,7 @@ export class SessionMetricsProjector { this.capacities.set(agent, { routeKey: undefined, generation: pending.generation + 1, + epoch: this.capacityEpoch, status: 'ready', }) return true diff --git a/packages/host/apiproxy/tests/api-proxy-models.spec.ts b/packages/host/apiproxy/tests/api-proxy-models.spec.ts index 722c2e7412..5a1d9f02cf 100644 --- a/packages/host/apiproxy/tests/api-proxy-models.spec.ts +++ b/packages/host/apiproxy/tests/api-proxy-models.spec.ts @@ -6,6 +6,7 @@ import { describe, expect, it, vi } from 'vitest' import { Context } from 'cordis' +import type { Fiber } from 'cordis' import AgentRegistry, { agentEvents, installAgentLlmTarget } from '@deepseek-ai/dsh-agent' import type { Agent, AgentLlmTargetRef } from '@deepseek-ai/dsh-agent' import LlmService, { LlmAdapter, ReasoningEffortId } from '@deepseek-ai/dsh-llm' @@ -99,9 +100,10 @@ const REASONING: LlmModelReasoningInfo = { defaultEffort: ReasoningEffortId('high'), } -async function hostContext(): Promise { +async function hostContext(onSessions?: (fiber: Fiber) => void): Promise { const ctx = new Context() - await ctx.plugin(SessionStore) + const sessionsFiber = await ctx.plugin(SessionStore) + onSessions?.(sessionsFiber) await ctx.plugin(SystemPrompt, { persona: '' }) await ctx.plugin(LlmService) await ctx.plugin(UserInteractionService) @@ -201,6 +203,15 @@ function settleCapacityCompletion(): Promise { return new Promise((resolve) => { setImmediate(resolve) }) } +function installDeferredAdapter( + ctx: Context, + adapter: DeferredCatalogAdapter, +): Fiber & PromiseLike { + return ctx.plugin(Object.assign((inner: Context) => { + inner.llm.registerAdapter(['deferred'], adapter) + }, { inject: ['llm'] })) +} + describe('Web session model selection', () => { it('groups successful providers, isolates failures, and preserves an unlisted current model', async () => { const { ctx, sessionId } = await harness({ @@ -509,4 +520,58 @@ describe('Web session model selection', () => { replacement.detach() await ctx.fiber.dispose() }) + + it('refreshes same-route capacity after adapter owner replacement', async () => { + const ctx = await hostContext() + const retiredAdapter = new DeferredCatalogAdapter() + const retiredFiber = await installDeferredAdapter(ctx, retiredAdapter) + const lifecycle = attachLifecycleSession(ctx, SessionId('capacity-adapter-lifecycle')) + const detachAgent = attachLifecycleAgent(ctx, lifecycle.session) + const api = createApiProxy(ctx, { provider: 'deepseek', model: 'deepseek-chat', cwd: '/tmp', workspaceRoot: '/tmp' }) + const controller = new AbortController() + const iterator = api.events.mux(request({}), controller.signal)[Symbol.asyncIterator]() + + expect((await nextMetrics(iterator)).contextWindow).toBeUndefined() + await vi.waitFor(() => { expect(retiredAdapter.pending).toHaveLength(1) }) + retiredAdapter.resolve(0, 64_000) + await settleCapacityCompletion() + expect((await nextMetrics(iterator)).contextWindow).toBe(64_000) + + await retiredFiber.dispose() + expect((await nextMetrics(iterator)).contextWindow).toBeUndefined() + const replacementAdapter = new DeferredCatalogAdapter() + const replacementFiber = await installDeferredAdapter(ctx, replacementAdapter) + expect((await nextMetrics(iterator)).contextWindow).toBeUndefined() + await vi.waitFor(() => { expect(replacementAdapter.pending).toHaveLength(1) }) + replacementAdapter.resolve(0, 128_000) + await settleCapacityCompletion() + expect((await nextMetrics(iterator)).contextWindow).toBe(128_000) + expect(lifecycle.session.requestHeader()?.config).toMatchObject({ + provider: 'deferred', + model: 'lifecycle-model', + }) + + controller.abort() + await iterator.return?.() + detachAgent() + lifecycle.detach() + await replacementFiber.dispose() + await ctx.fiber.dispose() + }) + + it('does not read the sessions service after its disposal status', async () => { + let sessionsFiber: Fiber | undefined + const ctx = await hostContext((fiber) => { sessionsFiber = fiber }) + if (sessionsFiber === undefined) throw new Error('sessions fiber missing') + createApiProxy(ctx, { provider: 'deepseek', model: 'deepseek-chat', cwd: '/tmp', workspaceRoot: '/tmp' }) + const sessions = ctx.get('sessions') + if (sessions === undefined) throw new Error('sessions service missing') + const list = vi.spyOn(sessions, 'list').mockImplementation(() => { + throw new Error('disposed sessions service read') + }) + + await expect(sessionsFiber.dispose()).resolves.toBeUndefined() + expect(list).not.toHaveBeenCalled() + await ctx.fiber.dispose() + }) }) diff --git a/packages/host/apiproxy/tests/session-metrics.spec.ts b/packages/host/apiproxy/tests/session-metrics.spec.ts index c92e5a76e4..3b3653aad8 100644 --- a/packages/host/apiproxy/tests/session-metrics.spec.ts +++ b/packages/host/apiproxy/tests/session-metrics.spec.ts @@ -33,6 +33,10 @@ function agent(session: Session): Agent { return { id: session.id, session } as Agent } +function settleAsyncWork(): Promise { + return new Promise((resolve) => { setImmediate(resolve) }) +} + describe('SessionMetricsProjector', () => { it('filters text/reasoning stream deltas while retaining usage, headers, and surface mutations', () => { const session = new Session(SessionId('metrics-filter')) @@ -187,6 +191,73 @@ describe('SessionMetricsProjector', () => { }) }) + it('retries a failed same-route capacity lookup only on the next snapshot', async () => { + const ctx = new Context() + const attempts: PromiseWithResolvers<{ context: { contextWindow: number } }>[] = [] + ctx.provide('llm', { + resolveModelInfo() { + const attempt = Promise.withResolvers<{ context: { contextWindow: number } }>() + attempts.push(attempt) + return attempt.promise + }, + }) + const session = new Session(SessionId('capacity-retry')) + const attached = agent(session) + const resolved = vi.fn() + const projector = new SessionMetricsProjector( + ctx, + () => ({ provider: 'test', model: 'alpha' }), + resolved, + ) + + expect(projector.snapshot(session, attached).contextWindow).toBeUndefined() + await vi.waitFor(() => { expect(attempts).toHaveLength(1) }) + attempts[0]?.reject(new Error('metadata temporarily unavailable')) + await settleAsyncWork() + expect(attempts).toHaveLength(1) + expect(resolved).not.toHaveBeenCalled() + + expect(projector.snapshot(session, attached).contextWindow).toBeUndefined() + await settleAsyncWork() + expect(attempts).toHaveLength(2) + attempts[1]?.resolve({ context: { contextWindow: 128_000 } }) + await vi.waitFor(() => { expect(resolved).toHaveBeenCalledOnce() }) + expect(projector.snapshot(session, attached).contextWindow).toBe(128_000) + }) + + it('invalidates same-route capacity and fences the prior epoch in flight', async () => { + const ctx = new Context() + const attempts: PromiseWithResolvers<{ context: { contextWindow: number } }>[] = [] + ctx.provide('llm', { + resolveModelInfo() { + const attempt = Promise.withResolvers<{ context: { contextWindow: number } }>() + attempts.push(attempt) + return attempt.promise + }, + }) + const session = new Session(SessionId('capacity-invalidation')) + const attached = agent(session) + const resolved = vi.fn() + const projector = new SessionMetricsProjector( + ctx, + () => ({ provider: 'test', model: 'alpha' }), + resolved, + ) + + expect(projector.snapshot(session, attached).contextWindow).toBeUndefined() + await vi.waitFor(() => { expect(attempts).toHaveLength(1) }) + projector.invalidateCapacities() + attempts[0]?.resolve({ context: { contextWindow: 64_000 } }) + await settleAsyncWork() + expect(resolved).not.toHaveBeenCalled() + + expect(projector.snapshot(session, attached).contextWindow).toBeUndefined() + await vi.waitFor(() => { expect(attempts).toHaveLength(2) }) + attempts[1]?.resolve({ context: { contextWindow: 128_000 } }) + await vi.waitFor(() => { expect(resolved).toHaveBeenCalledOnce() }) + expect(projector.snapshot(session, attached).contextWindow).toBe(128_000) + }) + it('starts a fresh capacity generation when an unavailable route returns', async () => { const ctx = new Context() const resolutions: ((contextWindow: number) => void)[] = []