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

Merge remote-tracking branch 'origin/token-meter-service' into compact-post-step-overflow-recovery

Browse Source 
# Conflicts:
#	docs/core-data-structures/compaction.md
#	docs/rfc/implemented/architecture/2026-07-15-replay-token-meter-service.i18n.yaml
#	docs/rfc/implemented/architecture/2026-07-15-replay-token-meter-service.md
#	docs/rfc/implemented/architecture/2026-07-15-replay-token-meter-service.zh.md
#	docs/rfc/implemented/feature/2026-06-18-compaction-capability-seam.md
#	examples/coding-agent/cordis.yml
#	packages/compact/compact-basic/README.md
#	packages/compact/compact-basic/src/automatic.ts
#	packages/compact/compact-basic/src/config.ts
#	packages/compact/compact-basic/src/index.ts
#	packages/compact/compact-basic/tests/compact-basic.spec.ts
This commit is contained in:
Hypatia May
2026-07-16 16:18:16 +08:00
parent d027ea0d10 c3ddbde3c9
commit 5ff491e3a2
36 changed files with 939 additions and 1292 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
packages/llm/README.md
View File

@@ -5,7 +5,7 @@ The LLM seam and its provider adapters. The interface package (`llm`) owns the a
| Package | Role | ctx key |
|---|---|---|
| `llm/` | Abstract LLM service + content-block vocabulary + chunk assembler | `ctx.llm` |
| `token-meter/` | Replay-aware, per-model request and surface token measurement | `ctx.tokenMeter` |
| `token-meter/` | Replay-aware request and surface token measurement | `ctx.tokenMeter` |
| `llm-deepseek/` | DeepSeek API adapter (hand-rolled fetch/SSE) | (registers on `ctx.llm`) |
| `llm-pi-ai/` | DeepSeek adapter via `@earendil-works/pi-ai` (design twin) | (registers on `ctx.llm`) |

35
packages/llm/token-meter/README.md
View File

