refactor(token-meter): make context occupancy durable projection state
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.
This commit is contained in:
@@ -12,7 +12,7 @@ export type {
|
||||
WorkspaceApi, WorkspaceId, WorkspaceView,
|
||||
CommandsApi, CommandDescriptor, SkillsApi, SkillEntry,
|
||||
ModelCatalogFailure, ModelCatalogModel, ModelProviderGroup, ModelReasoning,
|
||||
ModelReasoningEffort, ModelRequestTelemetry, ModelTarget, SessionModels, SessionProjectionsBlock,
|
||||
ModelReasoningEffort, ModelTarget, SessionModels,
|
||||
GoalsApi, GoalRef,
|
||||
} from '@deepseek-ai/dsh-host-apiproxy/api'
|
||||
export type { ToolCallView, ToolResultView } from '@deepseek-ai/dsh-tools/presentation'
|
||||
|
||||
@@ -35,6 +35,10 @@ function sleep(ms: number, signal: AbortSignal): Promise<void> {
|
||||
})
|
||||
}
|
||||
|
||||
/** Coarse connection state for the UI (audit C1): 'connected' after each generation's handshake,
|
||||
* 'reconnecting' the moment the generation fails (covers the whole backoff+retry span). */
|
||||
export type ConnectionState = 'connected' | 'reconnecting'
|
||||
|
||||
/** Frame sink callbacks: the Controller owns the physical streams; business dispatch belongs to
|
||||
* SessionManager. */
|
||||
export interface ConnectionSinks {
|
||||
@@ -42,8 +46,9 @@ export interface ConnectionSinks {
|
||||
onHostEnvelope?: (envelope: RpcRequest<HostFrame>) => void
|
||||
/** After each connection generation is established (both streams open + describe succeeded), first connect included. */
|
||||
onConnected?: () => void
|
||||
/** After every failed generation closes and before retry starts. Not emitted when the controller is stopped. */
|
||||
onDisconnected?: () => void
|
||||
/** Coarse state transitions (deduplicated: fires only on change). The initial pre-connect
|
||||
* span reports nothing — the UI treats "no state yet" as connecting, not as an outage. */
|
||||
onStateChange?: (state: ConnectionState) => void
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -58,6 +63,7 @@ export class ConnectionController {
|
||||
private attempt = 0
|
||||
private current: AbortController | null = null
|
||||
private running = false
|
||||
private lastState: ConnectionState | null = null
|
||||
private readonly config: Required<ConnectionConfig>
|
||||
|
||||
constructor(
|
||||
@@ -114,8 +120,8 @@ export class ConnectionController {
|
||||
if (gen === this.generation && !ac.signal.aborted) ac.abort()
|
||||
resolve()
|
||||
}
|
||||
void this.pumpStream(this.api.events.mux({}, ac.signal, muxOpened), this.sinks.onMuxEnvelope, ac.signal, settle)
|
||||
void this.pumpStream(this.api.events.host({}, ac.signal, hostOpened), this.sinks.onHostEnvelope, ac.signal, settle)
|
||||
void this.pumpStream(this.api.events.mux({}, ac.signal, muxOpened), this.sinks.onMuxEnvelope, settle)
|
||||
void this.pumpStream(this.api.events.host({}, ac.signal, hostOpened), this.sinks.onHostEnvelope, settle)
|
||||
})
|
||||
|
||||
try {
|
||||
@@ -132,6 +138,7 @@ export class ConnectionController {
|
||||
timeout.abort()
|
||||
if (ac.signal.aborted) throw new Error('generation aborted during readiness handshake')
|
||||
this.attempt = 0
|
||||
this.emitState('connected')
|
||||
this.callSink(this.sinks.onConnected)
|
||||
} catch {
|
||||
// Transport failure: treat as generation failure, fall through to the shared backoff.
|
||||
@@ -140,7 +147,7 @@ export class ConnectionController {
|
||||
|
||||
await failed
|
||||
if (!this.isRunning()) return
|
||||
this.callSink(this.sinks.onDisconnected)
|
||||
this.emitState('reconnecting')
|
||||
this.attempt += 1
|
||||
console.warn(`[web-runtime] connection lost, retry #${this.attempt}`)
|
||||
const idle = new AbortController()
|
||||
@@ -148,15 +155,20 @@ export class ConnectionController {
|
||||
}
|
||||
}
|
||||
|
||||
/** Deduplicated state emission (sink isolation applies). */
|
||||
private emitState(state: ConnectionState): void {
|
||||
if (this.lastState === state) return
|
||||
this.lastState = state
|
||||
this.callSink(() => this.sinks.onStateChange?.(state))
|
||||
}
|
||||
|
||||
private async pumpStream<F extends { type: string }>(
|
||||
stream: AsyncIterable<RpcRequest<F>>,
|
||||
sink: ((envelope: RpcRequest<F>) => void) | undefined,
|
||||
signal: AbortSignal,
|
||||
onEnd: () => void,
|
||||
): Promise<void> {
|
||||
try {
|
||||
for await (const envelope of stream) {
|
||||
if (signal.aborted) break
|
||||
if (envelope.payload.type === 'stream/error') break
|
||||
if (sink !== undefined) this.callSink(() => { sink(envelope) })
|
||||
}
|
||||
|
||||
@@ -444,6 +444,47 @@ function tokenUsageOf(log: readonly SessionEvent[]): FixtureTokenUsageProjection
|
||||
return totals
|
||||
}
|
||||
|
||||
/** Latest log-only capacity record, or undefined before any request ran. */
|
||||
function lastRequestContext(
|
||||
log: readonly SessionEvent[],
|
||||
): { provider: string; model: string; contextWindow: number } | undefined {
|
||||
const event = log.findLast(item => (item as { type: string }).type === 'request/context')
|
||||
return event === undefined
|
||||
? undefined
|
||||
: (event as unknown as { data: { provider: string; model: string; contextWindow: number } }).data
|
||||
}
|
||||
|
||||
/**
|
||||
* Fixture parallel of token-meter's request-pressure projection: the last
|
||||
* provider-reported prompt size paired with the last recorded capacity. The
|
||||
* two need not come from one request — see the token-meter README.
|
||||
*/
|
||||
function contextPressureOf(
|
||||
log: readonly SessionEvent[],
|
||||
): { pressureTokens: number; contextWindow?: number } {
|
||||
let pressureTokens = 0
|
||||
for (const event of log) {
|
||||
const item = event as unknown as {
|
||||
type: string
|
||||
data: { usage?: TokenUsage; chunk?: { type?: string; usage?: TokenUsage } }
|
||||
}
|
||||
const usage = item.type === 'assistant/chunk' && item.data.chunk?.type === 'usage'
|
||||
? item.data.chunk.usage
|
||||
: item.type === 'assistant/message'
|
||||
? item.data.usage
|
||||
: undefined
|
||||
if (usage === undefined) continue
|
||||
pressureTokens = usage.inputTokens
|
||||
+ (usage.cacheReadTokens ?? 0)
|
||||
+ (usage.cacheWriteTokens ?? 0)
|
||||
}
|
||||
const contextWindow = lastRequestContext(log)?.contextWindow
|
||||
return {
|
||||
pressureTokens,
|
||||
...contextWindow === undefined ? {} : { contextWindow },
|
||||
}
|
||||
}
|
||||
|
||||
function projectionValuesOf(log: readonly SessionEvent[]): Record<string, unknown> {
|
||||
const values: Record<string, unknown> = {}
|
||||
const titleEvent = log.findLast(item => (item as { type: string }).type === 'session/title')
|
||||
@@ -460,23 +501,32 @@ function projectionValuesOf(log: readonly SessionEvent[]): Record<string, unknow
|
||||
values['goal'] = backscanGoal(log)
|
||||
// Always present (token-meter composed): full-log provider billing.
|
||||
values['tokenUsage'] = tokenUsageOf(log)
|
||||
// Always present (token-meter composed): last request pressure and capacity.
|
||||
values['contextPressure'] = contextPressureOf(log)
|
||||
return values
|
||||
}
|
||||
|
||||
/** Host push-frame parallel: emit one session/projection frame per key the given event advanced. */
|
||||
function projectionFramesOf(id: SessionId, log: readonly SessionEvent[], event: SessionEvent): Extract<MuxFrame, { type: 'session/projection' }>[] {
|
||||
const type = (event as { type: string }).type
|
||||
// One usage sample advances both token-meter units.
|
||||
if (
|
||||
(type === 'assistant/chunk'
|
||||
&& (event as unknown as { data: { chunk?: { type?: string } } }).data.chunk?.type === 'usage')
|
||||
|| (type === 'assistant/message'
|
||||
&& (event as unknown as { data: { usage?: TokenUsage } }).data.usage !== undefined)
|
||||
) {
|
||||
return [
|
||||
{ type: 'session/projection', sessionId: id, key: 'tokenUsage', value: tokenUsageOf(log), seq: event.seq },
|
||||
{ type: 'session/projection', sessionId: id, key: 'contextPressure', value: contextPressureOf(log), seq: event.seq },
|
||||
]
|
||||
}
|
||||
if (type === 'request/context') {
|
||||
return [{
|
||||
type: 'session/projection',
|
||||
sessionId: id,
|
||||
key: 'tokenUsage',
|
||||
value: tokenUsageOf(log),
|
||||
key: 'contextPressure',
|
||||
value: contextPressureOf(log),
|
||||
seq: event.seq,
|
||||
}]
|
||||
}
|
||||
@@ -1136,20 +1186,16 @@ export function createFixtureApi(options: FixtureOptions = {}): ApiProxy {
|
||||
append(id, { type: 'plan/mode', data: { active: plan.wanted } })
|
||||
}
|
||||
append(id, { type: 'user/message', surfaceOp: 'append', data: userMessage(content) })
|
||||
// Capacity parallel of the host token-meter's request/context record:
|
||||
// log-only, appended inside the open turn, and deduplicated against the
|
||||
// route already recorded (the fixture never varies contextWindow).
|
||||
const target = modelTargets.get(id) ?? { provider: 'deepseek', model: 'deepseek-v4-flash' }
|
||||
emitMux({
|
||||
type: 'session/model-request',
|
||||
sessionId: id,
|
||||
// The fixture's durable transcript is historically zero-based, while
|
||||
// the real Agent's request telemetry opens turns at one.
|
||||
turn: turn + 1,
|
||||
step: 1,
|
||||
provider: target.provider,
|
||||
model: target.model,
|
||||
// No fixture token-meter is composed, so omit the request-pressure
|
||||
// numerator instead of substituting cumulative provider billing.
|
||||
contextWindow: 128_000,
|
||||
})
|
||||
if (lastRequestContext(logOf(id))?.model !== target.model) {
|
||||
append(id, {
|
||||
type: 'request/context',
|
||||
data: { provider: target.provider, model: target.model, contextWindow: 128_000 },
|
||||
})
|
||||
}
|
||||
startReply(
|
||||
id,
|
||||
turn,
|
||||
|
||||
@@ -5,7 +5,7 @@
|
||||
*/
|
||||
import type { Context } from 'cordis'
|
||||
import type { IApiClient } from './api.ts'
|
||||
import { ConnectionController, type ConnectionConfig, type ConnectionSinks } from './connection.ts'
|
||||
import { ConnectionController, type ConnectionConfig, type ConnectionSinks, type ConnectionState } from './connection.ts'
|
||||
import { FixtureApiClient } from './fixture.ts'
|
||||
import { WebApiClient } from './web-api-client.ts'
|
||||
|
||||
@@ -17,7 +17,7 @@ export type {
|
||||
ToolCallView, ToolResultView, WorkspaceApi, WorkspaceId, WorkspaceView,
|
||||
CommandsApi, CommandDescriptor, SkillsApi, SkillEntry,
|
||||
ModelCatalogFailure, ModelCatalogModel, ModelProviderGroup, ModelReasoning,
|
||||
ModelReasoningEffort, ModelRequestTelemetry, ModelTarget, SessionModels, SessionProjectionsBlock,
|
||||
ModelReasoningEffort, ModelTarget, SessionModels,
|
||||
RpcRequest, RpcResponse, RpcResult, RpcError, RpcErrorCode,
|
||||
ClientRequest, ServerResponse, ServerRequest, ClientResponse, RpcMessage, RpcReceipt,
|
||||
IApiClient, SessionId, SessionEvent, ContentBlock, StreamChunk,
|
||||
@@ -27,7 +27,7 @@ export { RpcId, AbstractApiClient, transportError } from './api.ts'
|
||||
|
||||
// Connection loop types are public through ConnectionHandle.start; the
|
||||
// controller remains package-internal.
|
||||
export type { ConnectionConfig, ConnectionSinks }
|
||||
export type { ConnectionConfig, ConnectionSinks, ConnectionState }
|
||||
|
||||
|
||||
/** Required services (none — this is the wire root). */
|
||||
|
||||
Reference in New Issue
Block a user