fix(host): refresh metric capacity metadata (round 6)

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

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<Session, UsageState>()
   private readonly capacities = new WeakMap<Agent, CapacityState>()
+  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

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<Context> {
+async function hostContext(onSessions?: (fiber: Fiber) => void): Promise<Context> {
   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<void> {
   return new Promise<void>((resolve) => { setImmediate(resolve) })
 }
 
+function installDeferredAdapter(
+  ctx: Context,
+  adapter: DeferredCatalogAdapter,
+): Fiber & PromiseLike<Fiber> {
+  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()
+  })
 })

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<void> {
+  return new Promise<void>((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)[] = []