@@ -1,29 +1,25 @@
# @deepseek-ai/dsh-token-meter
Replay-aware token measurement through `ctx.tokenMeter`. The service binds one stable meter to each configured model and advances isolated per-model/per-session folds from the durable session log. Compaction consumes it today; other pressure-sensitive plugins can reuse the same accounting without depending on `CompactService`.
Replay-aware token measurement through the singleton `ctx.tokenMeter` service. It advances one isolated fold per session from the durable log, so compaction and other pressure-sensitive plugins can share accounting without depending on `CompactService`.
## Profiles and configuration
The built-in `deepseek-v4-flash` and `deepseek-v4-pro` profiles each use a 128,000-token context window and four characters per estimated token. `models` merges overrides field-by-field, so changing only density keeps the built-in window. A custom model requires `contextWindow`; its `charsPerToken` defaults to `4`.
## Configuration
| Key | Default | Contract |
|---|---:|---|
| `models.<built-in>.contextWindow` | `128000` | Positive integer provider capacity. |
| `models.<model>.charsPerToken` | `4` | Positive finite heuristic density. |
| `contextWindow` | `128000` | Positive integer service-wide context capacity. |
Resolving an unknown model throws `TokenMeterError` with code `TOKEN_METER_MODEL_UNCONFIGURED` and preserves the exact model name. Direct-construction profile validation uses `TOKEN_METER_INVALID_CONFIG`; Loader mounts first apply the package's Schemastery shape validation. There is no universal fallback window.
The estimator intentionally uses one fixed heuristic: four characters per token plus structural overhead for roles, blocks, and request-envelope fields. `contextWindow` is the only deployment setting. Direct construction validates it; Loader mounts first apply the package's Schemastery shape validation. Unrecognized top-level keys are rejected.
## Measurement contract
`ctx.tokenMeter.resolve(model)` returns a `ModelTokenMeter` with three operations:
`ctx.tokenMeter` directly exposes two operations:
- `measure(session, requestHeader?)` returns scalar request pressure at one consumed-log revision.
- `measureSurface(session)` returns current surface nodes and their per-node prices at the same kind of revision.
- `estimateMessage(message)` prices one detached message under that profile.
- `measure(session, requestHeader?)` returns request pressure and the current priced surface at one consumed-log revision.
- `estimateMessage(message)` prices one message with the fixed heuristic.
Measurements are detached and deeply immutable. A caller that needs a consistent scalar/surface decision compares their `logRevision` values instead of copying the full history on every read.
`measure()` synchronizes once and returns one detached, deeply immutable snapshot. `totalTokens` is request-and-response pressure, while `surfaceTokens` is the surface-only heuristic total and equals the sum of `nodes[].tokens`. A `requestHeader` override affects pressure fields only; the surface fields still describe the current session. Every call clones the positional nodes, so measurement is O(surface).
The fold tracks request headers and deltas, step boundaries, surface appends and replacements, successful assistant messages, provider usage, and assistant-chunk provenance. Provider usage is reused only when the handle's model and the canonical request envelope match the successful-call anchor. Otherwise the complete current envelope and surface are repriced under the requested model. Surface changes remain signed relative to a matching anchor, including negative deltas after shrinking replacements.
The fold tracks request headers and deltas, step boundaries, surface appends and replacements, successful assistant messages, provider usage, and assistant-chunk provenance. Provider usage is reused only when the latest successful call's canonical request envelope matches the measured envelope and its total is no lower than that call's full heuristic anchor; a later success replaces the earlier anchor. Otherwise the complete current envelope and surface are estimated. Surface changes remain signed relative to a matching anchor, including negative deltas after shrinking replacements.
Usage accounting sums disjoint input, cache-read, cache-write, and output buckets; reasoning is not added again. Every successful call records an assistant anchor, including content-less calls. An explicit empty provenance list means a known empty provider stream, while absent legacy provenance conservatively treats the durable assistant output as provider output.
@@ -34,16 +30,12 @@ Usage accounting sums disjoint input, cache-read, cache-write, and output bucket
- name: '@deepseek-ai/dsh-compact-basic'
```
Both plugins have usable defaults for the bundled DeepSeek profiles. Custom deployments can override only the fields that differ:
Both plugins have usable defaults. A deployment with a different capacity configures the meter once:
```yaml
- name: '@deepseek-ai/dsh-token-meter'
config:
models:
deepseek-v4-flash:
charsPerToken: 2
local-model:
contextWindow: 32768
contextWindow: 32768
```
## Model Experience
@@ -52,6 +44,7 @@ Indirectly, through consumers such as `dsh-compact-basic`; the service itself ad
## Known Limitations and Deferred Work
- **Heuristic density still needs maintenance** — message content without provider usage is priced by configured character density plus structural overhead, not an exact provider tokenizer. CJK-heavy or provider-specific formats may need profile overrides.
- **Provider usage is only reusable for an identical canonical envelope** — prompt, prefix, tools, or call-config changes deliberately fall back to full heuristic repricing.
- **The fixed heuristic is approximate** — content without reusable provider usage is priced by character count plus structural overhead, not an exact provider tokenizer or request serializer.
- **Every measurement clones the current surface** — coherent immutable snapshots make reads O(surface), including below-threshold pressure checks.
- **Provider usage is only reusable for an identical canonical envelope** — prompt, prefix, tools, model, or call-config changes deliberately fall back to full heuristic estimation.
- **Legacy provenance is conservative** — assistant messages without `sourceEventSeqs` cannot distinguish provider output from listener rewrites, so the fold avoids claiming a known empty or exact chunk stream.

2
packages/llm/token-meter/package.json
View File

@@ -1,6 +1,6 @@
{
"name": "@deepseek-ai/dsh-token-meter",
"description": "Replay-aware per-model token measurement service (ctx.tokenMeter) for the DeepSeek Harness",
"description": "Replay-aware token measurement service (ctx.tokenMeter) for the DeepSeek Harness",
"version": "0.0.1",
"private": true,
"type": "module",

520
packages/llm/token-meter/src/index.ts
View File

@@ -1,192 +1,424 @@
/**
* Replay token-meter service with model-specific context capacity and pricing.
* Single replay-aware token-meter service for request and surface pressure.
*
* @module @deepseek-ai/dsh-token-meter
*/
import { Context, Service } from 'cordis'
import z from 'schemastery'
import { HarnessError, deepFreeze } from '@deepseek-ai/dsh-llm'
import type { Session } from '@deepseek-ai/dsh-session'
import { ReplayModelTokenMeter } from './replay.ts'
import type { ModelTokenProfile } from './replay.ts'
import { BlockAssembler, deepFreeze } from '@deepseek-ai/dsh-llm'
import type { ContentBlock, Message, TokenUsage } from '@deepseek-ai/dsh-llm'
import type { EpochHeader, Session, SessionEvent, SurfaceEvent } from '@deepseek-ai/dsh-session'
import { applyHeaderDelta, canonicalHeader, headerEquals, isSurfaceEvent } from '@deepseek-ai/dsh-session'
import type {
ModelTokenMeter,
ModelTokenMeterConfig,
TokenMeasurement,
TokenMeasurementBaseline,
TokenMeterConfig,
TokenSurfaceNode,
} from './types.ts'
export type * from './types.ts'
/** Exact error code for resolving a model without a configured profile. */
export const TOKEN_METER_MODEL_UNCONFIGURED = 'TOKEN_METER_MODEL_UNCONFIGURED'
/** Default service-wide provider context capacity. */
const DEFAULT_CONTEXT_WINDOW = 128_000
/** Exact error code for invalid token-meter configuration. */
export const TOKEN_METER_INVALID_CONFIG = 'TOKEN_METER_INVALID_CONFIG'
/** Complete public configuration key set. */
const TOKEN_METER_CONFIG_KEYS: ReadonlySet<string> = new Set(['contextWindow'])
/** Closed machine-routable token-meter failure taxonomy. */
export type TokenMeterErrorCode =
| typeof TOKEN_METER_MODEL_UNCONFIGURED
| typeof TOKEN_METER_INVALID_CONFIG
/** Fixed text-density estimate used until exact tokenization is needed. */
const CHARS_PER_TOKEN = 4
/** Built-in DeepSeek model profiles available with zero configuration. */
const BUILTIN_TOKEN_PROFILES: Readonly<Record<string, Readonly<ModelTokenProfile>>> = deepFreeze({
'deepseek-v4-flash': {
model: 'deepseek-v4-flash',
contextWindow: 128_000,
charsPerToken: 4,
},
'deepseek-v4-pro': {
model: 'deepseek-v4-pro',
contextWindow: 128_000,
charsPerToken: 4,
},
})
/** Per-block structural overhead for JSON framing and type tags. */
const BLOCK_OVERHEAD = 4
/** Typed token-meter failure with the affected model preserved for callers. */
export class TokenMeterError extends HarnessError {
declare readonly code: TokenMeterErrorCode
/** Exact model name involved in this error, when applicable. */
readonly model: string | undefined
/** Role-field framing overhead added to every priced message. */
const ROLE_OVERHEAD = 4
constructor(message: string, code: TokenMeterErrorCode, model?: string, options?: ErrorOptions) {
super(message, code, options)
this.name = 'TokenMeterError'
this.model = model
interface MeasurementAnchor {
readonly header: EpochHeader | undefined
readonly surfaceTokens: number
readonly baseline: Exclude<TokenMeasurementBaseline, { kind: 'none' }>
}
interface ReplayState {
consumedEvents: number
header: EpochHeader | undefined
surface: TokenSurfaceNode[]
surfaceTokens: number
stepStart: { turn: number; step: number; surfaceTokens: number } | undefined
anchor: MeasurementAnchor | undefined
}
interface PreparedSurfaceMutation {
readonly tokens: number
commit(state: ReplayState): void
}
/** Sum disjoint provider usage buckets without double-counting reasoning output. */
function usageTokens(usage: TokenUsage): number {
return usage.inputTokens
+ (usage.cacheReadTokens ?? 0)
+ (usage.cacheWriteTokens ?? 0)
+ usage.outputTokens
}
/** Compare optional envelopes so a headerless estimate can track later surface deltas. */
function optionalHeaderEquals(
left: EpochHeader | undefined,
right: EpochHeader | undefined,
): boolean {
if (left === undefined || right === undefined) return left === right
return headerEquals(left, right)
}
/** Reject stale or misspelled keys before defaults can hide them. */
function validateConfigKeys(config: TokenMeterConfig): void {
for (const key of Object.keys(config)) {
if (!TOKEN_METER_CONFIG_KEYS.has(key)) {
throw new Error(
`TokenMeterConfig: unknown key "${key}" (allowed: contextWindow)`,
)
}
}
}
/** Resolve and validate the one service-wide context capacity. */
function resolveContextWindow(config: TokenMeterConfig): number {
validateConfigKeys(config)
const contextWindow = config.contextWindow === undefined
? DEFAULT_CONTEXT_WINDOW
: config.contextWindow
if (!Number.isInteger(contextWindow) || contextWindow <= 0) {
throw new Error(
`TokenMeterConfig: contextWindow (${contextWindow}) must be a positive integer`,
)
}
return contextWindow
}
declare module 'cordis' {
interface Context {
tokenMeter: TokenMeterService
}
}
/** Validate and detach all configured model profiles. */
function resolveProfiles(config: TokenMeterConfig): readonly ModelTokenProfile[] {
const profiles = new Map<string, ModelTokenProfile>()
for (const profile of Object.values(BUILTIN_TOKEN_PROFILES)) {
profiles.set(profile.model, { ...profile })
}
const configuredValue: unknown = config.models
const configuredModels = configuredValue === undefined ? {} : configuredValue
if (typeof configuredModels !== 'object'
|| configuredModels === null
|| Array.isArray(configuredModels)) {
throw new TokenMeterError(
'TokenMeterConfig: models must be an object',
TOKEN_METER_INVALID_CONFIG,
)
}
for (const [model, override] of Object.entries(configuredModels as Record<string, unknown>)) {
if (model.length === 0) {
throw new TokenMeterError(
'TokenMeterConfig: model names must not be empty',
TOKEN_METER_INVALID_CONFIG,
model,
)
}
assertProfileObject(model, override)
const builtIn = profiles.get(model)
const contextWindow = override.contextWindow ?? builtIn?.contextWindow
const charsPerToken = override.charsPerToken ?? builtIn?.charsPerToken ?? 4
if (contextWindow === undefined) {
throw new TokenMeterError(
`TokenMeterConfig: custom model "${model}" requires contextWindow`,
TOKEN_METER_INVALID_CONFIG,
model,
)
}
assertPositiveInteger(model, 'contextWindow', contextWindow)
assertPositiveFinite(model, 'charsPerToken', charsPerToken)
profiles.set(model, { model, contextWindow, charsPerToken })
}
for (const profile of profiles.values()) {
assertPositiveInteger(profile.model, 'contextWindow', profile.contextWindow)
assertPositiveFinite(profile.model, 'charsPerToken', profile.charsPerToken)
}
return deepFreeze([...profiles.values()].map(profile => ({ ...profile })))
}
function assertProfileObject(model: string, value: unknown): asserts value is ModelTokenMeterConfig {
if (typeof value !== 'object' || value === null || Array.isArray(value)) {
throw new TokenMeterError(
`TokenMeterConfig: profile "${model}" must be an object`,
TOKEN_METER_INVALID_CONFIG,
model,
)
}
}
function assertPositiveInteger(model: string, name: string, value: number): void {
if (!Number.isInteger(value) || value <= 0) {
throw new TokenMeterError(
`TokenMeterConfig: ${model}.${name} (${value}) must be a positive integer`,
TOKEN_METER_INVALID_CONFIG,
model,
)
}
}
function assertPositiveFinite(model: string, name: string, value: number): void {
if (typeof value !== 'number' || !Number.isFinite(value) || value <= 0) {
throw new TokenMeterError(
`TokenMeterConfig: ${model}.${name} (${value}) must be a positive finite number`,
TOKEN_METER_INVALID_CONFIG,
model,
)
}
}
/** Concrete registry and replay owner for all configured model meters. */
/** Replay owner for one service-wide estimator and isolated per-session folds. */
export class TokenMeterService extends Service {
static Config: z<TokenMeterConfig> = z.object({
models: z.dict(z.object({
contextWindow: z.number(),
charsPerToken: z.number(),
})),
contextWindow: z.number().step(1).min(1).default(DEFAULT_CONTEXT_WINDOW),
})
private readonly meters = new Map<string, ReplayModelTokenMeter>()
/** Provider context-window capacity used by pressure consumers. */
readonly contextWindow: number
private readonly states = new WeakMap<Session, ReplayState>()
constructor(ctx: Context, config: TokenMeterConfig = {}) {
super(ctx, 'tokenMeter')
for (const profile of resolveProfiles(config)) {
this.meters.set(profile.model, new ReplayModelTokenMeter(profile))
}
this.contextWindow = resolveContextWindow(config)
// Readers catch up independently, while eager observation bounds ordinary
// read latency. A reader in an earlier listener consumes the new event;
// this listener then sees the same revision and performs no duplicate fold.
// read latency without creating state for sessions no consumer has read.
ctx.on('session/event', (session) => {
this._observe(session)
if (this.states.has(session)) this._sync(session)
})
}
/**
* Resolve one stable model-bound replay handle.
* @param model - exact routed model name.
* @throws {@link TokenMeterError} with `TOKEN_METER_MODEL_UNCONFIGURED` when no profile exists.
* @returns the configured handle for this model.
* Measure current request pressure and surface through the durable tail.
*
* Provider usage is reused only when the latest successful call's canonical
* request envelope matches `requestHeader` and its total is no lower than
* that call's full heuristic anchor; otherwise the complete envelope and
* surface are heuristically repriced.
*
* `requestHeader` affects request pressure only; surface fields always
* describe the current session surface. Every call clones those positional
* nodes, so measurement is O(surface).
*
* @param session - session to replay through its current durable tail.
* @param requestHeader - optional effective request envelope replacing the latest logged header.
* @returns a detached deeply immutable pressure and surface measurement.
*/
resolve(model: string): ModelTokenMeter {
const meter = this.meters.get(model)
if (meter === undefined) {
throw new TokenMeterError(
`token meter has no profile for model "${model}"`,
TOKEN_METER_MODEL_UNCONFIGURED,
model,
)
measure(session: Session, requestHeader?: EpochHeader): TokenMeasurement {
const state = this._sync(session)
const header = requestHeader === undefined
? state.header
: canonicalHeader(requestHeader)
const anchor = state.anchor
let baseline: TokenMeasurementBaseline
let surfaceDeltaTokens: number
if (anchor !== undefined && optionalHeaderEquals(anchor.header, header)) {
baseline = anchor.baseline
surfaceDeltaTokens = state.surfaceTokens - anchor.surfaceTokens
} else if (header === undefined && state.surfaceTokens === 0) {
baseline = { kind: 'none', tokens: 0 }
surfaceDeltaTokens = 0
} else {
baseline = {
kind: 'estimated',
tokens: this._estimateHeader(header) + state.surfaceTokens,
}
surfaceDeltaTokens = 0
}
return meter
return deepFreeze(structuredClone({
logRevision: state.consumedEvents,
baseline,
surfaceDeltaTokens,
totalTokens: Math.max(0, baseline.tokens + surfaceDeltaTokens),
surfaceTokens: state.surfaceTokens,
nodes: state.surface,
}))
}
/** Advance every configured model's isolated replay fold. */
private _observe(session: Session): void {
for (const meter of this.meters.values()) meter.observeIfActive(session)
/**
* Heuristically price one model-visible message.
* @param message - message to price without mutation.
* @returns content and role-framing tokens under the fixed service heuristic.
*/
estimateMessage(message: Message): number {
return this._estimateContent(message.content) + ROLE_OVERHEAD
}
/** Catch one session's fold up to the current durable tail. */
private _sync(session: Session): ReplayState {
let state = this.states.get(session)
if (state === undefined) {
state = {
consumedEvents: 0,
header: undefined,
surface: [],
surfaceTokens: 0,
stepStart: undefined,
anchor: undefined,
}
this.states.set(session, state)
}
while (state.consumedEvents < session.events.length) {
// eslint-disable-next-line @typescript-eslint/no-non-null-assertion -- contiguous session seqs index the durable log
const event = session.events[state.consumedEvents]!
this._foldEvent(session, state, event)
state.consumedEvents += 1
}
return state
}
/**
* Validate and prepare every fallible part before mutating replay state.
* A malformed event remains unread on every retry instead of partially
* applying the same mutation more than once.
*/
private _foldEvent(session: Session, state: ReplayState, event: SessionEvent): void {
let nextHeader = state.header
let nextStepStart = state.stepStart
let nextAnchor = state.anchor
switch (event.type) {
case 'request/header':
nextHeader = canonicalHeader(event.data.header)
break
case 'request/header-delta':
if (state.header === undefined) {
throw new Error(`token meter: request/header-delta at seq ${event.seq} has no preceding header`)
}
nextHeader = applyHeaderDelta(state.header, event.data)
break
case 'step/start':
if (state.stepStart !== undefined) {
throw new Error(
`token meter: step/start at seq ${event.seq} arrived before turn ${state.stepStart.turn}/step ${state.stepStart.step} ended`,
)
}
nextStepStart = { ...event.data, surfaceTokens: state.surfaceTokens }
break
case 'step/end':
if (state.stepStart === undefined
|| state.stepStart.turn !== event.data.turn
|| state.stepStart.step !== event.data.step) {
throw new Error(`token meter: step/end at seq ${event.seq} has no matching step/start boundary`)
}
nextStepStart = undefined
break
default:
break
}
const surface = isSurfaceEvent(event)
? this._prepareSurfaceMutation(session, state, event)
: undefined
if (event.type === 'assistant/message') {
const stepStart = state.stepStart
if (stepStart === undefined
|| stepStart.turn !== event.data.turn
|| stepStart.step !== event.data.step) {
throw new Error(`token meter: assistant/message at seq ${event.seq} has no matching step/start boundary`)
}
// assistant/message is surface-mandatory at every append/seed boundary.
// eslint-disable-next-line @typescript-eslint/no-non-null-assertion
const eventTokens = surface!.tokens
if (event.data.usage !== undefined && nextHeader !== undefined) {
const providerAssistantTokens = this._estimateProviderAssistant(
session,
event,
eventTokens,
)
const anchorSurfaceTokens = stepStart.surfaceTokens + providerAssistantTokens
const providerTokens = usageTokens(event.data.usage)
const estimatedAnchorTokens = this._estimateHeader(nextHeader) + anchorSurfaceTokens
nextAnchor = {
header: nextHeader,
surfaceTokens: anchorSurfaceTokens,
// Signed heuristic deltas remain conservative only from an anchor
// that is at least as large as the matching full heuristic price.
baseline: providerTokens >= estimatedAnchorTokens
? { kind: 'usage', tokens: providerTokens, usage: event.data.usage }
: { kind: 'estimated', tokens: estimatedAnchorTokens },
}
} else {
const anchorSurfaceTokens = stepStart.surfaceTokens + eventTokens
nextAnchor = {
header: nextHeader,
surfaceTokens: anchorSurfaceTokens,
baseline: {
kind: 'estimated',
tokens: this._estimateHeader(nextHeader) + anchorSurfaceTokens,
},
}
}
}
state.header = nextHeader
state.stepStart = nextStepStart
if (surface !== undefined) surface.commit(state)
state.anchor = nextAnchor
}
/** Validate one surface operation and return its allocation-light commit. */
private _prepareSurfaceMutation(
session: Session,
state: ReplayState,
event: SurfaceEvent,
): PreparedSurfaceMutation {
const tokens = this._estimateSurfaceEvent(session, event)
const op = event.surfaceOp
if (op === 'append') {
return {
tokens,
commit(target) {
target.surface.push({ seq: event.seq, tokens })
target.surfaceTokens += tokens
},
}
}
const startIdx = state.surface.findIndex(node => node.seq === op.start)
const endIdx = state.surface.findIndex(node => node.seq === op.end)
if (startIdx === -1 || endIdx === -1 || startIdx > endIdx) {
throw new Error(
`token meter: replace at seq ${event.seq} has invalid current range ${op.start}-${op.end}`,
)
}
const removedTokens = state.surface
.slice(startIdx, endIdx + 1)
.reduce((total, node) => total + node.tokens, 0)
return {
tokens,
commit(target) {
target.surface.splice(startIdx, endIdx - startIdx + 1, { seq: event.seq, tokens })
target.surfaceTokens += tokens - removedTokens
},
}
}
/** Price one current surface event exactly as it projects to a request. */
private _estimateSurfaceEvent(session: Session, event: SurfaceEvent): number {
const message = session.deriveEventMessage(event)
return message === null ? 0 : this.estimateMessage(message)
}
/**
* Reassemble provider output from exact chunk provenance for a usage anchor.
* Missing legacy provenance conservatively treats the durable output as the
* provider output; explicit empty provenance prices a known empty stream.
*/
private _estimateProviderAssistant(
session: Session,
event: SessionEvent<'assistant/message'>,
durableEventTokens: number,
): number {
const sourceSeqs = event.sourceEventSeqs
if (sourceSeqs === undefined) return durableEventTokens
const assembler = new BlockAssembler()
const seen = new Set<number>()
for (const seq of sourceSeqs) {
if (seq >= event.seq) {
throw new Error(`token meter: assistant/message at seq ${event.seq} source seq ${seq} is not earlier`)
}
if (seen.has(seq)) {
throw new Error(`token meter: assistant/message at seq ${event.seq} repeats source seq ${seq}`)
}
seen.add(seq)
// Session construction validates contiguous seqs, and the explicit
// earlier-than-assistant check above therefore guarantees existence.
const source = session.events[seq]
// eslint-disable-next-line @typescript-eslint/no-non-null-assertion
const sourceEvent = source!
if (sourceEvent.type !== 'assistant/chunk') {
throw new Error(`token meter: assistant/message at seq ${event.seq} source seq ${seq} is not assistant/chunk`)
}
if (sourceEvent.data.turn !== event.data.turn || sourceEvent.data.step !== event.data.step) {
throw new Error(`token meter: assistant/message at seq ${event.seq} source seq ${seq} belongs to another step`)
}
assembler.push(sourceEvent.data.chunk)
}
const providerMessage = assembler.message()
return providerMessage.content.length === 0 ? 0 : this.estimateMessage(providerMessage)
}
/** Price content blocks recursively under the fixed density heuristic. */
private _estimateContent(blocks: readonly ContentBlock[]): number {
let tokens = 0
for (const block of blocks) {
switch (block.type) {
case 'text':
case 'reasoning':
tokens += Math.ceil(block.text.length / CHARS_PER_TOKEN) + BLOCK_OVERHEAD
break
case 'tool-call':
tokens += Math.ceil(block.name.length / CHARS_PER_TOKEN)
+ Math.ceil(block.arguments.length / CHARS_PER_TOKEN)
+ BLOCK_OVERHEAD
break
case 'tool-result':
tokens += this._estimateContent(block.content) + BLOCK_OVERHEAD
break
default:
// ContentBlockMap is merge-extensible; unknown blocks retain a
// conservative structural JSON price under the fixed heuristic.
tokens += BLOCK_OVERHEAD + Math.ceil(JSON.stringify(block).length / CHARS_PER_TOKEN)
}
}
return tokens
}
/** Price the canonical non-surface request envelope. */
private _estimateHeader(header: EpochHeader | undefined): number {
if (header === undefined) return 0
let tokens = 0
for (const message of header.messagePrefix ?? []) tokens += this.estimateMessage(message)
if (header.system !== undefined) {
tokens += Math.ceil(header.system.length / CHARS_PER_TOKEN) + ROLE_OVERHEAD
}
if (header.tools !== undefined && header.tools.length > 0) {
tokens += Math.ceil(JSON.stringify(header.tools).length / CHARS_PER_TOKEN) + BLOCK_OVERHEAD
}
return tokens
}
}

367
packages/llm/token-meter/src/replay.ts
View File

@@ -1,367 +0,0 @@
/**
* Model-bound transactional replay of request headers, surface mutations, and
* successful-call token anchors.
*
* @module @deepseek-ai/dsh-token-meter/replay
*/
import { BlockAssembler, deepFreeze } from '@deepseek-ai/dsh-llm'
import type { ContentBlock, Message, TokenUsage } from '@deepseek-ai/dsh-llm'
import type { EpochHeader, Session, SessionEvent, SurfaceEvent } from '@deepseek-ai/dsh-session'
import { applyHeaderDelta, canonicalHeader, headerEquals, isSurfaceEvent } from '@deepseek-ai/dsh-session'
import type {
ModelTokenMeter,
TokenMeasurement,
TokenMeasurementBaseline,
TokenSurfaceMeasurement,
TokenSurfaceNode,
} from './types.ts'
/** Internal validated pricing profile. */
export interface ModelTokenProfile {
readonly model: string
readonly contextWindow: number
readonly charsPerToken: number
}
/** Per-block structural overhead for JSON framing and type tags. */
const BLOCK_OVERHEAD = 4
/** Role-field framing overhead added to every priced message. */
const ROLE_OVERHEAD = 4
interface UsageAnchor {
readonly header: EpochHeader
readonly surfaceTokens: number
readonly baseline: Exclude<TokenMeasurementBaseline, { kind: 'none' }>
}
interface ReplayState {
consumedEvents: number
header: EpochHeader | undefined
surface: TokenSurfaceNode[]
surfaceTokens: number
stepStart: { turn: number; step: number; surfaceTokens: number } | undefined
anchor: UsageAnchor | undefined
}
interface PreparedSurfaceMutation {
readonly tokens: number
commit(state: ReplayState): void
}
/** Sum disjoint provider usage buckets without double-counting reasoning output. */
function usageTokens(usage: TokenUsage): number {
return usage.inputTokens
+ (usage.cacheReadTokens ?? 0)
+ (usage.cacheWriteTokens ?? 0)
+ usage.outputTokens
}
/** One configured model's replay fold, weakly isolated by session identity. */
export class ReplayModelTokenMeter implements ModelTokenMeter {
readonly model: string
readonly contextWindow: number
readonly charsPerToken: number
private readonly states = new WeakMap<Session, ReplayState>()
constructor(profile: ModelTokenProfile) {
this.model = profile.model
this.contextWindow = profile.contextWindow
this.charsPerToken = profile.charsPerToken
}
/**
* Advance an already-read model/session fold without creating unused state.
* @param session - session whose durable tail advanced.
*/
observeIfActive(session: Session): void {
if (this.states.has(session)) this._sync(session)
}
/** @inheritdoc */
estimateMessage(message: Message): number {
return this._estimateContent(message.content) + ROLE_OVERHEAD
}
/** @inheritdoc */
measure(session: Session, requestHeader?: EpochHeader): TokenMeasurement {
const state = this._sync(session)
const header = requestHeader === undefined
? state.header
: canonicalHeader(requestHeader)
const anchor = state.anchor
let baseline: TokenMeasurementBaseline
let surfaceDeltaTokens: number
if (anchor !== undefined && header !== undefined && headerEquals(anchor.header, header)) {
baseline = anchor.baseline
surfaceDeltaTokens = state.surfaceTokens - anchor.surfaceTokens
} else if (header === undefined && state.surfaceTokens === 0) {
baseline = { kind: 'none', tokens: 0 }
surfaceDeltaTokens = 0
} else {
baseline = {
kind: 'estimated',
tokens: this._estimateHeader(header) + state.surfaceTokens,
}
surfaceDeltaTokens = 0
}
return deepFreeze(structuredClone({
model: this.model,
logRevision: state.consumedEvents,
baseline,
surfaceDeltaTokens,
totalTokens: Math.max(0, baseline.tokens + surfaceDeltaTokens),
}))
}
/** @inheritdoc */
measureSurface(session: Session): TokenSurfaceMeasurement {
const state = this._sync(session)
return deepFreeze(structuredClone({
model: this.model,
logRevision: state.consumedEvents,
totalTokens: state.surfaceTokens,
nodes: state.surface,
}))
}
/** Catch one session's fold up to the current durable tail. */
private _sync(session: Session): ReplayState {
let state = this.states.get(session)
if (state === undefined) {
state = {
consumedEvents: 0,
header: undefined,
surface: [],
surfaceTokens: 0,
stepStart: undefined,
anchor: undefined,
}
this.states.set(session, state)
}
while (state.consumedEvents < session.events.length) {
// eslint-disable-next-line @typescript-eslint/no-non-null-assertion -- contiguous session seqs index the durable log
const event = session.events[state.consumedEvents]!
this._foldEvent(session, state, event)
state.consumedEvents += 1
}
return state
}
/**
* Validate and prepare every fallible part before mutating replay state.
* A malformed event therefore remains the next unread event on every retry
* instead of applying a partial surface mutation twice.
*/
private _foldEvent(session: Session, state: ReplayState, event: SessionEvent): void {
let nextHeader = state.header
let nextStepStart = state.stepStart
let nextAnchor = state.anchor
switch (event.type) {
case 'request/header':
nextHeader = canonicalHeader(event.data.header)
break
case 'request/header-delta':
if (state.header === undefined) {
throw new Error(`token meter: request/header-delta at seq ${event.seq} has no preceding header`)
}
nextHeader = applyHeaderDelta(state.header, event.data)
break
case 'step/start':
if (state.stepStart !== undefined) {
throw new Error(
`token meter: step/start at seq ${event.seq} arrived before turn ${state.stepStart.turn}/step ${state.stepStart.step} ended`,
)
}
nextStepStart = { ...event.data, surfaceTokens: state.surfaceTokens }
break
case 'step/end':
if (state.stepStart === undefined
|| state.stepStart.turn !== event.data.turn
|| state.stepStart.step !== event.data.step) {
throw new Error(`token meter: step/end at seq ${event.seq} has no matching step/start boundary`)
}
nextStepStart = undefined
break
default:
break
}
const surface = isSurfaceEvent(event)
? this._prepareSurfaceMutation(session, state, event)
: undefined
if (event.type === 'assistant/message' && nextHeader?.config.model === this.model) {
const stepStart = state.stepStart
if (stepStart === undefined
|| stepStart.turn !== event.data.turn
|| stepStart.step !== event.data.step) {
throw new Error(`token meter: assistant/message at seq ${event.seq} has no matching step/start boundary`)
}
// assistant/message is surface-mandatory at every append/seed boundary.
// eslint-disable-next-line @typescript-eslint/no-non-null-assertion
const eventTokens = surface!.tokens
if (event.data.usage !== undefined) {
const providerAssistantTokens = this._estimateProviderAssistant(
session,
event,
eventTokens,
)
nextAnchor = {
header: nextHeader,
surfaceTokens: stepStart.surfaceTokens + providerAssistantTokens,
baseline: {
kind: 'usage',
tokens: usageTokens(event.data.usage),
usage: event.data.usage,
},
}
} else {
const anchorSurfaceTokens = stepStart.surfaceTokens + eventTokens
nextAnchor = {
header: nextHeader,
surfaceTokens: anchorSurfaceTokens,
baseline: {
kind: 'estimated',
tokens: this._estimateHeader(nextHeader) + anchorSurfaceTokens,
},
}
}
}
state.header = nextHeader
state.stepStart = nextStepStart
if (surface !== undefined) surface.commit(state)
state.anchor = nextAnchor
}
/** Validate one surface operation and return its allocation-light commit. */
private _prepareSurfaceMutation(
session: Session,
state: ReplayState,
event: SurfaceEvent,
): PreparedSurfaceMutation {
const tokens = this._estimateSurfaceEvent(session, event)
const op = event.surfaceOp
if (op === 'append') {
return {
tokens,
commit(target) {
target.surface.push({ seq: event.seq, tokens })
target.surfaceTokens += tokens
},
}
}
const startIdx = state.surface.findIndex(node => node.seq === op.start)
const endIdx = state.surface.findIndex(node => node.seq === op.end)
if (startIdx === -1 || endIdx === -1 || startIdx > endIdx) {
throw new Error(
`token meter: replace at seq ${event.seq} has invalid current range ${op.start}-${op.end}`,
)
}
const removedTokens = state.surface
.slice(startIdx, endIdx + 1)
.reduce((total, node) => total + node.tokens, 0)
return {
tokens,
commit(target) {
target.surface.splice(startIdx, endIdx - startIdx + 1, { seq: event.seq, tokens })
target.surfaceTokens += tokens - removedTokens
},
}
}
/** Price one current surface event exactly as it projects to a request. */
private _estimateSurfaceEvent(session: Session, event: SurfaceEvent): number {
const message = session.deriveEventMessage(event)
return message === null ? 0 : this.estimateMessage(message)
}
/**
* Reassemble provider output from exact chunk provenance for a usage anchor.
* Missing legacy provenance conservatively treats the durable output as the
* provider output; explicit empty provenance prices a known empty stream.
*/
private _estimateProviderAssistant(
session: Session,
event: SessionEvent<'assistant/message'>,
durableEventTokens: number,
): number {
const sourceSeqs = event.sourceEventSeqs
if (sourceSeqs === undefined) return durableEventTokens
const assembler = new BlockAssembler()
const seen = new Set<number>()
for (const seq of sourceSeqs) {
if (seq >= event.seq) {
throw new Error(`token meter: assistant/message at seq ${event.seq} source seq ${seq} is not earlier`)
}
if (seen.has(seq)) {
throw new Error(`token meter: assistant/message at seq ${event.seq} repeats source seq ${seq}`)
}
seen.add(seq)
// Session construction validates contiguous seqs, and the explicit
// earlier-than-assistant check above therefore guarantees existence.
const source = session.events[seq]
// eslint-disable-next-line @typescript-eslint/no-non-null-assertion
const sourceEvent = source!
if (sourceEvent.type !== 'assistant/chunk') {
throw new Error(`token meter: assistant/message at seq ${event.seq} source seq ${seq} is not assistant/chunk`)
}
if (sourceEvent.data.turn !== event.data.turn || sourceEvent.data.step !== event.data.step) {
throw new Error(`token meter: assistant/message at seq ${event.seq} source seq ${seq} belongs to another step`)
}
assembler.push(sourceEvent.data.chunk)
}
const providerMessage = assembler.message()
return providerMessage.content.length === 0 ? 0 : this.estimateMessage(providerMessage)
}
/** Price content blocks recursively under this model's density profile. */
private _estimateContent(blocks: readonly ContentBlock[]): number {
let tokens = 0
for (const block of blocks) {
switch (block.type) {
case 'text':
case 'reasoning':
tokens += Math.ceil(block.text.length / this.charsPerToken) + BLOCK_OVERHEAD
break
case 'tool-call':
tokens += Math.ceil(block.name.length / this.charsPerToken)
+ Math.ceil(block.arguments.length / this.charsPerToken)
+ BLOCK_OVERHEAD
break
case 'tool-result':
tokens += this._estimateContent(block.content) + BLOCK_OVERHEAD
break
default:
// ContentBlockMap is merge-extensible; unknown blocks retain a
// conservative structural JSON price under the selected profile.
tokens += BLOCK_OVERHEAD + Math.ceil(JSON.stringify(block).length / this.charsPerToken)
}
}
return tokens
}
/** Price the canonical non-surface request envelope. */
private _estimateHeader(header: EpochHeader | undefined): number {
if (header === undefined) return 0
let tokens = 0
for (const message of header.messagePrefix ?? []) tokens += this.estimateMessage(message)
if (header.system !== undefined) {
tokens += Math.ceil(header.system.length / this.charsPerToken) + ROLE_OVERHEAD
}
if (header.tools !== undefined && header.tools.length > 0) {
tokens += Math.ceil(JSON.stringify(header.tools).length / this.charsPerToken) + BLOCK_OVERHEAD
}
return tokens
}
}

74
packages/llm/token-meter/src/types.ts
View File

@@ -4,21 +4,12 @@
* @module @deepseek-ai/dsh-token-meter/types
*/
import type { Message, TokenUsage } from '@deepseek-ai/dsh-llm'
import type { EpochHeader, Session } from '@deepseek-ai/dsh-session'
/** Optional pricing fields for one configured model. */
export interface ModelTokenMeterConfig {
/** Provider context-window capacity in tokens. Required for a custom model. */
contextWindow?: number
/** Heuristic text density in characters per token. Defaults to `4`. */
charsPerToken?: number
}
import type { TokenUsage } from '@deepseek-ai/dsh-llm'
/** Token-meter plugin configuration. */
export interface TokenMeterConfig {
/** Built-in field overrides and custom model profiles, keyed by routed model name. */
models?: Record<string, ModelTokenMeterConfig>
/** Service-wide context-window capacity in tokens. Defaults to `128000`. */
contextWindow?: number
}
/** The baseline from which a signed surface delta produces current pressure. */
@@ -27,10 +18,8 @@ export type TokenMeasurementBaseline =
| { readonly kind: 'estimated'; readonly tokens: number }
| { readonly kind: 'usage'; readonly tokens: number; readonly usage: Readonly<TokenUsage> }
/** Detached immutable scalar pressure at one consumed session-log revision. */
/** Detached immutable request-pressure and surface snapshot at one consumed log revision. */
export interface TokenMeasurement {
/** Model profile used for every heuristic component. */
readonly model: string
/** Number of durable events consumed; equal to the next unread event seq. */
readonly logRevision: number
/** Provider or heuristic anchor used for this measurement. */
@@ -39,6 +28,10 @@ export interface TokenMeasurement {
readonly surfaceDeltaTokens: number
/** Non-negative current request-and-response pressure. */
readonly totalTokens: number
/** Total heuristic tokens across the current surface. */
readonly surfaceTokens: number
/** Current surface nodes in positional head-to-tail order. */
readonly nodes: readonly TokenSurfaceNode[]
}
/** One token-priced node in the current ordered session surface. */
@@ -48,54 +41,3 @@ export interface TokenSurfaceNode {
/** Heuristic tokens for the exact message projected by this node. */
readonly tokens: number
}
/** Detached immutable priced surface at one consumed session-log revision. */
export interface TokenSurfaceMeasurement {
/** Model profile used to price every node. */
readonly model: string
/** Number of durable events consumed; equal to the next unread event seq. */
readonly logRevision: number
/** Total heuristic tokens across the current surface. */
readonly totalTokens: number
/** Current surface nodes in positional head-to-tail order. */
readonly nodes: readonly TokenSurfaceNode[]
}
/** A model-bound replay meter returned by {@link TokenMeterService.resolve}. */
export interface ModelTokenMeter {
/** Routed model name bound to this handle. */
readonly model: string
/** Provider context-window capacity in tokens. */
readonly contextWindow: number
/** Heuristic text density in characters per token. */
readonly charsPerToken: number
/**
* Measure current request pressure through the session's durable tail.
*
* Provider usage is reused only when its routed model and canonical request
* envelope match `requestHeader`; otherwise the complete envelope and
* surface are heuristically repriced for this handle's model.
*
* @param session - session to replay through its current durable tail.
* @param requestHeader - optional effective request envelope replacing the latest logged header.
* @returns a detached deeply immutable pressure measurement.
*/
measure(session: Session, requestHeader?: EpochHeader): TokenMeasurement
/**
* Price the current surface for retention and replacement decisions.
*
* @param session - session to replay through its current durable tail.
* @returns a detached deeply immutable positional surface measurement.
*/
measureSurface(session: Session): TokenSurfaceMeasurement
/**
* Heuristically price one model-visible message.
*
* @param message - message to price without mutation.
* @returns content and role-framing tokens under this model profile.
*/
estimateMessage(message: Message): number
}

306
packages/llm/token-meter/tests/token-meter.spec.ts
View File

@@ -4,12 +4,8 @@ import { CallId } from '@deepseek-ai/dsh-llm'
import type { ContentBlock, Message, TokenUsage } from '@deepseek-ai/dsh-llm'
import SessionStore, { Session, SessionId, canonicalHeader } from '@deepseek-ai/dsh-session'
import type { EpochHeader } from '@deepseek-ai/dsh-session'
import TokenMeterService, {
TOKEN_METER_INVALID_CONFIG,
TOKEN_METER_MODEL_UNCONFIGURED,
TokenMeterError,
} from '@deepseek-ai/dsh-token-meter'
import type { ModelTokenMeter, TokenMeterConfig } from '@deepseek-ai/dsh-token-meter'
import TokenMeterService from '@deepseek-ai/dsh-token-meter'
import type { TokenMeasurement, TokenMeterConfig } from '@deepseek-ai/dsh-token-meter'
function header(model: string, extras: Omit<EpochHeader, 'config'> = {}): EpochHeader {
return canonicalHeader({ config: { model }, ...extras })
@@ -75,70 +71,34 @@ function meter(config: TokenMeterConfig = {}): TokenMeterService {
return new TokenMeterService(new Context(), config)
}
function expectSurfaceTotal(measurement: TokenMeasurement): void {
expect(measurement.nodes.reduce((total, node) => total + node.tokens, 0))
.toBe(measurement.surfaceTokens)
}
describe('TokenMeterService configuration and registration', () => {
it('provides immutable zero-config DeepSeek profiles', () => {
it('provides one zero-config context window', () => {
const service = meter()
expect(service.resolve('deepseek-v4-flash')).toMatchObject({
model: 'deepseek-v4-flash',
contextWindow: 128_000,
charsPerToken: 4,
})
expect(service.resolve('deepseek-v4-pro')).toMatchObject({
model: 'deepseek-v4-pro',
contextWindow: 128_000,
charsPerToken: 4,
})
expect(service.contextWindow).toBe(128_000)
})
it('merges built-in overrides field-wise and defaults custom density', () => {
const service = meter({
models: {
'deepseek-v4-flash': { charsPerToken: 2 },
custom: { contextWindow: 32_000 },
},
})
expect(service.resolve('deepseek-v4-flash')).toMatchObject({ contextWindow: 128_000, charsPerToken: 2 })
expect(service.resolve('deepseek-v4-pro')).toMatchObject({ contextWindow: 128_000, charsPerToken: 4 })
expect(service.resolve('custom')).toMatchObject({ contextWindow: 32_000, charsPerToken: 4 })
it('accepts one service-wide context-window override', () => {
expect(meter({ contextWindow: 32_000 }).contextWindow).toBe(32_000)
})
it('throws a typed exact-code error for unknown models', () => {
const service = meter()
let thrown: unknown
try {
service.resolve('unconfigured-model')
} catch (error: unknown) {
thrown = error
}
expect(thrown).toBeInstanceOf(TokenMeterError)
expect(thrown).toMatchObject({
code: TOKEN_METER_MODEL_UNCONFIGURED,
model: 'unconfigured-model',
})
expect((thrown as Error).message).toContain('unconfigured-model')
it.each(['models', 'contextWidow'])('rejects unknown top-level config key %s', (key) => {
expect(() => meter({ [key]: {} }))
.toThrow(`TokenMeterConfig: unknown key "${key}"`)
})
it.each([
[{ models: null }, /models must be an object/],
[{ models: [] }, /models must be an object/],
[{ models: { custom: {} } }, /requires contextWindow/],
[{ models: { '': { contextWindow: 1 } } }, /must not be empty/],
[{ models: { custom: { contextWindow: 0 } } }, /positive integer/],
[{ models: { custom: { contextWindow: 1.5 } } }, /positive integer/],
[{ models: { custom: { contextWindow: 1, charsPerToken: 0 } } }, /positive finite/],
[{ models: { custom: { contextWindow: 1, charsPerToken: Number.NaN } } }, /positive finite/],
[{ models: { custom: null } }, /must be an object/],
[{ models: { custom: [] } }, /must be an object/],
] as unknown as Array<[TokenMeterConfig, RegExp]>)('rejects invalid profile config %#', (config, pattern) => {
let thrown: unknown
try {
meter(config)
} catch (error: unknown) {
thrown = error
}
expect(thrown).toBeInstanceOf(TokenMeterError)
expect(thrown).toMatchObject({ code: TOKEN_METER_INVALID_CONFIG })
expect((thrown as Error).message).toMatch(pattern)
{ contextWindow: 0 },
{ contextWindow: -1 },
{ contextWindow: 1.5 },
{ contextWindow: Number.NaN },
{ contextWindow: null },
] as unknown as TokenMeterConfig[])('rejects invalid context capacity %#', (config) => {
expect(() => meter(config)).toThrow(/contextWindow .* positive integer/)
})
it('registers and unregisters ctx.tokenMeter with its plugin fiber', async () => {
@@ -151,9 +111,9 @@ describe('TokenMeterService configuration and registration', () => {
})
})
describe('ModelTokenMeter pricing', () => {
it('prices every built-in content shape and merge-extended blocks', () => {
const handle = meter({ models: { custom: { contextWindow: 100, charsPerToken: 2 } } }).resolve('custom')
describe('TokenMeterService pricing', () => {
it('prices every built-in content shape and merge-extended blocks with one fixed heuristic', () => {
const service = meter({ contextWindow: 100 })
const blocks: ContentBlock[] = [
{ type: 'text', text: 'abcd' },
{ type: 'reasoning', text: 'ab' },
@@ -166,55 +126,66 @@ describe('ModelTokenMeter pricing', () => {
},
{ type: 'future-block', payload: 'abcd' } as unknown as ContentBlock,
]
const estimated = handle.estimateMessage({ role: 'assistant', content: blocks })
const estimated = service.estimateMessage({ role: 'assistant', content: blocks })
expect(estimated).toBeGreaterThan(30)
expect(handle.estimateMessage(textMessage('abcd'))).toBe(10)
expect(service.estimateMessage(textMessage('abcd'))).toBe(9)
})
it('returns a detached deeply immutable empty measurement', () => {
const handle = meter().resolve('deepseek-v4-flash')
const service = meter()
const session = new Session(SessionId('empty'))
const result = handle.measure(session)
const result = service.measure(session)
expect(result).toEqual({
model: 'deepseek-v4-flash',
logRevision: 0,
baseline: { kind: 'none', tokens: 0 },
surfaceDeltaTokens: 0,
totalTokens: 0,
surfaceTokens: 0,
nodes: [],
})
expect(Object.isFrozen(result)).toBe(true)
expect(Object.isFrozen(result.baseline)).toBe(true)
expect(Object.isFrozen(result.nodes)).toBe(true)
expectSurfaceTotal(result)
expect(() => {
;(result as { totalTokens: number }).totalTokens = 1
}).toThrow(TypeError)
})
it('keeps earlier scalar and surface snapshots detached from later replay', () => {
const handle = meter().resolve('deepseek-v4-flash')
it('keeps an earlier unified snapshot detached from later replay', () => {
const service = meter()
const session = new Session(SessionId('detached'))
session.append('user/message', {
content: [{ type: 'text', text: 'first' }],
source: { kind: 'user' },
}, { surfaceOp: 'append' })
const scalar = handle.measure(session)
const surface = handle.measureSurface(session)
const scalarCopy = structuredClone(scalar)
const surfaceCopy = structuredClone(surface)
const snapshot = service.measure(session)
const snapshotCopy = structuredClone(snapshot)
expect(Object.isFrozen(snapshot.nodes)).toBe(true)
expect(Object.isFrozen(snapshot.nodes[0])).toBe(true)
expectSurfaceTotal(snapshot)
expect(() => {
;(snapshot.nodes as Array<{ seq: number; tokens: number }>).push({ seq: 99, tokens: 1 })
}).toThrow(TypeError)
expect(() => {
;(snapshot.nodes[0] as { seq: number; tokens: number }).tokens = 1
}).toThrow(TypeError)
session.append('user/message', {
content: [{ type: 'text', text: 'second' }],
source: { kind: 'user' },
}, { surfaceOp: 'append' })
expect(handle.measure(session).logRevision).toBe(2)
expect(handle.measureSurface(session).nodes).toHaveLength(2)
expect(scalar).toEqual(scalarCopy)
expect(surface).toEqual(surfaceCopy)
expect(scalar.logRevision).toBe(1)
expect(surface.nodes).toHaveLength(1)
const advanced = service.measure(session)
expect(advanced.logRevision).toBe(2)
expect(advanced.nodes).toHaveLength(2)
expectSurfaceTotal(advanced)
expect(snapshot).toEqual(snapshotCopy)
expect(snapshot.logRevision).toBe(1)
expect(snapshot.nodes).toHaveLength(1)
})
it('prices header, prefix, tools, and surface when no reusable usage exists', () => {
const handle = meter().resolve('deepseek-v4-flash')
const service = meter()
const session = new Session(SessionId('heuristic'))
session.append('user/message', {
content: [{ type: 'text', text: 'question' }],
@@ -225,10 +196,29 @@ describe('ModelTokenMeter pricing', () => {
messagePrefix: [textMessage('prefix')],
tools: [{ name: 'read', description: 'read', parameters: { type: 'object' } }],
}))
const result = handle.measure(session)
const result = service.measure(session)
expect(result.baseline.kind).toBe('estimated')
expect(result.totalTokens).toBeGreaterThan(handle.measureSurface(session).totalTokens)
expect(result.totalTokens).toBeGreaterThan(result.surfaceTokens)
expect(result.logRevision).toBe(session.events.length)
expectSurfaceTotal(result)
})
it('keeps request-header overrides out of the returned surface', () => {
const service = meter()
const session = new Session(SessionId('override-surface'))
session.append('user/message', {
content: [{ type: 'text', text: 'question' }],
source: { kind: 'user' },
}, { surfaceOp: 'append' })
const logged = service.measure(session)
const overridden = service.measure(session, header('another-model', {
system: 'large override '.repeat(100),
}))
expect(overridden.totalTokens).toBeGreaterThan(logged.totalTokens)
expect(overridden.surfaceTokens).toBe(logged.surfaceTokens)
expect(overridden.nodes).toEqual(logged.nodes)
expectSurfaceTotal(overridden)
})
})
@@ -242,7 +232,7 @@ describe('replay anchors and surface folds', () => {
}
it('uses disjoint provider usage and signed durable-output rewrites', () => {
const handle = meter().resolve('deepseek-v4-flash')
const service = meter()
const session = new Session(SessionId('usage'))
session.append('user/message', {
content: [{ type: 'text', text: 'before' }],
@@ -253,7 +243,7 @@ describe('replay anchors and surface folds', () => {
durableText: 'a much longer rewritten durable assistant answer',
usage: USAGE,
})
const result = handle.measure(session)
const result = service.measure(session)
expect(result.baseline).toMatchObject({ kind: 'usage', tokens: 34, usage: USAGE })
expect(result.surfaceDeltaTokens).toBeGreaterThan(0)
expect(result.totalTokens).toBe(34 + result.surfaceDeltaTokens)
@@ -262,21 +252,51 @@ describe('replay anchors and surface folds', () => {
}).toThrow(TypeError)
})
it('selects a heuristic anchor when provider usage would undercut its scale', () => {
const service = meter()
const session = new Session(SessionId('low-usage-anchor'))
const system = 'system context'
const requestHeader = header('deepseek-v4-flash', { system })
appendSuccessfulCall(session, requestHeader, {
providerText: 'abcd'.repeat(512),
usage: { inputTokens: 20, outputTokens: 7 },
})
const anchored = service.measure(session)
expect(anchored.baseline.kind).toBe('estimated')
const assistant = anchored.nodes[0]!.seq
session.append('user/message', {
content: [{ type: 'text', text: 'short' }],
source: { kind: 'plugin', plugin: 'test' },
}, {
surfaceOp: { op: 'replace', start: assistant, end: assistant },
sourceEventSeqs: [assistant],
})
const shrunken = service.measure(session)
expect(27 + shrunken.surfaceDeltaTokens).toBeLessThan(0)
expect(shrunken.totalTokens).toBeGreaterThan(0)
expect(shrunken.totalTokens).toBe(service.measure(
session,
header('different-model', { system }),
).totalTokens)
})
it('uses an estimated anchor when provider usage is absent', () => {
const handle = meter().resolve('deepseek-v4-flash')
const service = meter()
const session = new Session(SessionId('missing-usage'))
appendSuccessfulCall(session, header('deepseek-v4-flash', { system: 's' }), {
providerText: 'provider',
durableText: 'rewritten',
})
const anchored = handle.measure(session)
const anchored = service.measure(session)
expect(anchored.baseline.kind).toBe('estimated')
expect(anchored.surfaceDeltaTokens).toBe(0)
session.append('user/message', {
content: [{ type: 'text', text: 'later' }],
source: { kind: 'user' },
}, { surfaceOp: 'append' })
const advanced = handle.measure(session)
const advanced = service.measure(session)
expect(advanced.surfaceDeltaTokens).toBeGreaterThan(0)
})
@@ -295,24 +315,17 @@ describe('replay anchors and surface folds', () => {
usage: USAGE,
provenance: 'absent',
})
const handle = meter().resolve('deepseek-v4-flash')
expect(handle.measure(explicit).surfaceDeltaTokens).toBeGreaterThan(0)
expect(handle.measure(legacy).surfaceDeltaTokens).toBe(0)
const service = meter()
expect(service.measure(explicit).surfaceDeltaTokens).toBeGreaterThan(0)
expect(service.measure(legacy).surfaceDeltaTokens).toBe(0)
})
it('preserves one model anchor across another model success and reuses it after switching back', () => {
const service = meter({
models: {
alpha: { contextWindow: 1000 },
beta: { contextWindow: 1000, charsPerToken: 2 },
},
})
const alpha = service.resolve('alpha')
const beta = service.resolve('beta')
it('keeps only the latest successful request anchor across model switches', () => {
const service = meter({ contextWindow: 1_000 })
const session = new Session(SessionId('switch'))
const alphaHeader = header('alpha', { system: 'same envelope' })
appendSuccessfulCall(session, alphaHeader, { usage: USAGE, providerText: 'alpha' })
expect(alpha.measure(session).baseline.kind).toBe('usage')
expect(service.measure(session).baseline).toMatchObject({ kind: 'usage', tokens: 34 })
appendSuccessfulCall(session, header('beta'), {
turn: 1,
@@ -320,34 +333,33 @@ describe('replay anchors and surface folds', () => {
usage: { inputTokens: 100, outputTokens: 50 },
providerText: 'beta response',
})
expect(alpha.measure(session).baseline.kind).toBe('estimated')
expect(beta.measure(session).baseline).toMatchObject({ kind: 'usage', tokens: 150 })
expect(service.measure(session).baseline).toMatchObject({ kind: 'usage', tokens: 150 })
appendHeader(session, alphaHeader)
const switchedBack = alpha.measure(session)
expect(switchedBack.baseline).toMatchObject({ kind: 'usage', tokens: 34 })
expect(switchedBack.surfaceDeltaTokens).toBeGreaterThan(0)
const switchedBack = service.measure(session)
expect(switchedBack.baseline.kind).toBe('estimated')
expect(switchedBack.surfaceDeltaTokens).toBe(0)
})
it('invalidates usage for any canonical envelope change or explicit override', () => {
const handle = meter().resolve('deepseek-v4-flash')
const service = meter()
const session = new Session(SessionId('envelope'))
const anchoredHeader = header('deepseek-v4-flash', { system: 'one' })
appendSuccessfulCall(session, anchoredHeader, { usage: USAGE })
expect(handle.measure(session, { ...anchoredHeader, tools: [] }).baseline.kind).toBe('usage')
expect(handle.measure(session, header('deepseek-v4-flash', { system: 'two' })).baseline.kind)
expect(service.measure(session, { ...anchoredHeader, tools: [] }).baseline.kind).toBe('usage')
expect(service.measure(session, header('deepseek-v4-flash', { system: 'two' })).baseline.kind)
.toBe('estimated')
expect(handle.measure(session, header('deepseek-v4-pro', { system: 'one' })).baseline.kind)
expect(service.measure(session, header('deepseek-v4-pro', { system: 'one' })).baseline.kind)
.toBe('estimated')
expect(handle.measure(session, {
expect(service.measure(session, {
...anchoredHeader,
config: { ...anchoredHeader.config, temperature: 0.2 },
}).baseline.kind).toBe('estimated')
expect(handle.measure(session, {
expect(service.measure(session, {
...anchoredHeader,
messagePrefix: [textMessage('prefix')],
}).baseline.kind).toBe('estimated')
expect(handle.measure(session, {
expect(service.measure(session, {
...anchoredHeader,
tools: [{ name: 'read', description: 'read', parameters: { type: 'object' } }],
}).baseline.kind).toBe('estimated')
@@ -357,7 +369,7 @@ describe('replay anchors and surface folds', () => {
const session = new Session(SessionId('header-delta'))
appendHeader(session, header('deepseek-v4-flash'))
session.append('request/header-delta', { config: { model: 'deepseek-v4-pro' } })
const result = meter().resolve('deepseek-v4-flash').measure(session)
const result = meter().measure(session)
expect(result.baseline.kind).toBe('estimated')
expect(result.logRevision).toBe(2)
})
@@ -374,28 +386,27 @@ describe('replay anchors and surface folds', () => {
source: { kind: 'user' },
}, { surfaceOp: 'append' })
const seeded = new Session(SessionId('surface-seeded'), original.events)
const handle = service.resolve('deepseek-v4-flash')
const before = handle.measureSurface(seeded)
const beforeScalar = handle.measure(seeded)
const before = service.measure(seeded)
expect(before.nodes).toHaveLength(2)
expect(beforeScalar.surfaceDeltaTokens).toBeGreaterThan(0)
expect(before.surfaceDeltaTokens).toBeGreaterThan(0)
expectSurfaceTotal(before)
const first = seeded.surface.nodes[0]!.seq
seeded.append('user/message', {
content: [{ type: 'text', text: 'replacement' }],
source: { kind: 'plugin', plugin: 'test' },
}, { surfaceOp: { op: 'replace', start: first, end: first }, sourceEventSeqs: [first] })
const after = handle.measureSurface(seeded)
const afterScalar = handle.measure(seeded)
const after = service.measure(seeded)
expect(after.nodes).toHaveLength(2)
expect(after.nodes[0]!.seq).toBe(seeded.events.length - 1)
expect(after.logRevision).toBe(seeded.events.length)
expect(Object.isFrozen(after.nodes)).toBe(true)
expect(Object.isFrozen(after.nodes[0])).toBe(true)
expect(afterScalar.surfaceDeltaTokens).toBeLessThan(0)
expect(after.surfaceDeltaTokens).toBeLessThan(0)
expectSurfaceTotal(after)
expect(before.nodes).toHaveLength(2)
expect(before.logRevision).toBe(original.events.length)
expect(beforeScalar.surfaceDeltaTokens).toBeGreaterThan(0)
expect(before.surfaceDeltaTokens).toBeGreaterThan(0)
})
it('prices an empty assistant surface anchor as zero', () => {
@@ -405,26 +416,27 @@ describe('replay anchors and surface folds', () => {
durableText: '',
provenance: 'empty',
})
const surface = meter().resolve('deepseek-v4-flash').measureSurface(session)
const measurement = meter().measure(session)
const assistant = session.events.find(event => event.type === 'assistant/message')!
expect(surface.nodes).toEqual([{ seq: assistant.seq, tokens: 0 }])
expect(surface.totalTokens).toBe(0)
expect(measurement.nodes).toEqual([{ seq: assistant.seq, tokens: 0 }])
expect(measurement.surfaceTokens).toBe(0)
expectSurfaceTotal(measurement)
})
})
describe('malformed replay and listener lifecycle', () => {
function expectRepeatedFailure(handle: ModelTokenMeter, session: Session, pattern: RegExp): void {
expect(() => handle.measure(session)).toThrow(pattern)
expect(() => handle.measure(session)).toThrow(pattern)
function expectRepeatedFailure(service: TokenMeterService, session: Session, pattern: RegExp): void {
expect(() => service.measure(session)).toThrow(pattern)
expect(() => service.measure(session)).toThrow(pattern)
}
it('rejects a header delta before any snapshot transactionally', () => {
const session = new Session(SessionId('bad-delta'))
session.append('request/header-delta', { config: { model: 'deepseek-v4-flash' } })
expectRepeatedFailure(meter().resolve('deepseek-v4-flash'), session, /no preceding header/)
expectRepeatedFailure(meter(), session, /no preceding header/)
})
it('rejects a matching-model assistant without its step boundary transactionally', () => {
it('rejects an assistant without its step boundary transactionally', () => {
const session = new Session(SessionId('bad-step'))
appendHeader(session, header('deepseek-v4-flash'))
session.append('assistant/message', {
@@ -432,7 +444,7 @@ describe('malformed replay and listener lifecycle', () => {
step: 1,
content: [{ type: 'text', text: 'bad' }],
}, { surfaceOp: 'append', sourceEventSeqs: [] })
expectRepeatedFailure(meter().resolve('deepseek-v4-flash'), session, /no matching step\/start/)
expectRepeatedFailure(meter(), session, /no matching step\/start/)
})
it('clears completed step boundaries and rejects overlapping or late step events', () => {
@@ -440,7 +452,7 @@ describe('malformed replay and listener lifecycle', () => {
overlapping.append('step/start', { turn: 1, step: 1 })
overlapping.append('step/start', { turn: 1, step: 2 })
expectRepeatedFailure(
meter().resolve('deepseek-v4-flash'),
meter(),
overlapping,
/arrived before turn 1\/step 1 ended/,
)
@@ -455,7 +467,7 @@ describe('malformed replay and listener lifecycle', () => {
content: [],
}, { surfaceOp: 'append', sourceEventSeqs: [] })
expectRepeatedFailure(
meter().resolve('deepseek-v4-flash'),
meter(),
late,
/no matching step\/start/,
)
@@ -464,7 +476,7 @@ describe('malformed replay and listener lifecycle', () => {
mismatchedEnd.append('step/start', { turn: 1, step: 1 })
mismatchedEnd.append('step/end', { turn: 1, step: 2 })
expectRepeatedFailure(
meter().resolve('deepseek-v4-flash'),
meter(),
mismatchedEnd,
/step\/end .* no matching step\/start/,
)
@@ -509,7 +521,7 @@ describe('malformed replay and listener lifecycle', () => {
content: [{ type: 'text', text: 'bad' }],
usage: { inputTokens: 1, outputTokens: 1 },
}, { surfaceOp: 'append', sourceEventSeqs })
expect(() => meter().resolve('deepseek-v4-flash').measure(session)).toThrow(testCase.pattern)
expect(() => meter().measure(session)).toThrow(testCase.pattern)
}
})
@@ -528,7 +540,7 @@ describe('malformed replay and listener lifecycle', () => {
content: [],
usage: { inputTokens: 1, outputTokens: 0 },
}, { surfaceOp: 'append', sourceEventSeqs: [source, source] })
expect(() => meter().resolve('deepseek-v4-flash').measure(duplicate)).toThrow(/repeats source seq/)
expect(() => meter().measure(duplicate)).toThrow(/repeats source seq/)
const future = new Session(SessionId('future-source'))
future.append('step/start', { turn: 1, step: 1 })
@@ -539,7 +551,7 @@ describe('malformed replay and listener lifecycle', () => {
content: [],
usage: { inputTokens: 1, outputTokens: 0 },
}, { surfaceOp: 'append', sourceEventSeqs: [99] })
expect(() => meter().resolve('deepseek-v4-flash').measure(future)).toThrow(/is not earlier/)
expect(() => meter().measure(future)).toThrow(/is not earlier/)
})
it('does not partially apply a malformed assistant replacement', () => {
@@ -556,7 +568,7 @@ describe('malformed replay and listener lifecycle', () => {
content: [{ type: 'text', text: 'replacement' }],
}, { surfaceOp: { op: 'replace', start: head, end: head }, sourceEventSeqs: [head] })
expectRepeatedFailure(
meter().resolve('deepseek-v4-flash'),
meter(),
session,
/no matching step\/start/,
)
@@ -572,32 +584,32 @@ describe('malformed replay and listener lifecycle', () => {
content: [{ type: 'text', text: 'bad' }],
source: { kind: 'user' },
}, { surfaceOp: { op: 'replace', start: 99, end: 99 }, sourceEventSeqs: [0] })
expectRepeatedFailure(meter().resolve('deepseek-v4-flash'), session, /invalid current range/)
expectRepeatedFailure(meter(), session, /invalid current range/)
})
it('handles earlier-reader catch-up, eager observation, and service reload', async () => {
const ctx = new Context()
await ctx.plugin(SessionStore)
let handle: ModelTokenMeter | undefined
let activeMeter: TokenMeterService | undefined
const revisions: number[] = []
ctx.on('session/event', (session) => {
if (handle !== undefined) revisions.push(handle.measure(session).logRevision)
if (activeMeter !== undefined) revisions.push(activeMeter.measure(session).logRevision)
})
const firstFiber = await ctx.plugin(TokenMeterService)
handle = ctx.tokenMeter.resolve('deepseek-v4-flash')
activeMeter = ctx.tokenMeter
const session = ctx.sessions.create(SessionId('listener-order'))
handle.measure(session)
activeMeter.measure(session)
session.append('user/message', {
content: [{ type: 'text', text: 'one' }],
source: { kind: 'user' },
}, { surfaceOp: 'append' })
expect(revisions).toEqual([1])
expect(handle.measure(session).logRevision).toBe(1)
expect(activeMeter.measure(session).logRevision).toBe(1)
await firstFiber.dispose()
const secondFiber = await ctx.plugin(TokenMeterService)
handle = ctx.tokenMeter.resolve('deepseek-v4-flash')
expect(handle.measure(session).logRevision).toBe(1)
activeMeter = ctx.tokenMeter
expect(activeMeter.measure(session).logRevision).toBe(1)
await secondFiber.dispose()
})
})