Hypatia May
4819210142
Replace the transient `session/model-request` mux frame with ordinary durable session state. Occupancy now rides two last-wins projection fields instead of a non-replayable frame that needed removal tombstones and cross-stream fencing. The frame was the only non-replayable class on the mux stream. Because host and mux are independent SSE streams with no cross-stream order, a request emitted before a removal could arrive after `host/session-removed`, and a legitimate request for a new lifecycle reusing the same id could be fenced by a late removal. Fixing that needed a lifecycle generation on every frame; the frame itself was the problem. Removed: the `session/model-request` frame and schema, the `agent/model-request` core event, the ApiProxy measurement point, the client-side telemetry map and removal tombstone, and the synthetic `cancelled` open error used to signal reconnect through the error channel. Added: `request/context`, a log-only session event recording the registration-bound capacity of the route a request resolved to, appended beside `request/header` from the lookup that already prepared the call and skipped when the route is unchanged. Capacity stays out of `EpochHeader` because it is adapter metadata about a route, not an input the request was built from, so it must not join request reconstruction or header equality. The `contextPressure` projection pairs the newest provider-reported prompt size with the newest recorded capacity. The two are deliberately not one atomic request observation: switching models can pair a fresh capacity with the prior route's pressure until the next request reports usage. The figure is a user-facing reference, and this matches how the TUI status line has always computed occupancy.
114 lines
4.6 KiB
TypeScript
114 lines
4.6 KiB
TypeScript
/**
|
|
* Runtime plugin browser-half apply: slots + object services mounting over the
|
|
* connection handle, stream-loop sink wiring into the object layer, and the
|
|
* fiber-scoped loop teardown.
|
|
*/
|
|
import { Context } from 'cordis'
|
|
import { describe, expect, it } from 'vitest'
|
|
import type { ConnectionHandle } from '@deepseek-ai/dsh-client-connection/client'
|
|
import type { ConnectionSinks } from '@deepseek-ai/dsh-client-connection/client'
|
|
import * as RuntimeClient from '../src/client/index.ts'
|
|
import type { SessionsService } from '../src/client/sessions/service.ts'
|
|
import type { WorkspacesService } from '../src/client/workspaces/service.ts'
|
|
import { FakeApiClient, ok } from './fake-api.ts'
|
|
|
|
interface Bench {
|
|
ctx: Context
|
|
api: FakeApiClient
|
|
sinks: ConnectionSinks | undefined
|
|
stopped: number
|
|
}
|
|
|
|
async function mount(): Promise<Bench> {
|
|
const ctx = new Context()
|
|
const api = new FakeApiClient()
|
|
const bench: Bench = { ctx, api, sinks: undefined, stopped: 0 }
|
|
const handle: ConnectionHandle = {
|
|
api,
|
|
start: (sinks) => {
|
|
bench.sinks = sinks
|
|
return { stop: () => { bench.stopped += 1 } }
|
|
},
|
|
}
|
|
ctx.reflect.provide('connection', handle)
|
|
await ctx.plugin(RuntimeClient).await()
|
|
return bench
|
|
}
|
|
|
|
async function flushMicrotasks(): Promise<void> {
|
|
for (let i = 0; i < 12; i++) await Promise.resolve()
|
|
}
|
|
|
|
describe('runtime client apply', () => {
|
|
it('mounts slots, Sessions, and Workspaces and fans host frames into both managers', async () => {
|
|
const bench = await mount()
|
|
expect(bench.ctx.get('slots') !== undefined).toBe(true)
|
|
// The built-in 'root' declaration ships with this package's SlotsService
|
|
// (the SlotMap 'root' merge lives here since the slot-parity rework).
|
|
expect(bench.ctx.slots.spec('root')).toEqual({ kind: 'single', scope: 'root' })
|
|
const sessions = bench.ctx.get('sessions')
|
|
const workspaces = bench.ctx.get('workspaces')
|
|
expect(sessions !== undefined).toBe(true)
|
|
expect(workspaces !== undefined).toBe(true)
|
|
if (workspaces === undefined) throw new Error('WorkspacesService missing after runtime apply')
|
|
expect(bench.sinks).toBeDefined()
|
|
|
|
// Frame sinks reach the object layer: a host session-added lands in the list store.
|
|
bench.sinks?.onHostEnvelope?.({
|
|
rpcId: 'r1' as never,
|
|
payload: { type: 'host/session-added', blank: true, sessionId: 's-new' } as never,
|
|
})
|
|
await Promise.resolve()
|
|
expect((sessions as { list: { getSnapshot(): { ids: string[] } } }).list.getSnapshot().ids).toContain('s-new')
|
|
bench.sinks?.onHostEnvelope?.({
|
|
rpcId: 'r-workspace' as never,
|
|
payload: {
|
|
type: 'host/workspace-changed',
|
|
workspace: {
|
|
workspaceId: 'w-new', path: '/w/new', title: 'new', sessionIds: [],
|
|
createdAt: '2026-01-01T00:00:00.000Z', updatedAt: '2026-01-01T00:00:00.000Z',
|
|
},
|
|
} as never,
|
|
})
|
|
await Promise.resolve()
|
|
expect(workspaces.list.getSnapshot().items[0]?.workspaceId).toBe('w-new')
|
|
// Mux sink and onConnected route without throwing (manager semantics own the behavior).
|
|
bench.sinks?.onMuxEnvelope?.({ rpcId: 'r2' as never, payload: { type: 'stream/error', message: 'x' } as never })
|
|
bench.sinks?.onConnected?.()
|
|
})
|
|
|
|
it('selects the recent Workspace once when the first baselines have no current session', async () => {
|
|
const bench = await mount()
|
|
bench.api.onWorkspaceList = () => Promise.resolve(ok({
|
|
items: [{
|
|
workspaceId: 'w-recent', path: '/w/recent', title: 'recent', sessionIds: [],
|
|
createdAt: '2026-01-01T00:00:00.000Z', updatedAt: '2026-01-01T00:00:00.000Z',
|
|
}] as never[],
|
|
}))
|
|
bench.api.onList = () => Promise.resolve(ok({ items: [] }))
|
|
|
|
bench.sinks?.onConnected?.()
|
|
await flushMicrotasks()
|
|
|
|
const sessions = bench.ctx.get('sessions') as SessionsService
|
|
const workspaces = bench.ctx.get('workspaces') as WorkspacesService
|
|
expect(bench.api.callsOf('session.create')).toEqual([{ workspaceId: 'w-recent' }])
|
|
expect(sessions.list.getSnapshot().current).toBe('fk-new')
|
|
|
|
sessions.clear()
|
|
await workspaces.refresh()
|
|
await flushMicrotasks()
|
|
expect(sessions.list.getSnapshot().current).toBeUndefined()
|
|
expect(bench.api.callsOf('session.create')).toHaveLength(1)
|
|
})
|
|
|
|
it('stops the stream loop when the plugin fiber unloads', async () => {
|
|
const bench = await mount()
|
|
const fiber = [...bench.ctx.registry.values()].find(f => f.name?.includes('client'))
|
|
// Dispose the whole tree: the ctx.effect teardown must call loop.stop exactly once.
|
|
await bench.ctx.fiber.dispose()
|
|
expect(bench.stopped).toBe(1)
|
|
void fiber
|
|
})
|
|
})
|