fix(host): pair metric agents by lifecycle (round 5)
This commit is contained in:
@@ -419,6 +419,12 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
|
|||||||
: { provider: logged.provider, model: logged.model }
|
: { provider: logged.provider, model: logged.model }
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/** Pair a registry agent only with the exact Session lifecycle it owns. */
|
||||||
|
function metricsAgentFor(session: Session): Agent | undefined {
|
||||||
|
const agent = ctx.agents.get(session.id)
|
||||||
|
return agent?.session === session ? agent : undefined
|
||||||
|
}
|
||||||
|
|
||||||
/** 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
|
||||||
@@ -464,7 +470,7 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
|
|||||||
broadcast({
|
broadcast({
|
||||||
type: 'session/metrics',
|
type: 'session/metrics',
|
||||||
sessionId: current.id,
|
sessionId: current.id,
|
||||||
metrics: metricsProjector.snapshot(current, ctx.agents.get(current.id)),
|
metrics: metricsProjector.snapshot(current, metricsAgentFor(current)),
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
})
|
})
|
||||||
@@ -1186,7 +1192,7 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
|
|||||||
queue.push(frame({
|
queue.push(frame({
|
||||||
type: 'session/metrics',
|
type: 'session/metrics',
|
||||||
sessionId: session.id,
|
sessionId: session.id,
|
||||||
metrics: metricsProjector.snapshot(session, ctx.agents.get(session.id)),
|
metrics: metricsProjector.snapshot(session, metricsAgentFor(session)),
|
||||||
}))
|
}))
|
||||||
}
|
}
|
||||||
for (const pending of pendingQuestions.values()) {
|
for (const pending of pendingQuestions.values()) {
|
||||||
@@ -1243,7 +1249,7 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
|
|||||||
queue.push(frame({
|
queue.push(frame({
|
||||||
type: 'session/metrics',
|
type: 'session/metrics',
|
||||||
sessionId: session.id,
|
sessionId: session.id,
|
||||||
metrics: metricsProjector.snapshot(session, ctx.agents.get(session.id)),
|
metrics: metricsProjector.snapshot(session, metricsAgentFor(session)),
|
||||||
}))
|
}))
|
||||||
}),
|
}),
|
||||||
ctx.on('session/disposed', (session: Session) => {
|
ctx.on('session/disposed', (session: Session) => {
|
||||||
|
|||||||
@@ -454,4 +454,59 @@ describe('Web session model selection', () => {
|
|||||||
liveSession.detach()
|
liveSession.detach()
|
||||||
await ctx.fiber.dispose()
|
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()
|
||||||
|
})
|
||||||
})
|
})
|
||||||
|
|||||||
Reference in New Issue
Block a user