Merge branch 'master' into worktree/provider-credential-lifecycle

packages/llm/llm-retry/src/index.ts

@@ -5,9 +5,9 @@
  * @module @deepseek-ai/dsh-llm-retry
  */
 
-import type { Context } from 'cordis'
+import type { Context, Events } from 'cordis'
 import z from 'schemastery'
-import type { Agent, RequestErrorAction, RequestFailureContext } from '@deepseek-ai/dsh-agent'
+import type { Agent, RequestErrorAction } from '@deepseek-ai/dsh-agent'
 import type { LlmFailure, ResolvedRetryPolicy } from '@deepseek-ai/dsh-llm'
 import type { SessionEvent } from '@deepseek-ai/dsh-session'
 
@@ -172,12 +172,9 @@ export function apply(ctx: Context, config: Config = {}, internals: RetryInterna
   }
 
   async function recover(
-    agent: Agent,
-    context: RequestFailureContext,
-    signal: AbortSignal,
+    { agent, turn, step, provider, failure, retryPolicy: policy, signal }: Parameters<Events['agent/request-error']>[0],
     next: () => Promise<RequestErrorAction>,
   ): Promise<RequestErrorAction> {
-    const { turn, step, provider, failure, retryPolicy: policy } = context
     if (policy === undefined) return next()
     if (policy.mode === 'always') {
       if (signal.aborted || lifetime.signal.aborted) return
@@ -228,16 +225,14 @@ export function apply(ctx: Context, config: Config = {}, internals: RetryInterna
   }
 
   const disposeListener = ctx.on('agent/request-error', (
-    agent: Agent,
-    context: RequestFailureContext,
-    signal: AbortSignal,
+    payload,
     next: () => Promise<RequestErrorAction>,
   ) => {
     // A waterfall may have captured this callback before its registration was
     // removed. Lifetime cancellation must prevent that stale callback from
     // entering a downstream policy after disposal.
     if (lifetime.signal.aborted) return Promise.resolve<RequestErrorAction>(undefined)
-    return track(recover(agent, context, signal, next))
+    return track(recover(payload, next))
   })
 
   ctx.effect(() => async () => {

packages/llm/llm-retry/tests/retry.spec.ts

@@ -506,7 +506,7 @@ describe('provider-routed retry policy', () => {
     ;({ ctx: context } = await harness(adapter, {
       other: alwaysConfig({ initialDelayMs: 1, maxDelayMs: 1, jitterRatio: 0 }),
     }, (ctx) => {
-      ctx.on('agent/request', async (_agent, _turn, _step, _signal, next) => ({
+      ctx.on('agent/request', async (_payload, next) => ({
         ...await next(),
         provider: 'other',
       }))
@@ -543,7 +543,7 @@ describe('provider-routed retry policy', () => {
         backoff: { initialDelayMs: 1, maxDelayMs: 1 },
       }),
     }, (ctx) => {
-      ctx.on('agent/request', async (_agent, _turn, _step, _signal, next) => ({
+      ctx.on('agent/request', async (_payload, next) => ({
         ...await next(),
         provider: adapter.requests.length === 0 ? 'mock' : 'other',
       }))
@@ -881,7 +881,7 @@ describe('provider-routed retry policy', () => {
     context = mounted.ctx
     const downstream = Promise.withResolvers<RequestErrorAction>()
     const entered = Promise.withResolvers<undefined>()
-    context.on('agent/request-error', (agent) => {
+    context.on('agent/request-error', ({ agent }) => {
       agent.cancel({ kind: 'user' })
       entered.resolve(undefined)
       return downstream.promise
@@ -917,7 +917,7 @@ describe('provider-routed retry policy', () => {
     const captured = Promise.withResolvers<undefined>()
     let invokeCaptured: (() => Promise<void>) | undefined
     const mounted = await harness(adapter, {}, (ctx) => {
-      ctx.on('agent/request-error', (_agent, _context, _signal, next) => {
+      ctx.on('agent/request-error', (_payload, next) => {
         return new Promise<RequestErrorAction>((resolve) => {
           invokeCaptured = async () => { resolve(await next()) }
           captured.resolve(undefined)
@@ -926,7 +926,7 @@ describe('provider-routed retry policy', () => {
     })
     context = mounted.ctx
     let downstreamCalls = 0
-    context.on('agent/request-error', async (_agent, _context, _signal, next) => {
+    context.on('agent/request-error', async (_payload, next) => {
       downstreamCalls += 1
       return next()
     })
@@ -980,7 +980,7 @@ describe('provider-routed retry policy', () => {
       textResponse('must not run'),
     ])
     ;({ ctx: context } = await harness(adapter, { mock: policy }, (ctx) => {
-      ctx.on('agent/request-error', async (agent, _context, _signal, next) => {
+      ctx.on('agent/request-error', async ({ agent }, next) => {
         agent.cancel({ kind: 'user' })
         return next()
       })

packages/llm/token-meter/src/breakdown-projection.ts

@@ -33,10 +33,11 @@ const breakdownSchema = z.object({
  *
  * Envelope figures are last-wins per `request/header`; the message figure
  * rides {@link foldSurfaceProjection} — the same O(1) fold the occupancy
- * projection uses — so it equals `measure().surfaceTokens` at every event
- * boundary and compaction shrinks it by its logged shadow price, the way it
- * shrinks the next request. The state is a fixed handful of numbers, so the
- * persisted checkpoint stays O(1) over the session's life.
+ * projection uses — so fully metered logs equal `measure().surfaceTokens` at
+ * every event boundary and compaction shrinks the figure by its logged shadow
+ * price. A replacement without a claim preserves the previous total. The
+ * state is a fixed handful of numbers, so the persisted checkpoint stays
+ * O(1) over the session's life.
  */
 export const contextBreakdownProjectionDefinition:
 ProjectionDefinition<'contextBreakdown', ContextBreakdownState> = {

packages/llm/token-meter/src/surface-fold.ts

@@ -3,9 +3,10 @@
  * surface `measure()` serves and compaction plans against. The projection
  * units deliberately do NOT share this fold — their state must stay O(1)
  * for the persisted checkpoint, so they ride `surface-projection.ts`'s
- * shadow-price protocol instead. The two stay in agreement by construction:
- * both price through `estimate.ts`, and every logged shadow price is derived
- * from THIS fold's nodes by the replace producer.
+ * shadow-price protocol instead. Fully metered logs stay in agreement by
+ * construction: both price through `estimate.ts`, and every logged shadow
+ * price is derived from THIS fold's nodes by the replace producer. A
+ * projection replacement without a claim deliberately folds with zero delta.
  *
  * @module @deepseek-ai/dsh-token-meter/surface-fold
  */

packages/llm/token-meter/src/surface-projection.ts

@@ -10,7 +10,9 @@
  * heuristic price of the exact replaced range, so the fold keeps a running
  * total plus at most one pending claim and never retains per-node prices.
  * The counts are exact by construction: producers derive them from the same
- * fixed estimator this module prices appends with.
+ * fixed estimator this module prices appends with. A replacement without an
+ * armed claim folds with zero delta because bounded state cannot reconstruct
+ * the replaced range; this preserves replay at the cost of possible drift.
  *
  * @module @deepseek-ai/dsh-token-meter/surface-projection
  */
@@ -47,16 +49,19 @@ export interface SurfaceTokensFold {
  * Fold one committed event onto a running surface-token total.
  *
  * A shadow-price event arms a claim; any other event expires it, and a
- * surface `replace` must consume a claim naming its exact range — the
+ * surface `replace` consumes the claim naming its exact range — the
  * producers append the metering event and the replacement synchronously
  * adjacent, so a surviving claim always prices the very next event.
+ * A replace with no claim folds with zero delta because the bounded state
+ * cannot reconstruct the replaced range. An armed claim for another range
+ * still fails because the adjacent events contradict each other.
  * @param claim - the claim armed by the immediately preceding event, if any.
  * @param event - the next committed session event.
  * @returns the signed token delta and the claim state after this event.
- * @throws when a replacement arrives without a claim for its exact range —
- *   every in-repo replace producer meters its replacement, so an unpriced
- *   replacement is a shadow-price contract violation and must fail loud
- *   rather than let the total drift.
+ * @throws when a replacement arrives with an armed claim for a different
+ *   range — the metering event was adjacent, so this is a live producer's
+ *   shadow-price contract violation, not historical data, and must fail
+ *   loud rather than let the total drift.
  */
 export function foldSurfaceProjection(
   claim: ShadowPriceClaim | undefined,
@@ -74,10 +79,15 @@ export function foldSurfaceProjection(
   const tokens = message === null ? 0 : estimateMessage(message)
   const op = event.surfaceOp
   if (op === 'append') return { deltaTokens: tokens, claim: undefined }
-  if (claim === undefined || claim.start !== op.start || claim.end !== op.end) {
+  // Sessions recorded before the shadow-price protocol log replacements with
+  // no adjacent metering event; the bounded state cannot reconstruct the
+  // replaced range's price, so fold those neutrally — historical replay
+  // degrades to drift instead of failing.
+  if (claim === undefined) return { deltaTokens: 0, claim: undefined }
+  if (claim.start !== op.start || claim.end !== op.end) {
     throw new Error(
       `token surface: replace at seq ${event.seq} over range ${op.start}-${op.end} has no adjacent shadow price`
-      + (claim === undefined ? '' : ` (armed claim covers ${claim.start}-${claim.end})`),
+      + ` (armed claim covers ${claim.start}-${claim.end})`,
     )
   }
   return { deltaTokens: tokens - claim.tokens, claim: undefined }

packages/llm/token-meter/src/usage-projection.ts

@@ -155,9 +155,10 @@ ProjectionDefinition<'tokenUsage', TokenUsageState> = {
  * `projectedTokens` — the sample plus the surface's signed movement since it
  * was taken — so occupancy answers for the next request rather than the last
  * one. The total rides {@link foldSurfaceProjection}, so the state stays O(1)
- * and a replacement shrinks it by its logged shadow price. A usage sample is
- * stamped BEFORE the same event joins the surface, so an `assistant/message`
- * anchors against the surface its own request saw.
+ * and a replacement shrinks it by its logged shadow price. A replacement
+ * without a claim preserves the previous total. A usage sample is stamped
+ * BEFORE the same event joins the surface, so an `assistant/message` anchors
+ * against the surface its own request saw.
  */
 export const contextPressureProjectionDefinition:
 ProjectionDefinition<'contextPressure', ContextPressureState> = {

packages/llm/token-meter/tests/context-breakdown-projection.spec.ts

@@ -180,7 +180,7 @@ describe('contextBreakdown session projection', () => {
     expect(agree()).toBeLessThan(grown)
   })
 
-  it('fails loud on a replacement without an adjacent matching shadow price', () => {
+  it('folds a replacement without a claim at zero and fails on a mismatched claim', () => {
     const definition = contextBreakdownProjectionDefinition
     const replace = (start: number, end: number): SessionEvent => ({
       type: 'user/message',
@@ -206,15 +206,17 @@ describe('contextBreakdown session projection', () => {
     let state = definition.init()
     state = definition.apply(state, append(1))
     state = definition.apply(state, append(3))
-    // No metering event at all.
-    expect(() => definition.apply(state, replace(1, 3))).toThrow('no adjacent shadow price')
-    // A claim for a different range does not price this replacement.
+    // No metering event: the replacement contributes zero instead of throwing.
+    expect(definition.view(definition.apply(state, replace(1, 3))).messageTokens)
+      .toBe(definition.view(state).messageTokens)
+    // An adjacent claim for another range contradicts the replacement.
     const mismatched = definition.apply(state, meter(1, 1, 8))
     expect(() => definition.apply(mismatched, replace(1, 3))).toThrow('no adjacent shadow price')
-    // A claim expires after one intervening event instead of lingering.
+    // A claim expires after one intervening event, so replacement delta is zero.
     let expired = definition.apply(state, meter(1, 3, 8))
     expired = definition.apply(expired, { type: 'todo/write', seq: 9, time: 0, data: { todos: [] } } as unknown as SessionEvent)
-    expect(() => definition.apply(expired, replace(1, 3))).toThrow('no adjacent shadow price')
+    expect(definition.view(definition.apply(expired, replace(1, 3))).messageTokens)
+      .toBe(definition.view(state).messageTokens)
     // The armed claim prices exactly the next event's matching replacement.
     const armed = definition.apply(state, meter(1, 3, 8))
     expect(definition.view(definition.apply(armed, replace(1, 3))).messageTokens)

packages/llm/token-meter/tests/token-usage-projection.spec.ts

@@ -419,6 +419,25 @@ describe('contextPressure session projection', () => {
     expect(compacted.projectedTokens).toBeLessThan(beforeCompaction!)
   })
 
+  it('folds a replacement without a claim at zero', async () => {
+    const { ctx, session } = await harness()
+    const question = appendUser(session, 'a question from an unmetered log')
+    startStep(session, 1, 1)
+    usageChunk(session, { inputTokens: 100, outputTokens: 1 }, 1, 1)
+    session.append('step/end', { turn: 1, step: 1 })
+    const before = pressure(ctx, session)
+
+    session.append('user/message', createUserMessage({
+      content: [{ type: 'text', text: 'summary without a preceding claim' }],
+      source: { kind: 'plugin', plugin: 'test' },
+    }), {
+      surfaceOp: { op: 'replace', start: question, end: question },
+      sourceEventSeqs: [question],
+    })
+
+    expect(pressure(ctx, session)).toEqual(before)
+  })
+
   it('clamps a projection that heuristic error drove below zero', async () => {
     const { ctx, session } = await harness()
     recordContext(session, 'large', 128_000)