refactor(token-meter): merge measurement snapshots (round 1)

2026-07-16 14:38:24 +08:00
parent 19a56ec542
commit 6d96b3e4a8
14 changed files with 119 additions and 115 deletions
--- a/packages/compact/compact-basic/src/index.ts
+++ b/packages/compact/compact-basic/src/index.ts
@@ -123,13 +123,7 @@ export class BasicCompactService extends CompactService {

    let result: CompactionResult | null = null
    for (let attempt = 0; attempt <= this.config.compactionRetries; attempt += 1) {
-      const surface = meter.measureSurface(agent.session)
-      if (surface.logRevision !== measurement.logRevision) {
-        throw new Error(
-          `compaction: pressure revision ${measurement.logRevision} does not match surface revision ${surface.logRevision}`,
-        )
-      }
-      const range = selectCompactableRange(agent.session, surface, this.config.retainTokens)
+      const range = selectCompactableRange(agent.session, measurement, this.config.retainTokens)
      if (range === null) {
        /* v8 ignore else -- concrete replacement preserves a compactable checkpoint; subclass hooks cannot mutate it. */
        if (result === null) return null
--- a/packages/compact/compact-basic/src/region.ts
+++ b/packages/compact/compact-basic/src/region.ts
@@ -10,7 +10,7 @@ import {
  toolPairingBalancedBefore,
 } from '@deepseek-ai/dsh-compact'
 import type { CompactionResult } from '@deepseek-ai/dsh-compact'
-import type { TokenMeterService, TokenSurfaceMeasurement } from '@deepseek-ai/dsh-token-meter'
+import type { TokenMeasurement, TokenMeterService } from '@deepseek-ai/dsh-token-meter'
 import type { Session, SessionEvent } from '@deepseek-ai/dsh-session'
 import type { Agent } from '@deepseek-ai/dsh-agent'
 import { frameSummary } from './summarizer.ts'
@@ -25,16 +25,16 @@ interface RegionDependencies {
 * Resolve the next head-anchored range while retaining a priced recent tail
 * and never splitting an assistant tool-call/result pair.
 * @param session - session supplying authoritative current surface positions.
- * @param pricedSurface - same-revision surface measurement from the conversation meter.
+ * @param measurement - unified pressure and surface measurement from the conversation meter.
 * @param retainTokens - minimum recent tail budget retained verbatim.
 * @returns the inclusive positional seq range to compact, or `null`.
 */
 export function selectCompactableRange(
  session: Session,
-  pricedSurface: TokenSurfaceMeasurement,
+  measurement: TokenMeasurement,
  retainTokens: number,
 ): { start: number; end: number } | null {
-  const pricedNodes = pricedSurface.nodes
+  const pricedNodes = measurement.nodes
  if (pricedNodes.length === 0) return null

  const surfaceNodes = session.surface.nodes
@@ -115,8 +115,8 @@ export async function compactSurfaceRegion(
  try {
    // Capture after the lock event so any later durable append, including a
    // log-only one, invalidates the async selection before replacement.
-    const lockedSurface = dependencies.meter.measureSurface(session)
-    const selected = lockedSurface.nodes.slice(startIdx, endIdx + 1)
+    const lockedMeasurement = dependencies.meter.measure(session)
+    const selected = lockedMeasurement.nodes.slice(startIdx, endIdx + 1)
    if (selected.length !== shadowedSeqs.length
      || selected.some((node, index) => node.seq !== shadowedSeqs[index])) {
      throw new Error('compaction: selected surface changed before summarization began')
@@ -125,8 +125,8 @@ export async function compactSurfaceRegion(
    const text = renderTranscript(session.events, shadowedSeqs)
    const { summary, model, maxTokens } = await dependencies.summarize(text, agent, signal)

-    const currentSurface = dependencies.meter.measureSurface(session)
-    if (currentSurface.logRevision !== lockedSurface.logRevision) {
+    const currentMeasurement = dependencies.meter.measure(session)
+    if (currentMeasurement.logRevision !== lockedMeasurement.logRevision) {
      throw new Error('compaction: session log changed during summarization')
    }
    const framedSummary = frameSummary(summary)
--- a/packages/compact/compact-basic/tests/compact-basic.spec.ts
+++ b/packages/compact/compact-basic/tests/compact-basic.spec.ts
@@ -251,17 +251,15 @@ describe('pressure measurement and retention', () => {
    expect(await compactIfNeeded(compact, retained, MODEL, 'x'.repeat(100_000))).toBeNull()
  })

-  it('detects scalar/surface revision disagreement', async () => {
+  it('uses one unified measurement for each pressure-and-retention decision', async () => {
    const ctx = createContext()
-    const meter = ctx.tokenMeter
-    const original = meter.measureSurface.bind(meter)
-    vi.spyOn(meter, 'measureSurface').mockImplementation((session) => {
-      const measurement = original(session)
-      return { ...measurement, logRevision: measurement.logRevision - 1 }
-    })
    const compact = service(compactConfig, ctx)
+    const measure = vi.spyOn(ctx.tokenMeter, 'measure')
+    const stop = new Error('stop after first decision')
+    vi.spyOn(compact, 'compactRegion').mockRejectedValueOnce(stop)

-    await expect(compactIfNeeded(compact, conversation(4))).rejects.toThrow(/revision/)
+    await expect(compactIfNeeded(compact, conversation(4))).rejects.toBe(stop)
+    expect(measure).toHaveBeenCalledTimes(1)
  })

  it('bounds retries when a shrinking checkpoint remains above threshold', async () => {
@@ -303,7 +301,7 @@ describe('pressure measurement and retention', () => {
  it('rejects a priced surface that is not the current positional surface', () => {
    const ctx = createContext()
    const session = conversation(2)
-    const priced = ctx.tokenMeter.measureSurface(session)
+    const priced = ctx.tokenMeter.measure(session)
    expect(() => selectCompactableRange(session, {
      ...priced,
      nodes: priced.nodes.slice(1),
@@ -331,7 +329,7 @@ describe('pressure measurement and retention', () => {
    }, { surfaceOp: 'append' })
    session.append('step/end', { turn: 1, step: 1 })

-    const priced = ctx.tokenMeter.measureSurface(session)
+    const priced = ctx.tokenMeter.measure(session)
    expect(selectCompactableRange(session, priced, 1)).toBeNull()
  })
 })
@@ -474,8 +472,8 @@ describe('compaction region transaction', () => {
  it('rejects a meter snapshot that changed before summarization began', async () => {
    const ctx = createContext()
    const meter = ctx.tokenMeter
-    const original = meter.measureSurface.bind(meter)
-    vi.spyOn(meter, 'measureSurface').mockImplementationOnce((session) => {
+    const original = meter.measure.bind(meter)
+    vi.spyOn(meter, 'measure').mockImplementationOnce((session) => {
      const measurement = original(session)
      return { ...measurement, nodes: measurement.nodes.slice(1) }
    })
--- a/packages/cordis/tool-cordis/src/api-catalog.ts
+++ b/packages/cordis/tool-cordis/src/api-catalog.ts
@@ -227,7 +227,6 @@ export const SERVICE_API: readonly ServiceApiEntry[] = [
    summary: 'Replay owner for one service-wide estimator and isolated per-session folds.',
    methods: [
      'measure(session: Session, requestHeader?: EpochHeader): TokenMeasurement',
-      'measureSurface(session: Session): TokenSurfaceMeasurement',
      'estimateMessage(message: Message): number',
    ],
  },
@@ -992,16 +991,12 @@ export const TYPE_API: readonly TypeApiEntry[] = [
  },
  {
    name: 'TokenMeasurement',
-    declaration: 'export interface TokenMeasurement {\n    readonly logRevision: number;\n    readonly baseline: TokenMeasurementBaseline;\n    readonly surfaceDeltaTokens: number;\n    readonly totalTokens: number;\n}',
+    declaration: 'export interface TokenMeasurement {\n    readonly logRevision: number;\n    readonly baseline: TokenMeasurementBaseline;\n    readonly surfaceDeltaTokens: number;\n    readonly totalTokens: number;\n    readonly surfaceTokens: number;\n    readonly nodes: readonly TokenSurfaceNode[];\n}',
  },
  {
    name: 'TokenMeasurementBaseline',
    declaration: 'export type TokenMeasurementBaseline = {\n    readonly kind: \'none\';\n    readonly tokens: 0;\n} | {\n    readonly kind: \'estimated\';\n    readonly tokens: number;\n} | {\n    readonly kind: \'usage\';\n    readonly tokens: number;\n    readonly usage: Readonly<TokenUsage>;\n};',
  },
-  {
-    name: 'TokenSurfaceMeasurement',
-    declaration: 'export interface TokenSurfaceMeasurement {\n    readonly logRevision: number;\n    readonly totalTokens: number;\n    readonly nodes: readonly TokenSurfaceNode[];\n}',
-  },
  {
    name: 'TokenSurfaceNode',
    declaration: 'export interface TokenSurfaceNode {\n    readonly seq: number;\n    readonly tokens: number;\n}',
--- a/packages/llm/token-meter/README.md
+++ b/packages/llm/token-meter/README.md
@@ -12,13 +12,12 @@ The estimator intentionally uses one fixed heuristic: four characters per token

 ## Measurement contract

-`ctx.tokenMeter` directly exposes 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.
+- `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 latest successful call's canonical request envelope matches the measured envelope; 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.

@@ -46,5 +45,6 @@ Indirectly, through consumers such as `dsh-compact-basic`; the service itself ad
 ## Known Limitations and Deferred Work

 - **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.
--- a/packages/llm/token-meter/src/index.ts
+++ b/packages/llm/token-meter/src/index.ts
@@ -14,7 +14,6 @@ import type {
  TokenMeasurement,
  TokenMeasurementBaseline,
  TokenMeterConfig,
-  TokenSurfaceMeasurement,
  TokenSurfaceNode,
 } from './types.ts'

@@ -126,15 +125,19 @@ export class TokenMeterService extends Service {
  }

  /**
-   * Measure current request pressure through the session's durable tail.
+   * 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`; 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 measurement.
+   * @returns a detached deeply immutable pressure and surface measurement.
   */
  measure(session: Session, requestHeader?: EpochHeader): TokenMeasurement {
    const state = this._sync(session)
@@ -164,19 +167,7 @@ export class TokenMeterService extends Service {
      baseline,
      surfaceDeltaTokens,
      totalTokens: Math.max(0, baseline.tokens + surfaceDeltaTokens),
-    }))
-  }
-
-  /**
-   * 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 {
-    const state = this._sync(session)
-    return deepFreeze(structuredClone({
-      logRevision: state.consumedEvents,
-      totalTokens: state.surfaceTokens,
+      surfaceTokens: state.surfaceTokens,
      nodes: state.surface,
    }))
  }
--- a/packages/llm/token-meter/src/types.ts
+++ b/packages/llm/token-meter/src/types.ts
@@ -18,7 +18,7 @@ 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 {
  /** Number of durable events consumed; equal to the next unread event seq. */
  readonly logRevision: number
@@ -28,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. */
@@ -37,13 +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 {
-  /** 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/packages/llm/token-meter/tests/token-meter.spec.ts
+++ b/packages/llm/token-meter/tests/token-meter.spec.ts
@@ -5,7 +5,7 @@ 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 from '@deepseek-ai/dsh-token-meter'
-import type { TokenMeterConfig } 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 })
@@ -71,6 +71,11 @@ 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 one zero-config context window', () => {
    const service = meter()
@@ -135,36 +140,48 @@ describe('TokenMeterService pricing', () => {
      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', () => {
+  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 = service.measure(session)
-    const surface = service.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(service.measure(session).logRevision).toBe(2)
-    expect(service.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', () => {
@@ -181,8 +198,27 @@ describe('TokenMeterService pricing', () => {
    }))
    const result = service.measure(session)
    expect(result.baseline.kind).toBe('estimated')
-    expect(result.totalTokens).toBeGreaterThan(service.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)
  })
 })

@@ -320,27 +356,27 @@ describe('replay anchors and surface folds', () => {
      source: { kind: 'user' },
    }, { surfaceOp: 'append' })
    const seeded = new Session(SessionId('surface-seeded'), original.events)
-    const before = service.measureSurface(seeded)
-    const beforeScalar = service.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 = service.measureSurface(seeded)
-    const afterScalar = service.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', () => {
@@ -350,10 +386,11 @@ describe('replay anchors and surface folds', () => {
      durableText: '',
      provenance: 'empty',
    })
-    const surface = meter().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)
  })
 })