fix(host): keep metrics route lookup passive (round 3)
This commit is contained in:
@@ -405,6 +405,20 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
|
|||||||
return target
|
return target
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Read the best capacity route without taking ownership of foreign routing.
|
||||||
|
* Web agents expose their live selection; other agents expose only a route
|
||||||
|
* that already crossed the durable request-header boundary.
|
||||||
|
*/
|
||||||
|
function metricsRouteFor(agent: Agent): Pick<AgentLlmTarget, 'provider' | 'model'> | undefined {
|
||||||
|
const installed = targets.get(agent)
|
||||||
|
if (installed !== undefined) return installed.current
|
||||||
|
const logged = agent.session.requestHeader()?.config
|
||||||
|
return logged === undefined
|
||||||
|
? undefined
|
||||||
|
: { provider: logged.provider, model: logged.model }
|
||||||
|
}
|
||||||
|
|
||||||
/** Pre-publication setup used by both fresh and resumed Web agents. */
|
/** Pre-publication setup used by both fresh and resumed Web agents. */
|
||||||
function installTarget(agentCtx: Context): void {
|
function installTarget(agentCtx: Context): void {
|
||||||
const agent = agentCtx.agent
|
const agent = agentCtx.agent
|
||||||
@@ -423,7 +437,7 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
|
|||||||
let metricsDisposed = false
|
let metricsDisposed = false
|
||||||
const metricsProjector = new SessionMetricsProjector(
|
const metricsProjector = new SessionMetricsProjector(
|
||||||
ctx,
|
ctx,
|
||||||
agent => targetFor(agent).current,
|
metricsRouteFor,
|
||||||
(agent) => { scheduleMetrics(agent.session) },
|
(agent) => { scheduleMetrics(agent.session) },
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|||||||
@@ -21,12 +21,14 @@ interface UsageState {
|
|||||||
}
|
}
|
||||||
|
|
||||||
interface CapacityState {
|
interface CapacityState {
|
||||||
routeKey: string
|
routeKey: string | undefined
|
||||||
generation: number
|
generation: number
|
||||||
status: 'pending' | 'ready'
|
status: 'pending' | 'ready'
|
||||||
contextWindow?: number
|
contextWindow?: number
|
||||||
}
|
}
|
||||||
|
|
||||||
|
type CapacityTarget = Pick<AgentLlmTarget, 'provider' | 'model'>
|
||||||
|
|
||||||
interface TokenMeterLike {
|
interface TokenMeterLike {
|
||||||
measure(session: Session): { totalTokens: number }
|
measure(session: Session): { totalTokens: number }
|
||||||
}
|
}
|
||||||
@@ -75,6 +77,10 @@ function recordUsage(state: UsageState, turn: number, step: number, usage: Token
|
|||||||
state.cacheWriteTokens += usage.cacheWriteTokens ?? 0
|
state.cacheWriteTokens += usage.cacheWriteTokens ?? 0
|
||||||
}
|
}
|
||||||
|
|
||||||
|
function routeKeyFor(target: CapacityTarget | undefined): string | undefined {
|
||||||
|
return target === undefined ? undefined : `${target.provider}\u0000${target.model}`
|
||||||
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Projects durable cumulative usage and route-aware current context without
|
* Projects durable cumulative usage and route-aware current context without
|
||||||
* awaiting model metadata on the session append path.
|
* awaiting model metadata on the session append path.
|
||||||
@@ -85,12 +91,12 @@ export class SessionMetricsProjector {
|
|||||||
|
|
||||||
/**
|
/**
|
||||||
* @param ctx - Host context providing optional token-meter and LLM services.
|
* @param ctx - Host context providing optional token-meter and LLM services.
|
||||||
* @param targetFor - selected route owner for one attached Web agent.
|
* @param targetFor - side-effect-free selected or logged route lookup for one attached agent.
|
||||||
* @param onCapacityResolved - schedules a fresh live projection after exact-route metadata resolves.
|
* @param onCapacityResolved - schedules a fresh live projection after exact-route metadata resolves.
|
||||||
*/
|
*/
|
||||||
constructor(
|
constructor(
|
||||||
private readonly ctx: Context,
|
private readonly ctx: Context,
|
||||||
private readonly targetFor: (agent: Agent) => Pick<AgentLlmTarget, 'provider' | 'model'>,
|
private readonly targetFor: (agent: Agent) => CapacityTarget | undefined,
|
||||||
private readonly onCapacityResolved: (agent: Agent) => void,
|
private readonly onCapacityResolved: (agent: Agent) => void,
|
||||||
) {}
|
) {}
|
||||||
|
|
||||||
@@ -151,23 +157,23 @@ export class SessionMetricsProjector {
|
|||||||
|
|
||||||
private capacityFor(agent: Agent): number | undefined {
|
private capacityFor(agent: Agent): number | undefined {
|
||||||
const target = this.targetFor(agent)
|
const target = this.targetFor(agent)
|
||||||
const routeKey = `${target.provider}\u0000${target.model}`
|
const routeKey = routeKeyFor(target)
|
||||||
let state = this.capacities.get(agent)
|
let state = this.capacities.get(agent)
|
||||||
if (state === undefined || state.routeKey !== routeKey) {
|
if (state === undefined || state.routeKey !== routeKey) {
|
||||||
state = {
|
state = {
|
||||||
routeKey,
|
routeKey,
|
||||||
generation: (state?.generation ?? 0) + 1,
|
generation: (state?.generation ?? 0) + 1,
|
||||||
status: 'pending',
|
status: target === undefined ? 'ready' : 'pending',
|
||||||
}
|
}
|
||||||
this.capacities.set(agent, state)
|
this.capacities.set(agent, state)
|
||||||
this.resolveCapacity(agent, target, state)
|
if (target !== undefined) this.resolveCapacity(agent, target, state)
|
||||||
}
|
}
|
||||||
return state.status === 'ready' ? state.contextWindow : undefined
|
return state.status === 'ready' ? state.contextWindow : undefined
|
||||||
}
|
}
|
||||||
|
|
||||||
private resolveCapacity(
|
private resolveCapacity(
|
||||||
agent: Agent,
|
agent: Agent,
|
||||||
target: Pick<AgentLlmTarget, 'provider' | 'model'>,
|
target: CapacityTarget,
|
||||||
pending: CapacityState,
|
pending: CapacityState,
|
||||||
): void {
|
): void {
|
||||||
const llm = this.ctx.get('llm') as LlmLike | undefined
|
const llm = this.ctx.get('llm') as LlmLike | undefined
|
||||||
@@ -179,16 +185,27 @@ export class SessionMetricsProjector {
|
|||||||
.then(() => llm.resolveModelInfo(target.provider, target.model))
|
.then(() => llm.resolveModelInfo(target.provider, target.model))
|
||||||
.then(
|
.then(
|
||||||
(resolved) => {
|
(resolved) => {
|
||||||
if (this.capacities.get(agent)?.generation !== pending.generation) return
|
if (this.capacityResolutionIsStale(agent, pending)) return
|
||||||
const current = this.targetFor(agent)
|
|
||||||
if (`${current.provider}\u0000${current.model}` !== pending.routeKey) return
|
|
||||||
pending.status = 'ready'
|
pending.status = 'ready'
|
||||||
if (resolved.context !== undefined) pending.contextWindow = resolved.context.contextWindow
|
if (resolved.context !== undefined) pending.contextWindow = resolved.context.contextWindow
|
||||||
this.onCapacityResolved(agent)
|
this.onCapacityResolved(agent)
|
||||||
},
|
},
|
||||||
() => {
|
() => {
|
||||||
if (this.capacities.get(agent)?.generation === pending.generation) pending.status = 'ready'
|
if (!this.capacityResolutionIsStale(agent, pending)) pending.status = 'ready'
|
||||||
},
|
},
|
||||||
)
|
)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
private capacityResolutionIsStale(agent: Agent, pending: CapacityState): boolean {
|
||||||
|
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
|
||||||
|
// starts a fresh resolution even when it equals the route that disappeared.
|
||||||
|
this.capacities.set(agent, {
|
||||||
|
routeKey: undefined,
|
||||||
|
generation: pending.generation + 1,
|
||||||
|
status: 'ready',
|
||||||
|
})
|
||||||
|
return true
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -6,8 +6,8 @@
|
|||||||
|
|
||||||
import { describe, expect, it } from 'vitest'
|
import { describe, expect, it } from 'vitest'
|
||||||
import { Context } from 'cordis'
|
import { Context } from 'cordis'
|
||||||
import AgentRegistry, { agentEvents } from '@deepseek-ai/dsh-agent'
|
import AgentRegistry, { agentEvents, installAgentLlmTarget } from '@deepseek-ai/dsh-agent'
|
||||||
import type { Agent } from '@deepseek-ai/dsh-agent'
|
import type { Agent, AgentLlmTargetRef } from '@deepseek-ai/dsh-agent'
|
||||||
import LlmService, { LlmAdapter, ReasoningEffortId } from '@deepseek-ai/dsh-llm'
|
import LlmService, { LlmAdapter, ReasoningEffortId } from '@deepseek-ai/dsh-llm'
|
||||||
import type {
|
import type {
|
||||||
GenerateOptions, LlmCallConfig, LlmModelInfo, LlmModelReasoningInfo, LlmProviderInfo,
|
GenerateOptions, LlmCallConfig, LlmModelInfo, LlmModelReasoningInfo, LlmProviderInfo,
|
||||||
@@ -72,15 +72,7 @@ const REASONING: LlmModelReasoningInfo = {
|
|||||||
defaultEffort: ReasoningEffortId('high'),
|
defaultEffort: ReasoningEffortId('high'),
|
||||||
}
|
}
|
||||||
|
|
||||||
async function harness(logged?: {
|
async function hostContext(): Promise<Context> {
|
||||||
provider: string
|
|
||||||
model: string
|
|
||||||
reasoningEffort?: ReasoningEffortId
|
|
||||||
}): Promise<{
|
|
||||||
ctx: Context
|
|
||||||
agent: Agent
|
|
||||||
sessionId: SessionId
|
|
||||||
}> {
|
|
||||||
const ctx = new Context()
|
const ctx = new Context()
|
||||||
await ctx.plugin(SessionStore)
|
await ctx.plugin(SessionStore)
|
||||||
await ctx.plugin(SystemPrompt, { persona: '' })
|
await ctx.plugin(SystemPrompt, { persona: '' })
|
||||||
@@ -100,6 +92,19 @@ async function harness(logged?: {
|
|||||||
{ provider: 'duplicate', id: 'same', name: 'Same' },
|
{ provider: 'duplicate', id: 'same', name: 'Same' },
|
||||||
{ provider: 'duplicate', id: 'same', name: 'Same Again' },
|
{ provider: 'duplicate', id: 'same', name: 'Same Again' },
|
||||||
]))
|
]))
|
||||||
|
return ctx
|
||||||
|
}
|
||||||
|
|
||||||
|
async function harness(logged?: {
|
||||||
|
provider: string
|
||||||
|
model: string
|
||||||
|
reasoningEffort?: ReasoningEffortId
|
||||||
|
}): Promise<{
|
||||||
|
ctx: Context
|
||||||
|
agent: Agent
|
||||||
|
sessionId: SessionId
|
||||||
|
}> {
|
||||||
|
const ctx = await hostContext()
|
||||||
const session = ctx.sessions.create()
|
const session = ctx.sessions.create()
|
||||||
if (logged !== undefined) {
|
if (logged !== undefined) {
|
||||||
session.append('request/header', { header: { config: logged }, reason: 'initial' })
|
session.append('request/header', { header: { config: logged }, reason: 'initial' })
|
||||||
@@ -246,6 +251,7 @@ describe('Web session model selection', () => {
|
|||||||
it('publishes unknown capacity immediately on selection, then the exact selected route capacity', async () => {
|
it('publishes unknown capacity immediately on selection, then the exact selected route capacity', async () => {
|
||||||
const { ctx, sessionId } = await harness()
|
const { ctx, 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' })
|
||||||
|
expectValue(await api.sessions.models(request({ sessionId })))
|
||||||
const controller = new AbortController()
|
const controller = new AbortController()
|
||||||
const iterator = api.events.mux(request({}), controller.signal)[Symbol.asyncIterator]()
|
const iterator = api.events.mux(request({}), controller.signal)[Symbol.asyncIterator]()
|
||||||
|
|
||||||
@@ -264,4 +270,56 @@ describe('Web session model selection', () => {
|
|||||||
await iterator.return?.()
|
await iterator.return?.()
|
||||||
await ctx.fiber.dispose()
|
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()
|
||||||
|
})
|
||||||
})
|
})
|
||||||
|
|||||||
@@ -187,6 +187,37 @@ describe('SessionMetricsProjector', () => {
|
|||||||
})
|
})
|
||||||
})
|
})
|
||||||
|
|
||||||
|
it('starts a fresh capacity generation when an unavailable route returns', async () => {
|
||||||
|
const ctx = new Context()
|
||||||
|
const resolutions: ((contextWindow: number) => void)[] = []
|
||||||
|
ctx.provide('llm', {
|
||||||
|
resolveModelInfo() {
|
||||||
|
return new Promise<{ context: { contextWindow: number } }>((resolve) => {
|
||||||
|
resolutions.push((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
|
||||||
|
resolutions[0]?.(64_000)
|
||||||
|
await vi.waitFor(() => { expect(targetFor).toHaveBeenCalledTimes(2) })
|
||||||
|
expect(resolved).not.toHaveBeenCalled()
|
||||||
|
current = { provider: 'test', model: 'alpha' }
|
||||||
|
expect(projector.snapshot(session, attached).contextWindow).toBeUndefined()
|
||||||
|
await vi.waitFor(() => { expect(resolutions).toHaveLength(2) })
|
||||||
|
resolutions[1]?.(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 () => {
|
it('omits current context fields when measurement or model metadata is unavailable', async () => {
|
||||||
const ctx = new Context()
|
const ctx = new Context()
|
||||||
ctx.provide('tokenMeter', { measure: () => { throw new Error('unmeasurable') } })
|
ctx.provide('tokenMeter', { measure: () => { throw new Error('unmeasurable') } })
|
||||||
@@ -211,9 +242,10 @@ describe('SessionMetricsProjector', () => {
|
|||||||
const session = new Session(SessionId('optional-metrics'))
|
const session = new Session(SessionId('optional-metrics'))
|
||||||
assistant(session, 1, 0, { inputTokens: 7, outputTokens: 2 })
|
assistant(session, 1, 0, { inputTokens: 7, outputTokens: 2 })
|
||||||
const attached = agent(session)
|
const attached = agent(session)
|
||||||
|
const selected: { current?: AgentLlmTarget } = {}
|
||||||
const projector = new SessionMetricsProjector(
|
const projector = new SessionMetricsProjector(
|
||||||
ctx,
|
ctx,
|
||||||
() => ({ provider: 'test', model: 'no-service' }),
|
() => selected.current,
|
||||||
() => {},
|
() => {},
|
||||||
)
|
)
|
||||||
|
|
||||||
@@ -224,6 +256,7 @@ describe('SessionMetricsProjector', () => {
|
|||||||
cacheWriteTokens: 0,
|
cacheWriteTokens: 0,
|
||||||
})
|
})
|
||||||
expect(projector.snapshot(session, attached).contextWindow).toBeUndefined()
|
expect(projector.snapshot(session, attached).contextWindow).toBeUndefined()
|
||||||
|
selected.current = { provider: 'test', model: 'no-service' }
|
||||||
expect(projector.snapshot(session, attached).contextWindow).toBeUndefined()
|
expect(projector.snapshot(session, attached).contextWindow).toBeUndefined()
|
||||||
await Promise.resolve()
|
await Promise.resolve()
|
||||||
})
|
})
|
||||||
|
|||||||
Reference in New Issue
Block a user