refactor(web): publish transient model request capacity (round 1)
This commit is contained in:
Hypatia May
104
packages/host/apiproxy/tests/api-proxy-model-request.spec.ts
Normal file
104
packages/host/apiproxy/tests/api-proxy-model-request.spec.ts
Normal file
@@ -0,0 +1,104 @@
|
||||
import { describe, expect, it } from 'vitest'
|
||||
import { Context } from 'cordis'
|
||||
import AgentRegistry, { agentEvents } from '@deepseek-ai/dsh-agent'
|
||||
import type { Agent } from '@deepseek-ai/dsh-agent'
|
||||
import SessionStore, { SessionId } from '@deepseek-ai/dsh-session'
|
||||
import UserInteractionService from '@deepseek-ai/dsh-user-interaction'
|
||||
import type { MuxFrame, RpcRequest } from '@deepseek-ai/dsh-host-apiproxy/api'
|
||||
import { RpcId } from '@deepseek-ai/dsh-host-apiproxy/api'
|
||||
import { createApiProxy } from '../src/api-proxy.ts'
|
||||
|
||||
async function nextFrame<K extends MuxFrame['type']>(
|
||||
iterator: AsyncIterator<RpcRequest<MuxFrame>>,
|
||||
type: K,
|
||||
): Promise<Extract<MuxFrame, { type: K }>> {
|
||||
for (;;) {
|
||||
const next = await iterator.next()
|
||||
if (next.done) throw new Error(`mux ended before ${type}`)
|
||||
if (next.value.payload.type === type) {
|
||||
return next.value.payload as Extract<MuxFrame, { type: K }>
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
describe('ApiProxy model-request telemetry', () => {
|
||||
it('forwards only to open mux connections and never backfills history or reconnect baselines', async () => {
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SessionStore)
|
||||
await ctx.plugin(UserInteractionService)
|
||||
await ctx.plugin(AgentRegistry)
|
||||
const session = ctx.sessions.create(SessionId('model-request-telemetry'))
|
||||
const agent = {
|
||||
id: session.id,
|
||||
session,
|
||||
status: 'running',
|
||||
ctx,
|
||||
} as Agent
|
||||
ctx.agents.register(agent)
|
||||
const api = createApiProxy(ctx, {
|
||||
provider: 'test',
|
||||
model: 'alpha',
|
||||
cwd: '/tmp',
|
||||
workspaceRoot: '/tmp',
|
||||
})
|
||||
|
||||
const primaryAbort = new AbortController()
|
||||
const primary = api.events.mux(
|
||||
{ rpcId: RpcId('primary'), payload: {} },
|
||||
primaryAbort.signal,
|
||||
)[Symbol.asyncIterator]()
|
||||
expect((await nextFrame(primary, 'session/subscribed')).sessionId).toBe(session.id)
|
||||
expect((await nextFrame(primary, 'session/metrics')).metrics).not.toHaveProperty('contextWindow')
|
||||
|
||||
agentEvents(ctx, agent).emit('agent/model-request', 1, 2, {
|
||||
provider: 'test',
|
||||
model: 'alpha',
|
||||
contextWindow: 128_000,
|
||||
})
|
||||
expect(await nextFrame(primary, 'session/model-request')).toEqual({
|
||||
type: 'session/model-request',
|
||||
sessionId: session.id,
|
||||
turn: 1,
|
||||
step: 2,
|
||||
provider: 'test',
|
||||
model: 'alpha',
|
||||
contextWindow: 128_000,
|
||||
})
|
||||
|
||||
const history = await api.sessions.history({
|
||||
rpcId: RpcId('history'),
|
||||
payload: { sessionId: session.id },
|
||||
})
|
||||
if (!history.result.ok) throw new Error('history failed')
|
||||
expect(history.result.value.metrics).not.toHaveProperty('contextWindow')
|
||||
|
||||
const reconnectAbort = new AbortController()
|
||||
const reconnect = api.events.mux(
|
||||
{ rpcId: RpcId('reconnect'), payload: {} },
|
||||
reconnectAbort.signal,
|
||||
)[Symbol.asyncIterator]()
|
||||
expect((await nextFrame(reconnect, 'session/subscribed')).sessionId).toBe(session.id)
|
||||
expect((await nextFrame(reconnect, 'session/metrics')).metrics).not.toHaveProperty('contextWindow')
|
||||
|
||||
agentEvents(ctx, agent).emit('agent/model-request', 2, 1, {
|
||||
provider: 'test',
|
||||
model: 'without-capacity',
|
||||
})
|
||||
for (const iterator of [primary, reconnect]) {
|
||||
expect(await nextFrame(iterator, 'session/model-request')).toEqual({
|
||||
type: 'session/model-request',
|
||||
sessionId: session.id,
|
||||
turn: 2,
|
||||
step: 1,
|
||||
provider: 'test',
|
||||
model: 'without-capacity',
|
||||
})
|
||||
}
|
||||
|
||||
primaryAbort.abort()
|
||||
reconnectAbort.abort()
|
||||
await primary.return?.()
|
||||
await reconnect.return?.()
|
||||
await ctx.fiber.dispose()
|
||||
})
|
||||
})
|
||||
@@ -4,23 +4,20 @@
|
||||
* models, and the prompt-assembly boundary for a running selection change.
|
||||
*/
|
||||
|
||||
import { describe, expect, it, vi } from 'vitest'
|
||||
import { Context, FiberState } 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 { describe, expect, it } from 'vitest'
|
||||
import { Context } from 'cordis'
|
||||
import AgentRegistry, { agentEvents } from '@deepseek-ai/dsh-agent'
|
||||
import type { Agent } from '@deepseek-ai/dsh-agent'
|
||||
import LlmService, { LlmAdapter, ReasoningEffortId } from '@deepseek-ai/dsh-llm'
|
||||
import type {
|
||||
GenerateOptions, LlmCallConfig, LlmModelInfo, LlmModelReasoningInfo, LlmProviderInfo,
|
||||
LlmResolvedModelInfo, StreamChunk,
|
||||
} from '@deepseek-ai/dsh-llm'
|
||||
import SessionStore, { SessionId } from '@deepseek-ai/dsh-session'
|
||||
import type { Session } from '@deepseek-ai/dsh-session'
|
||||
import SystemPrompt from '@deepseek-ai/dsh-system-prompt'
|
||||
import UserInteractionService from '@deepseek-ai/dsh-user-interaction'
|
||||
import type { RpcRequest } from '@deepseek-ai/dsh-host-apiproxy/api/rpc'
|
||||
import { RpcId } from '@deepseek-ai/dsh-host-apiproxy/api/rpc'
|
||||
import type { MuxFrame } from '@deepseek-ai/dsh-host-apiproxy/api'
|
||||
import { createApiProxy } from '../src/api-proxy.ts'
|
||||
|
||||
let nextRpc = 1
|
||||
@@ -64,40 +61,6 @@ class CatalogAdapter extends LlmAdapter {
|
||||
}
|
||||
}
|
||||
|
||||
class DeferredCatalogAdapter extends CatalogAdapter {
|
||||
readonly pending: {
|
||||
result: PromiseWithResolvers<LlmResolvedModelInfo>
|
||||
signal: AbortSignal | undefined
|
||||
}[] = []
|
||||
|
||||
constructor() {
|
||||
super('Deferred', [
|
||||
{ provider: 'deferred', id: 'lifecycle-model', name: 'Lifecycle model' },
|
||||
])
|
||||
}
|
||||
|
||||
override resolveModel(
|
||||
_provider: string,
|
||||
_model: string,
|
||||
signal?: AbortSignal,
|
||||
): Promise<LlmResolvedModelInfo> {
|
||||
const result = Promise.withResolvers<LlmResolvedModelInfo>()
|
||||
this.pending.push({ result, signal })
|
||||
return result.promise
|
||||
}
|
||||
|
||||
resolve(index: number, contextWindow: number): void {
|
||||
const pending = this.pending[index]
|
||||
if (pending === undefined) throw new Error(`no pending resolution at index ${String(index)}`)
|
||||
pending.result.resolve({
|
||||
provider: 'deferred',
|
||||
id: 'lifecycle-model',
|
||||
name: 'Lifecycle model',
|
||||
context: { contextWindow },
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
const REASONING: LlmModelReasoningInfo = {
|
||||
efforts: [
|
||||
{ id: ReasoningEffortId('off'), name: 'Off' },
|
||||
@@ -107,18 +70,13 @@ const REASONING: LlmModelReasoningInfo = {
|
||||
defaultEffort: ReasoningEffortId('high'),
|
||||
}
|
||||
|
||||
async function hostContext(
|
||||
onSessions?: (fiber: Fiber) => void,
|
||||
onAgents?: (fiber: Fiber) => void,
|
||||
): Promise<Context> {
|
||||
async function hostContext(): Promise<Context> {
|
||||
const ctx = new Context()
|
||||
const sessionsFiber = await ctx.plugin(SessionStore)
|
||||
onSessions?.(sessionsFiber)
|
||||
await ctx.plugin(SessionStore)
|
||||
await ctx.plugin(SystemPrompt, { persona: '' })
|
||||
await ctx.plugin(LlmService)
|
||||
await ctx.plugin(UserInteractionService)
|
||||
const agentsFiber = await ctx.plugin(AgentRegistry)
|
||||
onAgents?.(agentsFiber)
|
||||
await ctx.plugin(AgentRegistry)
|
||||
ctx.llm.registerAdapter(['deepseek'], new CatalogAdapter('DeepSeek', [
|
||||
{ provider: 'deepseek', id: 'deepseek-chat', name: 'DeepSeek Chat' },
|
||||
{ provider: 'deepseek', id: 'deepseek-reasoner', name: 'DeepSeek Reasoner', description: 'Reasoning model' },
|
||||
@@ -164,65 +122,6 @@ function expectValue<T>(response: { result: { ok: true; value: T } | { ok: false
|
||||
return response.result.value
|
||||
}
|
||||
|
||||
async function nextMetrics(
|
||||
iterator: AsyncIterator<RpcRequest<MuxFrame>>,
|
||||
): Promise<Extract<MuxFrame, { type: 'session/metrics' }>['metrics']> {
|
||||
for (;;) {
|
||||
const next = await iterator.next()
|
||||
if (next.done) throw new Error('mux ended before a metrics frame')
|
||||
if (next.value.payload.type === 'session/metrics') return next.value.payload.metrics
|
||||
}
|
||||
}
|
||||
|
||||
function attachLifecycleSession(
|
||||
ctx: Context,
|
||||
sessionId: SessionId,
|
||||
withMarker = false,
|
||||
): { session: Session; detach: () => void } {
|
||||
const session = ctx.sessions.prepare(sessionId)
|
||||
session.append('request/header', {
|
||||
header: { config: { provider: 'deferred', model: 'lifecycle-model' } },
|
||||
reason: 'initial',
|
||||
})
|
||||
if (withMarker) {
|
||||
session.append('user/message', {
|
||||
content: [{ type: 'text', text: 'replacement marker' }],
|
||||
source: { kind: 'plugin', plugin: 'test' },
|
||||
}, { surfaceOp: 'append' })
|
||||
}
|
||||
const detach = ctx.sessions.enter(session)
|
||||
ctx.sessions.announce(session)
|
||||
return { session, detach }
|
||||
}
|
||||
|
||||
function attachLifecycleAgent(
|
||||
ctx: Context,
|
||||
session: Session,
|
||||
): () => void {
|
||||
const agent = {
|
||||
id: session.id,
|
||||
session,
|
||||
status: 'running',
|
||||
ctx,
|
||||
} as Agent
|
||||
const detach = ctx.agents.enter(agent, undefined)
|
||||
ctx.agents.announce(agent)
|
||||
return detach
|
||||
}
|
||||
|
||||
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({
|
||||
@@ -230,7 +129,12 @@ describe('Web session model selection', () => {
|
||||
model: 'private-preview',
|
||||
reasoningEffort: ReasoningEffortId('max'),
|
||||
})
|
||||
const api = createApiProxy(ctx, { provider: 'deepseek', model: 'deepseek-chat', cwd: '/tmp', workspaceRoot: '/tmp' })
|
||||
const api = createApiProxy(ctx, {
|
||||
provider: 'deepseek',
|
||||
model: 'deepseek-chat',
|
||||
cwd: '/tmp',
|
||||
workspaceRoot: '/tmp',
|
||||
})
|
||||
|
||||
const catalog = expectValue(await api.sessions.models(request({ sessionId })))
|
||||
expect(catalog.current).toEqual({
|
||||
@@ -271,7 +175,12 @@ describe('Web session model selection', () => {
|
||||
|
||||
it('accepts an advisory-unlisted model, rejects an unavailable provider, and switches only after the next assembly', async () => {
|
||||
const { ctx, agent, sessionId } = await harness()
|
||||
const api = createApiProxy(ctx, { provider: 'deepseek', model: 'deepseek-chat', cwd: '/tmp', workspaceRoot: '/tmp' })
|
||||
const api = createApiProxy(ctx, {
|
||||
provider: 'deepseek',
|
||||
model: 'deepseek-chat',
|
||||
cwd: '/tmp',
|
||||
workspaceRoot: '/tmp',
|
||||
})
|
||||
const seed: LlmCallConfig = { provider: 'seed', model: 'seed', temperature: 0.2 }
|
||||
const signal = new AbortController().signal
|
||||
|
||||
@@ -336,534 +245,4 @@ describe('Web session model selection', () => {
|
||||
.toEqual({ provider: 'deepseek', model: 'private-preview', reasoningEffort: 'max' })
|
||||
await ctx.fiber.dispose()
|
||||
})
|
||||
|
||||
it('publishes unknown capacity immediately on selection, then the exact selected route capacity', async () => {
|
||||
const { ctx, sessionId } = await harness()
|
||||
const api = createApiProxy(ctx, { provider: 'deepseek', model: 'deepseek-chat', cwd: '/tmp', workspaceRoot: '/tmp' })
|
||||
expectValue(await api.sessions.models(request({ sessionId })))
|
||||
const controller = new AbortController()
|
||||
const iterator = api.events.mux(request({}), controller.signal)[Symbol.asyncIterator]()
|
||||
|
||||
expect((await nextMetrics(iterator)).contextWindow).toBeUndefined()
|
||||
expect((await nextMetrics(iterator)).contextWindow).toBe(64_000)
|
||||
|
||||
expectValue(await api.sessions.selectModel(request({
|
||||
sessionId,
|
||||
provider: 'deepseek',
|
||||
model: 'private-preview',
|
||||
})))
|
||||
expect((await nextMetrics(iterator)).contextWindow).toBeUndefined()
|
||||
expect((await nextMetrics(iterator)).contextWindow).toBe(128_000)
|
||||
|
||||
controller.abort()
|
||||
await iterator.return?.()
|
||||
await ctx.fiber.dispose()
|
||||
})
|
||||
|
||||
it('uses logged capacity without installing Web routing while scheduling foreign metrics', async () => {
|
||||
const ctx = await hostContext()
|
||||
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]()
|
||||
const initialMetrics = nextMetrics(iterator)
|
||||
const session = ctx.sessions.create()
|
||||
expect((await initialMetrics).contextWindow).toBeUndefined()
|
||||
session.append('request/header', {
|
||||
header: { config: { provider: 'deepseek', model: 'private-preview' } },
|
||||
reason: 'change',
|
||||
})
|
||||
const foreign = {
|
||||
id: session.id,
|
||||
session,
|
||||
status: 'running',
|
||||
ctx,
|
||||
} as Agent
|
||||
const foreignTarget: AgentLlmTargetRef = {
|
||||
current: { provider: 'foreign', model: 'foreign-model' },
|
||||
assembled: undefined,
|
||||
}
|
||||
const disposeForeignTarget = installAgentLlmTarget(foreign.ctx, foreignTarget)
|
||||
const scheduledMetrics = nextMetrics(iterator)
|
||||
ctx.agents.register(foreign)
|
||||
|
||||
expect((await scheduledMetrics).contextWindow).toBeUndefined()
|
||||
expect((await nextMetrics(iterator)).contextWindow).toBe(128_000)
|
||||
expect((await ctx.systemPrompt.assemble()).variables)
|
||||
.toMatchObject({ provider: 'foreign', model: 'foreign-model' })
|
||||
const seed: LlmCallConfig = { provider: 'seed', model: 'seed', temperature: 0.2 }
|
||||
const signal = new AbortController().signal
|
||||
await expect(agentEvents(ctx, foreign).waterfall(
|
||||
'agent/request', 1, 0, signal, () => Promise.resolve(seed),
|
||||
)).resolves.toMatchObject({ provider: 'foreign', model: 'foreign-model' })
|
||||
|
||||
disposeForeignTarget()
|
||||
expect((await ctx.systemPrompt.assemble()).variables).not.toHaveProperty('provider')
|
||||
await expect(agentEvents(ctx, foreign).waterfall(
|
||||
'agent/request', 1, 1, signal, () => Promise.resolve(seed),
|
||||
)).resolves.toBe(seed)
|
||||
|
||||
controller.abort()
|
||||
await iterator.return?.()
|
||||
await ctx.fiber.dispose()
|
||||
})
|
||||
|
||||
it('drops capacity completion from a replaced agent that retains the exact session', async () => {
|
||||
const ctx = await hostContext()
|
||||
const deferred = new DeferredCatalogAdapter()
|
||||
ctx.llm.registerAdapter(['deferred'], deferred)
|
||||
const lifecycle = attachLifecycleSession(ctx, SessionId('capacity-agent-lifecycle'))
|
||||
const retire = 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(deferred.pending).toHaveLength(1) })
|
||||
retire()
|
||||
const detachLive = attachLifecycleAgent(ctx, lifecycle.session)
|
||||
expect((await nextMetrics(iterator)).contextWindow).toBeUndefined()
|
||||
await vi.waitFor(() => { expect(deferred.pending).toHaveLength(2) })
|
||||
|
||||
deferred.resolve(0, 64_000)
|
||||
await settleCapacityCompletion()
|
||||
deferred.resolve(1, 128_000)
|
||||
await settleCapacityCompletion()
|
||||
expect((await nextMetrics(iterator)).contextWindow).toBe(128_000)
|
||||
|
||||
controller.abort()
|
||||
await iterator.return?.()
|
||||
detachLive()
|
||||
lifecycle.detach()
|
||||
await ctx.fiber.dispose()
|
||||
})
|
||||
|
||||
it('drops capacity completion from a replaced session while its old agent remains live', async () => {
|
||||
const ctx = await hostContext()
|
||||
const deferred = new DeferredCatalogAdapter()
|
||||
ctx.llm.registerAdapter(['deferred'], deferred)
|
||||
const sessionId = SessionId('capacity-session-lifecycle')
|
||||
const retiredSession = attachLifecycleSession(ctx, sessionId)
|
||||
const retireAgent = attachLifecycleAgent(ctx, retiredSession.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(deferred.pending).toHaveLength(1) })
|
||||
retiredSession.detach()
|
||||
const liveSession = attachLifecycleSession(ctx, sessionId, true)
|
||||
expect((await nextMetrics(iterator)).contextWindow).toBeUndefined()
|
||||
deferred.resolve(0, 64_000)
|
||||
await settleCapacityCompletion()
|
||||
retireAgent()
|
||||
const detachLiveAgent = attachLifecycleAgent(ctx, liveSession.session)
|
||||
const scheduled = await nextMetrics(iterator)
|
||||
expect(scheduled.logRevision).toBe(2)
|
||||
expect(scheduled.contextWindow).toBeUndefined()
|
||||
await vi.waitFor(() => { expect(deferred.pending).toHaveLength(2) })
|
||||
deferred.resolve(1, 128_000)
|
||||
await settleCapacityCompletion()
|
||||
expect((await nextMetrics(iterator)).contextWindow).toBe(128_000)
|
||||
|
||||
controller.abort()
|
||||
await iterator.return?.()
|
||||
detachLiveAgent()
|
||||
liveSession.detach()
|
||||
await ctx.fiber.dispose()
|
||||
})
|
||||
|
||||
it('does not project retired agent capacity into replacement session snapshots', async () => {
|
||||
const ctx = await hostContext()
|
||||
const deferred = new DeferredCatalogAdapter()
|
||||
ctx.llm.registerAdapter(['deferred'], deferred)
|
||||
const sessionId = SessionId('capacity-snapshot-lifecycle')
|
||||
const retiredSession = attachLifecycleSession(ctx, sessionId)
|
||||
const retireAgent = attachLifecycleAgent(ctx, retiredSession.session)
|
||||
const api = createApiProxy(ctx, { provider: 'deepseek', model: 'deepseek-chat', cwd: '/tmp', workspaceRoot: '/tmp' })
|
||||
const primaryController = new AbortController()
|
||||
const primary = api.events.mux(request({}), primaryController.signal)[Symbol.asyncIterator]()
|
||||
|
||||
expect((await nextMetrics(primary)).contextWindow).toBeUndefined()
|
||||
await vi.waitFor(() => { expect(deferred.pending).toHaveLength(1) })
|
||||
deferred.resolve(0, 64_000)
|
||||
await settleCapacityCompletion()
|
||||
expect((await nextMetrics(primary)).contextWindow).toBe(64_000)
|
||||
|
||||
retiredSession.detach()
|
||||
const replacement = attachLifecycleSession(ctx, sessionId)
|
||||
const createdBaseline = await nextMetrics(primary)
|
||||
replacement.session.append('user/message', {
|
||||
content: [{ type: 'text', text: 'replacement marker' }],
|
||||
source: { kind: 'plugin', plugin: 'test' },
|
||||
}, { surfaceOp: 'append' })
|
||||
const scheduledFlush = await nextMetrics(primary)
|
||||
const reconnectController = new AbortController()
|
||||
const reconnect = api.events.mux(request({}), reconnectController.signal)[Symbol.asyncIterator]()
|
||||
const reconnectBaseline = await nextMetrics(reconnect)
|
||||
expect(createdBaseline.logRevision).toBe(1)
|
||||
for (const metrics of [scheduledFlush, reconnectBaseline]) {
|
||||
expect(metrics.logRevision).toBe(2)
|
||||
}
|
||||
expect({
|
||||
created: createdBaseline.contextWindow,
|
||||
scheduled: scheduledFlush.contextWindow,
|
||||
reconnect: reconnectBaseline.contextWindow,
|
||||
}).toEqual({ created: undefined, scheduled: undefined, reconnect: undefined })
|
||||
|
||||
retireAgent()
|
||||
const detachReplacementAgent = attachLifecycleAgent(ctx, replacement.session)
|
||||
expect((await nextMetrics(primary)).contextWindow).toBeUndefined()
|
||||
await vi.waitFor(() => { expect(deferred.pending).toHaveLength(2) })
|
||||
deferred.resolve(1, 128_000)
|
||||
await settleCapacityCompletion()
|
||||
expect((await nextMetrics(primary)).contextWindow).toBe(128_000)
|
||||
|
||||
primaryController.abort()
|
||||
reconnectController.abort()
|
||||
await primary.return?.()
|
||||
await reconnect.return?.()
|
||||
detachReplacementAgent()
|
||||
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('aborts pending capacity during adapter UNLOADING and refreshes after settlement', async () => {
|
||||
const ctx = await hostContext()
|
||||
const retiredAdapter = new DeferredCatalogAdapter()
|
||||
const releaseUnload = Promise.withResolvers<undefined>()
|
||||
const releaseCancellationWait = Promise.withResolvers<undefined>()
|
||||
const abortObserved = Promise.withResolvers<undefined>()
|
||||
const retiredFiber = await ctx.plugin(Object.assign((inner: Context) => {
|
||||
inner.llm.registerAdapter(['deferred'], retiredAdapter)
|
||||
inner.effect(
|
||||
() => () => releaseUnload.promise,
|
||||
'test: hold adapter unload',
|
||||
)
|
||||
inner.effect(() => () => {
|
||||
const signal = retiredAdapter.pending[0]?.signal
|
||||
if (signal === undefined) throw new Error('pending capacity signal missing')
|
||||
const cancellation = new Promise<void>((resolve) => {
|
||||
const finish = () => {
|
||||
abortObserved.resolve(undefined)
|
||||
resolve()
|
||||
}
|
||||
if (signal.aborted) finish()
|
||||
else signal.addEventListener('abort', finish, { once: true })
|
||||
})
|
||||
return Promise.race([cancellation, releaseCancellationWait.promise])
|
||||
}, 'test: await capacity cancellation')
|
||||
}, { inject: ['llm'] }))
|
||||
const lifecycle = attachLifecycleSession(ctx, SessionId('capacity-adapter-unloading'))
|
||||
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]()
|
||||
const resolveModelInfo = vi.spyOn(ctx.llm, 'resolveModelInfo')
|
||||
|
||||
expect((await nextMetrics(iterator)).contextWindow).toBeUndefined()
|
||||
await vi.waitFor(() => { expect(retiredAdapter.pending).toHaveLength(1) })
|
||||
expect(resolveModelInfo).toHaveBeenCalledOnce()
|
||||
const listSessions = vi.spyOn(ctx.sessions, 'list')
|
||||
listSessions.mockClear()
|
||||
const pendingFrame = iterator.next()
|
||||
const disposing = retiredFiber.dispose()
|
||||
try {
|
||||
await vi.waitFor(() => {
|
||||
expect(retiredAdapter.pending[0]?.signal?.aborted).toBe(true)
|
||||
})
|
||||
await abortObserved.promise
|
||||
expect(retiredFiber.state).toBe(FiberState.UNLOADING)
|
||||
retiredAdapter.resolve(0, 64_000)
|
||||
await settleCapacityCompletion()
|
||||
expect(resolveModelInfo).toHaveBeenCalledOnce()
|
||||
expect(listSessions).not.toHaveBeenCalled()
|
||||
const outcome = await Promise.race([
|
||||
pendingFrame.then(() => 'frame' as const),
|
||||
new Promise<'idle'>((resolve) => { setImmediate(() => { resolve('idle') }) }),
|
||||
])
|
||||
expect(outcome).toBe('idle')
|
||||
} finally {
|
||||
releaseCancellationWait.resolve(undefined)
|
||||
releaseUnload.resolve(undefined)
|
||||
await disposing
|
||||
}
|
||||
|
||||
expect(retiredFiber.state).toBe(FiberState.DISPOSED)
|
||||
const settledFrame = await pendingFrame
|
||||
if (settledFrame.done || settledFrame.value.payload.type !== 'session/metrics') {
|
||||
throw new Error('expected settled metrics refresh')
|
||||
}
|
||||
expect(settledFrame.value.payload.metrics.contextWindow).toBeUndefined()
|
||||
await settleCapacityCompletion()
|
||||
expect(resolveModelInfo).toHaveBeenCalledTimes(2)
|
||||
|
||||
const replacementAdapter = new DeferredCatalogAdapter()
|
||||
const replacementFiber = await installDeferredAdapter(ctx, replacementAdapter)
|
||||
expect((await nextMetrics(iterator)).contextWindow).toBeUndefined()
|
||||
await vi.waitFor(() => { expect(replacementAdapter.pending).toHaveLength(1) })
|
||||
expect(resolveModelInfo).toHaveBeenCalledTimes(3)
|
||||
replacementAdapter.resolve(0, 128_000)
|
||||
await settleCapacityCompletion()
|
||||
expect((await nextMetrics(iterator)).contextWindow).toBe(128_000)
|
||||
|
||||
controller.abort()
|
||||
await iterator.return?.()
|
||||
detachAgent()
|
||||
lifecycle.detach()
|
||||
await replacementFiber.dispose()
|
||||
await ctx.fiber.dispose()
|
||||
})
|
||||
|
||||
it('aborts pending capacity when the API proxy fiber is disposed', async () => {
|
||||
const ctx = await hostContext()
|
||||
const deferred = new DeferredCatalogAdapter()
|
||||
ctx.llm.registerAdapter(['deferred'], deferred)
|
||||
const lifecycle = attachLifecycleSession(ctx, SessionId('capacity-api-proxy-teardown'))
|
||||
const detachAgent = attachLifecycleAgent(ctx, lifecycle.session)
|
||||
const proxy = Promise.withResolvers<ReturnType<typeof createApiProxy>>()
|
||||
const proxyFiber = await ctx.plugin(Object.assign((inner: Context) => {
|
||||
proxy.resolve(createApiProxy(inner, {
|
||||
provider: 'deepseek',
|
||||
model: 'deepseek-chat',
|
||||
cwd: '/tmp',
|
||||
workspaceRoot: '/tmp',
|
||||
}))
|
||||
}, { inject: ['agents', 'sessions', 'userInteraction'] }))
|
||||
const api = await proxy.promise
|
||||
const controller = new AbortController()
|
||||
const iterator = api.events.mux(request({}), controller.signal)[Symbol.asyncIterator]()
|
||||
|
||||
expect((await nextMetrics(iterator)).contextWindow).toBeUndefined()
|
||||
await vi.waitFor(() => { expect(deferred.pending).toHaveLength(1) })
|
||||
expect(deferred.pending[0]?.signal?.aborted).toBe(false)
|
||||
|
||||
await proxyFiber.dispose()
|
||||
expect(deferred.pending[0]?.signal?.aborted).toBe(true)
|
||||
deferred.resolve(0, 64_000)
|
||||
await settleCapacityCompletion()
|
||||
const pendingFrame = iterator.next()
|
||||
const outcome = await Promise.race([
|
||||
pendingFrame.then(() => 'frame' as const),
|
||||
new Promise<'idle'>((resolve) => { setImmediate(() => { resolve('idle') }) }),
|
||||
])
|
||||
expect(outcome).toBe('idle')
|
||||
|
||||
controller.abort()
|
||||
await expect(pendingFrame).resolves.toMatchObject({ done: true })
|
||||
await iterator.return?.()
|
||||
detachAgent()
|
||||
lifecycle.detach()
|
||||
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()
|
||||
})
|
||||
|
||||
it('invalidates pending capacity before SessionStore teardown can reach its callback', async () => {
|
||||
let sessionsFiber: Fiber | undefined
|
||||
const ctx = await hostContext((fiber) => { sessionsFiber = fiber })
|
||||
if (sessionsFiber === undefined) throw new Error('sessions fiber missing')
|
||||
const deferred = new DeferredCatalogAdapter()
|
||||
ctx.llm.registerAdapter(['deferred'], deferred)
|
||||
const lifecycle = attachLifecycleSession(ctx, SessionId('capacity-session-store-teardown'))
|
||||
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(deferred.pending).toHaveLength(1) })
|
||||
const agents = ctx.get('agents')
|
||||
if (agents === undefined) throw new Error('agent registry missing')
|
||||
expect(agents.get(lifecycle.session.id)).toBeDefined()
|
||||
const getAgent = vi.spyOn(agents, 'get')
|
||||
|
||||
await sessionsFiber.dispose()
|
||||
getAgent.mockClear()
|
||||
deferred.resolve(0, 64_000)
|
||||
await settleCapacityCompletion()
|
||||
expect(getAgent).not.toHaveBeenCalled()
|
||||
|
||||
const pendingFrame = iterator.next()
|
||||
const outcome = await Promise.race([
|
||||
pendingFrame.then(() => 'frame' as const),
|
||||
new Promise<'idle'>((resolve) => { setImmediate(() => { resolve('idle') }) }),
|
||||
])
|
||||
expect(outcome).toBe('idle')
|
||||
|
||||
controller.abort()
|
||||
await expect(pendingFrame).resolves.toMatchObject({ done: true })
|
||||
await iterator.return?.()
|
||||
detachAgent()
|
||||
lifecycle.detach()
|
||||
await ctx.fiber.dispose()
|
||||
})
|
||||
|
||||
it('publishes unknown metrics after AgentRegistry terminal disposal with mux active', async () => {
|
||||
let agentsFiber: Fiber | undefined
|
||||
const ctx = await hostContext(undefined, (fiber) => { agentsFiber = fiber })
|
||||
if (agentsFiber === undefined) throw new Error('agent registry fiber missing')
|
||||
const deferred = new DeferredCatalogAdapter()
|
||||
ctx.llm.registerAdapter(['deferred'], deferred)
|
||||
const lifecycle = attachLifecycleSession(ctx, SessionId('capacity-agent-registry-disposed'))
|
||||
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(deferred.pending).toHaveLength(1) })
|
||||
deferred.resolve(0, 64_000)
|
||||
await settleCapacityCompletion()
|
||||
expect((await nextMetrics(iterator)).contextWindow).toBe(64_000)
|
||||
|
||||
await agentsFiber.dispose()
|
||||
const refresh = nextMetrics(iterator).then(
|
||||
metrics => ({ kind: 'metrics' as const, metrics }),
|
||||
() => ({ kind: 'error' as const }),
|
||||
)
|
||||
const outcome = await Promise.race([
|
||||
refresh,
|
||||
new Promise<{ kind: 'idle' }>((resolve) => {
|
||||
setImmediate(() => { resolve({ kind: 'idle' }) })
|
||||
}),
|
||||
])
|
||||
|
||||
controller.abort()
|
||||
await refresh
|
||||
await iterator.return?.()
|
||||
detachAgent()
|
||||
lifecycle.detach()
|
||||
await ctx.fiber.dispose()
|
||||
expect(outcome.kind).toBe('metrics')
|
||||
if (outcome.kind === 'metrics') {
|
||||
expect(outcome.metrics.contextWindow).toBeUndefined()
|
||||
expect(outcome.metrics.logRevision).toBe(1)
|
||||
}
|
||||
})
|
||||
|
||||
it.each(['agents', 'sessions'] as const)(
|
||||
'drops capacity completion while %s is unavailable during unload',
|
||||
async (serviceName) => {
|
||||
let sessionsFiber: Fiber | undefined
|
||||
let agentsFiber: Fiber | undefined
|
||||
const ctx = await hostContext(
|
||||
(fiber) => { sessionsFiber = fiber },
|
||||
(fiber) => { agentsFiber = fiber },
|
||||
)
|
||||
const heldFiber = serviceName === 'sessions' ? sessionsFiber : agentsFiber
|
||||
if (heldFiber === undefined) throw new Error(`${serviceName} fiber missing`)
|
||||
const unloadStarted = Promise.withResolvers<undefined>()
|
||||
const releaseUnload = Promise.withResolvers<undefined>()
|
||||
heldFiber.ctx.effect(() => () => {
|
||||
unloadStarted.resolve(undefined)
|
||||
return releaseUnload.promise
|
||||
}, `test: hold ${serviceName} unload`)
|
||||
const deferred = new DeferredCatalogAdapter()
|
||||
ctx.llm.registerAdapter(['deferred'], deferred)
|
||||
const lifecycle = attachLifecycleSession(ctx, SessionId(`capacity-${serviceName}-unloading`))
|
||||
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(deferred.pending).toHaveLength(1) })
|
||||
const agents = ctx.get('agents')
|
||||
if (agents === undefined) throw new Error('agent registry missing')
|
||||
const sessions = ctx.get('sessions')
|
||||
if (sessions === undefined) throw new Error('sessions service missing')
|
||||
const getAgent = vi.spyOn(agents, 'get')
|
||||
const getSession = vi.spyOn(sessions, 'get')
|
||||
const disposing = heldFiber.dispose()
|
||||
await unloadStarted.promise
|
||||
await vi.waitFor(() => { expect(ctx.get(serviceName)).toBeUndefined() })
|
||||
expect(heldFiber.state).toBe(FiberState.UNLOADING)
|
||||
|
||||
getAgent.mockClear()
|
||||
getSession.mockClear()
|
||||
deferred.resolve(0, 64_000)
|
||||
await settleCapacityCompletion()
|
||||
const agentReads = getAgent.mock.calls.length
|
||||
const sessionReads = getSession.mock.calls.length
|
||||
const pendingFrame = iterator.next()
|
||||
const outcome = await Promise.race([
|
||||
pendingFrame.then(() => 'frame' as const),
|
||||
new Promise<'idle'>((resolve) => { setImmediate(() => { resolve('idle') }) }),
|
||||
])
|
||||
|
||||
controller.abort()
|
||||
await expect(pendingFrame).resolves.toMatchObject({ done: true })
|
||||
await iterator.return?.()
|
||||
releaseUnload.resolve(undefined)
|
||||
await disposing
|
||||
detachAgent()
|
||||
lifecycle.detach()
|
||||
await ctx.fiber.dispose()
|
||||
expect(agentReads).toBe(0)
|
||||
expect(sessionReads).toBe(0)
|
||||
expect(outcome).toBe('idle')
|
||||
},
|
||||
)
|
||||
})
|
||||
|
||||
@@ -146,10 +146,9 @@ describe('sessions domain schemas', () => {
|
||||
cacheReadTokens: 4_000,
|
||||
cacheWriteTokens: 500,
|
||||
contextTokens: 8_000,
|
||||
contextWindow: 128_000,
|
||||
},
|
||||
modelTarget: { provider: 'deepseek', model: 'deepseek-v4-flash' },
|
||||
}).metrics?.contextWindow).toBe(128_000)
|
||||
}).metrics?.contextTokens).toBe(8_000)
|
||||
expect(() => sessionMetricsSchema.parse({
|
||||
logRevision: 1,
|
||||
projectionRevision: 0,
|
||||
@@ -158,15 +157,6 @@ describe('sessions domain schemas', () => {
|
||||
cacheReadTokens: 0,
|
||||
cacheWriteTokens: 0,
|
||||
})).toThrow()
|
||||
expect(() => sessionMetricsSchema.parse({
|
||||
logRevision: 1,
|
||||
projectionRevision: 0,
|
||||
uncachedInputTokens: 0,
|
||||
outputTokens: 0,
|
||||
cacheReadTokens: 0,
|
||||
cacheWriteTokens: 0,
|
||||
contextWindow: 0,
|
||||
})).toThrow()
|
||||
expect(sessionModelsRequestSchema.parse({ sessionId: 's1' }).sessionId).toBe('s1')
|
||||
expect(sessionModelsValueSchema.parse({
|
||||
current: { provider: 'deepseek', model: 'deepseek-v4-flash', reasoningEffort: 'max' },
|
||||
@@ -344,6 +334,23 @@ describe('events frame schemas', () => {
|
||||
cacheWriteTokens: 40,
|
||||
},
|
||||
},
|
||||
{
|
||||
type: 'session/model-request',
|
||||
sessionId: 's',
|
||||
turn: 2,
|
||||
step: 1,
|
||||
provider: 'deepseek',
|
||||
model: 'deepseek-chat',
|
||||
contextWindow: 128_000,
|
||||
},
|
||||
{
|
||||
type: 'session/model-request',
|
||||
sessionId: 's',
|
||||
turn: 3,
|
||||
step: 1,
|
||||
provider: 'deepseek',
|
||||
model: 'unknown-capacity',
|
||||
},
|
||||
{ type: 'approval/requested', sessionId: 's', approvalId: 'a', toolName: 'bash', callId: 'c', reason: 'r' },
|
||||
{ type: 'approval/resolved', sessionId: 's', approvalId: 'a', outcome: 'allowed-once' },
|
||||
{ type: 'question/requested', sessionId: 's', questions: [{ id: 'q', question: 'Q?', options: [{ label: 'L' }], multiSelect: true }] },
|
||||
@@ -358,6 +365,8 @@ describe('events frame schemas', () => {
|
||||
{ type: 'session/title', sessionId: 's', title: '', eventSeq: 0, updatedAt: 1 },
|
||||
{ type: 'session/title', sessionId: 's', title: 'x', eventSeq: -1, updatedAt: 1 },
|
||||
{ type: 'session/title', sessionId: 's', title: 'x', eventSeq: 0.5, updatedAt: 1 },
|
||||
{ type: 'session/model-request', sessionId: 's', turn: 0, step: 1, provider: 'p', model: 'm' },
|
||||
{ type: 'session/model-request', sessionId: 's', turn: 1, step: 1, provider: 'p', model: 'm', contextWindow: 0 },
|
||||
{ type: 'session/title', sessionId: 's', title: 'x', eventSeq: 0, updatedAt: 'now' },
|
||||
{ type: 'session/title', sessionId: 's', title: 'x', eventSeq: 0, updatedAt: Number.NaN },
|
||||
]) expect(() => muxFrameSchema.parse(invalid)).toThrow()
|
||||
|
||||
@@ -1,6 +1,5 @@
|
||||
import { describe, expect, it, vi } from 'vitest'
|
||||
import { describe, expect, it } from 'vitest'
|
||||
import { Context } from 'cordis'
|
||||
import type { Agent, AgentLlmTarget } from '@deepseek-ai/dsh-agent'
|
||||
import { Session, SessionId } from '@deepseek-ai/dsh-session'
|
||||
import { affectsSessionMetrics, SessionMetricsProjector } from '../src/session-metrics.ts'
|
||||
|
||||
@@ -29,16 +28,8 @@ function assistant(
|
||||
}, { surfaceOp: 'append' })
|
||||
}
|
||||
|
||||
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', () => {
|
||||
it('filters text/reasoning deltas while retaining usage, headers, and surface mutations', () => {
|
||||
const session = new Session(SessionId('metrics-filter'))
|
||||
const text = session.append('assistant/chunk', {
|
||||
turn: 1,
|
||||
@@ -58,13 +49,16 @@ describe('SessionMetricsProjector', () => {
|
||||
content: [{ type: 'text', text: 'question' }],
|
||||
source: { kind: 'user' },
|
||||
}, { surfaceOp: 'append' })
|
||||
const plain = session.append('step/start', { turn: 1, step: 1 })
|
||||
|
||||
expect(affectsSessionMetrics(text)).toBe(false)
|
||||
expect(affectsSessionMetrics(usage)).toBe(true)
|
||||
expect(affectsSessionMetrics(header)).toBe(true)
|
||||
expect(affectsSessionMetrics(surface)).toBe(true)
|
||||
expect(affectsSessionMetrics(plain)).toBe(false)
|
||||
})
|
||||
|
||||
it('reconciles usage by turn:step, keeps cache writes disjoint, and survives a surface replacement', () => {
|
||||
it('folds usage by turn and step while synchronous pressure follows surface replacement', () => {
|
||||
const ctx = new Context()
|
||||
ctx.provide('tokenMeter', {
|
||||
measure(session: Session) {
|
||||
@@ -83,11 +77,8 @@ describe('SessionMetricsProjector', () => {
|
||||
cacheWriteTokens: 8,
|
||||
})
|
||||
|
||||
const current: AgentLlmTarget = { provider: 'test', model: 'alpha' }
|
||||
const projector = new SessionMetricsProjector(ctx, () => current, () => {})
|
||||
const attached = agent(session)
|
||||
const before = projector.snapshot(session, attached)
|
||||
expect(before).toMatchObject({
|
||||
const projector = new SessionMetricsProjector(ctx)
|
||||
expect(projector.snapshot(session)).toMatchObject({
|
||||
uncachedInputTokens: 11,
|
||||
outputTokens: 3,
|
||||
cacheReadTokens: 89,
|
||||
@@ -104,8 +95,7 @@ describe('SessionMetricsProjector', () => {
|
||||
surfaceOp: { op: 'replace', start: first.seq, end: assistantSeq },
|
||||
sourceEventSeqs: [first.seq, assistantSeq],
|
||||
})
|
||||
const compacted = projector.snapshot(session, attached)
|
||||
expect(compacted).toMatchObject({
|
||||
expect(projector.snapshot(session)).toMatchObject({
|
||||
uncachedInputTokens: 11,
|
||||
outputTokens: 3,
|
||||
cacheReadTokens: 89,
|
||||
@@ -113,357 +103,38 @@ describe('SessionMetricsProjector', () => {
|
||||
contextTokens: 100,
|
||||
})
|
||||
|
||||
// A replayed usage event for the same step replaces the settled value.
|
||||
session.append('assistant/chunk', {
|
||||
turn: 1,
|
||||
step: 1,
|
||||
chunk: {
|
||||
type: 'usage',
|
||||
usage: {
|
||||
inputTokens: 12,
|
||||
outputTokens: 4,
|
||||
cacheReadTokens: 88,
|
||||
cacheWriteTokens: 9,
|
||||
},
|
||||
usage: { inputTokens: 12, outputTokens: 4, cacheReadTokens: 88, cacheWriteTokens: 9 },
|
||||
},
|
||||
})
|
||||
const replayed = projector.snapshot(session, attached)
|
||||
expect(replayed).toMatchObject({
|
||||
uncachedInputTokens: 12,
|
||||
outputTokens: 4,
|
||||
cacheReadTokens: 88,
|
||||
cacheWriteTokens: 9,
|
||||
contextTokens: 100,
|
||||
})
|
||||
assistant(session, 1, 2, {
|
||||
inputTokens: 1_000,
|
||||
outputTokens: 500,
|
||||
cacheReadTokens: 2_000,
|
||||
cacheWriteTokens: 3_000,
|
||||
})
|
||||
|
||||
const after = projector.snapshot(session, attached)
|
||||
expect(after).toMatchObject({
|
||||
logRevision: session.events.length,
|
||||
projectionRevision: 3,
|
||||
uncachedInputTokens: 1_012,
|
||||
outputTokens: 504,
|
||||
cacheReadTokens: 2_088,
|
||||
cacheWriteTokens: 3_009,
|
||||
contextTokens: 200,
|
||||
})
|
||||
expect(after.uncachedInputTokens).not.toBe(
|
||||
after.uncachedInputTokens + after.cacheReadTokens + after.cacheWriteTokens,
|
||||
)
|
||||
})
|
||||
|
||||
it('publishes only the selected route capacity when asynchronous resolutions race', async () => {
|
||||
const ctx = new Context()
|
||||
const resolutions = new Map<string, {
|
||||
signal: AbortSignal | undefined
|
||||
resolve(contextWindow: number): void
|
||||
}>()
|
||||
ctx.provide('tokenMeter', { measure: () => ({ totalTokens: 35_000 }) })
|
||||
ctx.provide('llm', {
|
||||
resolveModelInfo(_provider: string, model: string, signal?: AbortSignal) {
|
||||
return new Promise<{ context: { contextWindow: number } }>((resolve) => {
|
||||
resolutions.set(model, {
|
||||
signal,
|
||||
resolve(contextWindow) {
|
||||
resolve({ context: { contextWindow } })
|
||||
},
|
||||
})
|
||||
})
|
||||
},
|
||||
})
|
||||
const session = new Session(SessionId('capacity-race'))
|
||||
const attached = agent(session)
|
||||
let current: AgentLlmTarget = { provider: 'test', model: 'alpha' }
|
||||
const resolved = vi.fn()
|
||||
const projector = new SessionMetricsProjector(ctx, () => current, resolved)
|
||||
|
||||
expect(projector.snapshot(session, attached).contextWindow).toBeUndefined()
|
||||
await vi.waitFor(() => { expect(resolutions.has('alpha')).toBe(true) })
|
||||
expect(resolutions.get('alpha')?.signal?.aborted).toBe(false)
|
||||
current = { provider: 'test', model: 'beta' }
|
||||
expect(projector.snapshot(session, attached).contextWindow).toBeUndefined()
|
||||
expect(resolutions.get('alpha')?.signal?.aborted).toBe(true)
|
||||
await vi.waitFor(() => { expect(resolutions.has('beta')).toBe(true) })
|
||||
expect(resolutions.get('beta')?.signal?.aborted).toBe(false)
|
||||
|
||||
resolutions.get('alpha')?.resolve(64_000)
|
||||
await Promise.resolve()
|
||||
expect(resolved).not.toHaveBeenCalled()
|
||||
resolutions.get('beta')?.resolve(128_000)
|
||||
await vi.waitFor(() => { expect(resolved).toHaveBeenCalledOnce() })
|
||||
expect(projector.snapshot(session, attached)).toMatchObject({
|
||||
contextTokens: 35_000,
|
||||
contextWindow: 128_000,
|
||||
})
|
||||
})
|
||||
|
||||
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('aborts every active capacity on invalidation and resolves fresh generations', async () => {
|
||||
const ctx = new Context()
|
||||
const attempts: {
|
||||
result: PromiseWithResolvers<{ context: { contextWindow: number } }>
|
||||
signal: AbortSignal | undefined
|
||||
}[] = []
|
||||
ctx.provide('llm', {
|
||||
resolveModelInfo(_provider: string, _model: string, signal?: AbortSignal) {
|
||||
const result = Promise.withResolvers<{ context: { contextWindow: number } }>()
|
||||
attempts.push({ result, signal })
|
||||
return result.promise
|
||||
},
|
||||
})
|
||||
const firstSession = new Session(SessionId('capacity-invalidation-first'))
|
||||
const secondSession = new Session(SessionId('capacity-invalidation-second'))
|
||||
const firstAgent = agent(firstSession)
|
||||
const secondAgent = agent(secondSession)
|
||||
const resolved = vi.fn()
|
||||
const projector = new SessionMetricsProjector(
|
||||
ctx,
|
||||
() => ({ provider: 'test', model: 'alpha' }),
|
||||
resolved,
|
||||
)
|
||||
|
||||
expect(projector.snapshot(firstSession, firstAgent).contextWindow).toBeUndefined()
|
||||
expect(projector.snapshot(secondSession, secondAgent).contextWindow).toBeUndefined()
|
||||
await vi.waitFor(() => { expect(attempts).toHaveLength(2) })
|
||||
expect(attempts.map(attempt => attempt.signal?.aborted)).toEqual([false, false])
|
||||
|
||||
projector.invalidateCapacities()
|
||||
expect(attempts.map(attempt => attempt.signal?.aborted)).toEqual([true, true])
|
||||
attempts[0]?.result.resolve({ context: { contextWindow: 32_000 } })
|
||||
attempts[1]?.result.resolve({ context: { contextWindow: 64_000 } })
|
||||
await settleAsyncWork()
|
||||
expect(resolved).not.toHaveBeenCalled()
|
||||
|
||||
expect(projector.snapshot(firstSession, firstAgent).contextWindow).toBeUndefined()
|
||||
expect(projector.snapshot(secondSession, secondAgent).contextWindow).toBeUndefined()
|
||||
await vi.waitFor(() => { expect(attempts).toHaveLength(4) })
|
||||
expect(attempts.slice(2).map(attempt => attempt.signal?.aborted)).toEqual([false, false])
|
||||
attempts[2]?.result.resolve({ context: { contextWindow: 128_000 } })
|
||||
attempts[3]?.result.resolve({ context: { contextWindow: 256_000 } })
|
||||
await vi.waitFor(() => { expect(resolved).toHaveBeenCalledTimes(2) })
|
||||
expect(projector.snapshot(firstSession, firstAgent).contextWindow).toBe(128_000)
|
||||
expect(projector.snapshot(secondSession, secondAgent).contextWindow).toBe(256_000)
|
||||
|
||||
projector.dispose()
|
||||
expect(projector.snapshot(firstSession, firstAgent).contextWindow).toBeUndefined()
|
||||
expect(attempts).toHaveLength(4)
|
||||
})
|
||||
|
||||
it('skips adapter work invalidated before its deferred invocation', async () => {
|
||||
const ctx = new Context()
|
||||
const attempts: {
|
||||
result: PromiseWithResolvers<{ context: { contextWindow: number } }>
|
||||
signal: AbortSignal | undefined
|
||||
}[] = []
|
||||
const resolveModelInfo = vi.fn((
|
||||
_provider: string,
|
||||
_model: string,
|
||||
signal?: AbortSignal,
|
||||
) => {
|
||||
const result = Promise.withResolvers<{ context: { contextWindow: number } }>()
|
||||
attempts.push({ result, signal })
|
||||
return result.promise
|
||||
})
|
||||
ctx.provide('llm', { resolveModelInfo })
|
||||
const session = new Session(SessionId('capacity-pre-invocation-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()
|
||||
projector.invalidateCapacities()
|
||||
await settleAsyncWork()
|
||||
expect(resolveModelInfo).not.toHaveBeenCalled()
|
||||
expect(resolved).not.toHaveBeenCalled()
|
||||
|
||||
expect(projector.snapshot(session, attached).contextWindow).toBeUndefined()
|
||||
await vi.waitFor(() => { expect(attempts).toHaveLength(1) })
|
||||
expect(attempts[0]?.signal?.aborted).toBe(false)
|
||||
attempts[0]?.result.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: {
|
||||
signal: AbortSignal | undefined
|
||||
resolve(contextWindow: number): void
|
||||
}[] = []
|
||||
ctx.provide('llm', {
|
||||
resolveModelInfo(_provider: string, _model: string, signal?: AbortSignal) {
|
||||
return new Promise<{ context: { contextWindow: number } }>((resolve) => {
|
||||
resolutions.push({
|
||||
signal,
|
||||
resolve(contextWindow) {
|
||||
resolve({ context: { contextWindow } })
|
||||
},
|
||||
})
|
||||
})
|
||||
},
|
||||
})
|
||||
const session = new Session(SessionId('capacity-route-return'))
|
||||
const attached = agent(session)
|
||||
let current: AgentLlmTarget | undefined = { provider: 'test', model: 'alpha' }
|
||||
const resolved = vi.fn()
|
||||
const targetFor = vi.fn(() => current)
|
||||
const projector = new SessionMetricsProjector(ctx, targetFor, resolved)
|
||||
|
||||
expect(projector.snapshot(session, attached).contextWindow).toBeUndefined()
|
||||
await vi.waitFor(() => { expect(resolutions).toHaveLength(1) })
|
||||
current = undefined
|
||||
expect(projector.snapshot(session, attached).contextWindow).toBeUndefined()
|
||||
expect(resolutions[0]?.signal?.aborted).toBe(true)
|
||||
resolutions[0]?.resolve(64_000)
|
||||
await settleAsyncWork()
|
||||
expect(resolved).not.toHaveBeenCalled()
|
||||
current = { provider: 'test', model: 'alpha' }
|
||||
expect(projector.snapshot(session, attached).contextWindow).toBeUndefined()
|
||||
await vi.waitFor(() => { expect(resolutions).toHaveLength(2) })
|
||||
expect(resolutions[1]?.signal?.aborted).toBe(false)
|
||||
resolutions[1]?.resolve(128_000)
|
||||
await vi.waitFor(() => { expect(resolved).toHaveBeenCalledOnce() })
|
||||
expect(projector.snapshot(session, attached).contextWindow).toBe(128_000)
|
||||
})
|
||||
|
||||
it('omits current context fields when measurement or model metadata is unavailable', async () => {
|
||||
const ctx = new Context()
|
||||
ctx.provide('tokenMeter', { measure: () => { throw new Error('unmeasurable') } })
|
||||
ctx.provide('llm', { resolveModelInfo: () => Promise.reject(new Error('metadata unavailable')) })
|
||||
const session = new Session(SessionId('missing-metrics'))
|
||||
const attached = agent(session)
|
||||
const projector = new SessionMetricsProjector(
|
||||
ctx,
|
||||
() => ({ provider: 'test', model: 'missing' }),
|
||||
() => {},
|
||||
)
|
||||
const metrics = projector.snapshot(session, attached)
|
||||
expect(metrics.contextTokens).toBeUndefined()
|
||||
expect(metrics.contextWindow).toBeUndefined()
|
||||
await vi.waitFor(() => {
|
||||
expect(projector.snapshot(session, attached).contextWindow).toBeUndefined()
|
||||
})
|
||||
})
|
||||
|
||||
it('keeps optional usage buckets at zero and tolerates absent host services or detached agents', async () => {
|
||||
const ctx = new Context()
|
||||
const session = new Session(SessionId('optional-metrics'))
|
||||
assistant(session, 1, 0, { inputTokens: 7, outputTokens: 2 })
|
||||
const attached = agent(session)
|
||||
const selected: { current?: AgentLlmTarget } = {}
|
||||
const projector = new SessionMetricsProjector(
|
||||
ctx,
|
||||
() => selected.current,
|
||||
() => {},
|
||||
)
|
||||
assistant(session, 1, 2, { inputTokens: 1_000, outputTokens: 500 })
|
||||
|
||||
expect(projector.snapshot(session)).toMatchObject({
|
||||
uncachedInputTokens: 7,
|
||||
outputTokens: 2,
|
||||
cacheReadTokens: 0,
|
||||
cacheWriteTokens: 0,
|
||||
logRevision: session.events.length,
|
||||
projectionRevision: 2,
|
||||
uncachedInputTokens: 1_012,
|
||||
outputTokens: 504,
|
||||
cacheReadTokens: 88,
|
||||
cacheWriteTokens: 9,
|
||||
contextTokens: 200,
|
||||
})
|
||||
expect(projector.snapshot(session, attached).contextWindow).toBeUndefined()
|
||||
selected.current = { provider: 'test', model: 'no-service' }
|
||||
expect(projector.snapshot(session, attached).contextWindow).toBeUndefined()
|
||||
await Promise.resolve()
|
||||
})
|
||||
|
||||
it('publishes a resolved route with no advertised capacity as unknown', async () => {
|
||||
const ctx = new Context()
|
||||
ctx.provide('llm', { resolveModelInfo: () => Promise.resolve({}) })
|
||||
const session = new Session(SessionId('no-capacity'))
|
||||
const attached = agent(session)
|
||||
const resolved = vi.fn()
|
||||
const projector = new SessionMetricsProjector(
|
||||
ctx,
|
||||
() => ({ provider: 'test', model: 'metadata-without-context' }),
|
||||
resolved,
|
||||
)
|
||||
it('omits pressure when the token meter is absent or cannot measure the replay', () => {
|
||||
const session = new Session(SessionId('metrics-pressure-unknown'))
|
||||
const withoutMeter = new SessionMetricsProjector(new Context()).snapshot(session)
|
||||
expect(withoutMeter.contextTokens).toBeUndefined()
|
||||
|
||||
expect(projector.snapshot(session, attached).contextWindow).toBeUndefined()
|
||||
await vi.waitFor(() => { expect(resolved).toHaveBeenCalledOnce() })
|
||||
expect(projector.snapshot(session, attached).contextWindow).toBeUndefined()
|
||||
})
|
||||
|
||||
it('ignores stale resolution failures and route metadata after the target moves', async () => {
|
||||
const ctx = new Context()
|
||||
const resolutions = new Map<string, {
|
||||
resolve(value: { context: { contextWindow: number } }): void
|
||||
reject(error: Error): void
|
||||
}>()
|
||||
ctx.provide('llm', {
|
||||
resolveModelInfo(_provider: string, model: string) {
|
||||
return new Promise<{ context: { contextWindow: number } }>((resolve, reject) => {
|
||||
resolutions.set(model, { resolve, reject })
|
||||
})
|
||||
ctx.provide('tokenMeter', {
|
||||
measure() {
|
||||
throw new Error('unmeasurable replay')
|
||||
},
|
||||
})
|
||||
const session = new Session(SessionId('stale-capacity'))
|
||||
const attached = agent(session)
|
||||
let current: AgentLlmTarget = { provider: 'test', model: 'alpha' }
|
||||
const resolved = vi.fn()
|
||||
const targetFor = vi.fn(() => current)
|
||||
const projector = new SessionMetricsProjector(ctx, targetFor, resolved)
|
||||
|
||||
projector.snapshot(session, attached)
|
||||
await vi.waitFor(() => { expect(resolutions.has('alpha')).toBe(true) })
|
||||
current = { provider: 'test', model: 'route-moved-before-snapshot' }
|
||||
resolutions.get('alpha')?.resolve({ context: { contextWindow: 64_000 } })
|
||||
await vi.waitFor(() => { expect(targetFor).toHaveBeenCalledTimes(2) })
|
||||
expect(resolved).not.toHaveBeenCalled()
|
||||
|
||||
projector.snapshot(session, attached)
|
||||
await vi.waitFor(() => { expect(resolutions.has('route-moved-before-snapshot')).toBe(true) })
|
||||
current = { provider: 'test', model: 'beta' }
|
||||
projector.snapshot(session, attached)
|
||||
await vi.waitFor(() => { expect(resolutions.has('beta')).toBe(true) })
|
||||
resolutions.get('route-moved-before-snapshot')?.reject(new Error('stale failure'))
|
||||
await Promise.resolve()
|
||||
resolutions.get('beta')?.resolve({ context: { contextWindow: 128_000 } })
|
||||
await vi.waitFor(() => { expect(resolved).toHaveBeenCalledOnce() })
|
||||
expect(projector.snapshot(session, attached).contextWindow).toBe(128_000)
|
||||
expect(new SessionMetricsProjector(ctx).snapshot(session).contextTokens).toBeUndefined()
|
||||
})
|
||||
})
|
||||
|
||||
Reference in New Issue
Block a user