Coder/deepseek-harness
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(web): project usage and snapshot request context

Browse Source
This commit is contained in:
Hypatia May
2026-07-29 15:27:59 +08:00
parent e37cb23336
commit bf618dabf9
78 changed files with 748 additions and 934 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
packages/host/apiproxy/README.i18n.yaml
View File

@@ -2,5 +2,5 @@
# side as of the last confirmed-consistent state. Both languages carry equal authority;
# after editing either side, bring the other along and re-record with:
# pnpm run verify-translation-pairing --write packages/host/apiproxy/README.md
README.md: 5d5f1e4cff97756e650cca02abef221dd9bef8f5
README.zh.md: 3b70aec65189b4a9e71170c4194ea5850fc4cffb
README.md: 872d653a194bc08ec7a4a2125137af51d0105a52
README.zh.md: 5470b7680b1fea42cc3383ebcc121f05015f5226

4
packages/host/apiproxy/README.md
View File

@@ -22,9 +22,9 @@ Directory picking delegates to the composed `ctx.directoryPicker` backend ([the
`host.openPath` opens a filesystem path with the operating system's default application (`open` on macOS, `Invoke-Item` on Windows, `xdg-open` on Linux). The opener is injectable for tests. The browser carrier applies the same loopback, same-origin restriction as `host.pickDirectory`.
`session.history` pages on message boundaries. Its tail page (no `beforeSeq`) carries the generic `projections` baseline, including the `todos` whole-list value when that unit is mounted, plus separate `metrics`: full-log usage deduplicated by `(turn, step)` and current token-meter pressure. Older pages omit both session-level carriers. Live `session/metrics` mux frames carry monotonic log/projection revisions, so clients reject stale frames and preserve the counters while prepending older pages. Cache reads and writes remain disjoint buckets; the cache-hit denominator is uncached input plus cache reads.
`session.history` pages on message boundaries. Its tail page (no `beforeSeq`) carries only the generic `projections` baseline for registered units; older pages omit it. When token-meter is composed with the projection registry, full-log provider billing rides the ordinary `tokenUsage` key. Its usage chunks and final messages are deduplicated by `(turn, step)`, while cache reads and writes remain disjoint buckets. ApiProxy owns no token-specific history field, mux frame, projector, revision counter, or refresh queue.
Context capacity uses a distinct transient `session/model-request` mux frame emitted from the contained Agent notification after an observed request attempt returns its outer stream handle. This boundary does not prove provider I/O began. The frame carries turn, step, final provider/model, and optional capacity only to mux connections already open at that instant. `session.history`, mux subscription baselines, reconnects, and session restore never query or replay prior capacity; a frame without capacity explicitly clears the earlier connection-local value.
Request context uses a distinct transient `session/model-request` mux frame emitted from the contained Agent notification after an observed request attempt returns its outer stream handle. This boundary does not prove provider I/O began. In the same synchronous event boundary, ApiProxy optionally reads `tokenMeter.measure(session).totalTokens` once and combines it with capacity from that exact prepared call. The atomic frame carries turn, step, final provider/model, and optional `contextTokens`/`contextWindow` only to mux connections already open at that instant. Measurement failure omits only the numerator. `session.history`, mux subscription baselines, reconnects, and session restore never query or replay an earlier snapshot, and missing fields in a later frame replace rather than inherit prior values.
The `command.*` and `skill.*` domains expose the host command registry and skill catalog to clients. Every method addresses one session's agent by `sessionId` (a served session always has an Agent; `command.*` resumes cold sessions through the same path as `session.*`, while `skill.list` resolves the project root from the session header without touching the Agent registry). `command.execute` runs a slash-command line host-side with pure admission semantics: the response reports whether the line resolved to a handler plus the minted lifecycle `commandId` when it did (correlating the acknowledgment with the flow node), while the outcome rides the durably logged `command/run`/`command/done` lifecycle pair broadcast on the mux stream; the carrier's request signal cancels the running handler. `host/commands-changed` is the catalog invalidation frame: clients refetch `command.list` instead of diffing.

4
packages/host/apiproxy/README.zh.md
View File

@@ -22,9 +22,9 @@ Workspace 列表与 Session 列表是相互独立的重连基线。`workspace.cr
`host.openPath` 会用操作系统的默认应用打开一个文件系统路径（macOS 为 `open`，Windows 为 `Invoke-Item`，Linux 为 `xdg-open`）。打开器可在测试中注入。浏览器载体对其施加与 `host.pickDirectory` 相同的回环、同源限制。
`session.history` 按消息边界分页。其尾页（不带 `beforeSeq`）携带通用 `projections` 基线（挂载对应单元时包含 `todos` 整表值），以及独立的 `metrics`：按 `(turn, step)` 去重的完整日志用量与当前 token 计量压力。较早的页面省略这两种会话级载体。实时 `session/metrics` mux 帧携带单调递增的日志修订号与投影修订号，因此客户端会拒绝陈旧帧，并在向前加载较早页面时保留计数器。缓存读取与缓存写入保持为彼此独立的计数项；缓存命中率的分母是未缓存输入加缓存读取。
`session.history` 按消息边界分页。其尾页（不带 `beforeSeq`）只携带已注册单元的通用 `projections` 基线；较早页面则省略该基线。当 token-meter 与投影注册表组合时，完整日志中的提供方计费用量会通过普通 `tokenUsage` 键承载。系统按 `(turn, step)` 对其用量分片和最终消息去重，缓存读取与写入则仍是相互独立的计数项。ApiProxy 不拥有任何 token 专用的历史字段、mux 帧、投影器、修订计数器或刷新队列。
上下文容量使用独立的临时 `session/model-request` mux 帧。外层流调用为一次观察到的请求尝试返回句柄后，系统会根据 Agent 通知发出该帧，并收容通知失败。这个边界不能证明提供方 I/O 已开始。该帧携带轮次、步骤、最终提供方／模型与可选容量，且只发送给当时已经打开的 mux 连接。`session.history`、mux 订阅基线、重连和会话恢复绝不会查询或回放先前的容量；不带容量的帧会显式清除较早的连接本地值。
请求上下文使用独立的临时 `session/model-request` mux 帧。外层流调用为一次已观测的请求尝试返回句柄后，系统会根据 Agent 通知发出该帧，并收容通知失败。这个边界不能证明提供方 I/O 已开始。在同一同步事件边界内，ApiProxy 会可选地读取一次 `tokenMeter.measure(session).totalTokens`，并将结果与该次准备完成调用的容量合并。这个原子帧携带轮次、步骤、最终提供方／模型与可选的 `contextTokens`／`contextWindow`，且只发送给当时已经打开的 mux 连接。测量失败时只省略分子。`session.history`、mux 订阅基线、重连和会话恢复绝不会查询或回放更早的快照；后续帧中缺失的字段会清除对应的先前值，而不是继承它。
`command.*` 与 `skill.*` 领域向客户端暴露宿主命令注册表和技能目录。每个方法都通过 `sessionId` 寻址一个会话的 Agent（被服务的会话必有 Agent；`command.*` 经由与 `session.*` 相同的路径恢复冷会话，而 `skill.list` 从会话头解析项目根目录，不触碰 Agent 注册表）。`command.execute` 在宿主侧运行一条斜杠命令行，语义为纯准入：响应报告该行是否解析到处理器，并在解析到时回带铸造的生命周期 `commandId`（将本次确认与流节点关联）；结局经由持久落账并在 mux 流广播的 `command/run`/`command/done` 生命周期事件对承载；载体的请求信号可取消正在运行的处理器。`host/commands-changed` 是目录失效帧：客户端重新拉取 `command.list` 而不是做差分。

93
packages/host/apiproxy/src/api-proxy.ts
View File

@@ -51,7 +51,6 @@ import type {
AskUserQuestionAnswer, AskUserQuestionItem, AskUserQuestionRequest,
} from '@deepseek-ai/dsh-user-interaction'
import { UserInteractionError } from '@deepseek-ai/dsh-user-interaction'
import { affectsSessionMetrics, SessionMetricsProjector } from './session-metrics.ts'
import { DirectoryPickerError } from '@deepseek-ai/dsh-host-directory-picker'
import { openNativePath } from './native-path-opener.ts'
@@ -496,56 +495,34 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
for (const queue of muxQueues) queue.push(envelope)
}
const pendingMetricSessions = new Set<Session>()
let metricFlushScheduled = false
const metricsProjector = new SessionMetricsProjector(ctx)
/** Queue one full-log metrics publication after synchronous session listeners drain. */
function scheduleMetrics(session: Session): void {
if (muxQueues.size === 0) return
pendingMetricSessions.add(session)
if (metricFlushScheduled) return
metricFlushScheduled = true
queueMicrotask(() => {
metricFlushScheduled = false
const sessions = [...pendingMetricSessions]
pendingMetricSessions.clear()
for (const current of sessions) {
broadcast({
type: 'session/metrics',
sessionId: current.id,
metrics: metricsProjector.snapshot(current),
})
}
})
}
ctx.effect(() => {
const disposers = [
ctx.on('session/event', (session: Session, event: SessionEvent) => {
if (affectsSessionMetrics(event)) scheduleMetrics(session)
}),
ctx.on('agent/created', (agent: Agent) => { scheduleMetrics(agent.session) }),
ctx.on('agent/model-request', (agent, turn, step, request) => {
broadcast({
type: 'session/model-request',
sessionId: agent.session.id,
turn,
step,
provider: request.provider,
model: request.model,
...request.contextWindow === undefined
? {}
: { contextWindow: request.contextWindow },
})
}),
ctx.on('session/disposed', (session: Session) => { pendingMetricSessions.delete(session) }),
]
return () => {
pendingMetricSessions.clear()
for (const dispose of disposers) dispose()
}
}, 'api-proxy: session metrics')
return ctx.on('agent/model-request', (agent, turn, step, request) => {
const tokenMeter = ctx.get('tokenMeter') as {
measure(session: Session): { totalTokens: number }
} | undefined
let contextTokens: number | undefined
if (tokenMeter !== undefined) {
try {
contextTokens = tokenMeter.measure(agent.session).totalTokens
} catch {
// A malformed or temporarily unmeasurable replay omits only the
// numerator; this request still replaces stale telemetry.
}
}
broadcast({
type: 'session/model-request',
sessionId: agent.session.id,
turn,
step,
provider: request.provider,
model: request.model,
...contextTokens === undefined ? {} : { contextTokens },
...request.contextWindow === undefined
? {}
: { contextWindow: request.contextWindow },
})
})
}, 'api-proxy: model request telemetry')
// Projection change feed → session/projection push frames. The carrier
// mints the wire frame (the seam package holds no wire vocabulary); the
@@ -991,16 +968,12 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
return { event, ...view === undefined ? {} : { view } }
})
// Baseline rider: tail page only — loadOlder (beforeSeq present) is
// the one path that never needs fresh projection or metrics state.
// the one path that never needs fresh projection state.
const projections = beforeSeq === undefined ? projectionsFor(ctx, found.agent) : undefined
const metrics = beforeSeq === undefined
? metricsProjector.snapshot(found.agent.session)
: undefined
return ok(request, {
events: entries,
hasMore: page.hasMore,
...projections === undefined ? {} : { projections },
...metrics === undefined ? {} : { metrics },
})
},
@@ -1501,11 +1474,6 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
muxQueues.add(queue)
for (const session of ctx.sessions.list()) {
subscribeSession(queue, session)
queue.push(frame({
type: 'session/metrics',
sessionId: session.id,
metrics: metricsProjector.snapshot(session),
}))
}
for (const pending of pendingQuestions.values()) {
queue.push({
@@ -1556,11 +1524,6 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
}),
ctx.on('session/created', (session: Session) => {
subscribeSession(queue, session)
queue.push(frame({
type: 'session/metrics',
sessionId: session.id,
metrics: metricsProjector.snapshot(session),
}))
}),
ctx.on('session/disposed', (session: Session) => {
openCalls.delete(session.id)

4
packages/host/apiproxy/src/api/events.schema.ts
View File

@@ -11,7 +11,7 @@ import type { Wire } from './rpc.schema.ts'
import { rpcErrorSchema, rpcIdSchema } from './rpc.schema.ts'
import { approvalRequestIdSchema } from './approvals.schema.ts'
import {
contentBlockSchema, sessionEventSchema, sessionIdSchema, sessionMetricsSchema, toolEventViewSchema,
contentBlockSchema, sessionEventSchema, sessionIdSchema, toolEventViewSchema,
} from './sessions.schema.ts'
import { workspaceIdSchema, workspaceViewSchema } from './workspace.schema.ts'
@@ -37,7 +37,6 @@ const messageSchema = z.object({
export const muxFrameSchema = z.discriminatedUnion('type', [
z.object({ type: z.literal('session/event'), sessionId: sessionIdSchema, event: sessionEventSchema, view: toolEventViewSchema.optional() }),
z.object({ type: z.literal('session/subscribed'), sessionId: sessionIdSchema, lastSeq: z.number().int() }),
z.object({ type: z.literal('session/metrics'), sessionId: sessionIdSchema, metrics: sessionMetricsSchema }),
z.object({
type: z.literal('session/model-request'),
sessionId: sessionIdSchema,
@@ -45,6 +44,7 @@ export const muxFrameSchema = z.discriminatedUnion('type', [
step: z.number().int().positive(),
provider: z.string().min(1),
model: z.string().min(1),
contextTokens: z.number().int().nonnegative().optional(),
contextWindow: z.number().int().positive().optional(),
}),
z.object({ type: z.literal('approval/requested'), sessionId: sessionIdSchema, approvalId: approvalRequestIdSchema, toolName: z.string(), callId: z.string().optional(), reason: z.string().optional() }),

28
packages/host/apiproxy/src/api/events.ts
View File

@@ -13,7 +13,6 @@ import type { CallId } from '@deepseek-ai/dsh-llm/brand'
import type { SessionEvent, SessionId } from '@deepseek-ai/dsh-session/types'
import type { ToolCallView, ToolResultView } from '@deepseek-ai/dsh-tools/presentation'
import type { RpcError, RpcId, RpcRequest } from './rpc.ts'
import type { SessionMetrics } from './sessions.ts'
import type { WorkspaceView } from './workspace.ts'
// Client-side consumers take the render-intent vocabulary from the contract;
@@ -32,6 +31,18 @@ export type ToolEventView =
| { for: 'call'; view: ToolCallView }
| { for: 'result'; view: ToolResultView }
/** Atomic telemetry captured at one observed model-request boundary. */
export interface ModelRequestTelemetry {
turn: number
step: number
provider: string
model: string
/** Token-meter pressure measured synchronously for this exact request. */
contextTokens?: number
/** Registration-bound capacity from this exact prepared call. */
contextWindow?: number
}
/** Streaming face of the contract: the two SSE stream openers (mux + host). */
export interface EventsApi {
/**
@@ -58,23 +69,18 @@ export interface EventsApi {
export type MuxFrame =
| { type: 'session/event'; sessionId: SessionId; event: SessionEvent; view?: ToolEventView }
| { type: 'session/subscribed'; sessionId: SessionId; lastSeq: number }
| { type: 'session/metrics'; sessionId: SessionId; metrics: SessionMetrics }
/**
* One request attempt observed by this already-open mux connection after its
* final route and outer `llm/stream` handle were obtained. This does not prove
* provider I/O began. The frame is transient: mux baselines, reconnects, and
* session history never replay it. An absent `contextWindow` explicitly clears
* a capacity observed from an earlier request on the same connection.
* session history never replay it. The optional numerator and capacity are
* one atomic request snapshot; absent fields explicitly replace, rather
* than inherit from, the preceding request.
*/
| {
| ({
type: 'session/model-request'
sessionId: SessionId
turn: number
step: number
provider: string
model: string
contextWindow?: number
}
} & ModelRequestTelemetry)
| { type: 'approval/requested'; sessionId: SessionId; approvalId: ApprovalRequestId; toolName: string; callId?: CallId; reason?: string }
| { type: 'approval/resolved'; sessionId: SessionId; approvalId: ApprovalRequestId; outcome: ApprovalOutcome }
| { type: 'question/requested'; sessionId: SessionId; questions: AskUserQuestionItem[] }

6
packages/host/apiproxy/src/api/index.ts
View File

@@ -29,14 +29,16 @@ export interface ApiProxy {
// ---- Domain interfaces and payload entities ----
export type {
HistoryEntry, ModelCatalogFailure, ModelCatalogModel, ModelProviderGroup, ModelReasoning,
ModelReasoningEffort, ModelTarget, SessionMetrics, SessionModels, SessionProjectionsBlock,
ModelReasoningEffort, ModelTarget, SessionModels, SessionProjectionsBlock,
SessionsApi, SessionSummary,
} from './sessions.ts'
export type { DirectoryEntry, DirectoryListing, HostApi } from './host.ts'
export type { WorkspaceApi, WorkspaceId, WorkspaceView } from './workspace.ts'
export type { CommandsApi, CommandDescriptor } from './commands.ts'
export type { SkillsApi, SkillEntry } from './skills.ts'
export type { EventsApi, MuxFrame, HostFrame, ToolCallView, ToolEventView, ToolResultView } from './events.ts'
export type {
EventsApi, HostFrame, ModelRequestTelemetry, MuxFrame, ToolCallView, ToolEventView, ToolResultView,
} from './events.ts'
export type { GoalsApi, GoalId, GoalRef } from './goals.ts'
export type { ApprovalResponsePayload } from './approvals.ts'
export type { QuestionResponsePayload } from './questions.ts'

17
packages/host/apiproxy/src/api/sessions.schema.ts
View File

@@ -11,7 +11,7 @@ import type { RequestPayload, ResponseValue } from './rpc-map.ts'
import type { Wire } from './rpc.schema.ts'
import type {
HistoryEntry, ModelCatalogFailure, ModelCatalogModel, ModelProviderGroup, ModelReasoning,
ModelReasoningEffort, ModelTarget, SessionMetrics, SessionProjectionsBlock, SessionSummary,
ModelReasoningEffort, ModelTarget, SessionProjectionsBlock, SessionSummary,
} from './sessions.ts'
import type { ToolEventView } from './events.ts'
import type { WorkspaceId } from './workspace.ts'
@@ -151,23 +151,11 @@ export const sessionProjectionsBlockSchema = z.object({
values: z.record(z.string(), z.unknown()),
}) as unknown as z.ZodType<SessionProjectionsBlock>
/** Host-owned durable usage and current-pressure projection. */
export const sessionMetricsSchema = z.object({
logRevision: z.number().int().nonnegative(),
projectionRevision: z.number().int().nonnegative(),
uncachedInputTokens: z.number().nonnegative(),
outputTokens: z.number().nonnegative(),
cacheReadTokens: z.number().nonnegative(),
cacheWriteTokens: z.number().nonnegative(),
contextTokens: z.number().nonnegative().optional(),
}) satisfies z.ZodType<Wire<SessionMetrics>>
/** session.history response value (projections and metrics ride the tail page only). */
/** session.history response value (projections ride the tail page only). */
export const sessionHistoryValueSchema = z.object({
events: z.array(historyEntrySchema),
hasMore: z.boolean(),
projections: sessionProjectionsBlockSchema.optional(),
metrics: sessionMetricsSchema.optional(),
}) satisfies z.ZodType<Wire<ResponseValue<'session.history'>>>
/** session.models request payload. */
@@ -223,4 +211,3 @@ export const sessionCancelRequestSchema = z.object({
export const sessionCancelValueSchema = z.object({
accepted: z.literal(true),
}) satisfies z.ZodType<Wire<ResponseValue<'session.cancel'>>>

30
packages/host/apiproxy/src/api/sessions.ts
View File

@@ -35,29 +35,6 @@ export interface HistoryEntry {
view?: ToolEventView
}
/**
* Host-owned token metrics for one durable session revision. Provider usage
* buckets are cumulative across the full log; current context pressure
* describes the replayed request surface at this revision and is absent when
* the Host cannot measure it.
*/
export interface SessionMetrics {
/** Number of durable events included in this projection. */
logRevision: number
/** Monotone ordering within one Host process and mux subscription generation. */
projectionRevision: number
/** Cumulative uncached provider input. */
uncachedInputTokens: number
/** Cumulative provider output. */
outputTokens: number
/** Cumulative provider cache reads. */
cacheReadTokens: number
/** Cumulative provider cache writes; excluded from the Web cache-hit formula. */
cacheWriteTokens: number
/** Current request pressure from `ctx.tokenMeter.measure(session).totalTokens`. */
contextTokens?: number
}
/**
* The projection baseline riding the history tail page: one synchronous cut
* over every registered projection unit, read from the registry's watermark
@@ -211,17 +188,14 @@ export interface SessionsApi {
* the client needs a fresh baseline already pulls the tail page, and
* loadOlder (the only beforeSeq path) is the only path that never needs one.
* A deployment without the registry serves histories without the block.
* The same tail-only rule carries `metrics`, whose cumulative usage and
* current token-meter pressure are Host projections over the full log
* rather than products of the returned page. Live model capacity is
* connection-local telemetry and is never reconstructed here.
* Model-request telemetry is connection-local and is never reconstructed
* from history.
*/
history(request: RpcRequest<{ sessionId: SessionId; beforeSeq?: number; maxMessages?: number }>):
Promise<RpcResponse<{
events: HistoryEntry[]
hasMore: boolean
projections?: SessionProjectionsBlock
metrics?: SessionMetrics
}>>
/** Reads a fresh advisory model directory for this session. Provider lookups run independently. */

122
packages/host/apiproxy/src/session-metrics.ts
View File

@@ -1,122 +0,0 @@
/**
* Full-log usage and current-context projection for Web clients.
*
* @module @deepseek-ai/dsh-host-apiproxy/session-metrics
*/
import type { Context } from 'cordis'
import type { TokenUsage } from '@deepseek-ai/dsh-llm'
import type { Session, SessionEvent } from '@deepseek-ai/dsh-session'
import type { SessionMetrics } from './api/sessions.ts'
interface UsageState {
logRevision: number
projectionRevision: number
uncachedInputTokens: number
outputTokens: number
cacheReadTokens: number
cacheWriteTokens: number
byStep: Map<string, TokenUsage>
}
interface TokenMeterLike {
measure(session: Session): { totalTokens: number }
}
function usageFrom(event: SessionEvent): { turn: number; step: number; usage: TokenUsage } | undefined {
if (event.type === 'assistant/chunk' && event.data.chunk.type === 'usage') {
return { turn: event.data.turn, step: event.data.step, usage: event.data.chunk.usage }
}
if (event.type === 'assistant/message' && event.data.usage !== undefined) {
return { turn: event.data.turn, step: event.data.step, usage: event.data.usage }
}
return undefined
}
/**
* Whether an appended event can change cumulative usage or token-meter
* pressure. Text/reasoning stream deltas remain outside both projections.
* @param event - appended durable event.
* @returns true when the Host must publish a fresh metrics snapshot.
*/
export function affectsSessionMetrics(event: SessionEvent): boolean {
if (event.type === 'assistant/chunk') return event.data.chunk.type === 'usage'
if (event.type === 'request/header') return true
return 'surfaceOp' in event
}
function recordUsage(state: UsageState, turn: number, step: number, usage: TokenUsage): void {
const key = `${turn}:${step}`
const previous = state.byStep.get(key)
if (previous !== undefined) {
state.uncachedInputTokens -= previous.inputTokens
state.outputTokens -= previous.outputTokens
state.cacheReadTokens -= previous.cacheReadTokens ?? 0
state.cacheWriteTokens -= previous.cacheWriteTokens ?? 0
}
state.byStep.set(key, usage)
state.uncachedInputTokens += usage.inputTokens
state.outputTokens += usage.outputTokens
state.cacheReadTokens += usage.cacheReadTokens ?? 0
state.cacheWriteTokens += usage.cacheWriteTokens ?? 0
}
/** Projects durable cumulative usage and synchronous current context pressure. */
export class SessionMetricsProjector {
private readonly usage = new WeakMap<Session, UsageState>()
/** @param ctx - Host context providing an optional token-meter service. */
constructor(private readonly ctx: Context) {}
/**
* Read a fresh detached projection through the session's durable tail.
* @param session - authoritative durable log owner.
* @returns cumulative usage and any currently measurable pressure.
*/
snapshot(session: Session): SessionMetrics {
const state = this.syncUsage(session)
const tokenMeter = this.ctx.get('tokenMeter') as TokenMeterLike | undefined
let contextTokens: number | undefined
if (tokenMeter !== undefined) {
try {
contextTokens = tokenMeter.measure(session).totalTokens
} catch {
// A malformed or temporarily unmeasurable replay has no honest pressure value.
}
}
return {
logRevision: state.logRevision,
projectionRevision: state.projectionRevision++,
uncachedInputTokens: state.uncachedInputTokens,
outputTokens: state.outputTokens,
cacheReadTokens: state.cacheReadTokens,
cacheWriteTokens: state.cacheWriteTokens,
...contextTokens === undefined ? {} : { contextTokens },
}
}
private syncUsage(session: Session): UsageState {
let state = this.usage.get(session)
if (state === undefined) {
state = {
logRevision: 0,
projectionRevision: 0,
uncachedInputTokens: 0,
outputTokens: 0,
cacheReadTokens: 0,
cacheWriteTokens: 0,
byStep: new Map(),
}
this.usage.set(session, state)
}
while (state.logRevision < session.events.length) {
const event = session.events[state.logRevision]
/* v8 ignore next -- Session events are append-only and dense; logRevision is bounded by length. */
if (event === undefined) break
const usage = usageFrom(event)
if (usage !== undefined) recordUsage(state, usage.turn, usage.step, usage.usage)
state.logRevision++
}
return state
}
}

30
packages/host/apiproxy/tests/api-proxy-model-request.spec.ts
View File

@@ -1,4 +1,4 @@
import { describe, expect, it } from 'vitest'
import { describe, expect, it, vi } from 'vitest'
import { Context } from 'cordis'
import AgentRegistry, { agentEvents } from '@deepseek-ai/dsh-agent'
import type { Agent } from '@deepseek-ai/dsh-agent'
@@ -22,7 +22,7 @@ async function nextFrame<K extends MuxFrame['type']>(
}
describe('ApiProxy model-request telemetry', () => {
it('forwards only to open mux connections and never backfills history or reconnect baselines', async () => {
it('atomically measures the observed request, forwards only live, and degrades per field', async () => {
const ctx = new Context()
await ctx.plugin(SessionStore)
await ctx.plugin(UserInteractionService)
@@ -35,6 +35,8 @@ describe('ApiProxy model-request telemetry', () => {
ctx,
} as Agent
ctx.agents.register(agent)
const measure = vi.fn(() => ({ totalTokens: 321 }))
const removeTokenMeter = ctx.provide('tokenMeter' as never, { measure } as never)
const api = createApiProxy(ctx, {
provider: 'test',
model: 'alpha',
@@ -48,13 +50,13 @@ describe('ApiProxy model-request telemetry', () => {
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(measure).toHaveBeenCalledWith(session)
expect(await nextFrame(primary, 'session/model-request')).toEqual({
type: 'session/model-request',
sessionId: session.id,
@@ -62,6 +64,7 @@ describe('ApiProxy model-request telemetry', () => {
step: 2,
provider: 'test',
model: 'alpha',
contextTokens: 321,
contextWindow: 128_000,
})
@@ -70,7 +73,8 @@ describe('ApiProxy model-request telemetry', () => {
payload: { sessionId: session.id },
})
if (!history.result.ok) throw new Error('history failed')
expect(history.result.value.metrics).not.toHaveProperty('contextWindow')
expect(history.result.value).not.toHaveProperty('metrics')
expect(history.result.value).not.toHaveProperty('modelRequest')
const reconnectAbort = new AbortController()
const reconnect = api.events.mux(
@@ -78,8 +82,8 @@ describe('ApiProxy model-request telemetry', () => {
reconnectAbort.signal,
)[Symbol.asyncIterator]()
expect((await nextFrame(reconnect, 'session/subscribed')).sessionId).toBe(session.id)
expect((await nextFrame(reconnect, 'session/metrics')).metrics).not.toHaveProperty('contextWindow')
measure.mockImplementation(() => { throw new Error('unmeasurable replay') })
agentEvents(ctx, agent).emit('agent/model-request', 2, 1, {
provider: 'test',
model: 'without-capacity',
@@ -95,6 +99,22 @@ describe('ApiProxy model-request telemetry', () => {
})
}
removeTokenMeter()
agentEvents(ctx, agent).emit('agent/model-request', 3, 1, {
provider: 'test',
model: 'without-meter',
contextWindow: 64_000,
})
expect(await nextFrame(primary, 'session/model-request')).toEqual({
type: 'session/model-request',
sessionId: session.id,
turn: 3,
step: 1,
provider: 'test',
model: 'without-meter',
contextWindow: 64_000,
})
primaryAbort.abort()
reconnectAbort.abort()
await primary.return?.()

38
packages/host/apiproxy/tests/rpc-schemas.spec.ts
View File

@@ -10,7 +10,7 @@ import {
sessionCreateValueSchema, sessionEventSchema, sessionHistoryRequestSchema, sessionHistoryValueSchema,
sessionIdSchema, sessionListRequestSchema, sessionListValueSchema, sessionModelsRequestSchema,
sessionModelsValueSchema, sessionPromptRequestSchema, sessionPromptValueSchema,
sessionSelectModelRequestSchema, sessionSelectModelValueSchema, sessionSummarySchema, sessionMetricsSchema,
sessionSelectModelRequestSchema, sessionSelectModelValueSchema, sessionSummarySchema,
} from '../src/api/sessions.schema.ts'
import {
hostCreateDirectoryRequestSchema, hostCreateDirectoryValueSchema,
@@ -153,29 +153,13 @@ describe('sessions domain schemas', () => {
expect(sessionCreateValueSchema.parse({ sessionId: 's1' }).sessionId).toBe('s1')
expect(sessionHistoryRequestSchema.parse({ sessionId: 's1', beforeSeq: 3, maxMessages: 5 }).beforeSeq).toBe(3)
expect(() => sessionHistoryRequestSchema.parse({ sessionId: 's1', maxMessages: 0 })).toThrow()
expect(sessionHistoryValueSchema.parse({
const history = sessionHistoryValueSchema.parse({
events: [],
hasMore: false,
projections: { asOfSeq: 11, values: { todos: [] } },
metrics: {
logRevision: 12,
projectionRevision: 4,
uncachedInputTokens: 1_000,
outputTokens: 200,
cacheReadTokens: 4_000,
cacheWriteTokens: 500,
contextTokens: 8_000,
},
modelTarget: { provider: 'deepseek', model: 'deepseek-v4-flash' },
}).metrics?.contextTokens).toBe(8_000)
expect(() => sessionMetricsSchema.parse({
logRevision: 1,
projectionRevision: 0,
uncachedInputTokens: -1,
outputTokens: 0,
cacheReadTokens: 0,
cacheWriteTokens: 0,
})).toThrow()
})
expect(history.projections).toEqual({ asOfSeq: 11, values: { todos: [] } })
expect(sessionModelsRequestSchema.parse({ sessionId: 's1' }).sessionId).toBe('s1')
expect(sessionModelsValueSchema.parse({
current: { provider: 'deepseek', model: 'deepseek-v4-flash', reasoningEffort: 'max' },
@@ -377,18 +361,6 @@ describe('events frame schemas', () => {
const frames = [
{ type: 'session/event', sessionId: 's', event: { type: 't', seq: 0, time: 1, data: null } },
{ type: 'session/subscribed', sessionId: 's', lastSeq: -1 },
{
type: 'session/metrics',
sessionId: 's',
metrics: {
logRevision: 3,
projectionRevision: 1,
uncachedInputTokens: 100,
outputTokens: 20,
cacheReadTokens: 300,
cacheWriteTokens: 40,
},
},
{
type: 'session/model-request',
sessionId: 's',
@@ -396,6 +368,7 @@ describe('events frame schemas', () => {
step: 1,
provider: 'deepseek',
model: 'deepseek-chat',
contextTokens: 8_000,
contextWindow: 128_000,
},
{
@@ -419,6 +392,7 @@ describe('events frame schemas', () => {
expect(() => muxFrameSchema.parse({ type: 'unknown/frame' })).toThrow()
for (const invalid of [
{ 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', contextTokens: -1 },
{ type: 'session/model-request', sessionId: 's', turn: 1, step: 1, provider: 'p', model: 'm', contextWindow: 0 },
{ type: 'session/projection', sessionId: 's', key: '', value: null, seq: 0 },
{ type: 'session/projection', sessionId: 's', key: 'todos', value: null, seq: -1 },

143
packages/host/apiproxy/tests/session-metrics.spec.ts
View File

@@ -1,143 +0,0 @@
import { describe, expect, it } from 'vitest'
import { Context } from 'cordis'
import { createAssistantMessage, createUserMessage } from '@deepseek-ai/dsh-llm'
import { Session, SessionId } from '@deepseek-ai/dsh-session'
import { affectsSessionMetrics, SessionMetricsProjector } from '../src/session-metrics.ts'
function assistant(
session: Session,
turn: number,
step: number,
usage: {
inputTokens: number
outputTokens: number
cacheReadTokens?: number
cacheWriteTokens?: number
},
): void {
session.append('assistant/chunk', {
turn,
step,
chunk: { type: 'usage', usage },
})
session.append('assistant/message', {
turn,
step,
message: createAssistantMessage({
content: [{ type: 'text', text: `answer-${turn}-${step}` }],
source: { provider: 'test', model: 'alpha' },
}),
usage,
}, { surfaceOp: 'append' })
}
describe('SessionMetricsProjector', () => {
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,
step: 1,
chunk: { type: 'text-delta', index: 0, text: 'x' },
})
const usage = session.append('assistant/chunk', {
turn: 1,
step: 1,
chunk: { type: 'usage', usage: { inputTokens: 1, outputTokens: 1 } },
})
const header = session.append('request/header', {
header: { config: { provider: 'test', model: 'alpha' } },
reason: 'initial',
})
const surface = session.append('user/message', createUserMessage({
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('folds usage by turn and step while synchronous pressure follows surface replacement', () => {
const ctx = new Context()
ctx.provide('tokenMeter', {
measure(session: Session) {
return { totalTokens: session.surface.nodes.length * 100 }
},
})
const session = new Session(SessionId('metrics-fold'))
const first = session.append('user/message', createUserMessage({
content: [{ type: 'text', text: 'large old surface' }],
source: { kind: 'user' },
}), { surfaceOp: 'append' })
assistant(session, 1, 1, {
inputTokens: 11,
outputTokens: 3,
cacheReadTokens: 89,
cacheWriteTokens: 8,
})
const projector = new SessionMetricsProjector(ctx)
expect(projector.snapshot(session)).toMatchObject({
uncachedInputTokens: 11,
outputTokens: 3,
cacheReadTokens: 89,
cacheWriteTokens: 8,
contextTokens: 200,
})
const assistantSeq = session.surface.nodes.at(-1)
if (assistantSeq === undefined) throw new Error('assistant surface missing')
session.append('user/message', createUserMessage({
content: [{ type: 'text', text: 'compact summary' }],
source: { kind: 'plugin', plugin: 'test' },
}), {
surfaceOp: { op: 'replace', start: first.seq, end: assistantSeq },
sourceEventSeqs: [first.seq, assistantSeq],
})
expect(projector.snapshot(session)).toMatchObject({
uncachedInputTokens: 11,
outputTokens: 3,
cacheReadTokens: 89,
cacheWriteTokens: 8,
contextTokens: 100,
})
session.append('assistant/chunk', {
turn: 1,
step: 1,
chunk: {
type: 'usage',
usage: { inputTokens: 12, outputTokens: 4, cacheReadTokens: 88, cacheWriteTokens: 9 },
},
})
assistant(session, 1, 2, { inputTokens: 1_000, outputTokens: 500 })
expect(projector.snapshot(session)).toMatchObject({
logRevision: session.events.length,
projectionRevision: 2,
uncachedInputTokens: 1_012,
outputTokens: 504,
cacheReadTokens: 88,
cacheWriteTokens: 9,
contextTokens: 200,
})
})
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()
const ctx = new Context()
ctx.provide('tokenMeter', {
measure() {
throw new Error('unmeasurable replay')
},
})
expect(new SessionMetricsProjector(ctx).snapshot(session).contextTokens).toBeUndefined()
})
})